住宅代理怎么接进 Python 采集脚本：Requests、HTTPX、AIOHTTP 和 Playwright 配置实录

接入前先确定代理类型与认证方式

开始写配置代码前要先明确两件事：任务需要轮换出口还是固定出口，以及代理商提供的是账密认证还是白名单认证。

动态住宅代理通常按流量或带宽计费，每次请求或按会话轮换 IP，适合大批量采集、SERP 监控和广告验证。静态住宅代理提供固定 IP，分短效（小时或天级）和长效（月或年级），适合需要稳定出口的店铺后台登录、社媒账号运营和 TK 内容管理。

账密认证在代理 URL 中直接嵌入用户名和密码，格式为 http://username:password@host:port。白名单认证则需在服务商后台绑定服务器公网 IP，之后直接连接 host:port 即可免密使用。部分动态代理允许在用户名后缀添加参数控制粘性会话或目标国家，例如 user_session_abc123 或 user_country_us，具体格式参考服务商控制台提取说明。

Requests 配置：Session 绑定与 SOCKS5h 防泄漏

Requests 是最常用的 HTTP 客户端，通过 proxies 字典分别声明 HTTP 与 HTTPS 协议代理：

import requests

proxies = {
    'http': 'http://username:[email protected]:8000',
    'https': 'http://username:[email protected]:8000'
}

response = requests.get('https://httpbin.org/ip', proxies=proxies)
print(response.json())

如果需要复用连接池、保持 Cookie 或管理多个任务的独立会话,应将代理绑定到 requests.Session() 上：

session = requests.Session()
session.proxies.update(proxies)
response = session.get('https://example.com')

使用 SOCKS5 住宅代理时,必须先安装支持包 pip install 'requests[socks]'（基于 PySocks）。更重要的是,推荐使用 socks5h:// 前缀而非 socks5://：

proxies = {
    'http': 'socks5h://username:[email protected]:1080',
    'https': 'socks5h://username:[email protected]:1080'
}

socks5h 强制通过代理服务端远端解析 DNS,避免本地 DNS 查询暴露真实 IP 或网络特征。如果目标站点对 DNS 请求来源敏感,这一步能降低被识别为代理流量的风险。

HTTPX 配置：同步与异步统一接口

HTTPX 同时支持同步与异步请求,代理参数在实例化 Client 时通过 proxy 传入：

import httpx

proxy_url = 'http://username:[email protected]:8000'

# 同步客户端
with httpx.Client(proxy=proxy_url, follow_redirects=True) as client:
    response = client.get('https://httpbin.org/ip')
    print(response.json())

# 异步客户端
import asyncio

async def fetch():
    async with httpx.AsyncClient(proxy=proxy_url, follow_redirects=True) as client:
        response = await client.get('https://httpbin.org/ip')
        print(response.json())

asyncio.run(fetch())

HTTPX 支持 HTTP、HTTPS 和 SOCKS5 代理,但默认不自动跟随重定向,需显式设置 follow_redirects=True。如果任务中有大量并发请求或需要保持长连接,可以调整 limits 参数控制连接池大小和超时：

limits = httpx.Limits(max_keepalive_connections=20, max_connections=100)
client = httpx.Client(proxy=proxy_url, limits=limits, timeout=10.0)

AIOHTTP 配置：原生 HTTP 与 SOCKS5 扩展

AIOHTTP 是异步 HTTP 客户端,代理在发起请求时通过 proxy 参数传入。原生主要支持 HTTP 与 HTTPS 代理：

import aiohttp
import asyncio

async def fetch():
    proxy = 'http://proxy.example.com:8000'
    proxy_auth = aiohttp.BasicAuth('username', 'password')
    
    async with aiohttp.ClientSession() as session:
        async with session.get(
            'https://httpbin.org/ip',
            proxy=proxy,
            proxy_auth=proxy_auth
        ) as response:
            print(await response.json())

asyncio.run(fetch())

认证信息既可以通过 URL 内联 http://username:password@host:port,也可以显式声明 proxy_auth=aiohttp.BasicAuth('user', 'pass')。

