如何用 Python 编写代理检测器

用 Python 编写的 proxy checker（代理检测器）通过代理服务器向已知的目标 URL 发送 HTTP 请求并分析响应，从而验证代理服务器的可用性、速度、匿名性和地理位置。

代理检测基础

proxy checker 的核心是通过指定代理发送一个 HTTP 请求并评估结果。该过程通常使用 requests 库，它简化了 Python 中的 HTTP 交互。

基础 HTTP 代理检测

要检测基础的 HTTP 或 HTTPS 代理，请在 requests.get() 方法中配置 proxies 字典。可靠的目标 URL（例如 httpbin.org/ip 或 ipinfo.io/json）必不可少。这些服务会返回请求来源的公网 IP 地址，从而可以进行验证。

import requests
import time

def check_http_proxy(proxy_address: str, timeout: float = 10.0) -> dict:
    """
    检测 HTTP/HTTPS 代理的基本连通性和响应时间。

    Args:
        proxy_address: 代理字符串（例如 "http://user:pass@ip:port"）。
        timeout: 等待响应的最长秒数。

    Returns:
        包含代理状态、响应时间以及错误信息（若有）的字典。
    """
    proxies = {
        "http": proxy_address,
        "https": proxy_address,
    }
    target_url = "http://httpbin.org/ip" # httpbin 使用 http，ipinfo 使用 https
    start_time = time.time()
    try:
        response = requests.get(target_url, proxies=proxies, timeout=timeout)
        response.raise_for_status() # 对错误响应（4xx 或 5xx）抛出 HTTPError
        end_time = time.time()
        latency = round((end_time - start_time) * 1000, 2) # 延迟，单位 ms
        origin_ip = response.json().get('origin')

        return {
            "proxy": proxy_address,
            "status": "Alive",
            "latency_ms": latency,
            "origin_ip": origin_ip,
            "error": None
        }
    except requests.exceptions.Timeout:
        return {"proxy": proxy_address, "status": "Timeout", "latency_ms": None, "origin_ip": None, "error": "Request timed out"}
    except requests.exceptions.ConnectionError:
        return {"proxy": proxy_address, "status": "Dead", "latency_ms": None, "origin_ip": None, "error": "Connection failed"}
    except requests.exceptions.HTTPError as e:
        return {"proxy": proxy_address, "status": "Error", "latency_ms": None, "origin_ip": None, "error": f"HTTP Error: {e}"}
    except Exception as e:
        return {"proxy": proxy_address, "status": "Error", "latency_ms": None, "origin_ip": None, "error": f"An unexpected error occurred: {e}"}

# 使用示例：
# print(check_http_proxy("http://203.0.113.45:8080"))
# print(check_http_proxy("https://198.51.100.22:443"))

处理不同类型的代理

requests 库原生支持 HTTP 和 HTTPS 代理。对于 SOCKS 代理（SOCKS4 和 SOCKS5），需要额外依赖 requests[socks]。使用 pip install requests[socks] 安装。之后代理 URL 的协议头改为 socks5:// 或 socks4://。

# SOCKS5 代理示例
def check_socks_proxy(proxy_address: str, timeout: float = 10.0) -> dict:
    """
    检测 SOCKS 代理。需要已安装 'requests[socks]'。
    """
    proxies = {
        "http": f"socks5://{proxy_address}",
        "https": f"socks5://{proxy_address}",
    }
    target_url = "http://httpbin.org/ip"
    start_time = time.time()
    try:
        response = requests.get(target_url, proxies=proxies, timeout=timeout)
        response.raise_for_status()
        end_time = time.time()
        latency = round((end_time - start_time) * 1000, 2)
        origin_ip = response.json().get('origin')

        return {
            "proxy": proxy_address,
            "status": "Alive",
            "latency_ms": latency,
            "origin_ip": origin_ip,
            "error": None
        }
    except requests.exceptions.Timeout:
        return {"proxy": proxy_address, "status": "Timeout", "latency_ms": None, "origin_ip": None, "error": "Request timed out"}
    except requests.exceptions.ConnectionError:
        return {"proxy": proxy_address, "status": "Dead", "latency_ms": None, "origin_ip": None, "error": "Connection failed"}
    except Exception as e:
        return {"proxy": proxy_address, "status": "Error", "latency_ms": None, "origin_ip": None, "error": f"SOCKS error: {e}"}

