آموزش پایتون ProxyScrape: ساخت یک استخر پروکسی چرخشی آسنکرون با httpx و SOCKS5

Overview

این یک آموزش پیشرفته پایتون است. بهجای ارسال یک درخواست از طریق یک پروکسی، یک واکشیکننده کوچک میسازید که بین نقاط پایانی ProxyScrape در طرح شما میچرخد، همزمانی را محدود میکند، با عقبنشینی همراه با جیتر دوباره تلاش میکند، و نقاط پایانی ناموفق را در یک دوره خنکسازی قرار میدهد.

شما این کارها را انجام میدهید:

• نصب httpx با پشتیبانی SOCKS5.
• نگهداشتن اعتبارنامههای ProxyScrape در متغیرهای محیطی.
• مدلسازی نقاط پایانی HTTP و SOCKS5 خود بهصورت یک استخر.
• چرخش، محدودسازی، تلاش مجدد، و گزارش سلامت هر نقطه پایانی.
• تست دود قبل از بزرگکردن مقیاس.

ProxyScrape پروکسیهای مسکونی، دیتاسنتر و موبایل ارائه میدهد، از HTTP و SOCKS5 پشتیبانی میکند، و صورتحساب را بهصورت پرداخت بهازای مصرف یا ماهانه ثابت محاسبه میکند. انتخاب شما هزینه و تأخیر را تغییر میدهد، نه کد کلاینت را — کد زیر با نقاط پایانی مثل پیکربندی رفتار میکند.

قبل از نوشتن کد، نوع پروکسی را انتخاب کنید

پیشنیازها

• پایتون ۳.۱۰ یا جدیدتر.
• یک طرح ProxyScrape با اعتبارنامهها، بههمراه میزبان و پورت برای http و/یا socks5.
• دسترسی خروجی از دستگاه شما به میزبان و پورت ProxyScrape.

گامها

گام ۱ — ایجاد پروژه و نصب وابستگیها

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 بررسی کنید.

گام ۲ — نگهداشتن اعتبارنامهها و نقاط پایانی در متغیرهای محیطی

# 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("Set PROXYSCRAPE_HTTP_PORT and/or PROXYSCRAPE_SOCKS5_PORT")

مقادیر را در شل خود export کنید، یا آنها را از یک مدیر اسرار در 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 را خراب میکنند. هرگز این مقادیر را commit نکنید.

گام ۳ — قبل از بزرگکردن مقیاس، هر نقطه پایانی را تأیید کنید

این چکلیست را به ترتیب اجرا کنید. اگر گامی شکست خورد، قبل از دستزدن به کد آسنکرون آن را برطرف کنید.

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 را resolve کند. اگر کلاینت پایتون شما هم socks5:// و هم socks5h:// را ارائه میدهد، وقتی resolve محلی DNS شکست میخورد یا میخواهید DNS همراه با درخواست از دستگاه شما خارج شود، نوع h را ترجیح دهید.

گام ۴ — ساخت واکشیکننده چرخشی و همزمان

# fetch_async.py
from __future__ import annotations

import asyncio
import itertools
import random
import time

import httpx

from pool import POOL, ProxyEndpoint


class RotatingFetcher:
    """Round-robin fetcher with concurrency limits, backoff, and cooldowns."""

    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("Proxy pool is empty")
        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"retryable status {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"All {self._max_attempts} attempts failed for {url}") from last_error

    def report(self) -> str:
        active = {label: count for label, count in self._failures.items() if count}
        if not active:
            return "no endpoints have failed yet"
        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 بهعنوان وضعیتهای قابل تلاش مجدد میپیوندند، و بهعنوان شکست نقطه پایانی شمرده میشوند تا یک نقطه پایانی بد مدتی ترافیک دریافت نکند.
• جریمههای شکست بهصورت خطی رشد میکنند و در موفقیت ریست میشوند، که مانع میشود یک نقطه پایانی نامطمئن در یک حلقه تنگ دوباره تلاش شود.

گام ۵ — تست دود و تنظیم همزمانی

# 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 صفر بود، concurrency را ۴ واحد افزایش دهید و دوباره اجرا کنید.
2. وقتی تأخیر یا timeoutها شروع به افزایش کردند متوقف شوید — این سقف عملی برای طرحی است که خریدهاید.
3. اگر شکستها روی یک نقطه پایانی متمرکز شدند، آن نقطه پایانی را ناسالم در نظر بگیرید و بهجای تلاش سختتر با پشتیبانی تماس بگیرید.

نکته طرح: با صورتحساب پرداخت بهازای مصرف میتوانید همزمانی را برای یک انفجار بالا ببرید؛ یک طرح ماهانه ثابت معمولاً برای کارهای ثابت و زمانبندیشده مناسبتر است.

عیبیابی

خلاصه

• ProxyScrape از HTTP و SOCKS5 پشتیبانی میکند؛ هر نقطه پایانی از طرح خود را بهصورت یک ProxyEndpoint نمایش دهید و بین آنها بچرخید.
• یک پروکسی را به httpx.AsyncClient متصل کنید، کار در پرواز را با سمافور محدود کنید، و با عقبنشینی همراه با جیتر دوباره تلاش کنید.
• پاسخهای 407، 429 و 5xx را بهجای خطاهای کاربر، شکست نقطه پایانی در نظر بگیرید و نقطه پایانی را خنک کنید.
• قبل از بزرگکردن مقیاس، نقاط پایانی را با curl تأیید کنید، سپس همزمانی را بهتدریج افزایش دهید تا تأخیر — نه شانس — به محدودیت شما تبدیل شود.