Проксі-ротація для парсингу: пул, моніторинг, обхід блокувань

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

При промисловому парсингу одна IP-адреса швидко блокується: сайти фіксують високу частоту запитів і видають 403 або 429. Втрати даних при цьому сягають 40%. Ручна зміна проксі неефективна — потрібне автоматичне рішення. Ми будуємо систему, яка розподіляє запити по пулу адрес, виводить заблоковані на карантин і зважено обирає найнадійніші проксі. Досвід показує: грамотна ротація підвищує збір даних до 99,5%.

Як працює зважений вибір проксі?

Кожен проксі в пулі зберігає статистику успішних і невдалих запитів. Ротатор обчислює success_rate як відношення успішних запитів до загальної кількості. При виборі адреси використовується зважена випадковість: проксі з коефіцієнтом 0.92 обирається в 7 разів частіше, ніж з 0.5. Якщо проксі повертає помилку (403, 429, таймаут), він поміщається на карантин на 15 хвилин (налаштовується). Це запобігає повторним спробам через свідомо заблокований IP.

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

Як інтегрувати ротатор із aiohttp?

Інтеграція будується на asyncio та aiohttp. Функція fetch отримує проксі з ротатора, виконує запит і звітує про успіх. При помилці проксі йде на карантин, і наступний запит отримає іншу адресу. Це дозволяє обробляти до 1000 запитів на хвилину без блокувань.

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 

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 

Чому зважений вибір кращий за випадковий?

Випадковий вибір проксі перевантажує систему помилками: погані адреси отримують стільки ж запитів, скільки хороші. Зважений підхід, заснований на success_rate, аналогічний методу експоненційного згладжування — чим довше проксі працює стабільно, тим частіше він використовується. Це знижує кількість повторних запитів на 30% — зважений вибір у 1.5 раза швидший за випадковий. Додатково ми впроваджуємо моніторинг проксі: кожні 5 хвилин перевіряємо доступність пулу та автоматично виключаємо довго заблоковані адреси.

Порівняння пулів: резидентні vs ISP vs дата-центр

Резидентні проксі (наприклад, від Bright Data) дають найкращий обхід блокувань, але дорожчі — економія бюджету досягається за рахунок меншої кількості повторних запитів (до 40% економії). ISP-проксі (Oxylabs) — золота середина для парсингу інтернет-магазинів. Дата-центр підходить лише для швидких прототипів, де допустимі тимчасові блокування.

Що робити, якщо всі проксі в карантині?

Алгоритм перевіряє blocked_until для кожної адреси. Якщо всі недоступні, викидається виняток — це сигнал для оператора. На практиці ми рекомендуємо резервний пул із 5–10 проксі зі збільшеним часом очікування між запитами. При пулі з 20 резидентних проксі з налаштуваннями за замовчуванням парсер працює без простоїв у 99% випадків. У складних сценаріях автоматично збільшуємо quarantine_minutes до 30, щоб знизити навантаження на пул.

Етапи впровадження системи ротації

1. Аналіз цільових сайтів: визначення таргетингу, лімітів запитів, типів блокувань.
2. Вибір проксі-провайдера: підбір оптимального пулу (резидентні/ISP/мікс) під бюджет і завдання.
3. Розробка ротатора: реалізація зваженого вибору, карантину та моніторингу.
4. Інтеграція з парсером: підключення aiohttp, обробка помилок, логування.
5. Навантажувальне тестування: прогін 10 000 запитів, перевірка стабільності при пікових навантаженнях.

Вибір проксі-провайдера: порівняння за цільовими завданнями

Для більшості завдань достатньо 10–15 резидентних проксі від Smartproxy при адекватних затримках. Якщо потрібне глобальне покриття, обирайте Bright Data — їхній пул включає 72 млн IP.

Що входить у результат

• Архітектура системи ротації з документацією
• Вихідний код на Python із asyncio та інтеграцією з aiohttp
• Інструкція з розгортання (Docker, налаштування моніторингу)
• Двотижнева підтримка після впровадження

Ми займаємося розробкою систем парсингу більше 5 років (5+ років досвіду), реалізували 15+ проєктів із проксі-ротацією. Всі рішення проходять навантажувальне тестування та супроводжуються гарантією стабільності. Отримайте консультацію — надішлемо оцінку за 1 день. Замовте реалізацію — впровадимо систему за 2–3 дні.

Строки

Базова система — 2–3 робочі дні. Під ключ з адаптацією під ваш стек — до 5 днів.