2026年Python调用住宅代理API全指南：千万级数据采集反封禁实战

pip install httpx[http2] retrying fake-useragent

import httpx
from retrying import retry
from fake_useragent import UserAgent
import random
import logging

# 配置日志，方便千万级任务监控
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
ua = UserAgent()

# 模拟 ZVVQ 合作代理商 API 接口
PROXY_POOL_URL = "http://api.your_proxy_provider.com/get_ip?auth=SECRET&num=5"

class ScraperSystem:
    def __init__(self):
        self.proxy_pool = []
        self.current_proxy = None

    def refresh_proxy_pool(self):
        """批量获取 IP，降低 API 调用频率"""
        try:
            logging.info("正在更新代理池...")
            resp = httpx.get(PROXY_POOL_URL, timeout=5)
            if resp.status_code == 200:
                # 假设返回的是换行符分割的 IP:Port
                self.proxy_pool = resp.text.strip().split('
')
                logging.info(f"成功获取 {len(self.proxy_pool)} 个新住宅 IP")
        except Exception as e:
            logging.error(f"代理池更新失败: {e}")

    def get_proxy(self):
        """从本地池中弹出一个 IP"""
        if not self.proxy_pool:
            self.refresh_proxy_pool()
        if self.proxy_pool:
            p = self.proxy_pool.pop(0)
            return {"http://": f"http://{p}", "https://": f"http://{p}"}
        return None

    @retry(stop_max_attempt_number=5, wait_fixed=2000)
    def fetch_page(self, url):
        """核心采集逻辑，含自动重试"""
        proxy = self.get_proxy()
        if not proxy:
            raise Exception("暂无可用代理")

        # 构造高仿真 Header
        headers = {
            "User-Agent": ua.random,
            "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
            "Accept-Language": "en-US,en;q=0.5",
            "Upgrade-Insecure-Requests": "1",
            "Sec-Fetch-Dest": "document",
            "Sec-Fetch-Mode": "navigate",
            "Sec-Fetch-Site": "none",
        }

        # 使用 httpx 开启 HTTP/2 支持，绕过部分针对 HTTP/1.1 的封锁
        with httpx.Client(proxies=proxy, headers=headers, http2=True, timeout=12) as client:
            logging.info(f"正在使用代理 {proxy['http://']} 请求: {url}")
            response = client.get(url)
            
            if response.status_code == 200:
                return response.text
            elif response.status_code in [403, 429]:
                logging.warning(f"触发风控 ({response.status_code})，准备重试并换 IP...")
                raise Exception("Triggered anti-spider")
            else:
                raise Exception(f"HTTP 错误: {response.status_code}")

if __name__ == "__main__":
    scraper = ScraperSystem()
    target_tasks = ["https://example.com/item/1", "https://example.com/item/2"] # 模拟千万级列表
    
    for task in target_tasks:
        try:
            html = scraper.fetch_page(task)
            logging.info(f"成功抓取长度: {len(html)}")
        except Exception as e:
            logging.error(f"任务失败: {task}, 原因: {e}")