Python代理IP采集怎样减少无效请求？6步优化脚本效率

import threading
from collections import Counter
​
import requests
​
​
class FailureStats:
    """线程安全的失败分类计数器。"""
​
    def __init__(self):
        self._counter = Counter()
        self._lock = threading.Lock()
​
    def record(self, category: str) -> None:
        with self._lock:
            self._counter[category] += 1
​
    def report(self) -> dict:
        with self._lock:
            total = sum(self._counter.values())
            if not total:
                return {}
            return {
                k: {"count": v, "ratio": round(v / total, 4)}
                for k, v in self._counter.most_common()
            }
​
​
def classify(status: int | None = None, exc: BaseException | None = None) -> str:
    """把一次请求结果归到一个可决策的类别上。"""
    if status is not None:
        if 200 <= status < 300:
            return "ok"
        if status in (401, 403, 451):
            return "target_reject"
        if status == 404:
            return "not_found"
        if status == 429:
            return "rate_limited"
        if 500 <= status < 600:
            return "target_5xx"
        return f"http_{status}"
​
    # 注意判断顺序：ProxyError / SSLError / ConnectTimeout
    # 都是 ConnectionError 或 Timeout 的子类，先具体后笼统。
    if isinstance(exc, requests.exceptions.ProxyError):
        return "proxy_error"
    if isinstance(exc, requests.exceptions.SSLError):
        return "ssl_error"
    if isinstance(exc, requests.exceptions.ConnectTimeout):
        return "connect_timeout"
    if isinstance(exc, requests.exceptions.ReadTimeout):
        return "read_timeout"
    if isinstance(exc, requests.exceptions.ConnectionError):
        return "conn_error"
    return "unknown"

from concurrent.futures import ThreadPoolExecutor
​
import requests
​
# 建议换成自建的 echo 端点，公共服务本身会限速，
# 预检结果会被它的限速污染。
PRECHECK_URL = "https://your-own-echo.example.com/ip"
​
​
def precheck(proxy: str) -> str | None:
    try:
        resp = requests.head(
            PRECHECK_URL,
            proxies={"http": proxy, "https": proxy},
            timeout=(2, 3),
            allow_redirects=False,
        )
        return proxy if resp.status_code < 400 else None
    except requests.RequestException:
        return None
​
​
def precheck_batch(proxy_list: list[str], workers: int = 20) -> list[str]:
    with ThreadPoolExecutor(max_workers=workers) as pool:
        return [p for p in pool.map(precheck, proxy_list) if p]

resp = requests.get(
    target_url,
    proxies={"http": proxy, "https": proxy},
    timeout=(3, 15),   # (connect_timeout, read_timeout)
    headers=default_headers,
)

import aiohttp
​
timeout = aiohttp.ClientTimeout(
    total=30,
    sock_connect=3,
    sock_read=15,
)

import random
from enum import Enum
​
​
class Action(str, Enum):
    ACCEPT = "accept"
    SKIP = "skip"
    RETRY_SAME = "retry_same"
    RETRY_SWITCH = "retry_switch"
    DELAY_RETRY = "delay_retry"
​
​
POLICY: dict[str, tuple[Action, int]] = {
    "ok":              (Action.ACCEPT,       0),
    "target_reject":   (Action.SKIP,         0),
    "not_found":       (Action.SKIP,         0),
    "rate_limited":    (Action.DELAY_RETRY,  1),
    "target_5xx":      (Action.RETRY_SAME,   2),
    "proxy_error":     (Action.RETRY_SWITCH, 3),
    "conn_error":      (Action.RETRY_SWITCH, 3),
    "connect_timeout": (Action.RETRY_SWITCH, 2),
    "read_timeout":    (Action.RETRY_SWITCH, 2),
    "ssl_error":       (Action.RETRY_SWITCH, 1),
    "unknown":         (Action.RETRY_SWITCH, 1),
}
​
​
def decide(category: str, attempt: int) -> Action:
    action, max_attempts = POLICY.get(category, (Action.RETRY_SWITCH, 1))
    if action in (Action.ACCEPT, Action.SKIP):
        return action
    return action if attempt < max_attempts else Action.SKIP
​
​
def backoff(attempt: int, base: float = 1.0, cap: float = 30.0) -> float:
    """指数退避 + 抖动，避免多协程在同一时刻集中重试。"""
    exp = min(cap, base * (2 ** attempt))
    return exp * (0.5 + random.random() * 0.5)

def retry_after_seconds(resp, fallback: float = 60.0) -> float:
    value = resp.headers.get("Retry-After")
    if not value:
        return fallback
    try:
        return float(value)
    except ValueError:
        return fallback   # HTTP-date 格式，按需再解析

import threading
​
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
​
​
class SessionFactory:
    """按代理入口地址复用 Session，退池时同步关闭释放连接。"""
​
    def __init__(self, pool_connections: int = 10, pool_maxsize: int = 20):
        self._sessions: dict[str, requests.Session] = {}
        self._lock = threading.Lock()
        self._pool_connections = pool_connections
        self._pool_maxsize = pool_maxsize
