import requests
import time
import concurrent.futures
from dataclasses import dataclass
from typing import Dict, List, Optional
import statistics
@dataclass
class ProxyConfig:
    """代理服务器配置"""
    name: str          # 服务商名称
    http_proxy: str    # HTTP代理地址，格式：http://user:pass@host:port
    https_proxy: str   # HTTPS代理地址，格式：http://user:pass@host:port
@dataclass
class TestResult:
    """单次测试结果"""
    success: bool          # 是否成功
    status_code: int       # HTTP状态码
    connect_time: float    # 连接时间（秒）
    transfer_time: float   # 传输时间（秒）
    total_time: float      # 总耗时（秒）
    error_msg: str = ""    # 错误信息
class StaticProxyTester:
    """静态住宅IP代理测试器"""

    def __init__(self, target_url: str, num_rounds: int = 10, timeout: int = 10):
        """
        初始化测试器

        Args:
            target_url: 目标新闻网站URL
            num_rounds: 每个代理的测试轮次
            timeout: 请求超时时间（秒）
        """
        self.target_url = target_url
        self.num_rounds = num_rounds
        self.timeout = timeout

    def _single_request(self, proxy_config: ProxyConfig) -> TestResult:
        """执行单次请求测试[citation:1]"""
        proxies = {
            'http': proxy_config.http_proxy,
            'https': proxy_config.https_proxy
        }

        start_time = time.time()
        connect_time = 0
        transfer_time = 0

        try:
            # 发送请求并测量时间
            response = requests.get(
                self.target_url,
                proxies=proxies,
                timeout=self.timeout
            )

            # 计算连接时间和传输时间
            # 注意：requests的elapsed时间包含连接和传输，这里做简单拆分
            total_elapsed = response.elapsed.total_seconds()
            connect_time = total_elapsed * 0.3  # 估算连接时间占30%
            transfer_time = total_elapsed * 0.7  # 估算传输时间占70%

            return TestResult(
                success=response.status_code == 200,
                status_code=response.status_code,
                connect_time=connect_time,
                transfer_time=transfer_time,
                total_time=total_elapsed
            )

        except requests.exceptions.ConnectTimeout:
            return TestResult(
                success=False,
                status_code=0,
                connect_time=self.timeout,
                transfer_time=0,
                total_time=self.timeout,
                error_msg="连接超时"
            )
        except requests.exceptions.ReadTimeout:
            return TestResult(
                success=False,
                status_code=0,
                connect_time=connect_time,
                transfer_time=self.timeout - connect_time,
                total_time=self.timeout,
                error_msg="读取超时"
            )
        except requests.exceptions.ProxyError as e:
            return TestResult(
                success=False,
                status_code=0,
                connect_time=0,
                transfer_time=0,
                total_time=time.time() - start_time,
                error_msg=f"代理错误: {str(e)}"
            )
        except Exception as e:
            return TestResult(
                success=False,
                status_code=0,
                connect_time=0,
                transfer_time=0,
                total_time=time.time() - start_time,
                error_msg=f"其他错误: {str(e)}"
            )

    def test_proxy(self, proxy_config: ProxyConfig) -> Dict:
        """测试单个代理的多轮性能[citation:1]"""
        print(f"开始测试 {proxy_config.name}...")

        results = []
        success_count = 0

        # 执行多轮测试
        for round_num in range(1, self.num_rounds + 1):
            result = self._single_request(proxy_config)
            results.append(result)

            if result.success:
                success_count += 1
                print(f"  Round {round_num}: 成功, 耗时 {result.total_time:.3f}s")
            else:
                print(f"  Round {round_num}: 失败 - {result.error_msg}")

            # 轮次间短暂间隔，模拟真实场景
            if round_num < self.num_rounds:
                time.sleep(0.5)

        # 计算统计指标
        success_results = [r for r in results if r.success]

        if success_results:
            avg_connect = statistics.mean([r.connect_time for r in success_results])
            avg_transfer = statistics.mean([r.transfer_time for r in success_results])
            avg_total = statistics.mean([r.total_time for r in success_results])
        else:
            avg_connect = avg_transfer = avg_total = 0

        success_rate = (success_count / self.num_rounds) * 100

        # 计算稳定性评价
        stability = self._calculate_stability(success_rate, avg_total if success_results else float('inf'))

        return {
            'service': proxy_config.name,
            'success_rate': success_rate,
            'avg_connect_time': avg_connect,
            'avg_transfer_time': avg_transfer,
            'avg_total_time': avg_total,
            'stability': stability,
            'total_tests': self.num_rounds,
            'success_count': success_count,
            'fail_count': self.num_rounds - success_count,
            'detailed_results': results
        }

    def _calculate_stability(self, success_rate: float, avg_total_time: float) -> str:
        """根据成功率和平均耗时计算稳定性评价"""
        if success_rate >= 95 and avg_total_time < 1.5:
            return "⭐⭐⭐⭐⭐"
        elif success_rate >= 85 and avg_total_time < 2.0:
            return "⭐⭐⭐⭐"
        elif success_rate >= 75 and avg_total_time < 2.5:
            return "⭐⭐⭐"
        elif success_rate >= 60:
            return "⭐⭐"
        else:
            return "⭐"

    def run_all_tests(self, proxy_configs: List[ProxyConfig], parallel: bool = False) -> List[Dict]:
        """运行所有代理测试"""
        print(f"开始代理稳定性测试，目标URL: {self.target_url}")
        print(f"每个代理测试轮次: {self.num_rounds}")
        print("=" * 60)

        results = []

        if parallel:
            # 并行测试（适用于多个代理同时测试）
            with concurrent.futures.ThreadPoolExecutor(max_workers=len(proxy_configs)) as executor:
                future_to_proxy = {
                    executor.submit(self.test_proxy, config): config 
                    for config in proxy_configs
                }

                for future in concurrent.futures.as_completed(future_to_proxy):
                    result = future.result()
                    results.append(result)
        else:
            # 串行测试（更稳定，便于调试）
            for config in proxy_configs:
                result = self.test_proxy(config)
                results.append(result)

        return results

    def print_summary(self, results: List[Dict]):
        """打印测试结果摘要（类似你提供的表格格式）"""
        print("\n" + "=" * 60)
        print("代理性能测试结果汇总")
        print("=" * 60)
        print(f"{'服务商':<15} {'成功率':<8} {'平均连接时间(s)':<15} {'平均传输时间(s)':<15} {'平均总耗时(s)':<15} {'稳定性评价':<10}")
        print("-" * 80)

        for result in results:
            print(f"{result['service']:<15} "
                  f"{result['success_rate']:>6.1f}% "
                  f"{result['avg_connect_time']:>15.3f} "
                  f"{result['avg_transfer_time']:>15.3f} "
                  f"{result['avg_total_time']:>15.3f} "
                  f"{result['stability']:>10}")

        # 找出最佳服务商
        if results:
            best = max(results, key=lambda x: (x['success_rate'], -x['avg_total_time']))
            print("=" * 80)
            print(f"最佳表现: {best['service']} (成功率: {best['success_rate']:.1f}%, "
                  f"平均耗时: {best['avg_total_time']:.3f}s)")
