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