如何用实用脚本批量验证代理IP有效性？

本文目录导读：

1. 功能特点
2. 脚本源码 (Python 3)
3. 使用方法
4. 高级自定义
5. 输出示例

功能特点

• 多线程并发：显著加快验证速度。
• 自定义测试URL：默认使用 http://httpbin.org/ip（返回请求者的IP）,也可改为其他网站。
• 支持HTTP/HTTPS：自动检测代理协议。
• 详细输出：显示IP、端口、类型、延迟与状态。
• 结果保存：将有效代理保存到文件。

脚本源码 (Python 3)

import requests
import threading
import time
import queue
from concurrent.futures import ThreadPoolExecutor, as_completed
# ===== 配置参数 =====
PROXY_FILE = "proxies.txt"          # 输入：代理列表文件，每行格式 IP:Port
VALID_FILE = "valid_proxies.txt"    # 输出：有效的代理文件
TEST_URL = "http://httpbin.org/ip"  # 测试URL（应返回请求者IP）
TIMEOUT = 5                         # 超时时间（秒）
MAX_WORKERS = 30                    # 并发线程数
# ==================
def test_proxy(proxy):
    """测试单个代理是否可用，返回结果字典"""
    result = {
        "proxy": proxy,
        "valid": False,
        "type": "unknown",
        "response_time": None,
        "error": None
    }
    # 尝试 HTTP 代理
    try:
        proxies = {
            "http": f"http://{proxy}",
            "https": f"http://{proxy}"
        }
        start = time.time()
        resp = requests.get(TEST_URL, proxies=proxies, timeout=TIMEOUT)
        elapsed = time.time() - start
        if resp.status_code == 200:
            result["valid"] = True
            result["type"] = "http"
            result["response_time"] = round(elapsed, 3)
            return result
    except Exception as e:
        pass
    # 尝试 HTTPS 代理（有些代理只支持 HTTPS）
    try:
        proxies = {
            "http": f"https://{proxy}",
            "https": f"https://{proxy}"
        }
        start = time.time()
        resp = requests.get(TEST_URL, proxies=proxies, timeout=TIMEOUT)
        elapsed = time.time() - start
        if resp.status_code == 200:
            result["valid"] = True
            result["type"] = "https"
            result["response_time"] = round(elapsed, 3)
            return result
    except Exception as e:
        result["error"] = str(e)
    return result
def load_proxies(file_path):
    """从文件加载代理列表，去除空白和空行"""
    proxies = []
    try:
        with open(file_path, "r") as f:
            for line in f:
                line = line.strip()
                if line and ":" in line:
                    proxies.append(line)
    except FileNotFoundError:
        print(f"[错误] 找不到文件: {file_path}")
        exit(1)
    return proxies
def main():
    print("[*] 正在加载代理列表...")
    proxies = load_proxies(PROXY_FILE)
    total = len(proxies)
    print(f"[+] 共加载 {total} 个代理")
    if total == 0:
        print("[!] 没有代理需要测试，退出。")
        return
    valid_proxies = []
    tested = 0
    lock = threading.Lock()
    print(f"[*] 开始测试（线程数: {MAX_WORKERS}）...")
    print("=" * 60)
    start_time = time.time()
    # 使用线程池
    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
        futures = {executor.submit(test_proxy, proxy): proxy for proxy in proxies}
        for future in as_completed(futures):
            proxy = futures[future]
            try:
                result = future.result()
                with lock:
                    tested += 1
                    if result["valid"]:
                        valid_proxies.append(result)
                        print(f"[✓] {proxy:<20} 类型: {result['type']:<5} 延迟: {result['response_time']}s")
                    else:
                        error = result["error"] if result["error"] else "不可用"
                        print(f"[✗] {proxy:<20} 原因: {error[:30]}")
                    # 显示进度
                    percentage = (tested / total) * 100
                    print(f"[进度] {tested}/{total} ({percentage:.1f}%)", end="\r")
            except Exception as e:
                print(f"[!] 测试代理 {proxy} 时发生异常: {e}")
    elapsed = time.time() - start_time
    print("\n" + "=" * 60)
    print(f"[✓] 测试完成，耗时 {elapsed:.1f} 秒")
    print(f"[+] 有效代理: {len(valid_proxies)} / {total}")
    # 保存有效代理
    if valid_proxies:
        # 按响应时间排序
        valid_proxies.sort(key=lambda x: x["response_time"])
        with open(VALID_FILE, "w") as f:
            for p in valid_proxies:
                f.write(f"{p['proxy']}  {p['type']}  {p['response_time']}s\n")
        print(f"[+] 有效代理已保存至: {VALID_FILE}")
    # 打印前10个最快的代理
    if valid_proxies:
        print("\n[快速代理 TOP 10]:")
        for i, p in enumerate(valid_proxies[:10], 1):
            print(f"  {i}. {p['proxy']} ({p['type']}, {p['response_time']}s)")
if __name__ == "__main__":
    main()

使用方法

准备工作

• 安装依赖：pip install requests
• 准备代理列表文件 proxies.txt，每行一个 IP:端口，

  168.1.1:8080
  10.0.0.1:3128
  8.8.8.8:80

运行脚本

python proxy_validator.py

输出说明

• 表示有效代理，显示类型（http/https）和响应时间（秒）。
• 表示无效代理，显示失败原因（超时、连接拒绝等）。
• 最终生成 valid_proxies.txt，包含有效代理及其类型和延迟（按速度排序）。

高级自定义

• 更改测试URL：将 TEST_URL 改为你信任的网站（如 https://www.google.com）。
• 提高准确度：可设置 TEST_URL 为返回IP的API（如 http://ip-api.com/json）并验证返回IP是否为你所用代理的IP。
• 验证SOCKS代理：如需支持SOCKS5，安装 pip install requests[socks]，然后修改代理格式为 socks5://ip:port。
• 批量测试大量代理：减少 TIMEOUT 值（如2秒），增加 MAX_WORKERS（但注意不要超过系统限制或触发目标网站反爬）。

输出示例

[*] 正在加载代理列表...
[+] 共加载 120 个代理
[*] 开始测试（线程数: 30）...
============================================================
[✓] 192.168.1.1:8080       类型: http  延迟: 0.435s
[✗] 10.0.0.1:3128         原因: timed out
[✓] 8.8.8.8:80            类型: https 延迟: 1.202s
...
[进度] 120/120 (100.0%)
============================================================
[✓] 测试完成，耗时 12.3 秒
[+] 有效代理: 48 / 120
[+] 有效代理已保存至: valid_proxies.txt
[快速代理 TOP 10]:
  1. 192.168.1.1:8080 (http, 0.435s)
  2. 203.0.113.5:3128 (http, 0.521s)
  ...

这个脚本简单、高效、可扩展,适合日常批量验证代理使用。