Реализация прокси-ротации для парсинга

Реализация прокси-ротации для парсинга

Прокси-ротация — обязательный компонент промышленного парсинга. Один IP-адрес неизбежно попадёт в блок при высокой частоте запросов. Задача системы ротации — автоматически распределять запросы по пулу адресов и выводить из ротации заблокированные.

Архитектура ротатора

Request → Proxy Selector
              ↓
         [Proxy Pool]
         192.168.1.1:8080  ← статус: OK, 245 запросов
         10.0.0.5:3128     ← статус: OK, 198 запросов
         172.16.0.2:8080   ← статус: BLOCKED, на карантине
              ↓
         Target Site
              ↓
         Response Checker
         (статус 200 → success / 403/429 → fail)

Реализация на Python

import asyncio
import aiohttp
from dataclasses import dataclass, field
from datetime import datetime, timedelta

@dataclass
class ProxyEntry:
    url: str
    success: int = 0
    fail: int = 0
    blocked_until: datetime = field(default_factory=lambda: datetime.min)

    @property
    def is_available(self):
        return datetime.now() > self.blocked_until

    @property
    def success_rate(self):
        total = self.success + self.fail
        return self.success / total if total > 0 else 0.5

class ProxyRotator:
    def __init__(self, proxies: list[str], quarantine_minutes=15):
        self.pool = [ProxyEntry(url=p) for p in proxies]
        self.quarantine = timedelta(minutes=quarantine_minutes)
        self._lock = asyncio.Lock()

    async def get(self) -> ProxyEntry:
        async with self._lock:
            available = [p for p in self.pool if p.is_available]
            if not available:
                raise RuntimeError("All proxies are blocked")
            # взвешенный выбор по success rate
            weights = [p.success_rate for p in available]
            return random.choices(available, weights=weights)[0]

    async def report(self, proxy: ProxyEntry, success: bool):
        async with self._lock:
            if success:
                proxy.success += 1
            else:
                proxy.fail += 1
                proxy.blocked_until = datetime.now() + self.quarantine

Интеграция с aiohttp

async def fetch(session, url, rotator):
    proxy = await rotator.get()
    try:
        async with session.get(url, proxy=proxy.url, timeout=10) as resp:
            if resp.status in (403, 429, 503):
                await rotator.report(proxy, success=False)
                return None
            await rotator.report(proxy, success=True)
            return await resp.text()
    except Exception:
        await rotator.report(proxy, success=False)
        return None

Источники прокси

Платные провайдеры (рекомендуется для серьёзных задач):

• Bright Data — крупнейший пул резидентных прокси, 72M+ IP
• Oxylabs — хорошее покрытие по СНГ, ISP-прокси
• Smartproxy — мобильные и резидентные, разумные цены

Бесплатные списки — нестабильны, не подходят для промышленного парсинга.

Сроки

Система прокси-ротации с мониторингом: 2–3 рабочих дня.