如何写自动剔除失效代理脚本

wen 实用脚本 2026-06-26 30

本文目录导读：

如何写自动剔除失效代理脚本

基于HTTP代理的测试脚本
SOCKS5代理测试脚本
简单Shell脚本版本
使用说明

我来帮你编写一个自动剔除失效代理的脚本,以下是几种常见的实现方式：

基于HTTP代理的测试脚本

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
自动剔除失效代理脚本 - HTTP/HTTPS代理检测
"""
import requests
import threading
from queue import Queue
import time
import json
class ProxyTester:
    def __init__(self, test_url="http://httpbin.org/ip", timeout=5, max_workers=10):
        self.test_url = test_url
        self.timeout = timeout
        self.max_workers = max_workers
        self.valid_proxies = []
        self.invalid_proxies = []
        self.lock = threading.Lock()
    def test_single_proxy(self, proxy):
        """测试单个代理"""
        proxies = {
            "http": proxy,
            "https": proxy
        }
        try:
            start_time = time.time()
            response = requests.get(
                self.test_url,
                proxies=proxies,
                timeout=self.timeout
            )
            response_time = time.time() - start_time
            if response.status_code == 200:
                with self.lock:
                    self.valid_proxies.append({
                        "proxy": proxy,
                        "response_time": round(response_time, 2),
                        "ip": response.json().get("origin", "unknown")
                    })
                return True
        except:
            with self.lock:
                self.invalid_proxies.append(proxy)
            return False
    def test_proxies(self, proxy_list):
        """并发测试代理列表"""
        self.valid_proxies = []
        self.invalid_proxies = []
        queue = Queue()
        for proxy in proxy_list:
            queue.put(proxy)
        def worker():
            while not queue.empty():
                proxy = queue.get()
                self.test_single_proxy(proxy)
                queue.task_done()
        # 创建线程池
        threads = []
        for _ in range(min(self.max_workers, len(proxy_list))):
            thread = threading.Thread(target=worker)
            thread.start()
            threads.append(thread)
        # 等待所有任务完成
        queue.join()
        for thread in threads:
            thread.join()
        return self.valid_proxies, self.invalid_proxies
def load_proxies(file_path):
    """从文件加载代理列表"""
    proxies = []
    try:
        with open(file_path, 'r') as f:
            for line in f:
                proxy = line.strip()
                if proxy and not proxy.startswith('#'):
                    proxies.append(proxy)
    except FileNotFoundError:
        print(f"文件 {file_path} 不存在")
    return proxies
def save_proxies(proxies, file_path):
    """保存有效代理到文件"""
    with open(file_path, 'w') as f:
        for proxy in proxies:
            if isinstance(proxy, dict):
                f.write(f"{proxy['proxy']}\n")
            else:
                f.write(f"{proxy}\n")
def main():
    # 配置参数
    input_file = "proxies.txt"  # 输入代理文件
    valid_file = "valid_proxies.txt"  # 有效代理输出文件
    invalid_file = "invalid_proxies.txt"  # 无效代理输出文件
    # 加载代理
    proxies = load_proxies(input_file)
    if not proxies:
        print("没有找到要测试的代理")
        return
    print(f"加载了 {len(proxies)} 个代理")
    # 创建测试器
    tester = ProxyTester(
        test_url="http://httpbin.org/ip",
        timeout=5,
        max_workers=20
    )
    # 测试代理
    print("开始测试代理...")
    start_time = time.time()
    valid, invalid = tester.test_proxies(proxies)
    test_time = time.time() - start_time
    # 输出结果
    print(f"\n测试完成! 耗时: {test_time:.2f}秒")
    print(f"有效代理: {len(valid)}")
    print(f"无效代理: {len(invalid)}")
    # 保存结果
    if valid:
        save_proxies(valid, valid_file)
        print(f"有效代理已保存到 {valid_file}")
        # 按响应时间排序
        valid.sort(key=lambda x: x['response_time'])
        print("\n最快的前10个代理:")
        for i, proxy in enumerate(valid[:10], 1):
            print(f"{i}. {proxy['proxy']} - {proxy['response_time']}s - IP: {proxy['ip']}")
    if invalid:
        save_proxies(invalid, invalid_file)
        print(f"无效代理已保存到 {invalid_file}")
if __name__ == "__main__":
    main()

SOCKS5代理测试脚本

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
自动剔除失效代理脚本 - SOCKS5代理检测
"""
import socket
import socks
import requests
from concurrent.futures import ThreadPoolExecutor
class Socks5ProxyTester:
    def __init__(self, test_url="http://httpbin.org/ip", timeout=5):
        self.test_url = test_url
        self.timeout = timeout
    def test_proxy(self, proxy_str):
        """测试单个SOCKS5代理"""
        try:
            # 解析代理字符串 (格式: ip:port 或 user:pass@ip:port)
            if '@' in proxy_str:
                auth, addr = proxy_str.split('@')
                username, password = auth.split(':')
                ip, port = addr.split(':')
            else:
                ip, port = proxy_str.split(':')
                username = password = None
            port = int(port)
            # 设置SOCKS5代理
            socks.set_default_proxy(
                socks.SOCKS5,
                addr=ip,
                port=port,
                username=username,
                password=password,
                rdns=True
            )
            socket.socket = socks.socksocket
            # 测试连接
            start_time = time.time()
            response = requests.get(
                self.test_url,
                timeout=self.timeout
            )
            response_time = time.time() - start_time
            if response.status_code == 200:
                return {
                    "proxy": proxy_str,
                    "response_time": round(response_time, 2),
                    "ip": response.json().get("origin", "unknown"),
                    "valid": True
                }
        except:
            pass
        finally:
            # 重置socket
            socket.socket = socket._socketobject
        return {"proxy": proxy_str, "valid": False}
    def test_proxies(self, proxy_list):
        """并发测试SOCKS5代理列表"""
        results = {"valid": [], "invalid": []}
        with ThreadPoolExecutor(max_workers=20) as executor:
            futures = [executor.submit(self.test_proxy, proxy) 
                      for proxy in proxy_list]
            for future in futures:
                result = future.result()
                if result["valid"]:
                    results["valid"].append(result)
                else:
                    results["invalid"].append(result["proxy"])
        return results

