[求助] 如何用 Python 批量测试 IP 代理是否可用

import requests
import concurrent.futures
from typing import List, Dict
import time

def test_proxy(proxy: str, timeout: int = 5) -> Dict:
    """
    测试单个代理是否可用
    :param proxy: 代理地址，格式如 "http://1.2.3.4:8080"
    :param timeout: 超时时间（秒）
    :return: 包含测试结果的字典
    """
    proxies = {
        "http": proxy,
        "https": proxy
    }
    
    test_urls = [
        "http://httpbin.org/ip",  # 返回请求IP
        "https://httpbin.org/ip"
    ]
    
    result = {
        "proxy": proxy,
        "valid": False,
        "response_time": None,
        "error": None,
        "actual_ip": None
    }
    
    try:
        start_time = time.time()
        
        # 测试HTTP和HTTPS
        for url in test_urls:
            try:
                response = requests.get(
                    url, 
                    proxies=proxies, 
                    timeout=timeout,
                    headers={"User-Agent": "Mozilla/5.0"}
                )
                
                if response.status_code == 200:
                    data = response.json()
                    result["actual_ip"] = data.get("origin", "Unknown")
                    result["valid"] = True
                    break  # 只要一个成功就认为可用
                    
            except requests.exceptions.Timeout:
                result["error"] = "连接超时"
                continue
            except requests.exceptions.ProxyError:
                result["error"] = "代理错误"
                continue
            except Exception as e:
                result["error"] = str(e)
                continue
                
        result["response_time"] = round(time.time() - start_time, 2)
        
    except Exception as e:
        result["error"] = str(e)
    
    return result

def batch_test_proxies(proxy_list: List[str], max_workers: int = 10) -> List[Dict]:
    """
    批量测试代理
    :param proxy_list: 代理列表
    :param max_workers: 最大并发数
    :return: 测试结果列表
    """
    results = []
    
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        # 提交所有测试任务
        future_to_proxy = {
            executor.submit(test_proxy, proxy): proxy 
            for proxy in proxy_list
        }
        
        # 收集结果
        for future in concurrent.futures.as_completed(future_to_proxy):
            try:
                result = future.result()
                results.append(result)
            except Exception as e:
                results.append({
                    "proxy": future_to_proxy[future],
                    "valid": False,
                    "error": str(e)
                })
    
    return results

def save_results(results: List[Dict], filename: str = "proxy_test_results.txt"):
    """保存测试结果到文件"""
    with open(filename, "w", encoding="utf-8") as f:
        valid_count = 0
        total_count = len(results)
        
        f.write(f"代理测试报告 - 总计: {total_count}个\n")
        f.write("=" * 50 + "\n\n")
        
        for result in results:
            status = "✓ 可用" if result["valid"] else "✗ 不可用"
            f.write(f"代理: {result['proxy']}\n")
            f.write(f"状态: {status}\n")
            
            if result["valid"]:
                valid_count += 1
                f.write(f"响应IP: {result['actual_ip']}\n")
                f.write(f"响应时间: {result['response_time']}秒\n")
            else:
                f.write(f"错误: {result.get('error', '未知错误')}\n")
            
            f.write("-" * 30 + "\n")
        
        f.write(f"\n总结: {valid_count}/{total_count} 个代理可用 ({valid_count/total_count*100:.1f}%)\n")

# 使用示例
if __name__ == "__main__":
    # 你的代理列表，格式：协议://IP:端口
    proxies_to_test = [
        "http://123.45.67.89:8080",
        "http://111.222.333.444:3128",
        "socks5://127.0.0.1:1080",
        "https://proxy.example.com:443"
    ]
    
    print("开始测试代理...")
    results = batch_test_proxies(proxies_to_test, max_workers=5)
    
    # 显示结果
    print("\n测试结果:")
    for result in results:
        if result["valid"]:
            print(f"✓ {result['proxy']} - 可用 ({result['response_time']}秒)")
        else:
            print(f"✗ {result['proxy']} - 失败: {result.get('error', '未知错误')}")
    
    # 保存结果
    save_results(results)
    print(f"\n详细结果已保存到 proxy_test_results.txt")

这个脚本的核心思路是：用requests库通过代理访问测试网站（httpbin.org），根据响应判断代理是否可用。主要特点：

1. 支持HTTP/HTTPS/SOCKS代理：requests自动识别协议
2. 并发测试：用ThreadPoolExecutor同时测试多个代理，提高效率
3. 完整错误处理：捕获超时、代理错误等各种异常
4. 返回详细信息：包括实际IP、响应时间、错误原因
5. 结果保存：自动生成测试报告文件

使用很简单：把代理列表填到proxies_to_test里，运行就能看到哪些代理可用。测试网站可以换成你自己的目标网站，这样更准确。

建议用这个脚本定期检查代理池，及时剔除失效的代理。