hs-net

统一多引擎的增强型 HTTP 客户端，内置重试、选择器、信号中间件，同步异步全支持。

特性

• 多引擎切换 — httpx、aiohttp、curl-cffi、requests、requests-go，统一 API，一行切换
• 同步 & 异步 — Net（异步）和 SyncNet（同步），接口完全一致
• 智能选择器 — 内置 CSS、XPath、正则、JMESPath 四种数据提取（可选安装）
• 流式响应 — 支持分块下载大文件，所有引擎统一接口
• 自动重试 — 基于 tenacity，可配置次数、间隔、随机抖动
• 信号中间件 — 请求前、响应后、重试时三个钩子
• 统一异常 — 超时、连接失败、状态码错误等统一映射，切换引擎不影响错误处理
• 反爬支持 — curl-cffi 浏览器 TLS 指纹模拟 + 内置真实 User-Agent 池（浏览器 / 搜索引擎爬虫 / 微信）
• 轻量核心 — 核心仅依赖 httpx + tenacity，爬虫增强按需安装

安装

# 核心安装（默认 httpx 引擎，仅 2 个依赖）
pip install hs-net

# 爬虫增强（CSS/XPath 选择器 + JMESPath + 速率限制）
pip install hs-net[sp]

# 按需安装额外引擎
pip install hs-net[aiohttp]      # aiohttp 引擎
pip install hs-net[curl]         # curl-cffi 引擎（浏览器指纹模拟）
pip install hs-net[requests]     # requests 引擎
pip install hs-net[requests-go]  # requests-go 引擎
pip install hs-net[all]          # 全部引擎 + 爬虫增强

Python >= 3.10。默认安装仅包含 httpx + tenacity，选择器需安装 [sp]。User-Agent 池是内置的，无需额外依赖。

快速开始

异步

import asyncio
from hs_net import Net

async def main():
    async with Net() as net:
        resp = await net.get("https://example.com")
        print(resp.text)         # 纯 HTTP 客户端，无需额外依赖
        print(resp.json_data)    # JSON 响应自动解析

asyncio.run(main())

同步

from hs_net import SyncNet

with SyncNet() as net:
    resp = net.get("https://example.com")
    print(resp.status_code)  # 200
    print(resp.text[:100])   # 响应文本

数据提取（需要 pip install hs-net[sp]）

with SyncNet() as net:
    resp = net.get("https://example.com")
    resp.css("title::text").get()               # CSS 选择器
    resp.xpath("//h1/text()").get()             # XPath
    resp.re_first(r"价格: (\d+)元")             # 正则

    resp = net.get("https://api.example.com")
    resp.jmespath("data[?age > `18`].name")     # JMESPath（JSON）

引擎对比

引擎切换

# httpx（默认）
Net(engine="httpx")

# aiohttp
Net(engine="aiohttp")

# curl-cffi（支持浏览器指纹模拟）
Net(engine="curl_cffi", engine_options={"impersonate": "chrome"})

# requests（仅同步）
SyncNet(engine="requests")

# requests-go
Net(engine="requests_go")

快捷函数（无需实例化）

import hs_net

# 异步
resp = await hs_net.get("https://example.com")
resp = await hs_net.post("https://api.example.com/data", json_data={"key": "val"})

# 同步
resp = hs_net.sync_get("https://example.com")

# 指定引擎
resp = await hs_net.get("https://example.com", engine="curl_cffi")

快捷函数每次创建临时客户端，适合简单请求。需要复用连接、配置中间件时请使用 Net / SyncNet。

流式响应

分块下载大文件，不占内存：

# 异步
async with Net() as net:
    resp = await net.stream("GET", "https://example.com/large-file.zip")
    async with resp:
        async for chunk in resp:
            f.write(chunk)

# 同步
with SyncNet() as net:
    with net.stream("GET", "https://example.com/large-file.zip") as resp:
        for chunk in resp:
            f.write(chunk)

配置

from hs_net import Net, NetConfig

