Руководство ProxyScrape Python: создание асинхронного пула ротационных прокси с httpx и SOCKS5

Обзор

Это продвинутое руководство по Python. Вместо отправки одного запроса через один прокси вы создаёте небольшой загрузчик, который ротирует эндпоинты ProxyScrape из вашего тарифа, ограничивает параллелизм, повторяет попытки с джиттерным backoff и отправляет сбойные эндпоинты в cooldown.

Вы будете:

• Установите httpx с поддержкой SOCKS5.
• Храните учётные данные ProxyScrape в переменных окружения.
• Представьте свои HTTP- и SOCKS5-эндпоинты как пул.
• Ротируйте, ограничивайте, повторяйте запросы и сообщайте о работоспособности каждого эндпоинта.
• Проведите smoke-тест перед масштабированием.

ProxyScrape предлагает резидентные, дата-центровые и мобильные прокси, поддерживает HTTP и SOCKS5 и тарифицирует либо по мере использования, либо фиксированной месячной оплатой. Ваш выбор влияет на стоимость и задержку, но не на клиентский код — в приведённом ниже коде эндпоинты рассматриваются как конфигурация.

Выберите тип прокси перед написанием кода

Предварительные требования

• Python 3.10 или новее.
• Тариф ProxyScrape с учётными данными, а также хост и порт для http и/или socks5.
• Исходящий доступ с вашей машины к хосту и порту ProxyScrape.

Шаги

Шаг 1 — Создайте проект и установите зависимости

mkdir proxyscrape-async
cd proxyscrape-async
python -m venv .venv
source .venv/bin/activate
pip install "httpx[socks]"

Дополнение [socks] устанавливает socksio. Без него httpx выдаёт ошибку при первой настройке эндпоинта socks5://. В httpx 0.26 и новее используется аргумент proxy= (в старых версиях — proxies=), поэтому проверьте свою версию командой pip show httpx, если примеры не совпадают.

Шаг 2 — Храните учётные данные и эндпоинты в переменных окружения

# pool.py
import os
from dataclasses import dataclass
from urllib.parse import quote


@dataclass(frozen=True)
class ProxyEndpoint:
    label: str
    url: str


def endpoint_url(scheme: str, host: str, port: str, username: str, password: str) -> str:
    user = quote(username, safe="")
    secret = quote(password, safe="")
    return f"{scheme}://{user}:{secret}@{host}:{port}"


HOST = os.environ["PROXYSCRAPE_HOST"]
USERNAME = os.environ["PROXYSCRAPE_USERNAME"]
PASSWORD = os.environ["PROXYSCRAPE_PASSWORD"]

POOL: list[ProxyEndpoint] = []

if http_port := os.environ.get("PROXYSCRAPE_HTTP_PORT"):
    POOL.append(ProxyEndpoint("http-1", endpoint_url("http", HOST, http_port, USERNAME, PASSWORD)))

if socks_port := os.environ.get("PROXYSCRAPE_SOCKS5_PORT"):
    POOL.append(ProxyEndpoint("socks5-1", endpoint_url("socks5", HOST, socks_port, USERNAME, PASSWORD)))

if not POOL:
    raise SystemExit("Задайте PROXYSCRAPE_HTTP_PORT и/или PROXYSCRAPE_SOCKS5_PORT")

Экспортируйте значения в своей оболочке или внедряйте их из менеджера секретов в CI:

export PROXYSCRAPE_HOST="host-from-your-plan"
export PROXYSCRAPE_USERNAME="your-username"
export PROXYSCRAPE_PASSWORD="your-password"
export PROXYSCRAPE_HTTP_PORT="port-from-your-plan"
export PROXYSCRAPE_SOCKS5_PORT="port-from-your-plan"

Процентное кодирование через quote() важно: пароли, содержащие @, : или /, иначе нарушают разбор URL. Никогда не коммитьте эти значения.

Шаг 3 — Проверьте каждый эндпоинт перед масштабированием

Выполните этот чек-лист по порядку. Если шаг не удался, исправьте его, прежде чем трогать асинхронный код.

1. Проверьте эндпоинт HTTP/HTTPS.
2. Проверьте эндпоинт SOCKS5.
3. Убедитесь, что возвращённый IP — не ваш собственный IP.

curl -sS -x "http://$PROXYSCRAPE_USERNAME:$PROXYSCRAPE_PASSWORD@$PROXYSCRAPE_HOST:$PROXYSCRAPE_HTTP_PORT" "https://api.ipify.org?format=json"
curl -sS -x "socks5h://$PROXYSCRAPE_USERNAME:$PROXYSCRAPE_PASSWORD@$PROXYSCRAPE_HOST:$PROXYSCRAPE_SOCKS5_PORT" "https://api.ipify.org?format=json"

socks5h:// указывает curl, чтобы прокси выполнял разрешение DNS. Если ваш Python-клиент поддерживает и socks5://, и socks5h://, предпочитайте вариант с h, когда локальное разрешение DNS не работает или когда вы хотите, чтобы DNS-запрос покидал вашу машину вместе с запросом.

Шаг 4 — Соберите ротирующий конкурентный загрузчик

# fetch_async.py
from __future__ import annotations

import asyncio
import itertools
import random
import time

import httpx

from pool import POOL, ProxyEndpoint