简单Shell脚本版本

#!/bin/bash
# 自动剔除失效代理脚本 (Shell版本)
PROXY_FILE="proxies.txt"
VALID_FILE="valid_proxies.txt"
TIMEOUT=5
TEST_URL="http://httpbin.org/ip"
# 清空输出文件
> "$VALID_FILE"
# 检查代理文件
if [ ! -f "$PROXY_FILE" ]; then
    echo "错误: 代理文件 $PROXY_FILE 不存在"
    exit 1
fi
# 计数器
total=0
valid=0
invalid=0
echo "开始测试代理..."
echo "使用测试URL: $TEST_URL"
echo "----------------------------------------"
# 逐行读取代理并测试
while IFS= read -r proxy; do
    # 跳过注释和空行
    [[ -z "$proxy" || "$proxy" =~ ^# ]] && continue
    ((total++))
    # 测试代理 (支持HTTP和HTTPS)
    if curl -s -x "$proxy" --connect-timeout $TIMEOUT --max-time $((TIMEOUT * 2)) \
        -o /dev/null -w "%{http_code}" "$TEST_URL" | grep -q "200"; then
        echo "✓ [$total] 有效: $proxy"
        echo "$proxy" >> "$VALID_FILE"
        ((valid++))
    else
        echo "✗ [$total] 无效: $proxy"
        ((invalid++))
    fi
done < "$PROXY_FILE"
echo "----------------------------------------"
echo "测试完成!"
echo "总代理数: $total"
echo "有效代理: $valid"
echo "无效代理: $invalid"
echo "有效代理已保存到: $VALID_FILE"

使用说明

代理文件格式 (proxies.txt):

# HTTP/HTTPS代理
192.168.1.1:8080
192.168.1.2:3128
# 带认证的代理
user:pass@192.168.1.3:8080
# SOCKS5代理 (需要修改测试脚本)
192.168.1.4:1080

运行脚本:

# Python版本
python3 proxy_tester.py
# Shell版本 (需要先授权)
chmod +x proxy_tester.sh
./proxy_tester.sh

功能特点:

✅ 并发测试提高效率
✅ 支持HTTP/HTTPS/SOCKS5代理
✅ 自动识别有效和无效代理
✅ 输出响应时间进行排序
✅ 支持带认证的代理
✅ 错误处理和超时控制

选择适合你需求的版本,根据实际情况调整测试URL和超时时间即可。