Ручная смена прокси при работе с сотнями запросов — это потеря времени и денег. API-ротация позволяет автоматически переключать IP-адреса при блокировках, распределять нагрузку и масштабировать парсинг или мультиаккаунтинг. В этом руководстве разберём, как настроить автоматическую смену прокси для разных задач: от парсинга маркетплейсов до фарма аккаунтов в Facebook Ads.
Материал подойдёт как разработчикам, которые пишут парсеры на Python или Node.js, так и арбитражникам, использующим готовые инструменты с API-интеграцией.
Зачем нужна автоматизация смены прокси через API
Автоматическая ротация IP-адресов через API решает несколько критических задач, с которыми сталкиваются специалисты в разных областях:
Для парсинга маркетплейсов и сайтов: При сборе данных с Wildberries, Ozon или Авито каждый IP может сделать ограниченное количество запросов (обычно 50-200 в час). API-ротация позволяет автоматически переключаться на новый IP при достижении лимита или получении капчи, обеспечивая непрерывный сбор данных.
Для арбитража и мультиаккаунтинга: При работе с 20-50 рекламными кабинетами Facebook Ads или аккаунтами Instagram нужно изолировать каждый профиль. API позволяет программно назначать уникальный прокси каждому аккаунту в Dolphin Anty или AdsPower, автоматически пересоздавать сессии при блокировках.
Для автоматизации SMM: Сервисы массового постинга в Instagram, TikTok или VK должны распределять действия между IP-адресами, чтобы избежать rate limits. API даёт возможность динамически получать новые прокси для каждой сессии или группы аккаунтов.
Основные преимущества API-автоматизации по сравнению с ручной сменой:
- Скорость: Смена IP происходит за миллисекунды программно, без участия человека
- Масштабируемость: Можно управлять тысячами прокси одновременно через единый интерфейс
- Отказоустойчивость: Автоматическая замена нерабочих прокси без остановки процесса
- Гибкость: Настройка правил ротации под конкретную задачу: по времени, по количеству запросов, по географии
- Экономия: Оптимальное использование трафика через балансировку нагрузки
Типичный сценарий использования: вы парсите цены конкурентов на Wildberries. Без API вам нужно вручную отслеживать блокировки, заходить в панель прокси-провайдера, копировать новые данные, вставлять их в скрипт. С API всё это происходит автоматически: скрипт получает ошибку 429 (Too Many Requests), отправляет запрос к API прокси-сервиса, получает новый IP и продолжает работу.
Типы ротации прокси: sticky sessions vs автоматическая смена
Прежде чем настраивать автоматизацию, важно понимать разницу между типами ротации IP-адресов. Выбор стратегии зависит от вашей задачи.
Sticky Sessions (сессионные прокси)
При использовании sticky sessions один IP-адрес закрепляется за вашей сессией на определённое время (обычно от 5 до 30 минут). Смена происходит только после истечения времени сессии или по вашему API-запросу.
Когда использовать:
- Работа с аккаунтами в соцсетях (Instagram, Facebook) — частая смена IP вызывает подозрения
- Заполнение многостраничных форм, где нужно сохранять сессию
- Тестирование рекламы из определённого региона в течение сессии
- Парсинг сайтов с авторизацией, где смена IP приведёт к разлогину
Пример API-запроса для создания sticky session (обычно используется специальный формат логина):
// Формат: username-session-SESSIONID:password
// SESSIONID — любая строка, одинаковая = один IP
proxy = "username-session-abc123:password@gate.proxycove.com:8000"
// Все запросы с session-abc123 получат один IP на время сессии
// Для нового IP используйте другой SESSIONID: session-xyz789
Автоматическая ротация на каждый запрос
IP-адрес меняется при каждом новом подключении к прокси-серверу. Это стандартное поведение резидентных прокси без указания параметров сессии.
Когда использовать:
- Массовый парсинг без авторизации (цены, контакты, объявления)
- Обход агрессивных rate limits на публичных API
- Сбор данных с сайтов, которые банят IP после 10-20 запросов
- Проверка доступности контента из разных регионов
Пример использования в Python (каждый запрос = новый IP):
import requests
proxy = {
"http": "http://username:password@gate.proxycove.com:8000",
"https": "http://username:password@gate.proxycove.com:8000"
}
# Каждый запрос получит новый IP
for i in range(10):
response = requests.get("https://api.ipify.org", proxies=proxy)
print(f"Запрос {i+1}, IP: {response.text}")
Ротация по таймеру
Вы программно контролируете, когда менять IP: каждые N минут, после M запросов или при получении определённых ошибок. Это гибридный подход, который реализуется через API прокси-сервиса.
Когда использовать:
- Оптимизация расхода трафика — смена только при необходимости
- Работа с сайтами, которые отслеживают паттерны (слишком частая смена = бан)
- Балансировка между анонимностью и стабильностью сессии
| Тип ротации | Частота смены IP | Задачи | Расход трафика |
|---|---|---|---|
| Sticky Session | 5-30 минут | Мультиаккаунтинг, авторизация | Низкий |
| Авто-ротация | Каждый запрос | Парсинг, обход rate limits | Средний |
| По таймеру | Настраиваемая | Универсальная | Оптимизированный |
Основы работы с API прокси-сервисов
Большинство современных прокси-провайдеров предоставляют два способа управления: через веб-панель и через API. API даёт программный доступ к функциям: получение списка прокси, создание новых сессий, проверка баланса, статистика использования.
Типовые API-методы прокси-сервисов
Хотя каждый провайдер имеет свою документацию, стандартные методы обычно включают:
- GET /api/v1/proxy/list — получить список доступных прокси с фильтрацией по стране, типу
- POST /api/v1/proxy/rotate — принудительно сменить IP в активной сессии
- GET /api/v1/account/balance — проверить остаток трафика или денег на балансе
- GET /api/v1/stats — статистика использования: объём трафика, количество запросов, ошибки
- POST /api/v1/session/create — создать новую sticky session с параметрами (страна, город, длительность)
Аутентификация обычно происходит через API-ключ в заголовке запроса:
curl -X GET "https://api.provider.com/v1/proxy/list?country=US" \
-H "Authorization: Bearer YOUR_API_KEY" \
-H "Content-Type: application/json"
Ответ обычно приходит в JSON-формате:
{
"status": "success",
"data": {
"proxies": [
{
"ip": "123.45.67.89",
"port": 8000,
"country": "US",
"city": "New York",
"protocol": "http",
"username": "user123",
"password": "pass456"
}
],
"total": 150,
"available": 147
}
}
Управление сессиями через API
Для задач, требующих контроля над временем жизни IP (мультиаккаунтинг, работа с аккаунтами), используется создание именованных сессий через API. Это позволяет программно управлять десятками и сотнями изолированных IP-адресов.
Пример создания сессии с параметрами:
POST /api/v1/session/create
{
"country": "US",
"state": "California",
"session_duration": 600, // 10 минут
"session_id": "facebook_account_001"
}
// Ответ:
{
"status": "success",
"session": {
"id": "facebook_account_001",
"proxy": "gate.provider.com:8000",
"username": "user-session-facebook_account_001",
"password": "your_password",
"ip": "45.67.89.123",
"expires_at": "2024-01-15T15:30:00Z"
}
}
Теперь вы можете использовать этот прокси в своём скрипте или антидетект-браузере, и IP останется неизменным в течение 10 минут. Для продления сессии отправляется повторный запрос с тем же session_id.
Примеры автоматизации на Python: requests, Selenium, Scrapy
Python — самый популярный язык для парсинга и автоматизации. Рассмотрим примеры интеграции API-ротации прокси с основными инструментами.
Автоматическая смена прокси в requests
Библиотека requests используется для простых HTTP-запросов. Для автоматической ротации создадим класс-обёртку, который меняет прокси при ошибках:
import requests
import random
import time
class RotatingProxySession:
def __init__(self, proxy_list):
"""
proxy_list: список словарей с данными прокси
[{"http": "http://user:pass@ip:port", "https": "..."}]
"""
self.proxy_list = proxy_list
self.current_proxy = None
self.session = requests.Session()
self.rotate()
def rotate(self):
"""Выбрать случайный прокси из списка"""
self.current_proxy = random.choice(self.proxy_list)
self.session.proxies.update(self.current_proxy)
print(f"Переключились на прокси: {self.current_proxy['http']}")
def get(self, url, max_retries=3, **kwargs):
"""GET-запрос с автоматической ротацией при ошибках"""
for attempt in range(max_retries):
try:
response = self.session.get(url, timeout=10, **kwargs)
# Если получили блокировку — сменить прокси
if response.status_code in [403, 429, 503]:
print(f"Получили {response.status_code}, меняем прокси...")
self.rotate()
time.sleep(2)
continue
return response
except requests.exceptions.ProxyError:
print(f"Прокси не работает, попытка {attempt+1}/{max_retries}")
self.rotate()
time.sleep(2)
except requests.exceptions.Timeout:
print("Таймаут, меняем прокси...")
self.rotate()
time.sleep(2)
raise Exception(f"Не удалось выполнить запрос после {max_retries} попыток")
# Использование:
proxies = [
{"http": "http://user1:pass@gate1.com:8000", "https": "http://user1:pass@gate1.com:8000"},
{"http": "http://user2:pass@gate2.com:8000", "https": "http://user2:pass@gate2.com:8000"},
]
session = RotatingProxySession(proxies)
# Парсим Wildberries
for page in range(1, 50):
url = f"https://www.wildberries.ru/catalog/page={page}"
response = session.get(url)
print(f"Страница {page}: {response.status_code}")
Интеграция с Selenium для браузерной автоматизации
Selenium используется для парсинга сайтов с JavaScript и автоматизации действий в браузере. Для смены прокси нужно пересоздавать драйвер, так как настройки прокси задаются при инициализации:
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
import time
class SeleniumRotatingProxy:
def __init__(self, proxy_list):
self.proxy_list = proxy_list
self.driver = None
self.current_proxy_index = 0
def create_driver(self):
"""Создать новый драйвер с текущим прокси"""
if self.driver:
self.driver.quit()
proxy = self.proxy_list[self.current_proxy_index]
chrome_options = Options()
chrome_options.add_argument(f'--proxy-server={proxy}')
chrome_options.add_argument('--headless') # без GUI
self.driver = webdriver.Chrome(options=chrome_options)
print(f"Создан драйвер с прокси: {proxy}")
def rotate(self):
"""Переключиться на следующий прокси"""
self.current_proxy_index = (self.current_proxy_index + 1) % len(self.proxy_list)
self.create_driver()
def get_with_retry(self, url, max_retries=3):
"""Открыть URL с автоматической сменой прокси при ошибках"""
for attempt in range(max_retries):
try:
if not self.driver:
self.create_driver()
self.driver.get(url)
# Проверка на блокировку (например, поиск капчи)
if "captcha" in self.driver.page_source.lower():
print("Обнаружена капча, меняем прокси...")
self.rotate()
time.sleep(3)
continue
return self.driver.page_source
except Exception as e:
print(f"Ошибка: {e}, меняем прокси (попытка {attempt+1})")
self.rotate()
time.sleep(3)
raise Exception("Не удалось загрузить страницу")
# Использование:
proxies = [
"http://user:pass@gate1.com:8000",
"http://user:pass@gate2.com:8000",
]
bot = SeleniumRotatingProxy(proxies)
# Парсим Ozon
for i in range(10):
html = bot.get_with_retry(f"https://www.ozon.ru/category/page-{i}")
print(f"Получили HTML страницы {i}, длина: {len(html)}")
bot.driver.quit()
Scrapy с middleware для ротации прокси
Scrapy — фреймворк для масштабного парсинга. Ротация прокси реализуется через middleware, который автоматически применяется ко всем запросам:
# middlewares.py
import random
from scrapy.exceptions import IgnoreRequest
class RotatingProxyMiddleware:
def __init__(self, proxy_list):
self.proxy_list = proxy_list
@classmethod
def from_crawler(cls, crawler):
# Получить список прокси из настроек
proxy_list = crawler.settings.getlist('ROTATING_PROXY_LIST')
return cls(proxy_list)
def process_request(self, request, spider):
# Назначить случайный прокси каждому запросу
proxy = random.choice(self.proxy_list)
request.meta['proxy'] = proxy
spider.logger.info(f'Используем прокси: {proxy}')
def process_exception(self, request, exception, spider):
# При ошибке прокси — повторить с другим
proxy = random.choice(self.proxy_list)
spider.logger.warning(f'Ошибка прокси, переключаемся на: {proxy}')
request.meta['proxy'] = proxy
return request # повторить запрос
# settings.py
DOWNLOADER_MIDDLEWARES = {
'myproject.middlewares.RotatingProxyMiddleware': 350,
}
ROTATING_PROXY_LIST = [
'http://user:pass@gate1.com:8000',
'http://user:pass@gate2.com:8000',
'http://user:pass@gate3.com:8000',
]
# Повторять запросы при ошибках
RETRY_TIMES = 5
RETRY_HTTP_CODES = [403, 429, 500, 502, 503]
Теперь каждый запрос Scrapy автоматически получит случайный прокси из списка, а при ошибках будет повторяться с другим IP.
Автоматизация на Node.js: axios, Puppeteer, Playwright
Node.js популярен для создания парсеров и ботов благодаря асинхронности и хорошей интеграции с браузерными инструментами. Рассмотрим примеры ротации прокси в основных библиотеках.
Axios с автоматической ротацией
Axios — библиотека для HTTP-запросов. Создадим класс с пулом прокси и автоматической заменой при ошибках:
const axios = require('axios');
const HttpsProxyAgent = require('https-proxy-agent');
class RotatingProxyClient {
constructor(proxyList) {
this.proxyList = proxyList;
this.currentIndex = 0;
}
getProxy() {
const proxy = this.proxyList[this.currentIndex];
this.currentIndex = (this.currentIndex + 1) % this.proxyList.length;
return proxy;
}
async request(url, options = {}, maxRetries = 3) {
for (let i = 0; i < maxRetries; i++) {
const proxy = this.getProxy();
const agent = new HttpsProxyAgent(proxy);
try {
const response = await axios.get(url, {
...options,
httpsAgent: agent,
timeout: 10000
});
// Если получили блокировку — следующая попытка
if ([403, 429, 503].includes(response.status)) {
console.log(`Статус ${response.status}, меняем прокси...`);
continue;
}
return response.data;
} catch (error) {
console.log(`Ошибка с прокси ${proxy}: ${error.message}`);
if (i === maxRetries - 1) throw error;
}
}
}
}
// Использование:
const proxies = [
'http://user:pass@gate1.com:8000',
'http://user:pass@gate2.com:8000',
];
const client = new RotatingProxyClient(proxies);
(async () => {
for (let page = 1; page <= 20; page++) {
const data = await client.request(`https://api.example.com/products?page=${page}`);
console.log(`Страница ${page}: получено ${data.length} товаров`);
}
})();
Puppeteer с ротацией прокси
Puppeteer управляет браузером Chrome. Прокси задаётся при запуске браузера, поэтому для смены нужно пересоздавать инстанс:
const puppeteer = require('puppeteer');
class PuppeteerRotatingProxy {
constructor(proxyList) {
this.proxyList = proxyList;
this.currentIndex = 0;
this.browser = null;
}
async createBrowser() {
if (this.browser) await this.browser.close();
const proxy = this.proxyList[this.currentIndex];
console.log(`Запускаем браузер с прокси: ${proxy}`);
this.browser = await puppeteer.launch({
headless: true,
args: [`--proxy-server=${proxy}`]
});
}
rotate() {
this.currentIndex = (this.currentIndex + 1) % this.proxyList.length;
}
async scrape(url, maxRetries = 3) {
for (let i = 0; i < maxRetries; i++) {
try {
if (!this.browser) await this.createBrowser();
const page = await this.browser.newPage();
// Авторизация прокси (если требуется)
await page.authenticate({
username: 'your_username',
password: 'your_password'
});
await page.goto(url, { waitUntil: 'networkidle2', timeout: 30000 });
// Проверка на капчу
const content = await page.content();
if (content.includes('captcha')) {
console.log('Обнаружена капча, меняем прокси...');
this.rotate();
await this.createBrowser();
continue;
}
return content;
} catch (error) {
console.log(`Ошибка: ${error.message}, попытка ${i+1}`);
this.rotate();
await this.createBrowser();
}
}
throw new Error('Не удалось загрузить страницу');
}
}
// Использование:
const proxies = ['gate1.com:8000', 'gate2.com:8000'];
const scraper = new PuppeteerRotatingProxy(proxies);
(async () => {
const html = await scraper.scrape('https://www.avito.ru/moskva');
console.log(`Получили HTML длиной: ${html.length}`);
await scraper.browser.close();
})();
Playwright с поддержкой ротации
Playwright — современная альтернатива Puppeteer с лучшей производительностью. Настройка прокси аналогична:
const { chromium } = require('playwright');
async function scrapeWithRotation(urls, proxyList) {
let proxyIndex = 0;
for (const url of urls) {
const proxy = proxyList[proxyIndex];
const browser = await chromium.launch({
headless: true,
proxy: {
server: proxy,
username: 'your_user',
password: 'your_pass'
}
});
const page = await browser.newPage();
try {
await page.goto(url, { timeout: 30000 });
const title = await page.title();
console.log(`${url} → ${title} (прокси: ${proxy})`);
} catch (error) {
console.log(`Ошибка на ${url}: ${error.message}`);
}
await browser.close();
// Следующий прокси для следующего URL
proxyIndex = (proxyIndex + 1) % proxyList.length;
}
}
const urls = [
'https://www.wildberries.ru',
'https://www.ozon.ru',
'https://www.avito.ru'
];
const proxies = [
'http://gate1.com:8000',
'http://gate2.com:8000'
];
scrapeWithRotation(urls, proxies);
Интеграция API с антидетект-браузерами: Dolphin Anty, AdsPower
Для арбитражников и SMM-специалистов, работающих с мультиаккаунтингом, ручное назначение прокси каждому профилю в Dolphin Anty или AdsPower занимает часы. API этих браузеров позволяет автоматизировать создание профилей и привязку прокси.
Автоматизация Dolphin Anty через API
Dolphin Anty предоставляет локальный API (обычно на http://localhost:3001/v1.0), через который можно создавать профили, назначать прокси, запускать браузеры программно.
Пример скрипта на Python для массового создания профилей с уникальными прокси:
import requests
import json
DOLPHIN_API = "http://localhost:3001/v1.0"
API_TOKEN = "your_dolphin_api_token"
# Список прокси из вашего провайдера (получены через их API)
proxies = [
{"host": "gate1.com", "port": 8000, "login": "user1", "password": "pass1"},
{"host": "gate2.com", "port": 8000, "login": "user2", "password": "pass2"},
]
def create_profile_with_proxy(name, proxy):
"""Создать профиль в Dolphin с привязкой прокси"""
payload = {
"name": name,
"tags": ["Facebook Ads", "Auto-created"],
"proxy": {
"type": "http", # или socks5
"host": proxy["host"],
"port": proxy["port"],
"login": proxy["login"],
"password": proxy["password"]
},
"fingerprint": {
"os": "win",
"webRTC": {
"mode": "altered",
"fillBasedOnIp": True
},
"canvas": {
"mode": "noise"
}
}
}
headers = {
"Authorization": f"Bearer {API_TOKEN}",
"Content-Type": "application/json"
}
response = requests.post(
f"{DOLPHIN_API}/browser_profiles",
headers=headers,
data=json.dumps(payload)
)
if response.status_code == 200:
profile = response.json()
print(f"✓ Создан профиль: {name}, ID: {profile['id']}")
return profile['id']
else:
print(f"✗ Ошибка создания {name}: {response.text}")
return None
# Создать 50 профилей с ротацией прокси
for i in range(50):
proxy = proxies[i % len(proxies)] # циклическая ротация
profile_name = f"FB_Account_{i+1:03d}"
create_profile_with_proxy(profile_name, proxy)
Теперь у вас 50 профилей в Dolphin Anty, каждый с уникальным отпечатком браузера и прокси. Для запуска профиля программно:
def start_profile(profile_id):
"""Запустить профиль браузера"""
response = requests.get(
f"{DOLPHIN_API}/browser_profiles/{profile_id}/start",
headers={"Authorization": f"Bearer {API_TOKEN}"}
)
if response.status_code == 200:
data = response.json()
print(f"Профиль запущен, порт WebDriver: {data['automation']['port']}")
return data['automation']['port']
else:
print(f"Ошибка запуска: {response.text}")
# Запустить профиль и управлять через Selenium
port = start_profile("profile_id_here")
from selenium import webdriver
driver = webdriver.Remote(
command_executor=f'http://127.0.0.1:{port}',
options=webdriver.ChromeOptions()
)
driver.get("https://facebook.com")
Автоматизация AdsPower
AdsPower также предоставляет локальный API. Логика аналогична Dolphin, но с другими эндпоинтами:
import requests
ADSPOWER_API = "http://local.adspower.net:50325/api/v1"
def create_adspower_profile(name, proxy):
payload = {
"name": name,
"group_id": "0", # ID группы профилей
"domain_name": "facebook.com",
"open_urls": ["https://facebook.com"],
"repeat_config": ["0"],
"username": proxy["login"],
"password": proxy["password"],
"proxy_type": "http",
"proxy_host": proxy["host"],
"proxy_port": proxy["port"],
"proxy_user": proxy["login"],
"proxy_password": proxy["password"]
}
response = requests.post(
f"{ADSPOWER_API}/user/create",
json=payload
)
if response.json()["code"] == 0:
user_id = response.json()["data"]["id"]
print(f"✓ Создан профиль AdsPower: {name}, ID: {user_id}")
return user_id
else:
print(f"✗ Ошибка: {response.json()['msg']}")
# Создание профилей
for i, proxy in enumerate(proxies):
create_adspower_profile(f"TikTok_Account_{i+1}", proxy)
Такая автоматизация критически важна при работе с десятками аккаунтов. Вместо ручного копирования данных прокси в каждый профиль, вы запускаете скрипт и получаете готовую инфраструктуру за минуты.
Обработка ошибок и автоматический фоллбэк
При работе с прокси неизбежны ситуации: IP заблокирован целевым сайтом, прокси-сервер не отвечает, закончился трафик. Правильная обработка ошибок — ключ к стабильной автоматизации.
Типы ошибок и стратегии обработки
| Ошибка | Причина | Решение |
|---|---|---|
| HTTP 403 Forbidden | IP в бан-листе сайта | Сменить прокси, добавить задержку |
| HTTP 429 Too Many Requests | Превышен rate limit | Сменить IP, увеличить интервал |
| ProxyError / Timeout | Прокси-сервер не отвечает | Убрать из пула, взять следующий |
| 407 Proxy Authentication Required | Неверные логин/пароль | Проверить credentials, обновить |
| Капча на странице | Сайт детектировал бота | Сменить IP, использовать мобильные прокси |
Реализация умной системы retry
Вместо простого повтора запроса, создадим систему с экспоненциальной задержкой и чёрным списком нерабочих прокси:
import requests
import time
from collections import defaultdict
class SmartProxyRotator:
def __init__(self, proxy_list):
self.proxy_list = proxy_list
self.blacklist = set() # IP, которые не работают
self.error_count = defaultdict(int) # счётчик ошибок по IP
self.max_errors = 3 # после 3 ошибок — в чёрный список
def get_working_proxy(self):
"""Получить прокси, который не в чёрном списке"""
available = [p for p in self.proxy_list if p not in self.blacklist]
if not available:
# Все прокси в бане — очистить чёрный список
print("⚠ Все прокси заблокированы, сбрасываем чёрный список")
self.blacklist.clear()
self.error_count.clear()
available = self.proxy_list
return available[0]
def mark_error(self, proxy):
"""Отметить ошибку прокси"""
self.error_count[proxy] += 1
if self.error_count[proxy] >= self.max_errors:
self.blacklist.add(proxy)
print(f"✗ Прокси {proxy} добавлен в чёрный список")
def request_with_retry(self, url, max_retries=5):
"""Запрос с умными повторами"""
for attempt in range(max_retries):
proxy = self.get_working_proxy()
try:
# Экспоненциальная задержка: 1s, 2s, 4s, 8s...
if attempt > 0:
delay = 2 ** attempt
print(f"Ожидание {delay}с перед попыткой {attempt+1}")
time.sleep(delay)
response = requests.get(
url,
proxies={"http": proxy, "https": proxy},
timeout=15
)
# Успех — сбросить счётчик ошибок
if response.status_code == 200:
self.error_count[proxy] = 0
return response
# Блокировка — сменить прокси
elif response.status_code in [403, 429]:
print(f"Статус {response.status_code}, меняем прокси")
self.mark_error(proxy)
continue
except requests.exceptions.ProxyError:
print(f"ProxyError с {proxy}")
self.mark_error(proxy)
except requests.exceptions.Timeout:
print(f"Timeout с {proxy}")
self.mark_error(proxy)
raise Exception(f"Не удалось выполнить запрос после {max_retries} попыток")
# Использование:
proxies = [
"http://user:pass@gate1.com:8000",
"http://user:pass@gate2.com:8000",
"http://user:pass@gate3.com:8000",
]
rotator = SmartProxyRotator(proxies)
for i in range(100):
try:
response = rotator.request_with_retry(f"https://api.example.com/data?page={i}")
print(f"✓ Страница {i}: {len(response.text)} байт")
except Exception as e:
print(f"✗ Критическая ошибка на странице {i}: {e}")
Мониторинг и алерты
Для production-систем важно отслеживать здоровье прокси-пула в реальном времени. Добавьте логирование метрик:
import logging
from datetime import datetime
class ProxyMonitor:
def __init__(self):
self.stats = {
"total_requests": 0,
"successful": 0,
"failed": 0,
"proxy_errors": defaultdict(int),
"start_time": datetime.now()
}
# Настройка логирования
logging.basicConfig(
filename='proxy_rotation.log',
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
def log_request(self, proxy, success, error=None):
self.stats["total_requests"] += 1
if success:
self.stats["successful"] += 1
logging.info(f"✓ Успех с {proxy}")
else:
self.stats["failed"] += 1
self.stats["proxy_errors"][proxy] += 1
logging.error(f"✗ Ошибка с {proxy}: {error}")
def get_report(self):
uptime = datetime.now() - self.stats["start_time"]
success_rate = (self.stats["successful"] / self.stats["total_requests"] * 100) if self.stats["total_requests"] > 0 else 0
return f"""
=== Отчёт по прокси-ротации ===
Время работы: {uptime}
Всего запросов: {self.stats["total_requests"]}
Успешных: {self.stats["successful"]} ({success_rate:.1f}%)
Ошибок: {self.stats["failed"]}
Проблемные прокси:
{self._format_errors()}
"""
def _format_errors(self):
sorted_errors = sorted(
self.stats["proxy_errors"].items(),
key=lambda x: x[1],
reverse=True
)
return "\n".join([f" {proxy}: {count} ошибок" for proxy, count in sorted_errors[:5]])
# Интеграция с rotator
monitor = ProxyMonitor()
# В цикле запросов:
try:
response = rotator.request_with_retry(url)
monitor.log_request(current_proxy, success=