Python代理延迟检测：从单次请求到批量测试的完整解决方案

import requests
import time
import statistics
from typing import List, Dict, Optional
from urllib.parse import urlparse
​
class ProxyLatencyTester:
    def __init__(self, target_url: str, proxy_url: str, timeout: int = 10):
        """
        初始化延迟检测器
        
        :param target_url: 目标网站URL
        :param proxy_url: 代理服务器URL (格式: http://user:pass@host:port 或 http://host:port)
        :param timeout: 请求超时时间(秒)
        """
        self.target_url = target_url
        self.proxy_url = proxy_url
        self.timeout = timeout
        self.results = []
        
        # 验证代理URL格式
        parsed = urlparse(proxy_url)
        if not parsed.scheme:
            raise ValueError("代理URL必须包含协议 (如 http:// 或 https://)")
    
    def test_once(self) -> Optional[float]:
        """
        执行一次延迟测试
        
        :return: 延迟时间(秒)，如果测试失败则返回None
        """
        try:
            start_time = time.time()
            response = requests.get(
                self.target_url,
                proxies={
                    'http': self.proxy_url,
                    'https': self.proxy_url
                },
                timeout=self.timeout,
                headers={
                    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
                }
            )
            end_time = time.time()
            
            latency = end_time - start_time
            self.results.append(latency)
            
            # 检查响应状态码
            if response.status_code != 200:
                print(f"警告: 收到非200状态码: {response.status_code}")
                
            return latency
            
        except requests.exceptions.ProxyError as e:
            print(f"代理错误: {e}")
            return None
        except requests.exceptions.ConnectTimeout as e:
            print(f"连接超时: {e}")
            return None
        except requests.exceptions.ReadTimeout as e:
            print(f"读取超时: {e}")
            return None
        except requests.exceptions.RequestException as e:
            print(f"请求异常: {e}")
            return None
    
    def run_tests(self, num_tests: int = 5) -> Dict:
        """
        执行多次延迟测试并返回统计结果
        
        :param num_tests: 测试次数
        :return: 包含统计信息的字典
        """
        print(f"开始对 {self.target_url} 进行延迟测试，使用代理: {self.proxy_url}")
        print(f"计划执行 {num_tests} 次测试...")
        print("-" * 50)
        
        successful_tests = 0
        self.results = []
        
        for i in range(num_tests):
            print(f"测试 #{i+1}...", end=" ")
            latency = self.test_once()
            
            if latency is not None:
                print(f"延迟: {latency:.3f} 秒")
                successful_tests += 1
            else:
                print("测试失败")
            
            # 短暂暂停以避免过于频繁的请求
            if i < num_tests - 1:
                time.sleep(1)
        
        print("-" * 50)
        
        # 计算统计信息
        stats = {
            'total_tests': num_tests,
            'successful_tests': successful_tests,
            'success_rate': successful_tests / num_tests if num_tests > 0 else 0
        }
        
        if self.results:
            stats['min_latency'] = min(self.results)
            stats['max_latency'] = max(self.results)
            stats['avg_latency'] = statistics.mean(self.results)
            stats['median_latency'] = statistics.median(self.results)
            
            if len(self.results) > 1:
                stats['stdev_latency'] = statistics.stdev(self.results)
            else:
                stats['stdev_latency'] = 0
        else:
            stats['min_latency'] = None
            stats['max_latency'] = None
            stats['avg_latency'] = None
            stats['median_latency'] = None
            stats['stdev_latency'] = None
        
        return stats
    
    def print_stats(self, stats: Dict):
        """打印统计结果"""
        print(f"测试完成!")
        print(f"总测试次数: {stats['total_tests']}")
        print(f"成功次数: {stats['successful_tests']}")
        print(f"成功率: {stats['success_rate']:.2%}")
        
        if stats['successful_tests'] > 0:
            print(f"最小延迟: {stats['min_latency']:.3f} 秒")
            print(f"最大延迟: {stats['max_latency']:.3f} 秒")
            print(f"平均延迟: {stats['avg_latency']:.3f} 秒")
            print(f"中位数延迟: {stats['median_latency']:.3f} 秒")
            print(f"延迟标准差: {stats['stdev_latency']:.3f} 秒")
            
            # 延迟评价
            avg = stats['avg_latency']
            if avg < 0.5:
                rating = "极好"
            elif avg < 1.0:
                rating = "良好"
            elif avg < 2.0:
                rating = "一般"
            elif avg < 3.0:
                rating = "较差"
            else:
                rating = "极差"
                
            print(f"延迟评价: {rating}")
​
def main():
    # 配置参数
    TARGET_URL = "https://httpbin.org/ip"  # 一个返回IP信息的测试网站
    PROXY_URL = "http://your-proxy-username:your-proxy-password@your-proxy-host:your-proxy-port"
    TEST_COUNT = 5  # 测试次数
    
    # 创建测试器实例
    tester = ProxyLatencyTester(
        target_url=TARGET_URL,
        proxy_url=PROXY_URL,
        timeout=10
    )
    
    # 运行测试
    stats = tester.run_tests(num_tests=TEST_COUNT)
    
    # 打印结果
    print("\n测试结果统计:")
    print("=" * 50)
    tester.print_stats(stats)
​
if __name__ == "__main__":
    main()

pip install requests