feat(http): use httpx with pooled connections and browser headers

Add httpx and route Bilibili and Telegram requests through pooled clients with keepalive and explicit timeouts. Align Bilibili headers with data/raw-request.txt: Chrome 155 UA, sec-ch-ua, space.bilibili.com origin, folder-specific referer, and priority header. Close replaced clients when credentials change. Add transport contract tests and live read-only validation for nav, folder sync, WBI search, and Telegram getUpdates.

DEVELOPMENT.md

@@ -15,19 +15,19 @@
 | 开发语言 | Python 3.14 | 与现有 `favsort.py` 一致，复用 Cookie、扫码授权与收藏夹排序能力。 |
 | 包与环境管理 | `uv` | 管理 Python 版本、`.venv`、依赖同步及统一命令。 |
 | 后端服务 | Python 标准库 `ThreadingHTTPServer` | 个人单用户看板，接口与并发规模有限，不引入 Web 框架。 |
-| HTTP 客户端 | `urllib.request` | 现有 B 站请求代码已使用；无需额外依赖。 |
+| HTTP 客户端 | `httpx` | 连接池、keepalive、明确的 connect/read 超时；B 站请求头与 UA 对齐 `data/raw-request.txt`。 |
 | 数据库 | SQLite（WAL） | 单用户、单机、低并发写入；备份和迁移成本低。 |
 | 调度 | `threading`、`queue.Queue`、标准库时间工具 | 单进程内运行发现、同步、快照和清理任务。 |
 | 前端 | 原生 HTML/CSS/JavaScript | 无 Node.js、无构建链；前端仅渲染 `/api/*` JSON。 |
 | 图表 | 仓库内置 ECharts | 离线静态资源，Apache-2.0 许可。 |
-| 通知 | Telegram Bot long polling | 具备按钮回调能力，开发和本机运行不需要公网 URL。 |
+| 通知 | Telegram Bot long polling + `httpx` | 具备按钮回调能力，开发和本机运行不需要公网 URL。 |
 | 容器化 | 可选的后续部署方式 | 本地开发、调试和日常运行均不要求 Docker。 |
 
 ### 1.1 选型原则
 
 1. **复用现有能力。** 当前仓库的 `favsort.py` 已实现 B 站设备指纹 Cookie、扫码登录、收藏夹读取和排序逻辑；新功能应拆分复用，而不是另起技术栈。
 2. **保持单体。** 当前规模约数百到数千视频，每日采集量约千级请求；单进程、SQLite 和内置调度足够。
-3. **尽量标准库。** 当前没有第三方 Python 依赖，第一期不因便利性引入 Web 框架、ORM、任务队列或 Redis。
+3. **标准库优先，成熟依赖可用。** 当依赖能明确改善可靠性、性能或可维护性时允许引入；必须锁定在 `pyproject.toml` / `uv.lock`，隔离在适配层，并用契约测试覆盖。
 4. **本地优先。** 可在 Windows 上通过 `uv` 直接运行，不要求 WSL、Docker、数据库服务或宿主机 cron。
 5. **未来可部署。** 服务保持无状态代码与可挂载数据目录分离，未来可选用 Docker 与 Caddy/Nginx 部署，但不是开发前置条件。
 
@@ -344,6 +344,17 @@ PRAGMA synchronous = NORMAL;
 | 加入收藏 | `x/v3/fav/resource/deal` | 确认候选后收藏 |
 | 收藏夹排序 | `x/v3/fav/resource/sort` | 定位至同 UP 分组 |
 
+B 站请求统一使用 `httpx` 连接池，并按 `data/raw-request.txt` 对齐浏览器特征：
+
+- `user-agent`：Chrome 155 / Windows；
+- `sec-ch-ua`、`sec-ch-ua-mobile`、`sec-ch-ua-platform`；
+- `origin: https://space.bilibili.com`；
+- `referer`：收藏夹请求指向 `https://space.bilibili.com/<mid>/favlist?fid=<media_id>&ftype=create`；
+- `priority: u=1, i`；
+- 匿名请求仍剥离账号身份 Cookie，仅保留设备指纹。
+
+请求头集中在 `bilibili/auth.py::make_headers`，不要在业务层散落自定义 Header。
+
 ### 7.2 发现任务
 
 每小时扫描两个关键词、每个关键词前两页：

