IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

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

wen 实用脚本 26

本文目录导读：

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

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

我来提供几种批量校验代理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

性能优化建议

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

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

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

下一篇脚本如何持久化保存登录Cookie

相关文章

抱歉，评论功能暂时关闭!