AI 数据采集中的动态代理：路由、会话、重试与验证

{
  "url": "https://example.org/data/42",
  "source": "example.org",
  "route": {
    "mode": "rotating",
    "country": "US",
    "session_id": null
  },
  "attempt": 0,
  "max_attempts": 3
}

export PROXY_URL='http://user:password@proxy.example.com:8080'

curl --fail-with-body \
  --connect-timeout 10 \
  --max-time 30 \
  --proxy "$PROXY_URL" \
  https://example.org/

import os
import time
from email.utils import parsedate_to_datetime
from datetime import datetime, timezone

import requests

PROXY_URL = os.environ["PROXY_URL"]
TIMEOUT = (10, 30)       # 示例值：connect/read timeout
MAX_ATTEMPTS = 3         # 示例值，不是通用最佳值


def retry_after_seconds(response: requests.Response) -> float | None:
    value = response.headers.get("Retry-After")
    if not value:
        return None
    if value.isdigit():
        return max(0.0, float(value))
    try:
        retry_at = parsedate_to_datetime(value)
        if retry_at.tzinfo is None:
            retry_at = retry_at.replace(tzinfo=timezone.utc)
        return max(0.0, (retry_at - datetime.now(timezone.utc)).total_seconds())
    except (TypeError, ValueError, OverflowError):
        return None


def fetch(url: str) -> requests.Response:
    proxies = {"http": PROXY_URL, "https": PROXY_URL}
    last_error: Exception | None = None

    with requests.Session() as session:
        for attempt in range(MAX_ATTEMPTS):
            try:
                response = session.get(
                    url,
                    proxies=proxies,
                    timeout=TIMEOUT,
                    headers={"User-Agent": "AuthorizedDataCollector/1.0"},
                )

                if response.status_code == 407:
                    raise RuntimeError("Proxy authentication failed (HTTP 407)")

                if response.status_code == 429:
                    wait = retry_after_seconds(response)
                    if wait is None or attempt == MAX_ATTEMPTS - 1:
                        response.raise_for_status()
                    time.sleep(min(wait, 60))  # 60 秒仅为示例安全上限
                    continue

                response.raise_for_status()
                return response

            except (requests.Timeout, requests.ConnectionError) as exc:
                last_error = exc
                if attempt == MAX_ATTEMPTS - 1:
                    raise
                time.sleep(min(2 ** attempt, 8))

    raise RuntimeError("request failed") from last_error