ProxyScrape Python: Bangun Kumpulan Proxy Rotasi Asinkron dengan httpx dan SOCKS5

Ringkasan

Ini adalah tutorial Python lanjutan. Alih-alih mengirim satu permintaan melalui satu proxy, Anda membangun pengambil kecil yang berputar di antara endpoint ProxyScrape dalam paket Anda, membatasi konkurensi, mencoba ulang dengan backoff yang dijitter, dan menempatkan endpoint yang gagal dalam cooldown.

Anda akan:

• Instal httpx dengan dukungan SOCKS5.
• Simpan kredensial ProxyScrape dalam variabel lingkungan.
• Modelkan endpoint HTTP dan SOCKS5 Anda sebagai kumpulan.
• Rotasi, batasi, coba ulang, dan laporkan kesehatan per endpoint.
• Uji coba sebelum meningkatkan skala.

ProxyScrape menawarkan proxy residensial, pusat data, dan seluler, mendukung HTTP dan SOCKS5, dan menagih baik bayar sesuai pemakaian atau tetap bulanan. Pilihan Anda mengubah biaya dan latensi, bukan kode klien — kode di bawah memperlakukan endpoint sebagai konfigurasi.

Pilih jenis proxy sebelum Anda menulis kode

Prasyarat

• Python 3.10 atau lebih baru.
• Paket ProxyScrape dengan kredensial, plus host dan port untuk http dan/atau socks5.
• Akses keluar dari mesin Anda ke host dan port ProxyScrape.

Langkah-langkah

Langkah 1 — Buat proyek dan instal dependensi

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

Ekstra [socks] menginstal socksio. Tanpa itu, httpx akan memunculkan error saat pertama kali Anda mengonfigurasi endpoint socks5://. httpx 0.26 dan yang lebih baru menggunakan argumen proxy= (rilis lama menggunakan proxies=), jadi periksa versi Anda dengan pip show httpx jika contoh tidak cocok.

Langkah 2 — Simpan kredensial dan endpoint dalam variabel lingkungan

# 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")

Ekspor nilai-nilai di shell Anda, atau suntikkan dari manajer rahasia di 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"

Pengodean persen melalui quote() penting: kata sandi yang mengandung @, :, atau / jika tidak akan merusak penguraian URL. Jangan pernah komit nilai-nilai ini.

Langkah 3 — Verifikasi setiap endpoint sebelum meningkatkan skala

Jalankan daftar periksa ini secara berurutan. Jika ada langkah yang gagal, perbaiki sebelum menyentuh kode async.

1. Periksa endpoint HTTP/HTTPS.
2. Periksa endpoint SOCKS5.
3. Pastikan IP yang dikembalikan bukan IP Anda sendiri.

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:// memberitahu curl untuk membiarkan proxy menyelesaikan DNS. Jika klien Python Anda menawarkan baik socks5:// maupun socks5h://, pilih varian h ketika resolusi DNS lokal gagal atau ketika Anda ingin DNS meninggalkan mesin Anda bersama permintaan.

Langkah 4 — Bangun fetcher berputar dan konkuren

# 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()))

Catatan desain yang patut dipertahankan:

• Satu AsyncClient per endpoint. Di httpx, proxy terikat ke klien, bukan ke permintaan individu, jadi klien per endpoint adalah model rotasi yang paling sederhana dan benar.
• Semaphore membatasi permintaan yang sedang berlangsung tidak peduli berapa banyak URL yang Anda kirimkan.
• 407 dan 429 bergabung dengan 5xx sebagai status yang dapat dicoba ulang, dan mereka dihitung sebagai kegagalan endpoint sehingga endpoint yang buruk berhenti menerima lalu lintas untuk sementara.
• Penalti kegagalan tumbuh secara linear dan direset saat berhasil, yang mencegah endpoint yang tidak stabil untuk dicoba ulang dalam loop ketat.

Langkah 5 — Uji coba dan setel konkurensi

# 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())

Lalu setel:

1. Jalankan sekali. Jika failed adalah 0, naikkan concurrency sebesar 4 dan jalankan lagi.
2. Berhenti menambah ketika latensi atau timeout mulai naik — itulah batas praktis untuk paket yang Anda beli.
3. Jika kegagalan menumpuk pada satu endpoint, anggap endpoint itu tidak sehat dan hubungi dukungan alih-alih mencoba lebih keras.

Catatan paket: dengan penagihan bayar sesuai pemakaian, Anda dapat mendorong konkurensi naik untuk lonjakan; paket bulanan tetap biasanya lebih cocok untuk pekerjaan terjadwal yang stabil.

Pemecahan Masalah

Ringkasan

• ProxyScrape mendukung HTTP dan SOCKS5; representasikan setiap endpoint dari paket Anda sebagai ProxyEndpoint dan rotasi di antara mereka.
• Ikat proxy ke httpx.AsyncClient, batasi pekerjaan yang sedang berlangsung dengan semaphore, dan coba ulang dengan backoff yang dijitter.
• Perlakukan respons 407, 429, dan 5xx sebagai kegagalan endpoint daripada kesalahan pengguna, dan dinginkan endpoint tersebut.
• Verifikasi endpoint dengan curl sebelum meningkatkan skala, lalu naikkan konkurensi secara bertahap sampai latensi — bukan keberuntungan — menjadi batas Anda.