README.md

@@ -258,6 +258,8 @@ uv run python app.py auth --account read --timeout 180
 
 只读客户端对 `ids` / `infos` 默认剥离 `read` 身份 Cookie，仅保留该账号设备指纹；如果失败才使用完整 `read` Cookie 重试。写操作一律使用完整 `write` Cookie 与 CSRF。
 
+B 站请求通过 `httpx` 连接池发送，并按 `data/raw-request.txt` 对齐 Chrome 155 的 UA、`sec-ch-ua`、`origin`、`referer` 和 `priority` 等请求头；收藏夹请求的 `referer` 会带上目标 `media_id`。
+
 `x/v3/fav/resource/sort` 的实测语义为：
 
 ```text
@@ -278,6 +280,7 @@ uv run python tests/test_migrations.py      # 迁移、备份、回滚与保留
 uv run python tests/test_database.py        # 单写队列、事务与并发
 uv run python tests/test_repository.py      # 收藏夹/视频状态机与黑名单
 uv run python tests/test_bilibili_client.py # ids/infos/view/fans/nav 契约与重试
+uv run python tests/test_httpx_transport.py # httpx B 站/Telegram 传输层
 uv run python tests/test_classify.py        # 标题归一化、时长解析与打分
 uv run python tests/test_search.py          # WBI 签名、搜索解析与降级
 uv run python tests/test_candidates.py      # 候选池、去重、通知与 fav_action

pyproject.toml

@@ -10,7 +10,9 @@ readme = "README.md"
 requires-python = ">=3.14,<3.15"
 license = { text = "MIT" }
 authors = [{ name = "bili-fav contributors" }]
-dependencies = []
+dependencies = [
+    "httpx>=0.28.1",
+]
 
 [project.scripts]
 bili-fav = "favsort:main"

reaction_board/bilibili/auth.py

@@ -45,15 +45,16 @@ def strip_identity(cookie: str) -> str:
     return "; ".join(kept)
 
 
-def make_headers(cfg: dict, auth: bool = False) -> dict[str, str]:
+def make_headers(cfg: dict, auth: bool = False, *, referer: str | None = None) -> dict[str, str]:
-    """构造 B 站 Web 请求头；``auth=False`` 时去掉账号身份 cookie。"""
+    """构造与 ``data/raw-request.txt`` 对齐的 B 站 Web 请求头。"""
     return {
         "accept": "*/*",
         "accept-language": "zh-CN,zh;q=0.9,en-US;q=0.8,en;q=0.7",
         "cookie": cfg["cookie"] if auth else strip_identity(cfg["cookie"]),
         "dnt": "1",
-        "origin": "https://www.bilibili.com",
+        "origin": "https://space.bilibili.com",
-        "referer": "https://www.bilibili.com/",
+        "priority": "u=1, i",
+        "referer": referer or "https://space.bilibili.com/",
         "sec-ch-ua": '"Google Chrome";v="155", "Chromium";v="155", "Not(A:Brand";v="24"',
         "sec-ch-ua-mobile": "?0",
         "sec-ch-ua-platform": '"Windows"',

reaction_board/bilibili/client.py

@@ -17,6 +17,8 @@ import urllib.request
 from dataclasses import dataclass
 from typing import Any, Mapping, Protocol, Sequence
 
+import httpx
+
 from .auth import make_headers
 from .models import (
     FanStat,
@@ -129,6 +131,54 @@ class UrllibTransport:
             raise TransportError(str(exc)) from exc
 
 
+class HttpxTransport:
+    """基于 httpx 的连接池传输，默认用于 B 站请求。"""
+
+    def __init__(
+        self,
+        *,
+        timeout: float = 30,
+        client: httpx.Client | None = None,
+    ) -> None:
+        self.timeout = timeout
+        self._client = client or httpx.Client(
+            timeout=httpx.Timeout(timeout, connect=10),
+            limits=httpx.Limits(
+                max_keepalive_connections=10,
+                keepalive_expiry=30,
+            ),
+            follow_redirects=False,
+            trust_env=True,
+        )
+
+    def request(
+        self,
+        method: str,
+        url: str,
+        headers: Mapping[str, str],
+        data: bytes | None = None,
+        timeout: float = 30,
+    ) -> HTTPResponse:
+        try:
+            response = self._client.request(
+                method,
+                url,
+                headers=dict(headers),
+                content=data,
+                timeout=timeout,
+            )
+        except httpx.TransportError as exc:
+            raise TransportError(str(exc)) from exc
+        return HTTPResponse(
+            status=int(response.status_code),
+            headers=dict(response.headers),
+            body=response.content,
+        )
+
+    def close(self) -> None:
+        self._client.close()
+
+
 class Pacer:
     """请求间随机等待；第一个请求不等待。"""
 
@@ -185,7 +235,7 @@ class BilibiliClient:
         self.cookie = cookie or ""
         self.csrf = csrf or ""
         self.mid = mid
-        self.transport = transport or UrllibTransport()
+        self.transport = transport or HttpxTransport()
         self.pacer = pacer or Pacer(sleep=sleeper)
         self._sleeper = sleeper
         self.retries = retries
@@ -193,15 +243,22 @@ class BilibiliClient:
         self.timeout = timeout
         self.logger = logger
 
+    def close(self) -> None:
+        close = getattr(self.transport, "close", None)
+        if callable(close):
+            close()
+
     # ------------------------------------------------------------ core
 
-    def _headers(self, auth: bool) -> dict[str, str]:
+    def _headers(self, auth: bool, referer: str | None = None) -> dict[str, str]:
-        return make_headers({"cookie": self.cookie}, auth=auth)
+        return make_headers({"cookie": self.cookie}, auth=auth, referer=referer)
 
-    def _request_json(self, url: str, *, auth: bool) -> dict[str, Any]:
+    def _request_json(
+        self, url: str, *, auth: bool, referer: str | None = None
+    ) -> dict[str, Any]:
         self.pacer.wait()
         resp = self.transport.request(
-            "GET", url, self._headers(auth), data=None, timeout=self.timeout
+            "GET", url, self._headers(auth, referer), data=None, timeout=self.timeout
         )
         if resp.status >= 400:
             raise BilibiliHTTPError(resp.status, resp.body.decode("utf-8", "replace")[:200])
@@ -210,11 +267,13 @@ class BilibiliClient:
             raise BilibiliProtocolError("接口响应顶层不是 JSON 对象")
         return payload
 
-    def _request_with_retry(self, url: str, *, auth: bool) -> dict[str, Any]:
+    def _request_with_retry(
+        self, url: str, *, auth: bool, referer: str | None = None
+    ) -> dict[str, Any]:
         last: BilibiliError | None = None
         for attempt in range(self.retries):
             try:
-                return self._request_json(url, auth=auth)
+                return self._request_json(url, auth=auth, referer=referer)
             except (TransportError, BilibiliHTTPError) as exc:
                 last = exc
                 if not exc.retryable or attempt >= self.retries - 1:
@@ -238,17 +297,19 @@ class BilibiliClient:
             raise BilibiliProtocolError("接口响应缺少整数 code")
         return BilibiliAPIError(code, str(payload.get("message") or ""))
 
-    def get_json(self, url: str, *, auth: bool = False) -> dict[str, Any]:
+    def get_json(
+        self, url: str, *, auth: bool = False, referer: str | None = None
+    ) -> dict[str, Any]:
         """GET JSON；匿名失败时自动带完整 Cookie 重试一次（与 favsort 一致）。"""
         if auth:
-            payload = self._request_with_retry(url, auth=True)
+            payload = self._request_with_retry(url, auth=True, referer=referer)
             if payload.get("code") != 0:
                 raise self._api_error(payload)
             return payload
 
         anon_error: BilibiliError | None = None
         try:
-            payload = self._request_with_retry(url, auth=False)
+            payload = self._request_with_retry(url, auth=False, referer=referer)
             if payload.get("code") == 0:
                 return payload
             anon_error = self._api_error(payload)
@@ -257,7 +318,7 @@ class BilibiliClient:
 
         if self.logger is not None and anon_error is not None:
             self.logger.debug("bili.anon_fallback", error=str(anon_error))
-        payload = self._request_with_retry(url, auth=True)
+        payload = self._request_with_retry(url, auth=True, referer=referer)
         if payload.get("code") != 0:
             raise self._api_error(payload)
         return payload
@@ -282,7 +343,11 @@ class BilibiliClient:
         url = (
             f"{BASE}/x/v3/fav/resource/ids?media_id={int(media_id)}&platform=web"
         )
-        payload = self.get_json(url)
+        referer = (
+            f"https://space.bilibili.com/{self.mid}/favlist"
+            f"?fid={int(media_id)}&ftype=create"
+        )
+        payload = self.get_json(url, referer=referer)
         data = self._data_list(payload, "fav/resource/ids")
         out: list[FavId] = []
         for index, item in enumerate(data):
@@ -321,6 +386,10 @@ class BilibiliClient:
             return InfosResult(requested=(), items={}, missing=())
 
         fmid = self.mid if folder_mid is None else int(folder_mid)
+        referer = (
+            f"https://space.bilibili.com/{fmid}/favlist"
+            f"?fid={int(media_id)}&ftype=create"
+        )
         items: dict[int, FavResource] = {}
         for chunk in _chunks(requested, chunk_size):
             resources = ",".join(f"{aid}:2" for aid in chunk)
@@ -329,7 +398,7 @@ class BilibiliClient:
                 f"?resources={urllib.parse.quote(resources)}"
                 f"&platform=web&folder_mid={fmid}&folder_id={int(media_id)}"
             )
-            payload = self.get_json(url)
+            payload = self.get_json(url, referer=referer)
             data = self._data_list(payload, "fav/resource/infos")
             chunk_set = set(chunk)
             for item in data:
@@ -351,7 +420,11 @@ class BilibiliClient:
     def fav_folders(self, up_mid: int | None = None) -> list[FolderInfo]:
         mid = self.mid if up_mid is None else int(up_mid)
         url = f"{BASE}/x/v3/fav/folder/created/list-all?up_mid={mid}&jsonp=jsonp"
-        payload = self.get_json(url, auth=True)
+        payload = self.get_json(
+            url,
+            auth=True,
+            referer=f"https://space.bilibili.com/{mid}/favlist",
+        )
         data = payload.get("data") or {}
         if not isinstance(data, Mapping):
             raise BilibiliProtocolError("fav/folder/created/list-all data 不是对象")

reaction_board/bilibili/provider.py

@@ -33,6 +33,10 @@ class ClientProvider:
         credentials = getattr(self.store.get().bilibili, self.role)
         key = (credentials.mid, credentials.cookie, credentials.csrf)
         if self._client is None or key != self._key:
+            if self._client is not None:
+                close = getattr(self._client, "close", None)
+                if callable(close):
+                    close()
             self._client = self.factory(credentials)
             self._key = key
         return self._client

reaction_board/bilibili/write_client.py

@@ -18,9 +18,11 @@ from .client import (
 
 
 class BilibiliWriteClient(BilibiliClient):
-    def post_form(self, url: str, data: Mapping[str, Any]) -> dict[str, Any]:
+    def post_form(
+        self, url: str, data: Mapping[str, Any], *, referer: str | None = None
+    ) -> dict[str, Any]:
         self.pacer.wait()
-        headers = self._headers(auth=True)
+        headers = self._headers(auth=True, referer=referer)
         headers["content-type"] = "application/x-www-form-urlencoded"
         body = urllib.parse.urlencode(data).encode("utf-8")
         response = self.transport.request(
@@ -47,6 +49,10 @@ class BilibiliWriteClient(BilibiliClient):
                 "del_media_ids": "",
                 "csrf": self.csrf,
             },
+            referer=(
+                f"https://space.bilibili.com/{self.mid}/favlist"
+                f"?fid={int(folder_id)}&ftype=create"
+            ),
         )
 
     def sort(self, media_id: int, sort_spec: str) -> dict[str, Any]:
@@ -57,4 +63,8 @@ class BilibiliWriteClient(BilibiliClient):
                 "sort": sort_spec,
                 "csrf": self.csrf,
             },
+            referer=(
+                f"https://space.bilibili.com/{self.mid}/favlist"
+                f"?fid={int(media_id)}&ftype=create"
+            ),
         )

reaction_board/notify/telegram.py

@@ -10,6 +10,8 @@ import urllib.parse
 import urllib.request
 from typing import Any, Mapping, Protocol, Sequence
 
+import httpx
+
 MAX_CANDIDATES_PER_MESSAGE = 3
 
 
@@ -64,6 +66,49 @@ def format_candidate(candidate: Mapping[str, Any]) -> str:
     )
 
 
+class HttpxTelegramTransport:
+    """基于 httpx 连接池的 Telegram 传输。"""
+
+    def __init__(
+        self,
+        *,
+        timeout: float = 30,
+        client: httpx.Client | None = None,
+    ) -> None:
+        self.timeout = timeout
+        self._client = client or httpx.Client(
+            timeout=httpx.Timeout(timeout, connect=10),
+            limits=httpx.Limits(
+                max_keepalive_connections=5,
+                keepalive_expiry=30,
+            ),
+            follow_redirects=False,
+            trust_env=True,
+        )
+
+    def post(self, url: str, data: Mapping[str, Any], timeout: float = 30) -> Any:
+        try:
+            response = self._client.post(url, data=dict(data), timeout=timeout)
+        except httpx.TransportError as exc:
+            raise TelegramError(str(exc)) from exc
+        try:
+            payload = response.json()
+        except ValueError as exc:
+            raise TelegramError(
+                f"Telegram HTTP {response.status_code}: {response.text[:200]}"
+            ) from exc
+        if response.status_code >= 400 and not (
+            isinstance(payload, Mapping) and "ok" in payload
+        ):
+            raise TelegramError(
+                f"Telegram HTTP {response.status_code}: {str(payload)[:200]}"
+            )
+        return payload
+
+    def close(self) -> None:
+        self._client.close()
+
+
 class TelegramClient:
     def __init__(
         self,
@@ -81,11 +126,16 @@ class TelegramClient:
             raise ValueError("telegram.chat_id 不能为空")
         self.token = token
         self.chat_id = int(chat_id)
-        self.transport = transport or UrllibTelegramTransport()
+        self.transport = transport or HttpxTelegramTransport()
         self.timeout = timeout
         self.poll_timeout = poll_timeout
         self.logger = logger
 
+    def close(self) -> None:
+        close = getattr(self.transport, "close", None)
+        if callable(close):
+            close()
+
     def _call(
         self,
         method: str,

tests/test_auth.py

@@ -13,6 +13,7 @@ sys.stdout.reconfigure(encoding="utf-8")
 from reaction_board.bilibili.auth import (  # noqa: E402
     cookie_value,
     cookies_from_set_cookie,
+    make_headers,
     merge_cookie,
     parse_cookie,
     sessdata_expiry,
@@ -78,10 +79,30 @@ def test_qr_png():
         assert path.read_bytes().startswith(b"\x89PNG\r\n\x1a\n")
 
 
+def test_make_headers_matches_raw_request():
+    headers = make_headers({"cookie": "SESSDATA=x; buvid3=y"}, auth=False)
+    assert headers["origin"] == "https://space.bilibili.com"
+    assert headers["referer"] == "https://space.bilibili.com/"
+    assert headers["priority"] == "u=1, i"
+    assert headers["user-agent"].endswith("Chrome/155.0.0.0 Safari/537.36")
+    assert headers["sec-ch-ua"].startswith('"Google Chrome";v="155"')
+    assert "SESSDATA" not in headers["cookie"]
+    assert "buvid3=y" in headers["cookie"]
+
+    full = make_headers({"cookie": "SESSDATA=x"}, auth=True)
+    assert full["cookie"] == "SESSDATA=x"
+
+    custom = make_headers(
+        {"cookie": ""}, referer="https://space.bilibili.com/1/favlist?fid=2"
+    )
+    assert custom["referer"] == "https://space.bilibili.com/1/favlist?fid=2"
+
+
 def main():
     test_cookie_helpers()
     test_finish_updates_only_selected_account_and_rolls_back()
     test_qr_png()
+    test_make_headers_matches_raw_request()
     print("OK: 双账号 auth Cookie 合并、写回隔离与回滚测试全部通过")
 
 

tests/test_bilibili_client.py

@@ -99,6 +99,10 @@ def test_fav_ids_parse_and_anonymous_header():
     assert "media_id=3700023351" in call["url"]
     assert "SESSDATA" not in call["headers"]["cookie"]
     assert "buvid3=device" in call["headers"]["cookie"]
+    assert call["headers"]["origin"] == "https://space.bilibili.com"
+    assert call["headers"]["priority"] == "u=1, i"
+    assert "Chrome/155.0.0.0" in call["headers"]["user-agent"]
+    assert "fid=3700023351" in call["headers"]["referer"]
 
 
 def test_anonymous_failure_falls_back_to_full_cookie():

tests/test_httpx_transport.py

@@ -0,0 +1,145 @@
+# -*- coding: utf-8 -*-
+"""httpx 传输层契约测试（不联网）: uv run python tests/test_httpx_transport.py"""
+
+import os
+import sys
+
+import httpx
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+sys.stdout.reconfigure(encoding="utf-8")
+
+from reaction_board.bilibili.client import (  # noqa: E402
+    HttpxTransport,
+    TransportError,
+)
+from reaction_board.notify.telegram import (  # noqa: E402
+    HttpxTelegramTransport,
+    TelegramError,
+)
+
+
+def test_bilibili_httpx_transport():
+    captured = {}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        captured["method"] = request.method
+        captured["url"] = str(request.url)
+        captured["ua"] = request.headers["user-agent"]
+        return httpx.Response(
+            200,
+            json={"code": 0, "message": "OK", "data": []},
+            headers={"content-type": "application/json"},
+        )
+
+    client = httpx.Client(transport=httpx.MockTransport(handler))
+    transport = HttpxTransport(client=client)
+    try:
+        response = transport.request(
+            "GET",
+            "https://api.bilibili.com/x/test",
+            {"user-agent": "UA-TEST"},
+            data=None,
+            timeout=30,
+        )
+    finally:
+        transport.close()
+    assert response.status == 200
+    assert response.json()["code"] == 0
+    assert captured == {
+        "method": "GET",
+        "url": "https://api.bilibili.com/x/test",
+        "ua": "UA-TEST",
+    }
+
+
+def test_bilibili_httpx_transport_maps_errors():
+    def handler(request):
+        raise httpx.ConnectError("boom")
+
+    transport = HttpxTransport(
+        client=httpx.Client(transport=httpx.MockTransport(handler))
+    )
+    try:
+        transport.request("GET", "https://api.bilibili.com/x/test", {})
+    except TransportError as exc:
+        assert "boom" in str(exc)
+    else:
+        raise AssertionError("httpx 连接错误应映射为 TransportError")
+    finally:
+        transport.close()
+
+
+def test_telegram_httpx_transport():
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, json={"ok": True, "result": [{"update_id": 1}]})
+
+    transport = HttpxTelegramTransport(
+        client=httpx.Client(transport=httpx.MockTransport(handler))
+    )
+    try:
+        payload = transport.post(
+            "https://api.telegram.org/botx/getUpdates", {"timeout": 0}, 30
+        )
+    finally:
+        transport.close()
+    assert payload["ok"] is True
+    assert payload["result"][0]["update_id"] == 1
+
+
+def test_telegram_httpx_transport_error_shapes():
+    def bad_json(request):
+        return httpx.Response(
+            400, json={"ok": False, "description": "query is too old"}
+        )
+
+    transport = HttpxTelegramTransport(
+        client=httpx.Client(transport=httpx.MockTransport(bad_json))
+    )
+    try:
+        payload = transport.post("https://api.telegram.org/botx/x", {}, 30)
+    finally:
+        transport.close()
+    assert payload["description"] == "query is too old"
+
+    def html_error(request):
+        return httpx.Response(500, text="bad gateway")
+
+    transport = HttpxTelegramTransport(
+        client=httpx.Client(transport=httpx.MockTransport(html_error))
+    )
+    try:
+        transport.post("https://api.telegram.org/botx/x", {}, 30)
+    except TelegramError as exc:
+        assert "500" in str(exc)
+    else:
+        raise AssertionError("非 JSON 错误应抛出 TelegramError")
+    finally:
+        transport.close()
+
+    def network_error(request):
+        raise httpx.ConnectError("offline")
+
+    transport = HttpxTelegramTransport(
+        client=httpx.Client(transport=httpx.MockTransport(network_error))
+    )
+    try:
+        transport.post("https://api.telegram.org/botx/x", {}, 30)
+    except TelegramError as exc:
+        assert "offline" in str(exc)
+    else:
+        raise AssertionError("网络错误应抛出 TelegramError")
+    finally:
+        transport.close()
+
+
+def main():
+    test_bilibili_httpx_transport()
+    test_bilibili_httpx_transport_maps_errors()
+    test_telegram_httpx_transport()
+    test_telegram_httpx_transport_error_shapes()
+    print("OK: httpx B 站与 Telegram 传输层测试全部通过")
+
+
+if __name__ == "__main__":
+    main()

uv.lock

@@ -2,12 +2,93 @@ version = 1
 revision = 3
 requires-python = "==3.14.*"
 
+[[package]]
+name = "anyio"
+version = "4.15.1"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "idna" },
+    { name = "typing-extensions" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/a9/d2/f4d173e22df740bc37b1db102b386ba719b66e95b0f0d751f556b387e6d2/anyio-4.15.1.tar.gz", hash = "sha256:9f28306018cbd6d329e64a36d58256edff76dd996fe423bc957326e578b82a94", size = 276966, upload-time = "2026-09-05T10:42:39.44Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/12/b8/4bd346e22b28902df4d651910f5242c28d84e4a5c2435ca5c3f797ed7e2e/anyio-4.15.1-py3-none-any.whl", hash = "sha256:6152fdbbf9a77fdec97731721bebf7c4c44f7c29b424b0065826173efc7ed101", size = 132079, upload-time = "2026-09-05T10:42:37.923Z" },
+]
+
 [[package]]
 name = "bili-fav"
 version = "0.1.0"
 source = { editable = "." }
+dependencies = [
+    { name = "httpx" },
+]
 
 [package.metadata]
+requires-dist = [{ name = "httpx", specifier = ">=0.28.1" }]
 
 [package.metadata.requires-dev]
 dev = []
+
+[[package]]
+name = "certifi"
+version = "2026.10.23"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/fd/11/b524cbd9bbdaca58608e4cf4053fd463417549e1959fa8d3aa18319f9555/certifi-2026.10.23.tar.gz", hash = "sha256:a9895e6e2758b371e0545e321a3a8fd1fcfd9829ada8824a936dfb16aaaf36cd", size = 136225, upload-time = "2026-10-23T11:36:16.478Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/32/d1/a3652fe2c19134d0d15922177821d863cbd7e842d53a21bbe7e8e3a1b5ac/certifi-2026.10.23-py3-none-any.whl", hash = "sha256:e9b54de76e530bb6cf0ccd036787103acb58fd42ba930ea8bf948ddbaa2f61d1", size = 135082, upload-time = "2026-10-23T11:36:14.763Z" },
+]
+
+[[package]]
+name = "h11"
+version = "0.16.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/01/ee/02a2c011bdab74c6fb3c75474d40b3052059d95df7e73351460c8588d963/h11-0.16.0.tar.gz", hash = "sha256:4e35b956cf45792e4caa5885e69fba00bdbc6ffafbfa020300e549b208ee5ff1", size = 101250, upload-time = "2025-04-24T03:35:25.427Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/04/4b/29cac41a4d98d144bf5f6d33995617b185d14b22401f75ca86f384e87ff1/h11-0.16.0-py3-none-any.whl", hash = "sha256:63cf8bbe7522de3bf65932fda1d9c2772064ffb3dae62d55932da54b31cb6c86", size = 37515, upload-time = "2025-04-24T03:35:24.344Z" },
+]
+
+[[package]]
+name = "httpcore"
+version = "1.0.9"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "certifi" },
+    { name = "h11" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/06/94/82699a10bca87a5556c9c59b5963f2d039dbd239f25bc2a63907a05a14cb/httpcore-1.0.9.tar.gz", hash = "sha256:6e34463af53fd2ab5d807f399a9b45ea31c3dfa2276f15a2c3f00afff6e176e8", size = 85484, upload-time = "2025-04-24T22:06:22.219Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/7e/f5/f66802a942d491edb555dd61e3a9961140fd64c90bce1eafd741609d334d/httpcore-1.0.9-py3-none-any.whl", hash = "sha256:2d400746a40668fc9dec9810239072b40b4484b640a8c38fd654a024c7a1bf55", size = 78784, upload-time = "2025-04-24T22:06:20.566Z" },
+]
+
+[[package]]
+name = "httpx"
+version = "0.28.1"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "anyio" },
+    { name = "certifi" },
+    { name = "httpcore" },
+    { name = "idna" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/b1/df/48c586a5fe32a0f01324ee087459e112ebb7224f646c0b5023f5e79e9956/httpx-0.28.1.tar.gz", hash = "sha256:75e98c5f16b0f35b567856f597f06ff2270a374470a5c2392242528e3e3e42fc", size = 141406, upload-time = "2024-12-06T15:37:23.222Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/2a/39/e50c7c3a983047577ee07d2a9e53faf5a69493943ec3f6a384bdc792deb2/httpx-0.28.1-py3-none-any.whl", hash = "sha256:d909fcccc110f8c7faf814ca82a9a4d816bc5a6dbfea25d6591d6985b8ba59ad", size = 73517, upload-time = "2024-12-06T15:37:21.509Z" },
+]
+
+[[package]]
+name = "idna"
+version = "3.20"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/f5/08/8eea9d4b8302028f3abb2c0813953f7aec26d33b7a8960ed760e65ff29fa/idna-3.20.tar.gz", hash = "sha256:a7db850025b95ded1eae8a46181a1a6c56c92c96f0e2b005d9ff8dc0210cab44", size = 216463, upload-time = "2026-09-17T14:11:04.752Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/58/a2/bb081bab032533a855d44de1d56f8e8426114ff1ba5d1f07a438a0a654f8/idna-3.20-py3-none-any.whl", hash = "sha256:ab7ae7122974553370f0bdb919e1a960b2cd1bc1ef0276416d896db81c14582c", size = 69583, upload-time = "2026-09-17T14:11:03.168Z" },
+]
+
+[[package]]
+name = "typing-extensions"
+version = "4.16.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/f6/cc/6253133b5bb138fc3306cebfbda2c520f545d36b5be2c7255cc528bb45d6/typing_extensions-4.16.0.tar.gz", hash = "sha256:dc983d19a509c94dba722ee6abd33940f7c05a89e243c47e907eb4db6f1a43e5", size = 113555, upload-time = "2026-07-02T08:40:05.92Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/49/d3/b8441a820a491ddfc024b0b0cf0393375b75ea13866d9c66727e54c2fc80/typing_extensions-4.16.0-py3-none-any.whl", hash = "sha256:481caa481374e813c1b176ada14e97f1f67a4539ce9cfeb3f350d78d6370c2e8", size = 45571, upload-time = "2026-07-02T08:40:04.659Z" },
+]