在 vps 的主目錄創建一個 python 文件(例如:proxy_test.py),創建後,複製粘貼下方代碼,根據需要,自行修改代理的網址或路徑
import requests
import concurrent.futures
import time
import os
import json
from urllib3.exceptions import InsecureRequestWarning
# Suppress only the InsecureRequestWarning from urllib3
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
# --- Configuration Parameters ---
# **Updated URL: Using a more stable public proxy list URL**
PROXY_URL = 'https://raw.githubusercontent.com/TheSpeedX/PROXY-List/master/http.txt'
# Target used for checking proxy anonymity (must echo back request data)
ANONYMITY_TARGET = 'http://httpbin.org/get'
# Stricter, real-world target for final connectivity check (tests SSL/TLS)
FINAL_CONNECTIVITY_TARGET = 'https://www.google.com'
# **UPDATE:** Single proxy test total timeout (seconds). Set back to 5s.
TIMEOUT = 5
# **NEW/UPDATED:** Maximum acceptable total latency (seconds). Anything slower will be rejected.
MAX_LATENCY = 4.5
# Max concurrent threads - Adjusted to 200 for better stability
MAX_WORKERS = 180
# ------------------
def fetch_proxies(url):
"""
Fetches the proxy list from the specified URL.
Includes detailed headers to simulate a browser.
"""
print(f"嘗試從 {url} 獲取代理列表...")
try:
# Browser-like headers
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/129.0.0.0 Safari/537.36 Edg/129.0.0.0',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
'Referer': 'https://www.google.com/',
'Accept-Encoding': 'gzip, deflate, br',
'Connection': 'keep-alive',
}
# Disable SSL verification due to common proxy issues
response = requests.get(url, headers=headers, timeout=10, verify=False)
response.raise_for_status()
proxies = [p.strip() for p in response.text.splitlines() if p.strip()]
print(f"成功獲取到 {len(proxies)} 個代理。")
return proxies
except requests.exceptions.RequestException as e:
print(f"❌ 錯誤:獲取代理列表失敗,請檢查URL或網絡連接。錯誤信息:{e}")
return []
def get_real_ip():
"""
Fetches the client's real IP address for comparison using two methods
for improved robustness.
"""
# --- 嘗試方法 1: httpbin.org ---
try:
print("嘗試使用 httpbin.org 獲取真實IP...")
response = requests.get(ANONYMITY_TARGET, timeout=5)
response.raise_for_status()
real_ip = response.json().get('origin')
if real_ip:
# httpbin.org有時會返回多個IP地址,我們只取第一個
return real_ip.split(',')[0].strip()
except Exception as e:
print(f"httpbin.org 獲取失敗: {type(e).__name__}。嘗試備用方案...")
# --- 嘗試方法 2: api.ipify.org (備用方案) ---
try:
response = requests.get('https://api.ipify.org?format=json', timeout=5)
response.raise_for_status()
return response.json().get('ip')
except Exception as e:
print(f"api.ipify.org 獲取失敗: {type(e).__name__}。")
return None
def check_proxy(proxy, real_ip):
"""
Tests a single proxy for anonymity, speed (latency), and real-world connectivity.
Returns (proxy, True, reason, latency) if successful.
"""
ip_port = proxy.split(':')
if len(ip_port) != 2:
return proxy, False, "格式錯誤", None
proxy_ip = ip_port[0]
proxies = {
'http': f'http://{proxy}',
'https': f'http://{proxy}',
}
# --- 1. Anonymity Check (using ANONYMITY_TARGET) ---
try:
start_time = time.time()
# Use the global TIMEOUT for initial connection
response = requests.get(ANONYMITY_TARGET, proxies=proxies, timeout=TIMEOUT, verify=False)
# Basic connectivity check
if response.status_code != 200:
return proxy, False, "連接/狀態碼異常", None
# Anonymity check
response_json = response.json()
reported_ip = response_json.get('origin')
# Anonymity Logic: Reported IP must not be real IP AND must match proxy's IP
if not (reported_ip and reported_ip != real_ip and reported_ip.startswith(proxy_ip)):
return proxy, False, "透明/身份信息洩露", None
except requests.exceptions.Timeout:
return proxy, False, f"超時 (>{TIMEOUT}s)", None
except Exception as e:
return proxy, False, f"匿名性檢測失敗: {type(e).__name__}", None
# --- 2. Final Real-World Connectivity Check (using FINAL_CONNECTIVITY_TARGET) ---
try:
# Test against Google. Use the global TIMEOUT for final test.
requests.get(FINAL_CONNECTIVITY_TARGET, proxies=proxies, timeout=TIMEOUT, verify=False).raise_for_status()
# Calculate total latency since the start of the first test
total_latency = time.time() - start_time
# **SPEED CHECK:** Only accept proxies under the maximum latency threshold
if total_latency > MAX_LATENCY:
return proxy, False, f"速度太慢 (總延遲: {total_latency:.2f}s)", None
# Success if we reach here, status code is good, and speed is acceptable
return proxy, True, f"高匿/匿名/快速連通 (總延遲: {total_latency:.2f}s)", total_latency
except Exception as e:
# If it passed anonymity but failed the final Google test, it's blocked/incompatible
return proxy, False, f"被目標網站屏蔽: {type(e).__name__}", None
def batch_check(proxies, real_ip):
"""Uses multi-threading to batch test proxies."""
successful_proxies = []
total_proxies = len(proxies)
# Use ThreadPoolExecutor for concurrency
with concurrent.futures.ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
# Submit all tasks, passing the real IP to each check function
future_to_proxy = {executor.submit(check_proxy, proxy, real_ip): proxy for proxy in proxies}
# List to store (proxy, latency) for sorting
working_proxies_with_latency = []
# Retrieve results one by one, printing progress
for i, future in enumerate(concurrent.futures.as_completed(future_to_proxy)):
proxy = future_to_proxy[future]
try:
# The function returns 4 values: (proxy, is_working, reason, latency)
proxy, is_working, reason, latency = future.result()
except Exception as exc:
is_working = False
latency = None
reason = f"線程錯誤: {exc}"
# Print real-time progress
progress = f"[{i+1}/{total_proxies}]"
print(f"\r{progress} 正在檢測... 當前可用: {len(working_proxies_with_latency)}", end="", flush=True)
if is_working:
working_proxies_with_latency.append((proxy, latency))
# Sort proxies by latency (fastest first)
working_proxies_with_latency.sort(key=lambda x: x[1])
successful_proxies = [p[0] for p in working_proxies_with_latency]
# Clear progress output and newline
print("\n")
return successful_proxies
if __name__ == '__main__':
start_time = time.time()
# 0. Get the client's real IP for anonymity comparison
print("正在獲取您的真實IP地址用於匿名性對比...")
real_ip = get_real_ip()
if not real_ip:
print("❌ 警告:無法獲取真實IP,將跳過匿名性檢查,代理質量可能降低。")
real_ip = '0.0.0.0' # Use a fallback value
else:
print(f"✅ 您的真實IP地址是: {real_ip}")
# 1. Fetch proxy list
proxy_list = fetch_proxies(PROXY_URL)
if not proxy_list:
print("程序終止,未能獲取到代理列表。")
else:
# Updated output to reflect the new speed limit
print(f"開始測試 {len(proxy_list)} 個代理 (要求匿名性, 速度 < {MAX_LATENCY:.1f}s, 目標: {FINAL_CONNECTIVITY_TARGET}),並發數:{MAX_WORKERS}...")
# 2. Batch check
successful_proxies = batch_check(proxy_list, real_ip)
end_time = time.time()
# 3. Summary and saving results
print("\n--- 結果總結 ---")
print(f"總耗時:{end_time - start_time:.2f} 秒")
print(f"总代理數:{len(proxy_list)}")
print(f"可用(高匿/匿名/快速)代理數:{len(successful_proxies)}")
if len(proxy_list) > 0:
print(f"成功率:{len(successful_proxies)/len(proxy_list)*100:.2f}%")
output_file = 'working_proxies.txt'
with open(output_file, 'w') as f:
f.write('\n'.join(successful_proxies))
print(f"\n✅ 可用代理已保存到文件:{output_file} (已按延遲排序,最快在前)")
下面是 socks5 批量測試腳本
import requests
import concurrent.futures
import time
import os
import json
from urllib3.exceptions import InsecureRequestWarning
# 請注意:要使用SOCKS5代理,您必須先運行:pip install requests[socks]
# Suppress only the InsecureRequestWarning from urllib3
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
# --- 配置參數 ---
# **已更新:使用SOCKS5代理列表URL**
PROXY_URL = 'https://raw.githubusercontent.com/TheSpeedX/PROXY-List/master/socks5.txt'
# 用於檢查代理匿名性的目標 (必須返回請求數據)
ANONYMITY_TARGET = 'http://httpbin.org/get'
# 用於最終連通性檢查的嚴格目標 (測試 SSL/TLS)
FINAL_CONNECTIVITY_TARGET = 'https://www.google.com'
# 單個代理測試的連接超時時間(秒)。
TIMEOUT = 5
# 最大可接受總延遲(秒)。任何慢於此值的代理都將被拒絕。
MAX_LATENCY = 4.0
# 最大並發線程數 - 調整為180以獲得更好的穩定性
MAX_WORKERS = 180
# ------------------
def fetch_proxies(url):
"""
從指定的URL獲取代理列表。
"""
print(f"嘗試從 {url} 獲取代理列表...")
try:
# 瀏覽器模擬請求頭
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/129.0.0.0 Safari/537.36 Edg/129.0.0.0',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
'Referer': 'https://www.google.com/',
'Accept-Encoding': 'gzip, deflate, br',
'Connection': 'keep-alive',
}
# 禁用SSL驗證
response = requests.get(url, headers=headers, timeout=10, verify=False)
response.raise_for_status()
proxies = [p.strip() for p in response.text.splitlines() if p.strip()]
print(f"成功獲取到 {len(proxies)} 個代理。")
return proxies
except requests.exceptions.RequestException as e:
print(f"❌ 錯誤:獲取代理列表失敗,請檢查URL或網絡連接。錯誤信息:{e}")
return []
def get_real_ip():
"""
獲取客戶端的真實IP地址用於對比,使用雙重回退機制。
"""
# --- 嘗試方法 1: httpbin.org ---
try:
print("嘗試使用 httpbin.org 獲取真實IP...")
response = requests.get(ANONYMITY_TARGET, timeout=5)
response.raise_for_status()
real_ip = response.json().get('origin')
if real_ip:
# httpbin.org有時會返回多個IP地址,我們只取第一個
return real_ip.split(',')[0].strip()
except Exception as e:
print(f"httpbin.org 獲取失敗: {type(e).__name__}。嘗試備用方案...")
# --- 嘗試方法 2: api.ipify.org (備用方案) ---
try:
response = requests.get('https://api.ipify.org?format=json', timeout=5)
response.raise_for_status()
return response.json().get('ip')
except Exception as e:
print(f"api.ipify.org 獲取失敗: {type(e).__name__}。")
return None
def check_proxy(proxy, real_ip):
"""
測試單個代理的匿名性、速度和實際連通性。
"""
ip_port = proxy.split(':')
if len(ip_port) != 2:
return proxy, False, "格式錯誤", None
proxy_ip = ip_port[0]
# **關鍵修改:使用 socks5:// 協議**
proxies = {
'http': f'socks5://{proxy}',
'https': f'socks5://{proxy}',
}
# --- 1. 匿名性檢查 (使用 ANONYMITY_TARGET) ---
try:
start_time = time.time()
# 使用全局 TIMEOUT 進行初始連接
response = requests.get(ANONYMITY_TARGET, proxies=proxies, timeout=TIMEOUT, verify=False)
# 基本連通性檢查
if response.status_code != 200:
return proxy, False, "連接/狀態碼異常", None
# 匿名性檢查
response_json = response.json()
reported_ip = response_json.get('origin')
# 匿名性邏輯:報告的IP既不能是真實IP,也必須與代理IP匹配
if not (reported_ip and reported_ip != real_ip and reported_ip.startswith(proxy_ip)):
return proxy, False, "透明/身份信息洩露", None
except requests.exceptions.Timeout:
return proxy, False, f"超時 (>{TIMEOUT}s)", None
except Exception as e:
return proxy, False, f"匿名性檢測失敗: {type(e).__name__}", None
# --- 2. 最終實際連通性檢查 (使用 FINAL_CONNECTIVITY_TARGET) ---
try:
# 對 Google 進行測試。使用全局 TIMEOUT 進行最終測試。
requests.get(FINAL_CONNECTIVITY_TARGET, proxies=proxies, timeout=TIMEOUT, verify=False).raise_for_status()
# 計算從第一次測試開始的總延遲
total_latency = time.time() - start_time
# **速度檢查:** 只接受低於最大延遲閾值的代理
if total_latency > MAX_LATENCY:
return proxy, False, f"速度太慢 (總延遲: {total_latency:.2f}s)", None
# 成功通過所有檢查
return proxy, True, f"高匿/匿名/快速連通 (總延遲: {total_latency:.2f}s)", total_latency
except Exception as e:
# 如果通過了匿名性檢查,但在 Google 測試中失敗,則認為被目標網站屏蔽
return proxy, False, f"被目標網站屏蔽: {type(e).__name__}", None
def batch_check(proxies, real_ip):
"""使用多線程批量測試代理。"""
successful_proxies = []
total_proxies = len(proxies)
# 使用 ThreadPoolExecutor 實現並發
with concurrent.futures.ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
# 提交所有任務,並將真實IP傳遞給每個檢查函數
future_to_proxy = {executor.submit(check_proxy, proxy, real_ip): proxy for proxy in proxies}
# 用於存儲 (proxy, latency) 以進行排序
working_proxies_with_latency = []
# 逐個獲取結果,並打印進度
for i, future in enumerate(concurrent.futures.as_completed(future_to_proxy)):
proxy = future_to_proxy[future]
try:
# 函數返回 4 個值: (proxy, is_working, reason, latency)
proxy, is_working, reason, latency = future.result()
except Exception as exc:
is_working = False
latency = None
reason = f"線程錯誤: {exc}"
# 實時進度條輸出
progress = f"[{i+1}/{total_proxies}]"
print(f"\r{progress} 正在檢測... 當前可用: {len(working_proxies_with_latency)}", end="", flush=True)
if is_working:
working_proxies_with_latency.append((proxy, latency))
# 按延遲排序(最快的在前)
working_proxies_with_latency.sort(key=lambda x: x[1])
successful_proxies = [p[0] for p in working_proxies_with_latency]
# 清除進度條輸出,換行
print("\n")
return successful_proxies
if __name__ == '__main__':
start_time = time.time()
# 0. 獲取客戶端的真實IP用於匿名性對比
print("正在獲取您的真實IP地址用於匿名性對比...")
real_ip = get_real_ip()
if not real_ip:
print("❌ 警告:無法獲取真實IP,將跳過匿名性檢查,代理質量可能降低。")
real_ip = '0.0.0.0' # 使用一個備用值
else:
print(f"✅ 您的真實IP地址是: {real_ip}")
# 1. 獲取代理列表
proxy_list = fetch_proxies(PROXY_URL)
if not proxy_list:
print("程序終止,未能獲取到代理列表。")
else:
# 輸出更新後的速度限制
print(f"開始測試 {len(proxy_list)} 個代理 (要求匿名性, 速度 < {MAX_LATENCY:.1f}s, 目標: {FINAL_CONNECTIVITY_TARGET}),並發數:{MAX_WORKERS}...")
# 2. 批量測試
successful_proxies = batch_check(proxy_list, real_ip)
end_time = time.time()
# 3. 結果總結和保存
print("\n--- 結果總結 ---")
print(f"總耗時:{end_time - start_time:.2f} 秒")
print(f"总代理數:{len(proxy_list)}")
print(f"可用(高匿/匿名/快速)代理數:{len(successful_proxies)}")
if len(proxy_list) > 0:
print(f"成功率:{len(successful_proxies)/len(proxy_list)*100:.2f}%")
output_file = 'working_socks5_proxies.txt'
with open(output_file, 'w') as f:
f.write('\n'.join(successful_proxies))
print(f"\n✅ 可用代理已保存到文件:{output_file} (已按延遲排序,最快在前)")
執行前請檢查版本,安裝必須庫
檢查 Python 版本#
python3 --version
檢查 pip(Python 包管理器)版本#
pip3 --version
如果提示命令不存在需要執行:
sudo apt update
sudo apt install python3 python3-pip -y
然後安裝虛擬環境工具包
sudo apt update
sudo apt install python3-venv -y
進入主目錄創建虛擬環境
cd ~
python3 -m venv proxy_venv
激活進入環境(激活後 root 前面帶 (proxy_venv))
source proxy_venv/bin/activate
在環境中安裝庫
pip install requests
安裝後在環境內執行腳本
python proxy_test.py