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

# 方式1：JSON數(shù)組格式（推薦）
PROXY_CONFIG='["http://127.0.0.1:8080", "http://proxy2.com:8080", "http://user:pass@proxy3.com:8080"]'

# 方式2：JSON對(duì)象格式（支持更多配置）
PROXY_CONFIG='{"proxies": ["http://127.0.0.1:8080", "http://proxy2.com:8080"], "timeout": 10}'

# 方式3：逗號(hào)分隔格式（兼容舊版）
PROXY_CONFIG="http://127.0.0.1:8080,http://proxy2.com:8080,http://user:pass@proxy3.com:8080"

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

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
代理池可用性測(cè)試程序
測(cè)試代理的連通性、響應(yīng)時(shí)間、匿名性等指標(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:
    """代理測(cè)試結(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:
    """代理測(cè)試器"""
    
    def __init__(self, 
                 test_urls: List[str] = None,
                 timeout: int = 10,
                 max_workers: int = 20):
        """
        初始化代理測(cè)試器
        
        Args:
            test_urls: 測(cè)試URL列表
            timeout: 請(qǐng)求超時(shí)時(shí)間
            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:
        """同步測(cè)試單個(gè)代理"""
        start_time = time.time()
        
        try:
            # 解析代理格式
            proxy_dict = {
                'http': proxy,
                'https': proxy
            }
            
            # 測(cè)試連通性
            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:
        """異步測(cè)試單個(gè)代理"""
        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:
        """測(cè)試直連網(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]:
        """同步批量測(cè)試代理"""
        logger.info(f"開始同步測(cè)試 {len(proxies)} 個(gè)代理...")
        
        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"測(cè)試代理失敗: {e}")
        
        return results
    
    async def test_proxies_async(self, proxies: List[str]) -> List[ProxyTestResult]:
        """異步批量測(cè)試代理"""
        logger.info(f"開始異步測(cè)試 {len(proxies)} 個(gè)代理...")
        
        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"測(cè)試代理異常: {result}")
            
            return valid_results
    
    def generate_report(self, results: List[ProxyTestResult]) -> Dict:
        """生成測(cè)試報(bào)告"""
        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):
        """打印測(cè)試報(bào)告"""
        print("\n" + "="*60)
        print("?? 代理池測(cè)試報(bào)告")
        print("="*60)
        
        # 概要信息
        summary = report["summary"]
        print(f"?? 測(cè)試概要:")
        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)計(jì):")
            print(f"   平均響應(yīng)時(shí)間: {perf['avg_response_time']}")
            print(f"   最快響應(yīng)時(shí)間: {perf['min_response_time']}")
            print(f"   最慢響應(yīng)時(shí)間: {perf['max_response_time']}")
        
        # 匿名性統(tǒng)計(jì)
        if report["anonymity"]:
            print(f"\n?? 匿名性統(tǒng)計(jì):")
            for level, count in report["anonymity"].items():
                print(f"   {level}: {count}個(gè)")
        
        # 可用代理列表
        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:
                    # 兼容舊的逗號(hào)分隔格式
                    proxies = [p.strip() for p in proxy_config.replace('\n', ',').split(',') if p.strip()]
                    return proxies
            except json.JSONDecodeError:
                # 如果JSON解析失敗，回退到逗號(hào)分隔格式
                proxies = [p.strip() for p in proxy_config.replace('\n', ',').split(',') if p.strip()]
                return proxies
        
        # 如果環(huán)境變量沒(méi)有，嘗試從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="代理池可用性測(cè)試程序")
    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="測(cè)試直連網(wǎng)絡(luò)連通性")
    parser.add_argument("--timeout", "-t", type=int, default=10, help="請(qǐng)求超時(shí)時(shí)間(秒)")
    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="輸出報(bào)告到文件")
    
    args = parser.parse_args()
    
    # 獲取代理列表
    proxies = []
    
    if args.direct:
        # 測(cè)試直連模式
        tester = ProxyTester(timeout=args.timeout, max_workers=args.workers)
        logger.info("測(cè)試直連網(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"報(bào)告已保存到: {args.output}")
            except Exception as e:
                logger.error(f"保存報(bào)告失敗: {e}")
        return
    elif args.config:
        proxies = load_proxies_from_config(getattr(args, 'env_file', None))
        logger.info(f"從配置文件加載了 {len(proxies)} 個(gè)代理")
    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)} 個(gè)代理")
        except Exception as e:
            logger.error(f"讀取文件失敗: {e}")
            return
    elif args.proxies:
        proxies = args.proxies
        logger.info(f"使用命令行參數(shù)提供的 {len(proxies)} 個(gè)代理")
    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)} 個(gè)代理")
    
    if not proxies:
        logger.error("沒(méi)有找到可測(cè)試的代理")
        return
    
    # 創(chuàng)建測(cè)試器
    tester = ProxyTester(timeout=args.timeout, max_workers=args.workers)
    
    # 執(zhí)行測(cè)試
    if getattr(args, 'async_mode', False):
        results = asyncio.run(tester.test_proxies_async(proxies))
    else:
        results = tester.test_proxies_sync(proxies)
    
    # 生成報(bào)告
    report = tester.generate_report(results)
    
    # 打印報(bào)告
    tester.print_report(report)
    
    # 保存報(bào)告到文件
    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"報(bào)告已保存到: {args.output}")
        except Exception as e:
            logger.error(f"保存報(bào)告失敗: {e}")

if __name__ == "__main__":
    main()