当前位置： 首页 > news >正文

代理IP批量可用性检测 程序【python】

news 2025/10/24 9:51:58

前言

实现一个代理池测试工具，支持测试代理的连通性、响应时间和匿名性。用户需将代理配置写入.env文件，支持JSON数组、JSON对象和逗号分隔三种格式。测试过程会记录每个代理的状态（可用/不可用）、响应时间、HTTP状态码和匿名等级，并生成详细的测试报告，包括成功率统计、性能指标和可用代理列表。核心功能通过ProxyTester类实现，支持同步测试方式，提供全面的代理检测能力。

ip写到环境变量中，运行main.py即可
依赖库自行安装

输出效果

2025-10-23 21:30:41.145 | INFO     | __main__:test_proxies_sync:239 - 开始同步测试 2 个代理...
2025-10-23 21:30:42.236 | INFO     | __main__:test_proxies_sync:251 - ❌ [1/2] http://170.114.45.249:80 - 不可用 (1.09s)
2025-10-23 21:30:42.492 | INFO     | __main__:test_proxies_sync:251 - ✅ [2/2] http://127.0.0.1:10808 - 可用 (1.35s)============================================================
🔍 代理池测试报告
============================================================
📊 测试概要:总代理数: 2可用代理: 1 ✅不可用代理: 1 ❌成功率: 50.00%⚡ 性能统计:平均响应时间: 1.35s最快响应时间: 1.35s最慢响应时间: 1.35s🔒 匿名性统计:transparent: 1个✅ 可用代理列表:http://127.0.0.1:10808❌ 不可用代理列表:http://170.114.45.249:80 - HTTPSConnectionPool(host='httpbin.org', port=443): Max retries exceeded with url: /ip (Caused by ProxyError('Unable to connect to proxy', OSError('Tunnel connection failed: 400 Bad Request')))

.env

# 代理池配置 - 支持多种格式# 方式1：JSON数组格式（推荐）
PROXY_CONFIG='["http://127.0.0.1:8080", "http://proxy2.com:8080", "http://user:pass@proxy3.com:8080"]'# 方式2：JSON对象格式（支持更多配置）
PROXY_CONFIG='{"proxies": ["http://127.0.0.1:8080", "http://proxy2.com:8080"], "timeout": 10}'# 方式3：逗号分隔格式（兼容旧版）
PROXY_CONFIG="http://127.0.0.1:8080,http://proxy2.com:8080,http://user:pass@proxy3.com:8080"# 代理测试相关配置
PROXY_TEST_URL="https://httpbin.org/ip"
PROXY_TEST_TIMEOUT="5"

