Реалізація ротації проксі для веб-скрейпінгу
Ротація проксі — обов'язковий компонент промислового веб-скрейпінгу. Однієї 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")
# зважений вибір за успішністю
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 робочі дні.