# ====================== 使用示例 ======================
def main():
    """测试脚本使用示例"""

    # 1. 定义要测试的代理配置（请替换为你的实际代理信息）[citation:1]
    ]

    # 2. 创建测试器实例
    # 请将target_url替换为你要测试的实际新闻网站
    tester = StaticProxyTester(
        target_url="https://news.google.com/home?hl=zh-CN&gl=CN&ceid=CN:zh-Hans",  # 替换为目标新闻网站
        num_rounds=10,    # 每个代理测试10轮
        timeout=8         # 8秒超时
    )

    # 3. 运行测试（使用串行模式，更稳定）
    results = tester.run_all_tests(proxy_configs, parallel=False)

    # 4. 打印汇总结果
    tester.print_summary(results)

    # 5. 可选：保存详细结果到文件
    save_detailed_results(results, "proxy_test_results.txt")
def save_detailed_results(results: List[Dict], filename: str):
    """保存详细测试结果到文件"""
    with open(filename, 'w', encoding='utf-8') as f:
        f.write("静态住宅IP代理详细测试报告\n")
        f.write("=" * 60 + "\n\n")

        for result in results:
            f.write(f"\n服务商: {result['service']}\n")
            f.write(f"成功率: {result['success_rate']:.1f}% ")
            f.write(f"({result['success_count']}/{result['total_tests']})\n")
            f.write(f"平均连接时间: {result['avg_connect_time']:.3f}s\n")
            f.write(f"平均传输时间: {result['avg_transfer_time']:.3f}s\n")
            f.write(f"平均总耗时: {result['avg_total_time']:.3f}s\n")
            f.write(f"稳定性评价: {result['stability']}\n")
            f.write("-" * 40 + "\n")

            # 每轮详细结果
            for i, detail in enumerate(result['detailed_results'], 1):
                status = "成功" if detail.success else "失败"
                f.write(f"  轮次 {i:2d}: {status:3} | ")
                f.write(f"状态码: {detail.status_code:3} | ")
                f.write(f"总耗时: {detail.total_time:6.3f}s | ")
                if not detail.success:
                    f.write(f"错误: {detail.error_msg}")
                f.write("\n")
            f.write("\n")
