高并发分布式爬虫在 Web 端验证码封锁下的 IP 轮拨方案

"""
网帆代理动态代理IP管理器
支持按请求轮拨、按时间轮拨、混合轮拨三种策略
"""
import time
import threading
import requests
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Optional, List, Dict
from enum import Enum


class RotationStrategy(Enum):
    """IP轮拨策略"""
    PER_REQUEST = "per_request"   # 按请求轮拨
    PER_TIME = "per_time"         # 按时间轮拨
    HYBRID = "hybrid"             # 混合轮拨


@dataclass
class ProxyIP:
    """代理IP数据结构"""
    ip: str
    port: int
    city: str
    isp: str
    ip_type: str
    expire_at: float
    speed: int
    fail_count: int = 0
    last_used: float = 0.0
    request_count: int = 0


class fanProxyManager:
    """网帆代理IP管理器"""

    def __init__(
        self,
        api_key: str,
        strategy: RotationStrategy = RotationStrategy.HYBRID,
        rotation_interval: int = 300,  # 按时间轮拨周期（秒），默认5分钟
        max_fail_count: int = 3,       # 最大失败次数，超过则剔除
        max_request_per_ip: int = 50,  # 单IP最大请求数
        api_url: str = "https://api.fanproxy.com/api/proxies",
    ):
        self.api_key = api_key
        self.strategy = strategy
        self.rotation_interval = rotation_interval
        self.max_fail_count = max_fail_count
        self.max_request_per_ip = max_request_per_ip
        self.api_url = api_url

        self._ip_pool: List[ProxyIP] = []
        self._current_ip: Optional[ProxyIP] = None
        self._blacklist: set = set()
        self._lock = threading.Lock()
        self._stats = defaultdict(int)

    def fetch_proxies(self, count: int = 1, city: str = None) -> List[ProxyIP]:
        """从网帆代理API获取IP列表"""
        params = {
            "api_key": self.api_key,
            "count": count,
            "format": "json",
            "protocol": "http",
        }
        if city:
            params["city"] = city

        try:
            resp = requests.get(self.api_url, params=params, timeout=10)
            resp.raise_for_status()
            data = resp.json()

            proxies = []
            for item in data.get("data", []):
                proxy = ProxyIP(
                    ip=item["ip"],
                    port=item["port"],
                    city=item.get("city", ""),
                    isp=item.get("isp", ""),
                    ip_type=item.get("type", ""),
                    expire_at=item.get("expire_at", time.time() + 300),
                    speed=item.get("speed", 0),
                )
                proxies.append(proxy)

            self._stats["fetch_success"] += 1
            return proxies

        except Exception as e:
            self._stats["fetch_fail"] += 1
            print(f"[ProxyManager] 获取IP失败: {e}")
            return []

    def get_proxy(self) -> Optional[ProxyIP]:
        """获取一个可用代理IP（根据轮拨策略）"""
        with self._lock:
            now = time.time()

            # 按请求轮拨：每次获取新IP
            if self.strategy == RotationStrategy.PER_REQUEST:
                proxies = self.fetch_proxies(count=1)
                if proxies:
                    self._current_ip = proxies[0]
                    self._current_ip.last_used = now
                    self._stats["total_assigned"] += 1
                    return self._current_ip
                return None

            # 按时间轮拨：检查当前IP是否过期
            if self.strategy == RotationStrategy.PER_TIME:
                if (self._current_ip and
                    now < self._current_ip.expire_at and
                    self._current_ip.fail_count < self.max_fail_count and
                    self._current_ip.request_count < self.max_request_per_ip):
                    self._current_ip.last_used = now
                    return self._current_ip

                proxies = self.fetch_proxies(count=1)
                if proxies:
                    self._current_ip = proxies[0]
                    self._current_ip.last_used = now
                    self._stats["total_assigned"] += 1
                    return self._current_ip
                return None

            # 混合轮拨：优先复用，触发条件时轮换
            if self.strategy == RotationStrategy.HYBRID:
                if (self._current_ip and
                    now < self._current_ip.expire_at and
                    self._current_ip.fail_count < self.max_fail_count and
                    self._current_ip.request_count < self.max_request_per_ip):
                    self._current_ip.last_used = now
                    self._current_ip.request_count += 1
                    return self._current_ip

                # 需要轮换：从池中取或获取新IP
                self._current_ip = self._get_healthy_from_pool()
                if not self._current_ip:
                    proxies = self.fetch_proxies(count=1)
                    if proxies:
                        self._current_ip = proxies[0]
                        self._ip_pool.append(self._current_ip)

                if self._current_ip:
                    self._current_ip.last_used = now
                    self._current_ip.request_count += 1
                    self._stats["total_assigned"] += 1
                    return self._current_ip
                return None

    def _get_healthy_from_pool(self) -> Optional[ProxyIP]:
        """从IP池中获取健康IP"""
        now = time.time()
        # 清理过期和黑名单IP
        self._ip_pool = [
            ip for ip in self._ip_pool
            if ip.expire_at > now
            and ip.ip not in self._blacklist
            and ip.fail_count < self.max_fail_count
            and ip.request_count < self.max_request_per_ip
        ]
        if self._ip_pool:
            # 选择请求次数最少的IP（负载均衡）
            return min(self._ip_pool, key=lambda x: x.request_count)
        return None

    def mark_success(self, ip: str):
        """标记请求成功"""
        with self._lock:
            for proxy in self._ip_pool:
                if proxy.ip == ip:
                    proxy.fail_count = 0
                    self._stats["success"] += 1
                    break

    def mark_fail(self, ip: str, reason: str = ""):
        """标记请求失败"""
        with self._lock:
            for proxy in self._ip_pool:
                if proxy.ip == ip:
                    proxy.fail_count += 1
                    if proxy.fail_count >= self.max_fail_count:
                        self._blacklist.add(ip)
                        self._stats["blacklisted"] += 1
                    break
            self._stats["fail"] += 1
            if reason:
                self._stats[f"fail_{reason}"] += 1

    def release_ip(self, ip: str):
        """主动释放IP（通过API通知网帆代理）"""
        try:
            requests.post(
                f"{self.api_url}/release",
                json={"api_key": self.api_key, "ip": ip},
                timeout=5
            )
        except Exception:
            pass

    def get_stats(self) -> Dict:
        """获取统计信息"""
        return dict(self._stats)