main.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
代理池可用性测试程序
测试代理的连通性、响应时间、匿名性等指标python main.py --direct --timeout 5 python main.py --config --timeout 5 
"""import asyncio
import aiohttp
import time
import json
import argparse
from typing import List, Dict, Optional, Tuple
from dataclasses import dataclass
from concurrent.futures import ThreadPoolExecutor
import requests
from loguru import logger@dataclass
class ProxyTestResult:"""代理测试结果"""proxy: stris_available: boolresponse_time: floatstatus_code: Optional[int] = Noneerror_message: Optional[str] = Noneanonymity_level: Optional[str] = None  # transparent, anonymous, elitereal_ip: Optional[str] = Noneproxy_ip: Optional[str] = Noneclass ProxyTester:"""代理测试器"""def __init__(self, test_urls: List[str] = None,timeout: int = 10,max_workers: int = 20):"""初始化代理测试器Args:test_urls: 测试URL列表timeout: 请求超时时间max_workers: 最大并发数"""self.test_urls = test_urls or ["https://httpbin.org/ip","https://api.ipify.org?format=json","https://ifconfig.me/ip"]self.timeout = timeoutself.max_workers = max_workers# 获取本机真实IPself.real_ip = self._get_real_ip()logger.info(f"本机真实IP: {self.real_ip}")def _get_real_ip(self) -> Optional[str]:"""获取本机真实IP"""try:response = requests.get("https://httpbin.org/ip", timeout=10)if response.status_code == 200:return response.json().get("origin", "").split(",")[0].strip()except Exception as e:logger.warning(f"获取真实IP失败: {e}")return Nonedef test_proxy_sync(self, proxy: str) -> ProxyTestResult:"""同步测试单个代理"""start_time = time.time()try:# 解析代理格式proxy_dict = {'http': proxy,'https': proxy}# 测试连通性response = requests.get(self.test_urls[0],proxies=proxy_dict,timeout=self.timeout,headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'})response_time = time.time() - start_timeif response.status_code == 200:# 解析响应获取代理IPtry:data = response.json()proxy_ip = data.get("origin", "").split(",")[0].strip()except:proxy_ip = None# 判断匿名性anonymity_level = self._check_anonymity(proxy_ip)return ProxyTestResult(proxy=proxy,is_available=True,response_time=response_time,status_code=response.status_code,anonymity_level=anonymity_level,real_ip=self.real_ip,proxy_ip=proxy_ip)else:return ProxyTestResult(proxy=proxy,is_available=False,response_time=response_time,status_code=response.status_code,error_message=f"HTTP {response.status_code}")except Exception as e:response_time = time.time() - start_timereturn ProxyTestResult(proxy=proxy,is_available=False,response_time=response_time,error_message=str(e))async def test_proxy_async(self, session: aiohttp.ClientSession, proxy: str) -> ProxyTestResult:"""异步测试单个代理"""start_time = time.time()try:async with session.get(self.test_urls[0],proxy=proxy,timeout=aiohttp.ClientTimeout(total=self.timeout),headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'}) as response:response_time = time.time() - start_timeif response.status == 200:try:data = await response.json()proxy_ip = data.get("origin", "").split(",")[0].strip()except:proxy_ip = Noneanonymity_level = self._check_anonymity(proxy_ip)return ProxyTestResult(proxy=proxy,is_available=True,response_time=response_time,status_code=response.status,anonymity_level=anonymity_level,real_ip=self.real_ip,proxy_ip=proxy_ip)else:return ProxyTestResult(proxy=proxy,is_available=False,response_time=response_time,status_code=response.status,error_message=f"HTTP {response.status}")except Exception as e:response_time = time.time() - start_timereturn ProxyTestResult(proxy=proxy,is_available=False,response_time=response_time,error_message=str(e))def _check_anonymity(self, proxy_ip: Optional[str]) -> str:"""检查代理匿名性"""if not proxy_ip or not self.real_ip:return "unknown"if proxy_ip == self.real_ip:return "transparent"  # 透明代理，暴露真实IPelse:return "anonymous"    # 匿名代理，隐藏真实IPdef test_direct_connection(self) -> ProxyTestResult:"""测试直连网络连通性"""start_time = time.time()try:response = requests.get(self.test_urls[0],timeout=self.timeout,headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'})response_time = time.time() - start_timeif response.status_code == 200:try:data = response.json()real_ip = data.get("origin", "").split(",")[0].strip()except:real_ip = Nonereturn ProxyTestResult(proxy="直连模式",is_available=True,response_time=response_time,status_code=response.status_code,anonymity_level="direct",real_ip=real_ip,proxy_ip=real_ip)else:return ProxyTestResult(proxy="直连模式",is_available=False,response_time=response_time,status_code=response.status_code,error_message=f"HTTP {response.status_code}")except Exception as e:response_time = time.time() - start_timereturn ProxyTestResult(proxy="直连模式",is_available=False,response_time=response_time,error_message=str(e))def test_proxies_sync(self, proxies: List[str]) -> List[ProxyTestResult]:"""同步批量测试代理"""logger.info(f"开始同步测试 {len(proxies)} 个代理...")results = []with ThreadPoolExecutor(max_workers=self.max_workers) as executor:futures = [executor.submit(self.test_proxy_sync, proxy) for proxy in proxies]for i, future in enumerate(futures):try:result = future.result()results.append(result)status = "✅" if result.is_available else "❌"logger.info(f"{status} [{i+1}/{len(proxies)}] {result.proxy} - "f"{'可用' if result.is_available else '不可用'} "f"({result.response_time:.2f}s)")except Exception as e:logger.error(f"测试代理失败: {e}")return resultsasync def test_proxies_async(self, proxies: List[str]) -> List[ProxyTestResult]:"""异步批量测试代理"""logger.info(f"开始异步测试 {len(proxies)} 个代理...")connector = aiohttp.TCPConnector(limit=self.max_workers)async with aiohttp.ClientSession(connector=connector) as session:tasks = [self.test_proxy_async(session, proxy) for proxy in proxies]results = await asyncio.gather(*tasks, return_exceptions=True)# 处理结果valid_results = []for i, result in enumerate(results):if isinstance(result, ProxyTestResult):valid_results.append(result)status = "✅" if result.is_available else "❌"logger.info(f"{status} [{i+1}/{len(proxies)}] {result.proxy} - "f"{'可用' if result.is_available else '不可用'} "f"({result.response_time:.2f}s)")else:logger.error(f"测试代理异常: {result}")return valid_resultsdef generate_report(self, results: List[ProxyTestResult]) -> Dict:"""生成测试报告"""total_count = len(results)available_count = sum(1 for r in results if r.is_available)unavailable_count = total_count - available_countif available_count > 0:avg_response_time = sum(r.response_time for r in results if r.is_available) / available_countmin_response_time = min(r.response_time for r in results if r.is_available)max_response_time = max(r.response_time for r in results if r.is_available)else:avg_response_time = min_response_time = max_response_time = 0# 按匿名性分类anonymity_stats = {}for result in results:if result.is_available and result.anonymity_level:anonymity_stats[result.anonymity_level] = anonymity_stats.get(result.anonymity_level, 0) + 1report = {"summary": {"total_proxies": total_count,"available_proxies": available_count,"unavailable_proxies": unavailable_count,"success_rate": f"{(available_count / total_count * 100):.2f}%" if total_count > 0 else "0%"},"performance": {"avg_response_time": f"{avg_response_time:.2f}s","min_response_time": f"{min_response_time:.2f}s","max_response_time": f"{max_response_time:.2f}s"},"anonymity": anonymity_stats,"available_proxies": [r.proxy for r in results if r.is_available],"unavailable_proxies": [{"proxy": r.proxy, "error": r.error_message} for r in results if not r.is_available]}return reportdef print_report(self, report: Dict):"""打印测试报告"""print("\n" + "="*60)print("🔍 代理池测试报告")print("="*60)# 概要信息summary = report["summary"]print(f"📊 测试概要:")print(f"   总代理数: {summary['total_proxies']}")print(f"   可用代理: {summary['available_proxies']} ✅")print(f"   不可用代理: {summary['unavailable_proxies']} ❌")print(f"   成功率: {summary['success_rate']}")# 性能信息if summary['available_proxies'] > 0:perf = report["performance"]print(f"\n⚡ 性能统计:")print(f"   平均响应时间: {perf['avg_response_time']}")print(f"   最快响应时间: {perf['min_response_time']}")print(f"   最慢响应时间: {perf['max_response_time']}")# 匿名性统计if report["anonymity"]:print(f"\n🔒 匿名性统计:")for level, count in report["anonymity"].items():print(f"   {level}: {count}个")# 可用代理列表if report["available_proxies"]:print(f"\n✅ 可用代理列表:")for proxy in report["available_proxies"]:print(f"   {proxy}")# 不可用代理列表if report["unavailable_proxies"]:print(f"\n❌ 不可用代理列表:")for item in report["unavailable_proxies"]:print(f"   {item['proxy']} - {item['error']}")print("="*60)def load_proxies_from_config(env_file: Optional[str] = None) -> List[str]:"""从配置文件加载代理列表"""try:import sysimport osfrom dotenv import load_dotenvimport json# 加载环境变量if env_file:load_dotenv(env_file)else:load_dotenv()# 尝试从环境变量直接读取proxy_config = os.getenv("PROXY_CONFIG", "")if proxy_config:try:# 支持JSON格式配置if proxy_config.strip().startswith('[') or proxy_config.strip().startswith('{'):proxy_data = json.loads(proxy_config)if isinstance(proxy_data, list):return proxy_dataelif isinstance(proxy_data, dict) and 'proxies' in proxy_data:return proxy_data['proxies']else:# 兼容旧的逗号分隔格式proxies = [p.strip() for p in proxy_config.replace('\n', ',').split(',') if p.strip()]return proxiesexcept json.JSONDecodeError:# 如果JSON解析失败，回退到逗号分隔格式proxies = [p.strip() for p in proxy_config.replace('\n', ',').split(',') if p.strip()]return proxies# 如果环境变量没有，尝试从config.py加载sys.path.append(os.path.dirname(__file__))from config import PROXY_LISTreturn PROXY_LISTexcept Exception as e:logger.warning(f"从配置文件加载代理失败: {e}")return []def main():"""主函数"""parser = argparse.ArgumentParser(description="代理池可用性测试程序")parser.add_argument("--proxies", "-p", nargs="+", help="代理列表")parser.add_argument("--file", "-f", help="从文件读取代理列表")parser.add_argument("--config", "-c", action="store_true", help="从config.py读取代理列表")parser.add_argument("--env-file", "-e", help="指定环境文件路径 (如 .env.test)")parser.add_argument("--direct", "-d", action="store_true", help="测试直连网络连通性")parser.add_argument("--timeout", "-t", type=int, default=10, help="请求超时时间(秒)")parser.add_argument("--workers", "-w", type=int, default=20, help="最大并发数")parser.add_argument("--async-mode", "-a", action="store_true", help="使用异步模式")parser.add_argument("--output", "-o", help="输出报告到文件")args = parser.parse_args()# 获取代理列表proxies = []if args.direct:# 测试直连模式tester = ProxyTester(timeout=args.timeout, max_workers=args.workers)logger.info("测试直连网络连通性...")direct_result = tester.test_direct_connection()report = tester.generate_report([direct_result])tester.print_report(report)if args.output:try:with open(args.output, 'w', encoding='utf-8') as f:json.dump(report, f, ensure_ascii=False, indent=2)logger.info(f"报告已保存到: {args.output}")except Exception as e:logger.error(f"保存报告失败: {e}")returnelif args.config:proxies = load_proxies_from_config(getattr(args, 'env_file', None))logger.info(f"从配置文件加载了 {len(proxies)} 个代理")elif args.file:try:with open(args.file, 'r', encoding='utf-8') as f:proxies = [line.strip() for line in f if line.strip()]logger.info(f"从文件 {args.file} 加载了 {len(proxies)} 个代理")except Exception as e:logger.error(f"读取文件失败: {e}")returnelif args.proxies:proxies = args.proxieslogger.info(f"使用命令行参数提供的 {len(proxies)} 个代理")else:# 使用示例代理proxies = ["http://127.0.0.1:8080","http://127.0.0.1:8081","http://proxy1.example.com:8080","http://proxy2.example.com:8080"]logger.info(f"使用示例代理列表 {len(proxies)} 个代理")if not proxies:logger.error("没有找到可测试的代理")return# 创建测试器tester = ProxyTester(timeout=args.timeout, max_workers=args.workers)# 执行测试if getattr(args, 'async_mode', False):results = asyncio.run(tester.test_proxies_async(proxies))else:results = tester.test_proxies_sync(proxies)# 生成报告report = tester.generate_report(results)# 打印报告tester.print_report(report)# 保存报告到文件if args.output:try:with open(args.output, 'w', encoding='utf-8') as f:json.dump(report, f, ensure_ascii=False, indent=2)logger.info(f"报告已保存到: {args.output}")except Exception as e:logger.error(f"保存报告失败: {e}")if __name__ == "__main__":main()
http://www.dtcms.com/a/520485.html

相关文章：

  • [学习日记]看书笔记
  • 专有软件使用Linux内核的用户头文件违反GPL吗？| 开源合规场景
  • Python 实现 Pelco-D 协议云台控制（win与ubuntu）
  • MEMS结构光在3D人脸识别中的系统优化
  • 东莞网站建设流程响应式网站制作价格
  • 做网站优化的好处做网站客户总是要退款
  • [人工智能-大模型-51]：Transformer、大模型、Copilot、具身智能、多模态、空间智能、世界模型，什么意思，它们不同点和联系
  • 鸿蒙：简单实现列表下拉刷新+上拉加载
  • [nanoGPT] ChatGPT 的 LLM 的全栈实现 | 快速上手
  • 公司公司手机网站制作互联网营销的特点
  • 做网站苏州营销型网站有什么特点
  • 【计算机网络】408计算机网络高分指南：物理层编码与调制技术精讲
  • 怎样做一个企业的网站建站个人网站备案 网站名称
  • AR巡检系统：打开工业智能运维的下一幕
  • wang域名注册网站厚瑜网站建设
  • 网站建设夬金手指花总南昌做网站哪家好
  • 洪梅网站建设微信 host 微网站模版
  • 网站建设要用到的技术有哪些行业网站建设报价
  • 建网站所需材料生物信息网站建设
  • 【EE初阶】JVM
  • 深度学习------YOLOv5《第一篇》
  • 手机网站无法访问的解决方法文字图片制作网站
  • 叙述一个网站开发流程住房建设和城乡管理局官网
  • HarmonyOS 分布式与 AI 集成：构建智能协同应用的进阶实践
  • Trae x 图片素描MCP一键将普通图片转换为多风格素描效果
  • 游艇网站建设方案网页给别人做的 网站后续收费
  • UE5 外轮廓线,边缘,边界
  • Jackson 序列化的隐性成本
  • ProcDump 学习笔记（6.5）：指定转储文件路径与命名策略
  • STM32项目分享：智能植物灌溉系统