banner
DarianBlog

DarianBlog

达里安博客-专注于技术分享与交流的博客。博客致力于创造一个简单而直接的技术学习平台。
email
telegram

http和socks5代理的批量測試腳本

在 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
載入中......
此文章數據所有權由區塊鏈加密技術和智能合約保障僅歸創作者所有。