代理
hs-net 内置代理归一化服务（ProxyService），统一处理 HTTP/SOCKS/认证代理，对外暴露纯 HTTP 本地代理。
章节概览
简单代理
直接传字符串，支持 HTTP、HTTPS、SOCKS4、SOCKS5 及认证格式：
simple_proxy.py
from hs_net import SyncNet
# HTTP 代理
with SyncNet(proxy="http://proxy-host:8080") as net:
resp = net.get("https://example.com")
# HTTPS 代理
with SyncNet(proxy="https://proxy-host:8443") as net:
resp = net.get("https://example.com")
# SOCKS5 代理
with SyncNet(proxy="socks5://proxy-host:1080") as net:
resp = net.get("https://example.com")
# SOCKS4 代理
with SyncNet(proxy="socks4://proxy-host:1080") as net:
resp = net.get("https://example.com")
# 带认证的代理
with SyncNet(proxy="http://user:password@proxy-host:8080") as net:
resp = net.get("https://example.com")
# 带认证的 SOCKS5
with SyncNet(proxy="socks5://user:password@proxy-host:1080") as net:
resp = net.get("https://example.com")
完整示例见 examples/proxy/fixed_proxy.py。
支持的代理格式
代理认证
代理认证通过 URL 中的 user:password@ 部分传递，不同协议的认证方式由 hs-net 自动处理：
HTTP/HTTPS 代理认证
使用 HTTP Basic Auth，hs-net 自动编码为 Proxy-Authorization 头：
http_auth.py
# 用户名 + 密码
with SyncNet(proxy="http://admin:secret123@proxy-host:8080") as net:
resp = net.get("https://example.com")
# HTTPS 代理同理
with SyncNet(proxy="https://admin:secret123@proxy-host:8443") as net:
resp = net.get("https://example.com")
SOCKS5 代理认证
SOCKS5 支持用户名/密码认证（RFC 1929），hs-net 自动完成认证握手：
socks5_auth.py
with SyncNet(proxy="socks5://myuser:mypass@proxy-host:1080") as net:
resp = net.get("https://example.com")
SOCKS4 代理认证
SOCKS4 仅支持 user_id（无密码），通过用户名字段传递：
socks4_auth.py
with SyncNet(proxy="socks4://myuser@proxy-host:1080") as net:
resp = net.get("https://example.com")
认证格式汇总
:::warning 特殊字符
密码中包含 @、:、/ 等特殊字符时，需要进行 URL 编码：
from urllib.parse import quote
password = quote("p@ss:word/123", safe="")
proxy = f"http://user:{password}@proxy-host:8080"
# => http://user:p%40ss%3Aword%2F123@proxy-host:8080
ProxyService
对于更复杂的代理需求，使用 ProxyService：
列表轮换
list_proxy.py
from hs_net import Net, ProxyService
# 轮询模式（默认）
proxy = ProxyService([
"http://proxy1:8080",
"socks5://proxy2:1080",
"http://user:pass@proxy3:8080",
], strategy="round_robin")
async with Net(proxy=proxy) as net:
# 每次 switch() 切换到下一个代理
resp = await net.get("https://example.com")
net.proxy_service.switch() # 切换到下一个
resp = await net.get("https://example.com")
random_proxy.py
# 随机模式
proxy = ProxyService([
"http://proxy1:8080",
"socks5://proxy2:1080",
], strategy="random")
with SyncNet(proxy=proxy) as net:
resp = net.get("https://example.com")
net.proxy_service.switch() # 随机切换
完整示例见 examples/proxy/switch_proxy.py。
自定义代理源
实现 ProxyProvider 接口，对接自己的代理源：
custom_provider.py
from hs_net import Net, ProxyService, ProxyProvider
class MyProvider(ProxyProvider):
"""从代理 API 获取代理地址。"""
def get_proxy(self) -> str:
# 同步获取代理
return "socks5://1.2.3.4:1080"
async def async_get_proxy(self) -> str:
# 异步获取代理（可选，默认回退到 get_proxy）
return "socks5://1.2.3.4:1080"
proxy = ProxyService(provider=MyProvider())
async with Net(proxy=proxy) as net:
resp = await net.get("https://example.com")
await net.proxy_service.async_switch() # 异步切换代理
完整示例见 examples/proxy/custom_provider.py。
同步 vs 异步
ProxyProvider 提供两个方法：
get_proxy() — 同步版本，必须实现async_get_proxy() — 异步版本，可选，默认回退到 get_proxy()
Net（异步客户端）启动和切换代理时调用 async_get_proxy()，SyncNet 调用 get_proxy()。
对于纯计算的 Provider（如 FixedProxyProvider、ListProxyProvider），只实现 get_proxy() 即可。
需要异步 I/O 时（如调用代理池 API），应覆写 async_get_proxy()。
代理池 API
使用内置的 ApiProxyProvider 对接代理池 API，自动获取代理地址：
api_proxy_simple.py
from hs_net import ApiProxyProvider, Net, ProxyService
# 最简单 — API 直接返回代理地址文本
provider = ApiProxyProvider("https://api.pool.com/get")
svc = ProxyService(provider=provider)
async with Net(proxy=svc) as net:
resp = await net.get("https://example.com")
从 JSON 响应中提取代理地址：
api_proxy_json.py
from httpx import Response as HttpxResponse
from hs_net import ApiProxyProvider, Net, ProxyService
def parse_proxy(resp: HttpxResponse) -> str:
data = resp.json()
item = data["data"][0]
return f"http://{item['ip']}:{item['port']}"
provider = ApiProxyProvider(
"https://api.pool.com/get?num=1",
parser=parse_proxy,
)
svc = ProxyService(provider=provider)
async with Net(proxy=svc) as net:
resp = await net.get("https://example.com")
await svc.async_switch() # 从 API 获取新代理并切换
访问代理池 API 需要翻墙时，指定 proxy 参数：
api_proxy_with_proxy.py
provider = ApiProxyProvider(
"https://api.overseas-pool.com/get",
proxy="http://127.0.0.1:7897", # 通过本地 Clash 访问 API
parser=parse_proxy,
)
ApiProxyProvider 参数：
完整示例见 examples/proxy/api_proxy.py。
代理链（中转代理）
通过中转代理访问上游代理，适用于上游代理无法直连的场景（如通过本地 Clash 中转访问海外代理）：
transit_proxy.py
from hs_net import Net, ProxyService
# 通过本地 Clash (127.0.0.1:7897) 中转访问海外 SOCKS5 代理
proxy = ProxyService(
"socks5://overseas-proxy:1080",
transit="http://127.0.0.1:7897",
)
async with Net(proxy=proxy) as net:
resp = await net.get("https://example.com")
流量路径：客户端 → 本地 Clash → 海外代理 → 目标网站
完整示例见 examples/proxy/transit_proxy.py。
域名路由
通过 rules 参数按域名路由到不同代理，支持通配符匹配和 "direct" 直连：
domain_rules.py
from hs_net import Net, ProxyService
svc = ProxyService(
"http://default-proxy:8080", # 未匹配域名走默认代理
rules={
"*.cn": "direct", # 国内网站直连
"*.google.com": "socks5://proxy1:1080", # Google 走 SOCKS5
"github.com": "http://proxy2:8080", # GitHub 走 HTTP 代理
},
)
async with Net(proxy=svc) as net:
await net.get("https://baidu.cn") # → 直连
await net.get("https://www.google.com") # → proxy1
await net.get("https://github.com") # → proxy2
await net.get("https://example.com") # → default-proxy
同步客户端同理：
domain_rules_sync.py
from hs_net import ProxyService, SyncNet
svc = ProxyService(
"http://default-proxy:8080",
rules={"*.cn": "direct"},
)
with SyncNet(proxy=svc) as net:
net.get("https://baidu.cn") # → 直连
net.get("https://example.com") # → default-proxy
域名匹配规则：
匹配优先级
规则按 dict 插入顺序匹配，先匹配先命中。如果多个规则都能匹配同一个域名，只有第一个生效。
switch() / async_switch() 在 rules 模式下只切换未匹配域名的默认代理，不影响已匹配的规则。
仅域名匹配
rules 只支持域名级别匹配，不支持路径（如 example.com/api）。原因是 HTTPS 流量走 CONNECT 隧道，路径在 TLS 加密内，代理层无法看到。
完整示例见 examples/proxy/domain_rules.py。
身份路由
通过 identity_extractor 参数，根据请求中的身份信息（cookies、headers、URL 参数等）自动为每个身份分配并绑定代理。同一身份始终使用同一个代理（sticky），不同身份使用不同代理。
典型场景：多账号并发采集，每个账号需要独立的代理 IP。
identity_routing.py
from hs_net import Net, ProxyService, RequestModel
from hs_net.proxy import ListProxyProvider
# 身份提取器：从请求中提取身份标识
def my_extractor(request: RequestModel) -> str | None:
if request.cookies and "session" in request.cookies:
return request.cookies["session"]
return None
# 代理池 + 身份提取器
svc = ProxyService(
provider=ListProxyProvider([
"http://proxy1:8080",
"http://proxy2:8080",
"http://proxy3:8080",
]),
identity_extractor=my_extractor,
)
async with Net(proxy=svc) as net:
# 同一身份 → 同一代理（sticky）
await net.get(url, cookies={"session": "user_a"}) # → proxy1
await net.get(url, cookies={"session": "user_a"}) # → proxy1
# 不同身份 → 不同代理
await net.get(url, cookies={"session": "user_b"}) # → proxy2
# 无身份 → 默认代理
await net.get(url)
身份提取器的签名为 Callable[[RequestModel], str | None]，可以从请求的任意字段提取身份：
extractor_examples.py
# 从 headers 提取
def from_header(req: RequestModel) -> str | None:
return (req.headers or {}).get("Authorization")
# 从 URL 参数提取
def from_params(req: RequestModel) -> str | None:
return (req.url_params or {}).get("token")
# 从 cookies 提取
def from_cookie(req: RequestModel) -> str | None:
return (req.cookies or {}).get("session")
身份路由可以和域名路由、代理链一起使用：
identity_with_rules.py
svc = ProxyService(
provider=ListProxyProvider(proxies),
identity_extractor=my_extractor,
transit="http://127.0.0.1:7897", # 代理链
rules={"*.cn": "direct"}, # 国内直连
)
代理选择优先级
域名路由（rules）> 身份路由（identity_extractor）> 默认上游代理。
即使请求带有身份标识，如果域名命中了 rules 规则，仍然走规则指定的代理或直连。
完整示例见 examples/proxy/identity_routing.py。
异步代理
异步客户端搭配 ProxyService 使用，循环验证所有异步引擎均支持代理：
async_proxy.py
import asyncio
from hs_net import EngineEnum, Net, ProxyService
PROXY = "http://127.0.0.1:8080"
TEST_URL = "http://ip-api.com/json/"
ASYNC_ENGINES = [
EngineEnum.HTTPX,
EngineEnum.AIOHTTP,
EngineEnum.CURL_CFFI,
EngineEnum.REQUESTS_GO,
]
async def main():
svc = ProxyService(PROXY)
for engine in ASYNC_ENGINES:
async with Net(proxy=svc, engine=engine, retries=0, timeout=30) as net:
resp = await net.get(TEST_URL)
ip = resp.json_data["query"]
print(f"{engine.value:12s} -> IP: {ip}")
if __name__ == "__main__":
asyncio.run(main())
完整示例见 examples/proxy/async_proxy.py。
代理工作原理
当使用 ProxyService 时，hs-net 会在本地启动一个 HTTP 代理服务器（127.0.0.1 随机端口），自动处理：
- 协议转换：将 SOCKS4/SOCKS5 代理统一为 HTTP 代理
- 认证处理：自动注入代理认证头
- 代理链：支持通过中转代理连接上游
- 代理池 API：内置
ApiProxyProvider，支持同步/异步获取代理
- 域名路由：通过
rules 参数按域名匹配不同代理或直连
- 身份路由：通过
identity_extractor 按请求身份自动分配并 sticky 绑定代理
- 热切换：调用
switch() / async_switch() 即可切换上游代理，无需重建连接
所有引擎只需连接本地 HTTP 代理，无需各自处理 SOCKS 协议和认证。
ProxyService 生命周期
当你传入 proxy="http://..." 字符串时，Net/SyncNet 内部创建 ProxyService 并在 close() 时自动停止。
当你传入 proxy=svc（ProxyService 实例）时，close() 不会停止它，你可以跨多个客户端复用同一个 ProxyService：
reuse_proxy_service.py
from hs_net import EngineEnum, ProxyService, SyncNet
svc = ProxyService("socks5://proxy:1080")
# 同一个 ProxyService 跨多个引擎复用
for engine in [EngineEnum.HTTPX, EngineEnum.CURL_CFFI]:
with SyncNet(proxy=svc, engine=engine) as net:
resp = net.get("https://example.com")
svc.switch() # close() 不会停止 svc，switch 正常工作
# ProxyService 出作用域后自动清理
