🐍可靠的 Python 代理抓取需要同时控制：代理身份、连接池、超时、重试、页面分类、地域验证和资源清理。仅把代理 URL 填进 proxies 参数并不等于生产可用。
本文分别展示 Requests、HTTPX 和 Playwright 的实现边界。所有示例只应用于获得授权的公开数据、QA、监控或自有系统。
环境变量
PROXY_URL=http://USER:PASSWORD@PROXY_HOST:PORT
IP_CHECK_URL=https://iprobe.io/json
TARGET_URL=https://example.com
生产环境中应把主机、用户名和密码分开存储，避免异常堆栈输出完整 URL。
Requests 同步示例
import os
import requests
proxy = os.environ['PROXY_URL']
proxies = {
'http': proxy,
'https': proxy,
}
response = requests.get(
os.getenv('IP_CHECK_URL', 'https://iprobe.io/json'),
proxies=proxies,
timeout=(10, 20),
)
response.raise_for_status()
print(response.text)
timeout=(10, 20) 分别限制连接和读取等待。不要省略超时，否则故障路由可能长期占用工作线程。
使用 Session 复用连接
from contextlib import closing
import requests
with closing(requests.Session()) as session:
session.proxies.update(proxies)
session.headers.update({
'User-Agent': 'AuthorizedDataCollector/1.0',
'Accept-Language': 'en-US,en;q=0.9',
})
response = session.get('https://example.com', timeout=(10, 20))
连接复用会影响代理轮换。即使凭证表示轮换，持续连接也可能继续使用原有出口。需要每个任务新身份时，应明确连接和 Session 生命周期。
HTTPX 异步示例
import asyncio
import os
import httpx
async def fetch(url: str) -> dict:
timeout = httpx.Timeout(connect=10, read=20, write=10, pool=10)
limits = httpx.Limits(max_connections=20, max_keepalive_connections=10)
async with httpx.AsyncClient(
proxy=os.environ['PROXY_URL'],
timeout=timeout,
limits=limits,
follow_redirects=True,
) as client:
response = await client.get(url)
return {
'status': response.status_code,
'final_url': str(response.url),
'body': response.text,
}
print(asyncio.run(fetch('https://iprobe.io/json')))
异步并不意味着无限并发。应通过队列或信号量限制同一目标的并发。
并发控制
import asyncio
semaphore = asyncio.Semaphore(5)
async def bounded_fetch(url: str):
async with semaphore:
return await fetch(url)
并发值应依据目标允许速率、代理账户限制、CPU、连接数和有效输出率逐步调整。
轮换与粘性代理
代理服务通常通过用户名中的会话参数或不同端点实现轮换。
from urllib.parse import quote
def proxy_url(host: str, user: str, password: str) -> str:
return f"http://{quote(user)}:{quote(password)}@{host}"
独立任务： 每个 URL 或任务使用新的轮换身份。
状态任务： 登录、分页、表单和购物车应使用同一粘性会话，并保持同一个客户端或浏览器 Context。
不要在重试函数内部无条件生成新会话，否则状态任务会在中途换 IP。
SOCKS 与远程 DNS
安装 SOCKS 支持：
pip install 'httpx[socks]'
不同库对 SOCKS5 本地或代理端 DNS 的配置方式不同。需要远程解析时，验证客户端是否支持 socks5h 语义，不要只根据 URL 字符串推测。
Playwright 浏览器回退
import os
from playwright.async_api import async_playwright
async def browser_fetch(url: str) -> str:
async with async_playwright() as p:
browser = await p.chromium.launch(
headless=True,
proxy={
'server': os.environ['PROXY_SERVER'],
'username': os.environ.get('PROXY_USERNAME'),
'password': os.environ.get('PROXY_PASSWORD'),
},
)
try:
context = await browser.new_context(
locale='en-US',
timezone_id='America/New_York',
)
try:
page = await context.new_page()
await page.goto(url, wait_until='domcontentloaded', timeout=45_000)
return await page.locator('body').inner_text()
finally:
await context.close()
finally:
await browser.close()
只有在数据依赖 JavaScript 或交互时才使用浏览器。静态 HTTP 更便宜、更容易调试。
页面验证
不要只判断 response.status_code == 200。还要检查：
- 最终 URL
- Content-Type
- 页面标题
- 业务字段
- 挑战、登录或同意页标记
- 地区、语言和币种
- 响应长度异常
from dataclasses import dataclass
@dataclass
class Validation:
valid: bool
page_class: str
reason: str | None = None
def validate(status: int, body: str) -> Validation:
lower = body.lower()
if status == 407:
return Validation(False, 'proxy_auth', '代理凭证被拒绝')
if 'captcha' in lower:
return Validation(False, 'challenge', '发现挑战页面')
if '<title>example' not in lower:
return Validation(False, 'unexpected', '缺少预期页面标记')
return Validation(True, 'expected')
实际项目不要只依赖一个字符串，应结合 URL、结构和业务 Schema。
错误分类与重试
from enum import Enum
class RetryClass(str, Enum):
STOP = 'stop'
BACKOFF = 'backoff'
NEW_ROUTE = 'new_route'
PARSER_REVIEW = 'parser_review'
建议规则：
- 407：停止，修复凭证
- 401/明确拒绝：停止，复核授权
- 429：退避并降低频率
- 连接重置或临时 5xx：有限重试
- 错地区：检查市场配置，不直接当成成功
- 解析失败：保存证据并更新解析器
指数退避与抖动
import asyncio
import random
async def sleep_before_retry(attempt: int) -> None:
base = min(2 ** attempt, 30)
await asyncio.sleep(random.uniform(0, base))
每个任务还应限制最大尝试次数、总耗时和最大字节数。
结构化日志
{
"jobId": "job-001",
"attempt": 1,
"proxyMode": "sticky",
"requestedCountry": "US",
"observedCountry": "US",
"status": 200,
"pageClass": "expected",
"durationMs": 1840,
"downloadBytes": 182220,
"businessOutput": "valid"
}
不要记录完整密码、可复用会话 ID、敏感页面内容或不必要的个人数据。
生产指标
- 有效业务记录率
- 每条有效记录的字节数
- 每条有效记录的成本
- 407、连接失败和 429 比例
- 错地区率
- 挑战页比例
- 中位数和 p95 时延
- 重试膨胀系数
上线检查清单
- 所有客户端都有超时。
- 并发有上限。
- Session 生命周期与代理会话一致。
- 浏览器只作为必要回退。
- HTTP 状态和页面分类分别记录。
- 重试按错误类型决策。
- 凭证已脱敏。
- 有原始证据与内容哈希。
- 指标以有效结果为中心。
- 已确认数据访问权限。
相关内容