1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
| import requests
import time
import random
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
from fake_useragent import UserAgent
BASE_URL = "https://www.xxx.com"
VOTE_URL = f"{BASE_URL}/api/vote"
TARGET_REGISTRATION_ID = 347
# ==================== 核心參數調整 ====================
TOTAL_USERS = 5000
MAX_WORKERS = 30
DELAY_MIN = 0.5
DELAY_MAX = 2.0
MAX_RETRIES = 5
TIMEOUT_SEC = 3
# ====================================================
ua_generator = UserAgent()
PROXIES_LIST = []
proxy_lock = threading.Lock()
def fetch_proxies():
"""多重備援抓取機制：GitHub -> ProxyScrape -> 本地應急 IP"""
global PROXIES_LIST
print("\n🔄 正在動態更新全球代理 IP 清單...")
# 備援 1：GitHub 開源清單
try:
url = "https://raw.githubusercontent.com/TheSpeedX/SOCKS-List/master/http.txt"
response = requests.get(url, timeout=8)
if response.status_code == 200:
raw_ips = response.text.replace('\r', '').split('\n')
with proxy_lock:
PROXIES_LIST = [ip.strip() for ip in raw_ips if ip.strip() and ":" in ip]
print(f"✅ [來源: GitHub] 成功補給！目前庫存：{len(PROXIES_LIST)} 個 IP")
return
except Exception as e:
print(f"⚠️ [來源: GitHub] 連線失敗或被拒絕 ({e})，正在切換備援來源...")
# 備援 2：ProxyScrape API
try:
url = "https://api.proxyscrape.com/v2/?request=displayproxies&protocol=http&timeout=10000&country=all&ssl=all&anonymity=all"
response = requests.get(url, timeout=8)
if response.status_code == 200:
raw_ips = response.text.replace('\r', '').split('\n')
with proxy_lock:
PROXIES_LIST = [ip.strip() for ip in raw_ips if ip.strip() and ":" in ip]
print(f"✅ [來源: ProxyScrape] 成功補給！目前庫存：{len(PROXIES_LIST)} 個 IP")
return
except Exception as e:
print(f"⚠️ [來源: ProxyScrape] 連線亦失敗 ({e})。")
# 備援 3：如果全部斷網，啟用本地應急靜態 IP (避免程式直接終止)
with proxy_lock:
if not PROXIES_LIST:
print("🚨 警告：所有線上代理 API 皆無法連線！啟用本地應急 IP 庫...")
# 這裡塞入一組固定的公共 DNS 或常用開放代理作為最後防線，至少不讓程式閃退
PROXIES_LIST = [
"1.1.1.1:8080", "8.8.8.8:8080", "114.114.114.114:80",
"45.133.107.192:81", "124.105.42.227:8081"
]
def get_random_proxy_dict():
global PROXIES_LIST
with proxy_lock:
if not PROXIES_LIST:
return None, None
proxy_ip = random.choice(PROXIES_LIST)
return proxy_ip, {
"http": f"http://{proxy_ip}",
"https": f"http://{proxy_ip}"
}
def remove_bad_proxy(proxy_ip):
global PROXIES_LIST
with proxy_lock:
if proxy_ip in PROXIES_LIST:
PROXIES_LIST.remove(proxy_ip)
# 庫存過低時，在背景悄悄啟動補貨，不卡主線程
if len(PROXIES_LIST) < 10 and len(PROXIES_LIST) % 2 == 0:
threading.Thread(target=fetch_proxies, daemon=True).start()
def build_random_headers():
return {
"Accept": "application/json, */*",
"Content-Type": "application/json",
"Origin": BASE_URL,
"Referer": f"{BASE_URL}/vote",
"User-Agent": ua_generator.random,
}
def simulate_user(user_id: int):
registration_id = TARGET_REGISTRATION_ID
session = requests.Session()
proxy_ip, proxies = get_random_proxy_dict()
for attempt in range(1, MAX_RETRIES + 1):
headers = build_random_headers()
try:
session.get(f"{BASE_URL}/vote", headers=headers, proxies=proxies, timeout=TIMEOUT_SEC)
time.sleep(random.uniform(DELAY_MIN, DELAY_MAX))
response = session.post(
VOTE_URL,
headers=headers,
json={"registrationId": registration_id},
proxies=proxies,
timeout=TIMEOUT_SEC
)
if response.status_code == 200:
return {"user": user_id, "status": 200, "proxy": proxy_ip, "attempts": attempt}
elif response.status_code == 429:
if proxy_ip: remove_bad_proxy(proxy_ip)
proxy_ip, proxies = get_random_proxy_dict()
time.sleep(1)
else:
return {"user": user_id, "status": response.status_code, "proxy": proxy_ip, "attempts": attempt}
except Exception:
if proxy_ip: remove_bad_proxy(proxy_ip)
proxy_ip, proxies = get_random_proxy_dict()
if attempt == MAX_RETRIES:
return {"user": user_id, "status": "TIMEOUT_FAILED", "proxy": proxy_ip, "attempts": attempt}
return {"user": user_id, "status": "FAILED", "proxy": proxy_ip, "attempts": MAX_RETRIES}
def main():
fetch_proxies()
results = []
print(f"🚀 大規模任務啟動！目標總數: {TOTAL_USERS}，併發線程: {MAX_WORKERS}")
with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
futures = [executor.submit(simulate_user, i) for i in range(1, TOTAL_USERS + 1)]
for future in as_completed(futures):
result = future.result()
results.append(result)
if result['status'] == 200:
print(f"🟢 [User {result['user']}] 投票成功! IP: {result['proxy']}")
elif result['user'] % 50 == 0:
print(f"📊 目前進度：已處理 {len(results)}/{TOTAL_USERS}...")
success_count = len([r for r in results if r['status'] == 200])
print("\n" + "="*40)
print(f"🏁 最終大規模執行結果:")
print(f"總發送任務: {TOTAL_USERS}")
print(f"實際成功數: {success_count} / {TOTAL_USERS}")
print(f"成功率: {(success_count/TOTAL_USERS)*100:.2f}%")
print("="*40)
if __name__ == "__main__":
main()
|