"""
验证码检测与自动重试引擎
检测目标网站的验证码触发，自动轮换IP并重试
"""
import time
import random
import requests
from typing import Optional, Callable
from urllib.parse import urlparse


class CaptchaDetector:
    """验证码检测器"""

    # 常见验证码页面特征
    CAPTCHA_KEYWORDS = [
        "captcha", "verify", "验证", "人机验证",
        "安全验证", "滑动验证", "请完成验证",
        "challenge", "security check", "robot"
    ]

    CAPTCHA_STATUS_CODES = {403, 429}

    @classmethod
    def detect(cls, response: requests.Response) -> bool:
        """检测响应是否为验证码页面"""
        # 状态码检测
        if response.status_code in cls.CAPTCHA_STATUS_CODES:
            return True

        # 内容特征检测
        try:
            text = response.text.lower()
            for keyword in cls.CAPTCHA_KEYWORDS:
                if keyword in text:
                    return True
        except Exception:
            pass

        # 响应体过短（验证码页面通常较小）
        if len(response.content) < 500 and response.status_code == 200:
            content_type = response.headers.get("Content-Type", "")
            if "text/html" in content_type:
                return True

        return False


class RetryEngine:
    """自动重试引擎"""

    def __init__(
        self,
        proxy_manager: fanProxyManager,
        max_retries: int = 3,
        base_delay: float = 1.0,
        max_delay: float = 30.0,
        backoff_factor: float = 2.0,
    ):
        self.proxy_manager = proxy_manager
        self.max_retries = max_retries
        self.base_delay = base_delay
        self.max_delay = max_delay
        self.backoff_factor = backoff_factor

    def request_with_retry(
        self,
        url: str,
        method: str = "GET",
        headers: dict = None,
        data: dict = None,
        timeout: int = 15,
        on_captcha: Callable = None,
    ) -> Optional[requests.Response]:
        """带验证码检测和IP轮拨的请求"""
        default_headers = {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                          "AppleWebKit/537.36 (KHTML, like Gecko) "
                          "Chrome/120.0.0.0 Safari/537.36",
            "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
            "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
        }
        if headers:
            default_headers.update(headers)

        for attempt in range(self.max_retries + 1):
            # 获取代理IP
            proxy_ip = self.proxy_manager.get_proxy()
            if not proxy_ip:
                print(f"[RetryEngine] 无可用代理IP，等待重试...")
                time.sleep(self.base_delay)
                continue

            proxy_url = f"http://{proxy_ip.ip}:{proxy_ip.port}"
            proxies = {"http": proxy_url, "https": proxy_url}

            try:
                response = requests.request(
                    method=method,
                    url=url,
                    headers=default_headers,
                    data=data,
                    proxies=proxies,
                    timeout=timeout,
                    allow_redirects=True,
                )

                # 检测验证码
                if CaptchaDetector.detect(response):
                    print(f"[RetryEngine] 检测到验证码 "
                          f"(IP: {proxy_ip.ip}, 尝试: {attempt+1}/{self.max_retries+1})")

                    # 标记IP失败
                    self.proxy_manager.mark_fail(proxy_ip.ip, "captcha")

                    # 释放当前IP
                    self.proxy_manager.release_ip(proxy_ip.ip)

                    # 回调通知
                    if on_captcha:
                        on_captcha(url, proxy_ip.ip, attempt)

                    # 指数退避 + 随机抖动
                    delay = min(
                        self.base_delay * (self.backoff_factor ** attempt),
                        self.max_delay
                    )
                    delay += random.uniform(0, delay * 0.1)
                    time.sleep(delay)
                    continue

                # 请求成功
                self.proxy_manager.mark_success(proxy_ip.ip)
                return response

            except requests.exceptions.Timeout:
                print(f"[RetryEngine] 请求超时 (IP: {proxy_ip.ip})")
                self.proxy_manager.mark_fail(proxy_ip.ip, "timeout")
                time.sleep(self.base_delay)

            except requests.exceptions.ConnectionError:
                print(f"[RetryEngine] 连接失败 (IP: {proxy_ip.ip})")
                self.proxy_manager.mark_fail(proxy_ip.ip, "connection")
                time.sleep(self.base_delay)

            except Exception as e:
                print(f"[RetryEngine] 请求异常: {e} (IP: {proxy_ip.ip})")
                self.proxy_manager.mark_fail(proxy_ip.ip, "error")
                time.sleep(self.base_delay)

        print(f"[RetryEngine] 达到最大重试次数: {url}")
        return None