​
    def _build(self, proxy: str) -> requests.Session:
        session = requests.Session()
        session.proxies = {"http": proxy, "https": proxy}
        adapter = HTTPAdapter(
            pool_connections=self._pool_connections,
            pool_maxsize=self._pool_maxsize,
            # 底层重试全部关掉，重试策略由上层的 POLICY 统一决定，
            # 否则两套重试会叠乘，实际请求数远超预期。
            max_retries=Retry(total=0, backoff_factor=0, status_forcelist=[]),
        )
        session.mount("http://", adapter)
        session.mount("https://", adapter)
        return session
​
    def get(self, proxy: str) -> requests.Session:
        with self._lock:
            if proxy not in self._sessions:
                self._sessions[proxy] = self._build(proxy)
            return self._sessions[proxy]
​
    def drop(self, proxy: str) -> None:
        with self._lock:
            session = self._sessions.pop(proxy, None)
        if session is not None:
            session.close()

# ❌ 错误：每个 task 各自持有一个新信号量，等于没有限流
for i, url in enumerate(urls):
    sem = asyncio.Semaphore(MAX_PER_IP)
    tasks.append(fetch(sessions[i % len(sessions)], url, sem))

import asyncio
​
import aiohttp
​
MAX_PER_IP = 2
MAX_TOTAL = 30
TIMEOUT = aiohttp.ClientTimeout(total=30, sock_connect=3, sock_read=15)
​
​
async def fetch(session, url, ip_sem, total_sem):
    async with total_sem, ip_sem:
        try:
            async with session.get(url, timeout=TIMEOUT) as resp:
                if resp.status >= 400:
                    return None, classify(status=resp.status)
                return await resp.text(), "ok"
        except asyncio.TimeoutError:
            return None, "read_timeout"
        except aiohttp.ClientError as exc:
            return None, classify(exc=exc)
​
​
async def run_batch(urls: list[str], sessions: dict[str, aiohttp.ClientSession]):
    total_sem = asyncio.Semaphore(MAX_TOTAL)
    ip_sems = {proxy: asyncio.Semaphore(MAX_PER_IP) for proxy in sessions}
​
    keys = list(sessions)
    tasks = [
        fetch(sessions[keys[i % len(keys)]], url,
              ip_sems[keys[i % len(keys)]], total_sem)
        for i, url in enumerate(urls)
    ]
    return await asyncio.gather(*tasks)

import threading
import time
from collections import defaultdict
​
​
class IPPool:
    def __init__(self, proxies, fail_threshold: int = 3, cooldown: int = 300):
        self._available = list(proxies)
        self._failed: dict[str, float] = {}
        self._fail_count: dict[str, int] = defaultdict(int)
        self._cursor = 0
        self._fail_threshold = fail_threshold
        self._cooldown = cooldown
        self._lock = threading.Lock()
​
    def acquire(self) -> str | None:
        """轮询取用，避免总是命中列表头部的同一个 IP。"""
        with self._lock:
            self._recover_locked()
            if not self._available:
                return None
            self._cursor = (self._cursor + 1) % len(self._available)
            return self._available[self._cursor]
​
    def mark_success(self, proxy: str) -> None:
        with self._lock:
            self._fail_count[proxy] = 0
​
    def mark_fail(self, proxy: str) -> None:
        with self._lock:
            self._fail_count[proxy] += 1
            if self._fail_count[proxy] >= self._fail_threshold:
                if proxy in self._available:
                    self._available.remove(proxy)
                    self._cursor = 0
                self._failed[proxy] = time.time()
​
    def _recover_locked(self) -> None:
        now = time.time()
        for proxy, failed_at in list(self._failed.items()):
            if now - failed_at > self._cooldown:
                self._failed.pop(proxy)
                self._fail_count[proxy] = 0
                self._available.append(proxy)
​
    @property
    def stats(self) -> dict:
        with self._lock:
            return {"available": len(self._available), "cooling": len(self._failed)}

import time
​
import requests
​
​
def fetch_with_policy(
    url: str,
    pool: IPPool,
    factory: SessionFactory,
    stats: FailureStats,
    max_rounds: int = 5,
) -> str | None:
    attempt = 0
    proxy = pool.acquire()
​
    for _ in range(max_rounds):
        if proxy is None:
            stats.record("pool_exhausted")
            return None
​
        session = factory.get(proxy)
        try:
            resp = session.get(url, timeout=(3, 15))
            category = classify(status=resp.status_code)
        except requests.RequestException as exc:
            category = classify(exc=exc)
            resp = None
​
        stats.record(category)
​
        if category == "ok":
            pool.mark_success(proxy)
            return resp.text
​
        action = decide(category, attempt)
        attempt += 1
​
        if action is Action.SKIP:
            return None
        if action is Action.DELAY_RETRY:
            time.sleep(retry_after_seconds(resp) if resp is not None else 60.0)
            continue
        if action is Action.RETRY_SAME:
            time.sleep(backoff(attempt))
            continue
        if action is Action.RETRY_SWITCH:
            pool.mark_fail(proxy)
            factory.drop(proxy)
            proxy = pool.acquire()
            time.sleep(backoff(attempt, base=0.5))
            continue
​
    return None