# 代理池配置 - 支持多種格式

# 方式1：JSON數(shù)組格式（推薦）
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"

# 代理測試相關(guān)配置
PROXY_TEST_URL="https://httpbin.org/ip"
PROXY_TEST_TIMEOUT="5"

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
代理池可用性測試程序
測試代理的連通性、響應(yīng)時間、匿名性等指標(biāo)

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:
    """代理測試結(jié)果"""
    proxy: str
    is_available: bool
    response_time: float
    status_code: Optional[int] = None
    error_message: Optional[str] = None
    anonymity_level: Optional[str] = None  # transparent, anonymous, elite
    real_ip: Optional[str] = None
    proxy_ip: Optional[str] = None

class ProxyTester:
    """代理測試器"""
    
    def __init__(self, 
                 test_urls: List[str] = None,
                 timeout: int = 10,
                 max_workers: int = 20):
        """
        初始化代理測試器
        
        Args:
            test_urls: 測試URL列表
            timeout: 請求超時時間
            max_workers: 最大并發(fā)數(shù)
        """
        self.test_urls = test_urls or [
            "https://httpbin.org/ip",
            "https://api.ipify.org?format=json",
            "https://ifconfig.me/ip"
        ]
        self.timeout = timeout
        self.max_workers = max_workers
        
        # 獲取本機(jī)真實(shí)IP
        self.real_ip = self._get_real_ip()
        logger.info(f"本機(jī)真實(shí)IP: {self.real_ip}")
    
    def _get_real_ip(self) -> Optional[str]:
        """獲取本機(jī)真實(shí)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"獲取真實(shí)IP失敗: {e}")
        return None
    
    def 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_time
            
            if response.status_code == 200:
                # 解析響應(yīng)獲取代理IP
                try:
                    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_time
            return 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_time
                
                if response.status == 200:
                    try:
                        data = await 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,
                        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_time
            return 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"  # 透明代理，暴露真實(shí)IP
        else:
            return "anonymous"    # 匿名代理，隱藏真實(shí)IP
    
    def test_direct_connection(self) -> ProxyTestResult:
        """測試直連網(wǎng)絡(luò)連通性"""
        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_time
            
            if response.status_code == 200:
                try:
                    data = response.json()
                    real_ip = data.get("origin", "").split(",")[0].strip()
                except:
                    real_ip = None
                
                return 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_time
            return 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 results
    
    async 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)
            
            # 處理結(jié)果
            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_results
    
    def 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_count
        
        if available_count > 0:
            avg_response_time = sum(r.response_time for r in results if r.is_available) / available_count
            min_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) + 1
        
        report = {
            "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 report
    
    def print_report(self, report: Dict):
        """打印測試報告"""
        print("\n" + "="*60)
        print("?? 代理池測試報告")
        print("="*60)
        
        # 概要信息
        summary = report["summary"]
        print(f"?? 測試概要:")
        print(f"   總代理數(shù): {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? 性能統(tǒng)計:")
            print(f"   平均響應(yīng)時間: {perf['avg_response_time']}")
            print(f"   最快響應(yīng)時間: {perf['min_response_time']}")
            print(f"   最慢響應(yīng)時間: {perf['max_response_time']}")
        
        # 匿名性統(tǒng)計
        if report["anonymity"]:
            print(f"\n?? 匿名性統(tǒng)計:")
            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 sys
        import os
        from dotenv import load_dotenv
        import json
        
        # 加載環(huán)境變量
        if env_file:
            load_dotenv(env_file)
        else:
            load_dotenv()
        
        # 嘗試從環(huán)境變量直接讀取
        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_data
                    elif 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 proxies
            except json.JSONDecodeError:
                # 如果JSON解析失敗，回退到逗號分隔格式
                proxies = [p.strip() for p in proxy_config.replace('\n', ',').split(',') if p.strip()]
                return proxies
        
        # 如果環(huán)境變量沒有，嘗試從config.py加載
        sys.path.append(os.path.dirname(__file__))
        from config import PROXY_LIST
        return PROXY_LIST
        
    except Exception as e:
        logger.warning(f"從配置文件加載代理失敗: {e}")
        return []

def main():
    """主函數(shù)"""
    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="指定環(huán)境文件路徑 (如 .env.test)")
    parser.add_argument("--direct", "-d", action="store_true", help="測試直連網(wǎng)絡(luò)連通性")
    parser.add_argument("--timeout", "-t", type=int, default=10, help="請求超時時間(秒)")
    parser.add_argument("--workers", "-w", type=int, default=20, help="最大并發(fā)數(shù)")
    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("測試直連網(wǎng)絡(luò)連通性...")
        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}")
        return
    elif 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}")
            return
    elif args.proxies:
        proxies = args.proxies
        logger.info(f"使用命令行參數(shù)提供的 {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
    
    # 創(chuàng)建測試器
    tester = ProxyTester(timeout=args.timeout, max_workers=args.workers)
    
    # 執(zhí)行測試
    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()