脚本怎样批量校验代理IP可用性

wen 实用脚本 2026-06-26 251

本文目录导读：

脚本怎样批量校验代理IP可用性

Python脚本方案（推荐）
Shell脚本方案
Python脚本（带进度条和统计）
使用说明
性能优化建议

我来提供几种批量校验代理IP可用性的脚本方案：

Python脚本方案（推荐）

基础版

import requests
import threading
from queue import Queue
import time
class ProxyChecker:
    def __init__(self, proxy_list, test_url='http://httpbin.org/ip', timeout=5):
        self.proxy_list = proxy_list
        self.test_url = test_url
        self.timeout = timeout
        self.valid_proxies = []
        self.queue = Queue()
        self.lock = threading.Lock()
    def check_proxy(self, proxy):
        """检查单个代理"""
        try:
            proxies = {
                'http': f'http://{proxy}',
                'https': f'http://{proxy}'
            }
            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:
                return {
                    'proxy': proxy,
                    'speed': round(response_time, 2),
                    'ip': response.json().get('origin', '')
                }
        except:
            return None
    def worker(self):
        """工作线程"""
        while not self.queue.empty():
            proxy = self.queue.get()
            result = self.check_proxy(proxy)
            if result:
                with self.lock:
                    self.valid_proxies.append(result)
                print(f"✓ {proxy} - 速度: {result['speed']}s")
            else:
                print(f"✗ {proxy} - 无效")
            self.queue.task_done()
    def check_batch(self, max_workers=10):
        """批量检查"""
        for proxy in self.proxy_list:
            self.queue.put(proxy)
        threads = []
        for _ in range(min(max_workers, len(self.proxy_list))):
            t = threading.Thread(target=self.worker)
            t.start()
            threads.append(t)
        self.queue.join()
        for t in threads:
            t.join()
        return self.valid_proxies
# 使用示例
def batch_check_proxies():
    # 读取代理列表
    proxies = [
        '192.168.1.1:8080',
        '192.168.1.2:3128',
        # 更多代理...
    ]
    checker = ProxyChecker(proxies)
    valid_proxies = checker.check_batch(max_workers=20)
    # 保存有效代理
    with open('valid_proxies.txt', 'w') as f:
        for p in valid_proxies:
            f.write(f"{p['proxy']}\n")
    return valid_proxies

高级版（带多种验证）

import aiohttp
import asyncio
import json
from typing import List, Dict
class AsyncProxyChecker:
    def __init__(self, proxy_list: List[str]):
        self.proxy_list = proxy_list
        self.test_urls = [
            'http://httpbin.org/ip',
            'http://ip-api.com/json',
            'http://checkip.amazonaws.com'
        ]
    async def check_single_proxy(self, session, proxy: str) -> Dict:
        """异步检查单个代理"""
        results = {}
        for test_url in self.test_urls:
            try:
                async with session.get(
                    test_url,
                    proxy=f'http://{proxy}',
                    timeout=aiohttp.ClientTimeout(total=5),
                    ssl=False
                ) as response:
                    if response.status == 200:
                        text = await response.text()
                        results[test_url] = {
                            'status': 'ok',
                            'response': text[:100]  # 截取前100字符
                        }
                    else:
                        results[test_url] = {'status': 'error', 'code': response.status}
            except Exception as e:
                results[test_url] = {'status': 'error', 'msg': str(e)}
        # 判断代理是否有效（至少一个测试URL成功）
        is_valid = any(r['status'] == 'ok' for r in results.values())
        return {
            'proxy': proxy,
            'valid': is_valid,
            'details': results
        }
    async def batch_check(self, max_concurrent: int = 20) -> List[Dict]:
        """批量异步检查"""
        connector = aiohttp.TCPConnector(limit=max_concurrent)
        timeout = aiohttp.ClientTimeout(total=10)
        async with aiohttp.ClientSession(
            connector=connector,
            timeout=timeout
        ) as session:
            tasks = []
            for proxy in self.proxy_list:
                task = self.check_single_proxy(session, proxy)
                tasks.append(task)
            results = await asyncio.gather(*tasks)
            return results
# 使用示例
async def main():
    proxies = ['192.168.1.1:8080', '192.168.1.2:3128']
    checker = AsyncProxyChecker(proxies)
    results = await checker.batch_check(max_concurrent=50)
    valid_proxies = [r for r in results if r['valid']]
    print(f"有效代理: {len(valid_proxies)}/{len(proxies)}")
    with open('valid_proxies.json', 'w') as f:
        json.dump(valid_proxies, f, indent=2)
# 运行
asyncio.run(main())

Shell脚本方案