# 方式 1：构造函数参数
net = Net(
    engine="httpx",
    base_url="https://api.example.com/v1",
    timeout=30.0,
    retries=5,
    retry_delay=1.0,
    user_agent="chrome",
    proxy="http://127.0.0.1:7890",
    verify=False,
    raise_status=True,
    allow_redirects=True,
    concurrency=10,
    headers={"Accept-Language": "zh-CN"},
    cookies={"token": "abc123"},
    engine_options={"http2": True},
)

# 方式 2：NetConfig 对象
config = NetConfig(
    engine="curl_cffi",
    retries=3,
    user_agent="random",
    headers={"Authorization": "Bearer token"},
    engine_options={"impersonate": "chrome"},
)
net = Net(config=config)

每次请求也可以覆盖全局配置：

async with Net(timeout=10, retries=3) as net:
    # 这次请求用不同的超时、代理、UA
    resp = await net.get(
        "https://example.com",
        params={"q": "python"},
        timeout=30.0,
        proxy="http://127.0.0.1:7890",
        user_agent="MyBot/1.0",
        headers={"X-Custom": "value"},
        cookies={"session": "xyz"},
        verify=False,
        retries=5,
        retry_delay=1.0,
        raise_status=False,
        allow_redirects=True,
    )

    # POST 还支持 json_data / form_data / files
    resp = await net.post(
        "https://api.example.com/upload",
        json_data={"key": "value"},
        # form_data={"field": "value"},
        # files={"file": ("name.txt", b"content", "text/plain")},
    )

参数优先级：请求方法参数 > 构造函数参数 > NetConfig 默认值

User-Agent

默认就是当前版本的桌面 Chrome，不用配置。要换用内置的真实 UA 池，传快捷方式即可（不区分大小写）：

Net(user_agent="random")       # 任意桌面浏览器
Net(user_agent="chrome")       # 指定浏览器：chrome / firefox / edge / safari
Net(user_agent="mobile")       # 任意移动端浏览器
Net(user_agent="android")      # 指定平台：android / ios
Net(user_agent="bot")          # 任意搜索引擎爬虫
Net(user_agent="baiduspider")  # 指定爬虫，见下表
Net(user_agent="wechat")       # 微信内置浏览器
Net(user_agent="MyBot/1.0")    # 完整字符串则原样使用

random 和 mobile 是桌面、移动端各自的随机入口。爬虫和微信有特定用途，需要显式指定，不会被随机取到。不在上表中的字符串一律当作自定义 UA 使用。

from hs_net import DEFAULT_USER_AGENT, user_agent_shortcuts

DEFAULT_USER_AGENT      # 当前默认 UA
user_agent_shortcuts()  # 列出全部快捷方式（含别名）

UA 数据内置在包里（约 13 KB，无额外依赖），由 scripts/gen_ua_data.py 从四个上游源合并生成：

信号中间件

async with Net() as net:

    @net.on_request_before
    async def add_auth(req_data):
        req_data.headers["Authorization"] = "Bearer token"
        return req_data

    @net.on_response_after
    async def log_response(resp):
        print(f"{resp.status_code} {resp.url}")

    @net.on_request_retry
    async def on_retry(exc):
        print(f"重试: {exc}")

    resp = await net.get("https://example.com")

错误处理

所有引擎的异常统一映射，切换引擎不影响错误处理代码：

from hs_net import (
    Net, StatusException, TimeoutException,
    ConnectionException, RetryExhausted, RequestException,
)

async with Net() as net:
    try:
        resp = await net.get("https://httpbin.org/status/404")
    except TimeoutException as e:
        print(f"超时: timeout={e.timeout}s")
    except ConnectionException as e:
        print(f"连接失败: {e.url}")
    except RetryExhausted as e:
        print(f"{e.attempts} 次重试失败: {e.last_exception}")
    except StatusException as e:
        print(f"HTTP {e.code}")
    except RequestException as e:
        print(f"请求异常: {e}")

文档

完整文档见 docs/ 目录，本地预览：

cd docs
pnpm install
pnpm run dev

License

MIT