如何在 Python 中使用住宅代理(requests、httpx、Scrapy)

一份在 Python 中使用住宅代理的实战指南:认证、轮换 vs sticky IP、地理定位、重试,以及 requests、httpx、Scrapy 的完整示例。

把一个住宅代理接进 Python 脚本,一旦你见过一次怎么做,就是五分钟的活。摩擦从来不在代码,而在那些没人写下来的细节:凭证如何编码定位、什么时候轮换 vs 钉住一个 IP、如何处理代理特有的错误,以及 requests、httpx、Scrapy 之间那些细小的库差异。

这是我当初希望能有的一份指南。给你真正能跑的复制粘贴示例,覆盖你最可能在用的三个库,再加上那些能把一段能跑的片段变成一个挺得过生产的爬虫的部分。

下面的一切都用 Shifter 住宅网关:一个端点,p.shifter.io:443,所有定位都编码在用户名里。如果你用的是另一家供应商,形状是一样的;换掉主机和凭证格式即可。

你首先必须理解的那一件事

在住宅网关上,代理用户名同时承载你的认证和你的定位。你不是换端点来切换国家或会话,而是改用户名这个字符串。一个用户名长这样:

customer-USERNAME-country-us-sid-abc123-ttl-600

从左往右读:账号 id,然后是 flag。country-us 定位到美国。sid-abc123 钉住一个 sticky 会话。ttl-600 把那个 IP 保持 600 秒。去掉 sid/ttl,每个请求就轮换到一个新 IP。密码是固定的。这就是全部的心智模型,剩下的只是把这个字符串塞进每个库的代理槽位。

把凭证放在环境变量里,绝不要硬编码:

import os
USER = os.environ["SHIFTER_USER"]   # 你的账号用户名
PASS = os.environ["SHIFTER_PASS"]
GATEWAY = "p.shifter.io:443"

requests:6 行版本

requests 接受一个 proxies 字典。http 和 https 两个键都指向同一个 http:// 代理 URL,这是对的,requests 会通过 CONNECT 把 HTTPS 经它隧道转发。

import os, requests

USER = os.environ["SHIFTER_USER"]
PASS = os.environ["SHIFTER_PASS"]
GATEWAY = "p.shifter.io:443"

def proxy(country="us"):
    url = f"http://{USER}-country-{country}:{PASS}@{GATEWAY}"
    return {"http": url, "https": url}

r = requests.get("https://api.ipify.org?format=json", proxies=proxy("us"), timeout=30)
print(r.json())   # {'ip': '<一个美国住宅 IP>'}

跑两次,你会拿到两个不同的 IP,因为没有 sid,每个请求都轮换。这是默认行为,对大多数抓取来说,正是你想要的。

轮换 vs sticky,落到代码上

这是让人栽跟头的区别,所以这里讲具体。(概念版见sticky vs 轮换住宅代理。)

轮换(每个请求一个新 IP)是默认,只要省掉 sid:

for _ in range(3):
    r = requests.get("https://api.ipify.org", proxies=proxy("us"), timeout=30)
    print(r.text)   # 三个不同的 IP

Sticky(跨多个请求保持同一个 IP)需要在用户名里加一个会话 id 和一个 TTL。当一个流程横跨多个必须看起来像同一个用户的请求时用它——先登录,然后是它后面的页面:

def sticky_proxy(country="us", session="s1", ttl=600):
    url = f"http://{USER}-country-{country}-sid-{session}-ttl-{ttl}:{PASS}@{GATEWAY}"
    return {"http": url, "https": url}

s = requests.Session()
p = sticky_proxy(session="checkout-42", ttl=600)
for path in ("/login", "/cart", "/checkout"):
    r = s.get(f"https://shop.example{path}", proxies=p, timeout=30)
    # 三个请求共享一个 IP,最多 600 秒

会话 id 由你自己选,每个逻辑会话用任意一个唯一字符串。同一个字符串在 TTL 过期前返回同一个 IP,之后会在保持你其它 flag 的前提下轮换。