"""
分布式爬虫Worker节点
集成IP轮拨 + 验证码检测 + 自动重试
"""
import time
import queue
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed


class CrawlerWorker:
    """爬虫Worker节点"""

    def __init__(
        self,
        proxy_manager: fanProxyManager,
        task_queue: queue.Queue,
        result_queue: queue.Queue,
        concurrency: int = 20,
        request_interval: float = 0.5,
    ):
        self.proxy_manager = proxy_manager
        self.task_queue = task_queue
        self.result_queue = result_queue
        self.concurrency = concurrency
        self.request_interval = request_interval
        self.retry_engine = RetryEngine(proxy_manager, max_retries=3)
        self._running = False

    def crawl_single(self, url: str) -> dict:
        """采集单个URL"""
        start_time = time.time()

        response = self.retry_engine.request_with_retry(
            url=url,
            timeout=15,
            on_captcha=self._on_captcha_detected,
        )

        result = {
            "url": url,
            "success": response is not None,
            "status_code": response.status_code if response else None,
            "content_length": len(response.content) if response else 0,
            "elapsed_ms": int((time.time() - start_time) * 1000),
            "timestamp": time.time(),
        }

        if response and response.status_code == 200:
            result["content"] = response.text
            result["encoding"] = response.encoding

        return result

    def _on_captcha_detected(self, url: str, ip: str, attempt: int):
        """验证码检测回调"""
        print(f"[Worker] 验证码触发 → URL: {url[:50]}... "
              f"IP: {ip} 尝试: {attempt+1}")

    def run(self):
        """启动Worker"""
        self._running = True
        print(f"[Worker] 启动，并发数: {self.concurrency}")

        with ThreadPoolExecutor(max_workers=self.concurrency) as executor:
            futures = {}

            while self._running or not self.task_queue.empty():
                try:
                    url = self.task_queue.get(timeout=5)
                    future = executor.submit(self.crawl_single, url)
                    futures[future] = url

                    # 控制请求间隔
                    time.sleep(self.request_interval)

                except queue.Empty:
                    continue

                # 处理已完成的任务
                done = [f for f in futures if f.done()]
                for future in done:
                    url = futures.pop(future)
                    try:
                        result = future.result()
                        self.result_queue.put(result)
                    except Exception as e:
                        print(f"[Worker] 任务异常: {url} - {e}")

        print(f"[Worker] 停止，统计: {self.proxy_manager.get_stats()}")


# ==================== 启动示例 ====================

if __name__ == "__main__":
    # 初始化IP管理器（混合轮拨策略）
    proxy_manager = fanProxyManager(
        api_key="your_api_key",
        strategy=RotationStrategy.HYBRID,
        rotation_interval=300,      # 5分钟轮拨
        max_fail_count=3,           # 失败3次剔除
        max_request_per_ip=50,      # 单IP最多50次请求
    )

    # 创建任务队列和结果队列
    task_queue = queue.Queue()
    result_queue = queue.Queue()

    # 填充任务（示例URL）
    seed_urls = [
        "https://example.com/page/1",
        "https://example.com/page/2",
        "https://example.com/page/3",
        # ... 更多URL
    ]
    for url in seed_urls:
        task_queue.put(url)

    # 启动Worker
    worker = CrawlerWorker(
        proxy_manager=proxy_manager,
        task_queue=task_queue,
        result_queue=result_queue,
        concurrency=20,
        request_interval=0.5,
    )

    # 启动采集
    worker_thread = threading.Thread(target=worker.run)
    worker_thread.start()
    worker_thread.join()

    # 输出结果统计
    success_count = sum(1 for _ in iter(lambda: result_queue.get_nowait(), None)
                        if result_queue.qsize() > 0)
    print(f"\n采集完成，成功: {success_count} 条")
    print(f"IP管理统计: {proxy_manager.get_stats()}")