Python 移动代理配置指南

export PROXY_URL="http://USERNAME-mbl-us-rot-auto10:[email protected]:7000"

# 同一个字符串同时适用于 http:// 和 https:// 目标——
# 客户端会自动为 TLS 建立 CONNECT 隧道。

import os
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

PROXY = os.environ["PROXY_URL"]

session = requests.Session()
session.proxies = {"http": PROXY, "https": PROXY}

retry = Retry(
    total=4,
    backoff_factor=1.5,                    # 1.5 秒、3 秒、6 秒、12 秒
    status_forcelist=(429, 500, 502, 503, 504),
    respect_retry_after_header=True,
    allowed_methods=frozenset(["GET", "HEAD"]),
)
# pool_maxsize 让隧道保持存活，而不是每个请求都重新连接。
session.mount("https://", HTTPAdapter(max_retries=retry, pool_maxsize=20))
session.mount("http://", HTTPAdapter(max_retries=retry, pool_maxsize=20))

# 在任务写入任何数据之前，先断言出口位置。
who = session.get("https://ipinfo.io/json", timeout=30).json()
print(f"exit {who['ip']} — {who['country']} — {who['org']}")

r = session.get("https://example.com", timeout=30)
print(r.status_code, len(r.content))

Session 加 pool_maxsize，就是"复用一条隧道"与"每个请求一条隧道"的区别。

import os, asyncio, httpx

PROXY = os.environ["PROXY_URL"]

# 同步
with httpx.Client(proxy=PROXY, timeout=30.0) as client:
    print(client.get("https://api.ipify.org?format=json").json())

# 异步——请限制并发数，否则每个任务都会打开一条隧道。
async def main(urls):
    limits = httpx.Limits(max_connections=10, max_keepalive_connections=10)
    sem = asyncio.Semaphore(10)
    async with httpx.AsyncClient(proxy=PROXY, limits=limits, timeout=30.0) as c:
        async def one(u):
            async with sem:
                r = await c.get(u)
                return u, r.status_code, len(r.content)
        return await asyncio.gather(*(one(u) for u in urls))

print(asyncio.run(main(["https://example.com"] * 5)))

httpx 0.26 及以上版本使用 proxy=，旧版本使用 proxies=。请显式限制并发数——不受限的异步请求正是客户端被限流的原因。

import os, asyncio, aiohttp

PROXY = os.environ["PROXY_URL"]

async def main():
    conn = aiohttp.TCPConnector(limit=10)
    async with aiohttp.ClientSession(connector=conn) as s:
        # aiohttp 的 proxy= 要写在每次调用上，而不是会话上。
        async with s.get("https://api.ipify.org?format=json", proxy=PROXY) as r:
            print(r.status, await r.json())

asyncio.run(main())

常见陷阱：在 ClientSession 上设置 proxy 不起作用——它必须设置在每个请求上。

import os, re

def with_country(url: str, cc: str) -> str:
    return re.sub(r"-(us|gb|de|fr|es|nl|pl|ch|pa|am|ge)-", f"-{cc}-", url, count=1)

def with_session(url: str, sid: str) -> str:
    if "-sid-" in url:
        return re.sub(r"-sid-[a-z0-9_]+", f"-sid-{sid}", url)
    return url.replace("-rot-", f"-sid-{sid}-rot-")

base = os.environ["PROXY_URL"]
print(with_country(base, "de"))
print(with_session(with_country(base, "fr"), "worker07"))

路由信息都在用户名字符串中，所以切换市场只是一次字符串替换，而不是重新部署。