地理定位

因为定位住在用户名里,地理就只是另一个 flag。国家是常用的;州、城市和 ASN 用法一样:

def geo_proxy(country, city=None):
    parts = [USER, "country", country]
    if city:
        parts += ["city", city.lower().replace(" ", "_")]
    url = f"http://{'-'.join(parts)}:{PASS}@{GATEWAY}"
    return {"http": url, "https": url}

requests.get("https://example.com", proxies=geo_proxy("de"))              # 德国
requests.get("https://example.com", proxies=geo_proxy("us", "new york"))  # 纽约市

城市名用下划线代替空格(new_york)。flag 可以自由组合;顺序对网关来说无所谓。

httpx:同样的思路,支持 async

httpx 是你想要 async 或 HTTP/2 时的现代之选。代理设在 client 上。注意:在当前的 httpx 里参数是 proxy=(单数);旧版本用的是 proxies=。

import os, httpx

USER = os.environ["SHIFTER_USER"]
PASS = os.environ["SHIFTER_PASS"]
GATEWAY = "p.shifter.io:443"

def proxy_url(country="us"):
    return f"http://{USER}-country-{country}:{PASS}@{GATEWAY}"

# 同步
with httpx.Client(proxy=proxy_url("us"), timeout=30) as client:
    print(client.get("https://api.ipify.org").text)

async 版本才是 httpx 真正发挥价值的地方——把许多请求并发铺开,每个都经一个新的轮换 IP:

import asyncio, httpx

async def fetch(client, url):
    r = await client.get(url, timeout=30)
    return r.status_code, r.text[:80]

async def main(urls):
    async with httpx.AsyncClient(proxy=proxy_url("us")) as client:
        return await asyncio.gather(*(fetch(client, u) for u in urls))

urls = ["https://api.ipify.org"] * 10
print(asyncio.run(main(urls)))   # 10 个并发请求,轮换 IP

这就是十几行里做出十个并发的住宅请求。留意你的并发——更多不一定更快,而且用一阵 IP 去猛锤一个目标,仍然可能触发行为检测。

Scrapy:一个代理中间件

Scrapy 是做大规模 crawl 的重量级选手。挂代理的干净办法是按请求设置 request.meta["proxy"],Scrapy 内置的 HttpProxyMiddleware 会读它,并从 URL 里的凭证处理 Proxy-Authorization 头。

一个轮换地理、给每个请求一个新 IP 的小中间件:

import os

class ResidentialProxyMiddleware:
    def __init__(self):
        self.user = os.environ["SHIFTER_USER"]
        self.password = os.environ["SHIFTER_PASS"]
        self.gateway = "p.shifter.io:443"

    def process_request(self, request, spider):
        country = request.meta.get("country", "us")
        request.meta["proxy"] = (
            f"http://{self.user}-country-{country}:"
            f"{self.password}@{self.gateway}"
        )

在 settings.py 里启用它(它必须在默认代理中间件之前运行):

DOWNLOADER_MIDDLEWARES = {
    "myproject.middlewares.ResidentialProxyMiddleware": 350,
    "scrapy.downloadermiddlewares.httpproxy.HttpProxyMiddleware": 400,
}

现在任何 spider 都经住宅 IP 路由,你还可以用 Request(url, meta={"country": "gb"}) 按请求定位。要做 sticky 会话,就像 requests 示例那样用一个 sid/ttl 构造用户名,并把它设到 meta["proxy"] 上。

处理那些真的会发生的错误

一个无视代理错误的爬虫,在演示里能跑,过一夜就挂了。三个你会遇到的:

407 Proxy Authentication Required——用户名/密码错了,或者用了一个不被识别的 flag(country 或 asn 拼错了)。修凭证字符串;重试没用。
429 Too Many Requests——是目标在给你限速。退避、放慢、轮换 IP(你不做 sticky 时网关本来就按请求在轮换)。
502 Bad Gateway——当前没有 IP 匹配你的过滤条件(通常是 country+city+asn 这种地理约束太紧)。放宽一个 flag 再试。