if __name__ == "__main__":
    # 安装必要库：pip install requests
    main()

import requests
import time
from bs4 import BeautifulSoup
import logging
# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class StaticResidentialCrawler:
    def __init__(self, proxy_host, proxy_port, username, password):
        """
        初始化静态住宅代理爬虫。
        参数示例（需替换为从后台获取的实际值）：
        proxy_host: geo.net
        proxy_port: 2333
        username: your_username
        password: your_password
        """
        self.proxies = {
            'http': f'http://{username}:{password}@{proxy_host}:{proxy_port}',
            'https': f'http://{username}:{password}@{proxy_host}:{proxy_port}',
        }
        self.session = requests.Session()
        self.session.proxies.update(self.proxies)  # 会话级代理，所有请求使用同一IP
        self.session.headers.update({
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
        })

    def fetch_news(self, url, max_retries=3):
        """使用静态IP抓取新闻页面"""
        for attempt in range(max_retries):
            try:
                logger.info(f"[尝试 {attempt+1}] 抓取: {url}")
                # 添加随机延迟，模拟人工
                time.sleep(2)
                resp = self.session.get(url, timeout=30)
                resp.raise_for_status()

                # 简单验证：返回内容是否有效
                if len(resp.text) < 1024:
                    logger.warning("返回内容过短，可能遭遇拦截。")
                    continue

                logger.info(f"抓取成功，状态码: {resp.status_code}")
                return resp.text

            except requests.exceptions.RequestException as e:
                logger.error(f"请求失败: {e}")
                if attempt == max_retries - 1:
                    return None
                time.sleep(5)  # 重试前等待
        return None

# 1. 初始化爬虫（请替换为您的真实代理信息）
crawler = StaticResidentialCrawler(
    proxy_host='geo.net',
    proxy_port=2333,
    username='your_username_here',
    password='your_password_here'
)
# 2. 目标新闻网址（示例：某国际新闻网站科技板块）
target_url = "https://example-news.com/technology"
# 3. 执行抓取
html_content = crawler.fetch_news(target_url)
if html_content:
    # 4. 使用BeautifulSoup解析新闻标题和链接

    soup = BeautifulSoup(html_content, 'html.parser')
    # 此处需根据目标网站的实际HTML结构编写解析规则
    news_items = []
    for article in soup.select('article.news-card'):  # 假设的CSS选择器
        title_elem = article.select_one('h2 a')
        if title_elem:
            news_items.append({
                'title': title_elem.text.strip(),
                'link': title_elem['href'],
                'source': target_url
            })

    print(f"解析到 {len(news_items)} 条新闻。")
    for item in news_items[:5]:  # 打印前5条
        print(f"- {item['title']}")
else:
    print("新闻抓取失败。")