class RotatingFetcher:
    """Загрузчик с круговой ротацией, ограничениями параллелизма, backoff и cooldown."""

    def __init__(
        self,
        pool: list[ProxyEndpoint],
        concurrency: int = 8,
        max_attempts: int = 4,
        timeout: float = 20.0,
        cooldown_seconds: float = 45.0,
    ) -> None:
        if not pool:
            raise ValueError("Пул прокси пуст")
        self._pool = pool
        self._cycle = itertools.cycle(pool)
        self._semaphore = asyncio.Semaphore(concurrency)
        self._max_attempts = max_attempts
        self._cooldown_seconds = cooldown_seconds
        self._failures: dict[str, int] = {}
        self._cooldown_until: dict[str, float] = {}
        self._clients = {
            item.label: httpx.AsyncClient(
                proxy=item.url,
                timeout=timeout,
                follow_redirects=True,
                headers={"User-Agent": "Mozilla/5.0 (compatible; ProxyScoutTutorial/1.0)"},
            )
            for item in pool
        }

    def _pick(self) -> ProxyEndpoint:
        now = time.monotonic()
        for _ in range(len(self._pool) * 2):
            candidate = next(self._cycle)
            if self._cooldown_until.get(candidate.label, 0.0) <= now:
                return candidate
        return min(self._pool, key=lambda item: self._cooldown_until.get(item.label, 0.0))

    def _record_failure(self, label: str) -> None:
        self._failures[label] = self._failures.get(label, 0) + 1
        penalty = self._cooldown_seconds * self._failures[label]
        self._cooldown_until[label] = time.monotonic() + penalty

    def _record_success(self, label: str) -> None:
        self._failures[label] = 0
        self._cooldown_until[label] = 0.0

    @staticmethod
    def _backoff(attempt: int) -> float:
        return min((2 ** attempt) + random.uniform(0, 0.5), 30.0)

    async def fetch(self, url: str) -> httpx.Response:
        last_error: Exception | None = None
        for attempt in range(1, self._max_attempts + 1):
            item = self._pick()
            try:
                async with self._semaphore:
                    response = await self._clients[item.label].get(url)
                if response.status_code in {407, 429} or response.status_code >= 500:
                    raise httpx.HTTPStatusError(
                        f"статус для повтора {response.status_code}",
                        request=response.request,
                        response=response,
                    )
                self._record_success(item.label)
                return response
            except Exception as exc:
                last_error = exc
                self._record_failure(item.label)
                if attempt < self._max_attempts:
                    await asyncio.sleep(self._backoff(attempt))
        raise RuntimeError(f"Все {self._max_attempts} попыток не удались для {url}") from last_error

    def report(self) -> str:
        active = {label: count for label, count in self._failures.items() if count}
        if not active:
            return "пока ни один эндпоинт не сбоил"
        return ", ".join(f"{label}={count}" for label, count in sorted(active.items()))

    async def aclose(self) -> None:
        await asyncio.gather(*(client.aclose() for client in self._clients.values()))

Заметки по дизайну, которые стоит сохранить:

• Один AsyncClient на эндпоинт. В httpx прокси привязан к клиенту, а не к отдельному запросу, поэтому клиент на эндпоинт — самая простая корректная модель ротации.
• Семафор ограничивает количество запросов в полёте, сколько бы URL вы ни отправили.
• 407 и 429 вместе с 5xx считаются статусами для повтора, и они учитываются как сбои эндпоинта, поэтому плохой эндпоинт на время перестаёт получать трафик.
• Штрафы за сбои растут линейно и сбрасываются при успехе, что не даёт нестабильному эндпоинту попадать в плотный цикл повторных попыток.

Шаг 5 — Smoke-тест и настройка параллелизма

# run_smoke_test.py
import asyncio

import httpx

from fetch_async import RotatingFetcher
from pool import POOL


async def main() -> None:
    fetcher = RotatingFetcher(pool=POOL, concurrency=8)
    urls = [f"https://example.com/?page={index}" for index in range(1, 21)]
    try:
        results = await asyncio.gather(*(fetcher.fetch(url) for url in urls), return_exceptions=True)
        ok = sum(1 for item in results if isinstance(item, httpx.Response))
        print(f"success={ok} failed={len(results) - ok}")
        print(f"endpoint health: {fetcher.report()}")
    finally:
        await fetcher.aclose()


if __name__ == "__main__":
    asyncio.run(main())

Затем настройте:

1. Запустите один раз. Если failed равен 0, увеличьте concurrency на 4 и запустите снова.
2. Остановите увеличение, когда задержка или таймауты начнут расти — это практический потолок для купленного тарифа.
3. Если сбои сосредоточены на одном эндпоинте, считайте этот эндпоинт неработоспособным и обратитесь в поддержку вместо более настойчивых повторов.

Примечание о тарифе: при оплате по мере использования можно поднять параллелизм для всплеска; фиксированный месячный тариф обычно лучше подходит для стабильных запланированных задач.

Устранение неполадок

Итог

• ProxyScrape поддерживает HTTP и SOCKS5; представьте каждый эндпоинт из вашего тарифа как ProxyEndpoint и ротируйте их.
• Привяжите прокси к httpx.AsyncClient, ограничьте работу в полёте с помощью семафора и повторяйте попытки с джиттерным backoff.
• Считайте ответы 407, 429 и 5xx сбоями эндпоинта, а не ошибками пользователя, и отправляйте эндпоинт в cooldown.
• Проверьте эндпоинты через curl перед масштабированием, затем постепенно увеличивайте параллелизм, пока пределом не станет задержка, а не удача.