一个会退避、并干净放弃的最小重试包装:

import time, requests

def get_with_retry(url, proxies, tries=4):
    for attempt in range(tries):
        try:
            r = requests.get(url, proxies=proxies, timeout=30)
            if r.status_code in (429, 502, 503):
                raise requests.exceptions.RequestException(f"status {r.status_code}")
            r.raise_for_status()
            return r
        except requests.exceptions.RequestException as e:
            if attempt == tries - 1:
                raise
            sleep = 2 ** attempt           # 1s, 2s, 4s
            print(f"retry {attempt+1}: {e}; sleeping {sleep}s")
            time.sleep(sleep)

指数退避加上按请求轮换,能处理绝大多数瞬时失败。如果你是持续而不是断续地被封,那问题就不在重试,而在 IP 质量或请求行为,见抓取时如何避免被封。

验证它确实在工作

在你信任一份配置之前,确认两件事:IP 在变(轮换),并且它在正确的国家(地理)。一个快速检查:

import requests

r = requests.get("http://ip-api.com/json", proxies=proxy("de"), timeout=30)
data = r.json()
print(data["query"], data["countryCode"])   # 期望一个 DE IP

不带 sid 调用几次,你应该看到不同的 IP,全都是 DE。如果国家不对,检查你的 flag 拼写。如果 IP 从不变化,说明你在用户名里不小心留了一个 sid。

关于 SOCKS5 的一点说明

上面的一切都用 HTTP 代理,这是网页抓取的正确默认。如果你的工作负载需要 SOCKS5(非 HTTP 流量,或一个期望它的工具),同一个网关也会说它——把协议方案改成 socks5h://,并安装 requests[socks](或用 httpx 的 SOCKS extra)。取舍在HTTP vs SOCKS5 代理里;对普通抓取,就用 HTTP。

常见问题

每个国家都要用一个不同的端点吗? 不。一个端点,p.shifter.io:443,管所有的。国家、城市、会话都通过用户名字符串切换,而不是主机。这是网关模型的核心。

为什么在 requests 里 http 和 https 两个键都设成一个 http:// URL? 因为 requests 通过一个 CONNECT 隧道、经一个 HTTP 代理发送 HTTPS。代理 URL 的协议方案描述的是你怎么跟代理说话(HTTP),而不是跟目标。这是正确且标准的,别把 https 键设成 https://。

怎么让每个请求都轮换 IP? 在用户名里省掉 sid。没有会话 id,网关会自动每个请求给你一个新 IP。你不用管一个代理列表,池子的轮换在服务端。

怎么在一个多步骤流程里保持同一个 IP? 在用户名里加 sid-<你的 id>-ttl-<秒数>,并在流程的每个请求里复用它。同一个 id,同一个 IP,直到 TTL 过期。

我的请求很慢。是代理的问题吗? 通常不是。住宅 IP 相比直连会加一些延迟,但规模上的慢,更常见的是并发太高、没复用连接(用一个 Session/Client)、或者目标本身慢。先做性能分析,再来怪代理。

这对 aiohttp、urllib3 或 pycurl 也适用吗? 适用。任何接受一个带认证的 http://user:pass@host:port 代理 URL 的客户端都行,编码在凭证里的定位是完全一样的。requests、httpx、Scrapy 只是最常见的三个。

收尾

每个库里的模式都一样:构造 http://USER-flags:PASS@p.shifter.io:443 这个 URL,把它放进代理槽位,省掉 sid 就轮换、加上就钉住。地理是一个 flag,错误是一个小重试循环,并发就是你的客户端本来就在做的事。

从上面的片段开始,把它们指向住宅网关,你就能在一个下午里写出产生产形态的代理代码。计划和每 GB 价格在定价页,完整的 flag 参考在网关文档。