# 使用示例：
# print(check_socks_proxy("user:[email protected]:1080"))

评估代理质量

除了基本连通性之外，代理的质量还取决于其速度、匿名性和地理位置。

速度与延迟

延迟通过发送请求到收到响应第一个字节之间的时间来衡量。可以用 time.time() 函数记录开始和结束时间，两者之差即为延迟。通常以毫秒为单位报告。

# 延迟计算已包含在 check_http_proxy 和 check_socks_proxy 函数中。
# latency = round((end_time - start_time) * 1000, 2) # 延迟，单位 ms

匿名等级

代理匿名性指代理隐藏客户端原始 IP 地址与身份的有效程度。它通过分析代理可能添加或修改的特定 HTTP 头来判断。可以使用 httpbin.org/headers 或 ipinfo.io/json 等服务来查看这些请求头。

• Elite/高匿：代理隐藏客户端 IP 地址，且不发送 X-Forwarded-For、Via 或 Proxy-Connection 等任何标识性请求头。目标服务器只能看到代理的 IP。
• 匿名（Anonymous）：代理隐藏客户端 IP 地址，但可能发送带有代理 IP 的 Via 或 X-Forwarded-For 等请求头，表明使用了代理。客户端的原始 IP 不会被直接暴露。
• 透明（Transparent）：代理在 X-Forwarded-For 等请求头中转发客户端的 IP 地址。目标服务器知道请求来自代理，并能识别原始客户端 IP。

要检测匿名性，请向一个会回显请求头的 URL 发送请求，然后检查响应。您还需要知道自己（不走代理时）的公网 IP 地址以作对比。

import requests

def get_my_ip():
    """返回客户端不经代理时的公网 IP 地址。"""
    try:
        response = requests.get("http://httpbin.org/ip", timeout=5)
        response.raise_for_status()
        return response.json().get('origin')
    except requests.exceptions.RequestException:
        return None

def check_anonymity(proxy_address: str, my_ip: str, timeout: float = 10.0) -> str:
    """
    检测代理的匿名等级。
    """
    proxies = {
        "http": proxy_address,
        "https": proxy_address,
    }
    target_url = "http://httpbin.org/headers" # 返回全部请求头
    try:
        response = requests.get(target_url, proxies=proxies, timeout=timeout)
        response.raise_for_status()
        headers = response.json().get('headers', {})

        # 检查 X-Forwarded-For、Via、Proxy-Connection 请求头
        if 'X-Forwarded-For' in headers and headers['X-Forwarded-For'] == my_ip:
            return "Transparent"
        if 'Via' in headers or 'X-Forwarded-For' in headers:
            return "Anonymous"

        # 如果来源 IP（来自 httpbin.org/ip）不同于 my_ip 且不存在标识性请求头。
        # 这需要通过代理对 httpbin.org/ip 做一次单独检查。
        # 若要更精确，可将通过代理从 httpbin.org/ip 得到的 'origin' IP
        # 与 my_ip 比较，然后再检查请求头。

        # 简化判断：若无常见标识性请求头，则视为 Elite（需谨慎验证）
        return "Elite"

    except requests.exceptions.RequestException:
        return "Unknown (Proxy Dead or Error)"

# 使用示例：
# my_public_ip = get_my_ip()
# if my_public_ip:
#     print(f"My IP: {my_public_ip}")
#     print(check_anonymity("http://203.0.113.45:8080", my_public_ip))

地理位置

地理位置用于确定代理服务器的物理位置（国家、城市、地区）。ipinfo.io/json 等服务会根据请求来源 IP 以 JSON 响应提供这些信息。

import requests

def get_proxy_geolocation(proxy_address: str, timeout: float = 10.0) -> dict:
    """
    获取代理的地理位置数据。
    """
    proxies = {
        "http": proxy_address,
        "https": proxy_address,
    }
    target_url = "https://ipinfo.io/json" # ipinfo.io 适合做地理位置查询
    try:
        response = requests.get(target_url, proxies=proxies, timeout=timeout)
        response.raise_for_status()
        data = response.json()
        return {
            "country": data.get('country'),
            "region": data.get('region'),
            "city": data.get('city'),
            "org": data.get('org'),
            "timezone": data.get('timezone')
        }
    except requests.exceptions.RequestException:
        return {"country": None, "region": None, "city": None, "org": None, "timezone": None}