若要支持 SOCKS5 代理,通常需要配合第三方连接器扩展包（如 aiohttp-socks）注入自定义 TCPConnector：

from aiohttp_socks import ProxyConnector

connector = ProxyConnector.from_url('socks5://username:[email protected]:1080')

async with aiohttp.ClientSession(connector=connector) as session:
    async with session.get('https://httpbin.org/ip') as response:
        print(await response.json())

Playwright 配置：全局与上下文级代理隔离

Playwright 等无头浏览器自动化框架适合需要渲染 JavaScript 或模拟真实浏览器行为的采集任务。代理可以在启动浏览器时全局配置,也可以在创建上下文时按任务或账号隔离配置。

全局代理配置在 launch 时传入：

from playwright.sync_api import sync_playwright

with sync_playwright() as p:
    browser = p.chromium.launch(
        proxy={
            'server': 'http://proxy.example.com:8000',
            'username': 'username',
            'password': 'password'
        }
    )
    page = browser.new_page()
    page.goto('https://httpbin.org/ip')
    print(page.content())
    browser.close()

如果不同任务需要分配独立出口 IP,推荐在 new_context 中按上下文配置：

browser = p.chromium.launch()

context_1 = browser.new_context(
    proxy={
        'server': 'http://proxy1.example.com:8000',
        'username': 'user1',
        'password': 'pass1'
    }
)
page_1 = context_1.new_page()
page_1.goto('https://httpbin.org/ip')

context_2 = browser.new_context(
    proxy={
        'server': 'http://proxy2.example.com:8000',
        'username': 'user2',
        'password': 'pass2'
    }
)
page_2 = context_2.new_page()
page_2.goto('https://httpbin.org/ip')

这种隔离方式适合多账号运营或需要不同地区出口的并行采集任务。

白名单认证与账密认证的代码差异

白名单认证绑定服务器公网 IP 后可免密直接连接,代理 URL 中不包含用户名和密码：

# Requests
proxies = {
    'http': 'http://proxy.example.com:8000',
    'https': 'http://proxy.example.com:8000'
}

# HTTPX
client = httpx.Client(proxy='http://proxy.example.com:8000')

# AIOHTTP
proxy = 'http://proxy.example.com:8000'
async with session.get(url, proxy=proxy) as response:
    pass

# Playwright
proxy={'server': 'http://proxy.example.com:8000'}

账密认证则必须在 URL 中内联或通过专用参数传递认证信息。部分动态代理允许在用户名后缀添加参数控制粘性会话或目标国家,格式因服务商而异,需参考控制台提取说明。

静态代理通常直连固定端口或独占地址,每个 IP 对应一个端口或独立的主机名,不需要在用户名中添加会话标识。

出口验证与异常重试

采集脚本启动或切换代理后,应首先请求回显接口验证出口 IP 是否符合预期：

import requests

def verify_proxy(proxies):
    try:
        response = requests.get(
            'https://httpbin.org/ip',
            proxies=proxies,
            timeout=10
        )
        origin_ip = response.json().get('origin')
        print(f"当前出口 IP: {origin_ip}")
        return origin_ip
    except requests.exceptions.ProxyError:
        print("代理连接失败")
        return None
    except requests.exceptions.ConnectTimeout:
        print("代理连接超时")
        return None

验证内容包括：

返回的 origin IP 是否与预期住宅出口归属地一致
地理位置是否符合任务目标国家或地区
静态 IP 是否与服务商提供的地址一致

对于可能因代理节点临时故障或目标站点限流导致的连接失败,应使用 try-except 捕获 ProxyError、ConnectTimeout 等异常,配合指数退避重试：

import time

def fetch_with_retry(url, proxies, max_retries=3):
    for attempt in range(max_retries):
        try:
            response = requests.get(url, proxies=proxies, timeout=10)
            return response
        except (requests.exceptions.ProxyError, requests.exceptions.ConnectTimeout) as e:
            wait_time = 2 ** attempt
            print(f"第 {attempt + 1} 次重试,等待 {wait_time} 秒")
            time.sleep(wait_time)
    raise Exception("超过最大重试次数")