#!/bin/bash
# 代理列表文件
PROXY_FILE="proxies.txt"
# 测试URL
TEST_URL="http://httpbin.org/ip"
# 超时时间（秒）
TIMEOUT=5
# 并发数
CONCURRENT=20
# 输出文件
OUTPUT_FILE="valid_proxies.txt"
# 清空输出文件
> $OUTPUT_FILE
# 函数：测试单个代理
test_proxy() {
    local proxy=$1
    local start_time=$(date +%s%N)
    # 使用curl测试代理
    response=$(curl -s -x $proxy \
        --connect-timeout $TIMEOUT \
        --max-time $TIMEOUT \
        -o /dev/null \
        -w "%{http_code}" \
        $TEST_URL 2>/dev/null)
    local end_time=$(date +%s%N)
    local duration=$(( ($end_time - $start_time) / 1000000 ))
    if [ "$response" = "200" ]; then
        echo "$proxy" >> $OUTPUT_FILE
        echo "✓ $proxy - ${duration}ms"
    else
        echo "✗ $proxy - 无效"
    fi
}
# 读取代理列表
while IFS= read -r proxy; do
    if [ -n "$proxy" ]; then
        # 使用&在后台运行，实现并发
        test_proxy "$proxy" &
        # 控制并发数
        running=$(jobs -p | wc -l)
        while [ $running -ge $CONCURRENT ]; do
            sleep 0.1
            running=$(jobs -p | wc -l)
        done
    fi
done < "$PROXY_FILE"
# 等待所有任务完成
wait
echo "完成！有效代理已保存到 $OUTPUT_FILE"

Python脚本（带进度条和统计）

import requests
from concurrent.futures import ThreadPoolExecutor, as_completed
from tqdm import tqdm
import time
import json
class ProxyChecker:
    def __init__(self):
        self.results = {
            'valid': [],
            'invalid': [],
            'stats': {'total': 0, 'valid': 0, 'invalid': 0}
        }
    def check_proxy(self, proxy):
        """测试单个代理"""
        proxy_url = f'http://{proxy}'
        result = {
            'proxy': proxy,
            'timestamp': time.time(),
            'sites': {}
        }
        test_sites = {
            'baidu': ('http://www.baidu.com', '百度'),
            'google': ('https://www.google.com', '谷歌'),
            'ipapi': ('http://ip-api.com/json', 'IP查询')
        }
        for site_name, (url, desc) in test_sites.items():
            try:
                start = time.time()
                response = requests.get(
                    url,
                    proxies={'http': proxy_url, 'https': proxy_url},
                    timeout=10,
                    headers={'User-Agent': 'Mozilla/5.0'}
                )
                elapsed = time.time() - start
                result['sites'][site_name] = {
                    'status': response.status_code,
                    'time': round(elapsed, 2),
                    'site': desc
                }
            except Exception as e:
                result['sites'][site_name] = {
                    'status': 'error',
                    'error': str(e)
                }
        # 判断有效性（至少能访问百度或谷歌）
        valid_sites = [
            site for site, data in result['sites'].items()
            if data.get('status') == 200
        ]
        result['valid'] = len(valid_sites) > 0
        return result
    def batch_check(self, proxy_list, max_workers=50):
        """批量检查"""
        self.results['stats']['total'] = len(proxy_list)
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            futures = {
                executor.submit(self.check_proxy, proxy): proxy 
                for proxy in proxy_list
            }
            with tqdm(total=len(proxy_list), desc="检查代理") as pbar:
                for future in as_completed(futures):
                    proxy = futures[future]
                    try:
                        result = future.result()
                        if result['valid']:
                            self.results['valid'].append(result)
                            self.results['stats']['valid'] += 1
                        else:
                            self.results['invalid'].append(result)
                            self.results['stats']['invalid'] += 1
                    except Exception as e:
                        self.results['invalid'].append({
                            'proxy': proxy,
                            'error': str(e)
                        })
                        self.results['stats']['invalid'] += 1
                    pbar.update(1)
                    pbar.set_postfix(
                        valid=self.results['stats']['valid'],
                        invalid=self.results['stats']['invalid']
                    )
        return self.results
# 使用示例
def main():
    # 从文件读取代理列表
    with open('proxy_list.txt', 'r') as f:
        proxies = [line.strip() for line in f if line.strip()]
    checker = ProxyChecker()
    results = checker.batch_check(proxies)
    # 保存结果
    with open('check_results.json', 'w') as f:
        json.dump(results, f, indent=2, ensure_ascii=False)
    # 输出统计
    print(f"\n检查完成:")
    print(f"总代理数: {results['stats']['total']}")
    print(f"有效代理: {results['stats']['valid']}")
    print(f"无效代理: {results['stats']['invalid']}")
    print(f"有效率: {results['stats']['valid']/results['stats']['total']*100:.2f}%")
if __name__ == '__main__':
    main()

使用说明

环境准备

# Python方案
pip install requests aiohttp tqdm
# Shell方案
# 需要安装curl（通常预装）

代理列表格式

ip:port
192.168.1.1:8080
192.168.1.2:3128

运行脚本

# Python脚本
python proxy_checker.py
# Shell脚本
chmod +x proxy_checker.sh
./proxy_checker.sh

性能优化建议

并发数控制：根据网络环境调整并发线程数
超时设置：根据实际需求设置合适超时时间
测试网站：选择稳定、响应快的测试网站
结果缓存：定期检查并更新代理可用性

需要根据具体使用场景调整参数和实现方式。

本文地址： https://www.bmjup.com/post/9081.html