# 使用示例：
# geo_data = get_proxy_geolocation("http://203.0.113.45:8080")
# print(geo_data)

构建并发检测器

对于大批量列表，逐个顺序检测代理效率很低。借助多线程实现并发，可以同时检测多个代理。Python 的 concurrent.futures.ThreadPoolExecutor 让这变得简单。

import concurrent.futures
import csv
import time

# 假定上文已定义 check_http_proxy、check_anonymity、get_proxy_geolocation

def comprehensive_proxy_check(proxy_address: str, my_ip: str, timeout: float = 10.0) -> dict:
    """
    对单个代理执行全面检测。
    """
    result = check_http_proxy(proxy_address, timeout) # 基本连通性与延迟
    if result["status"] == "Alive":
        # 仅当代理可用时才继续检测匿名性和地理位置
        anonymity = check_anonymity(proxy_address, my_ip, timeout)
        geolocation = get_proxy_geolocation(proxy_address, timeout)
        result.update({"anonymity": anonymity})
        result.update(geolocation)
    else:
        result.update({"anonymity": None, "country": None, "region": None, "city": None, "org": None, "timezone": None})
    return result

def run_concurrent_checker(proxy_list_file: str, output_file: str, max_workers: int = 10, timeout: float = 10.0):
    """
    从文件读取代理，并发检测，并将结果写入 CSV。
    """
    proxies_to_check = []
    try:
        with open(proxy_list_file, 'r') as f:
            for line in f:
                proxy = line.strip()
                if proxy:
                    proxies_to_check.append(proxy)
    except FileNotFoundError:
        print(f"Error: Proxy list file '{proxy_list_file}' not found.")
        return

    my_ip = get_my_ip()
    if not my_ip:
        print("Could not determine client's public IP. Anonymity checks might be inaccurate.")

    results = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        future_to_proxy = {executor.submit(comprehensive_proxy_check, proxy, my_ip, timeout): proxy for proxy in proxies_to_check}
        for future in concurrent.futures.as_completed(future_to_proxy):
            proxy_address = future_to_proxy[future]
            try:
                result = future.result()
                results.append(result)
                print(f"Checked {proxy_address}: Status={result['status']}, Latency={result['latency_ms']}ms, Anonymity={result['anonymity']}")
            except Exception as exc:
                print(f"Proxy {proxy_address} generated an exception: {exc}")
                results.append({"proxy": proxy_address, "status": "Error", "error": str(exc)})

    # 将结果写入 CSV
    if results:
        fieldnames = list(results[0].keys())
        with open(output_file, 'w', newline='') as csvfile:
            writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
            writer.writeheader()
            writer.writerows(results)
        print(f"Results written to {output_file}")
    else:
        print("No proxies checked or no results to write.")

# 使用示例：
# 创建名为 'proxies.txt' 的文件，每行一个代理，例如：
# http://user:[email protected]:8080
# socks5://user:[email protected]:1080
# http://203.0.113.3:80
# run_concurrent_checker('proxies.txt', 'proxy_results.csv', max_workers=20, timeout=15)

健壮的错误处理与最佳实践

• 具体异常：捕获具体的 requests.exceptions（如 Timeout、ConnectionError、HTTPError），以便给出细粒度反馈。捕获通用的 Exception 应作为最后手段。
• User-Agent：在请求中加入 User-Agent 头以模拟浏览器，避免被目标服务屏蔽。
  headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'}
• 目标 URL 选择：选择可靠、快速且公开可用、能提供所需信息（IP、请求头、地理位置）的服务。避免高频轰炸单一服务，以免 IP 被封。
• 输入/输出：从文本文件读取代理列表（每行一个代理），并将结果写入 CSV 或 JSON 等结构化格式，便于分析和集成。
• 代理认证：确保代理 URL 的认证格式正确（例如 http://user:pass@ip:port）。
• 重试：对于生产级检测器，可考虑为瞬时错误实现重试逻辑。为此可用 requests.adapters.HTTPAdapter 和 urllib3.util.retry.Retry 扩展 requests 库。