refactor(favsort): send bilibili requests through httpx

Reuse a pooled httpx client for favsort GET and POST requests, share the raw-request-derived browser headers, and add folder-specific referers for ids, infos and sort. Add transport tests for header parity and anonymous-to-authenticated fallback.

DEVELOPMENT.md

@@ -654,7 +654,7 @@ uv run python tests/test_moves.py
 - `app.py` 新增 `discover` / `candidates`；`run` 注册发现与候选清理任务，并启动 Telegram poller；
 - 真实只读联调：4 次 WBI 搜索、168 条结果、1 条候选（score=6），Telegram 通知发送成功；
 - 真实 `deal` / `sort` 尚未在收藏夹执行，等待用户点击 Telegram 按钮或授权测试视频后验证；
-- 自动化测试见 `tests/test_classify.py`、`test_search.py`、`test_candidates.py`、`test_discovery.py`、`test_fav_position.py`、`test_favorites.py`、`test_telegram.py`。
+- 自动化测试见 `tests/test_classify.py`、`test_search.py`、`test_candidates.py`、`test_discovery.py`、`test_fav_position.py`、`test_favorites.py`、`test_telegram.py`、`test_httpx_transport.py`、`test_favsort_http.py`。
 
 ### P2：本地看板
 

README.md

@@ -275,6 +275,7 @@ sort=A:2:B:2  →  将 B 插到 A 之后
 ```powershell
 uv run python tests/test_auth.py            # Cookie 合并、设备指纹、Set-Cookie 解析
 uv run python tests/test_moves.py           # 排序移动脚本：随机排列与头部用例
+uv run python tests/test_favsort_http.py    # favsort httpx 请求头与匿名回退
 uv run python tests/test_config.py          # 配置加载、校验、ENV 覆盖与热加载
 uv run python tests/test_migrations.py      # 迁移、备份、回滚与保留期
 uv run python tests/test_database.py        # 单写队列、事务与并发

favsort.py

@@ -28,7 +28,8 @@ import random
 import sys
 import time
 import urllib.parse
-import urllib.request
+
+import httpx
 
 sys.stdout.reconfigure(encoding="utf-8")
 
@@ -95,41 +96,67 @@ def load_config(require_auth=True):
 
 # ---------------------------------------------------------------- http
 
-def _get_json(cfg, url, auth):
+_HTTP_CLIENT: httpx.Client | None = None
-    req = urllib.request.Request(url, headers=make_headers(cfg, auth=auth))
-    with urllib.request.urlopen(req, timeout=30) as r:
-        return json.loads(r.read().decode("utf-8", "replace"))
 
 
-def get_json(cfg, url, auth=False):
+def _http_client() -> httpx.Client:
+    global _HTTP_CLIENT
+    if _HTTP_CLIENT is None:
+        _HTTP_CLIENT = httpx.Client(
+            timeout=httpx.Timeout(30, connect=10),
+            limits=httpx.Limits(
+                max_keepalive_connections=5,
+                keepalive_expiry=30,
+            ),
+            follow_redirects=False,
+            trust_env=True,
+        )
+    return _HTTP_CLIENT
+
+
+def _get_json(cfg, url, auth, referer=None):
+    response = _http_client().get(
+        url,
+        headers=make_headers(cfg, auth=auth, referer=referer),
+        timeout=30,
+    )
+    response.raise_for_status()
+    return response.json()
+
+
+def get_json(cfg, url, auth=False, referer=None):
     """只读请求：默认用去掉账号身份的 cookie。
 
     万一匿名请求被拒（HTTP 错误，或 code != 0），自动带完整 cookie 重试一次，
     保证功能不会因为剥离身份而回退。
     """
     if auth:
-        return _get_json(cfg, url, auth=True)
+        return _get_json(cfg, url, auth=True, referer=referer)
     try:
-        r = _get_json(cfg, url, auth=False)
+        r = _get_json(cfg, url, auth=False, referer=referer)
         if r.get("code") == 0:
             return r
     except Exception:  # noqa: BLE001  网络错误 / 412 风控都走回退
         pass
-    return _get_json(cfg, url, auth=True)
+    return _get_json(cfg, url, auth=True, referer=referer)
 
 
-def post_form(cfg, url, data):
+def post_form(cfg, url, data, referer=None):
-    headers = make_headers(cfg, auth=True)
+    headers = make_headers(cfg, auth=True, referer=referer)
     headers["content-type"] = "application/x-www-form-urlencoded"
-    req = urllib.request.Request(
+    response = _http_client().post(
-        url, data=urllib.parse.urlencode(data).encode(), headers=headers, method="POST"
+        url, headers=headers, data=data, timeout=30
     )
-    with urllib.request.urlopen(req, timeout=30) as r:
+    response.raise_for_status()
-        return json.loads(r.read().decode("utf-8", "replace"))
+    return response.json()
 
 
 def fetch_ids(cfg, media_id):
-    r = get_json(cfg, API["ids"].format(base=BASE, mid=media_id))
+    referer = (
+        f"https://space.bilibili.com/{cfg['mid']}/favlist"
+        f"?fid={media_id}&ftype=create"
+    )
+    r = get_json(cfg, API["ids"].format(base=BASE, mid=media_id), referer=referer)
     if r.get("code") != 0:
         raise SystemExit(f"ids 失败: code={r.get('code')} msg={r.get('message')}")
     return r.get("data") or []
@@ -137,13 +164,17 @@ def fetch_ids(cfg, media_id):
 
 def fetch_infos(cfg, media_id, folder_mid, id_list):
     items, calls = [], 0
+    referer = (
+        f"https://space.bilibili.com/{folder_mid}/favlist"
+        f"?fid={media_id}&ftype=create"
+    )
     for i in range(0, len(id_list), CHUNK):
         chunk = id_list[i:i + CHUNK]
         res = ",".join(f"{x['id']}:{x['type']}" for x in chunk)
         url = API["infos"].format(
             base=BASE, res=urllib.parse.quote(res), fmid=folder_mid, mid=media_id
         )
-        r = get_json(cfg, url)
+        r = get_json(cfg, url, referer=referer)
         calls += 1
         if r.get("code") != 0:
             raise SystemExit(f"infos 失败 @{i}: code={r.get('code')} msg={r.get('message')}")
@@ -310,7 +341,12 @@ def load_plan(media_id):
 def cmd_folders(cfg, args):
     # 注意：list-all 是"列出我自己的收藏夹"，匿名调用只会返回公开收藏夹（实测 11 -> 3），
     # 因此这里必须带完整 cookie（仍然只是 GET，不会改动任何数据）。
-    r = get_json(cfg, API["folders"].format(base=BASE, mid=cfg["mid"]), auth=True)
+    r = get_json(
+        cfg,
+        API["folders"].format(base=BASE, mid=cfg["mid"]),
+        auth=True,
+        referer=f"https://space.bilibili.com/{cfg['mid']}/favlist",
+    )
     if r.get("code") != 0:
         raise SystemExit(f"获取收藏夹列表失败: {r.get('code')} {r.get('message')}")
     lst = (r.get("data") or {}).get("list") or []
@@ -431,6 +467,10 @@ def cmd_run(cfg, args):
     work = list(actual)
     total = len(moves)
     long_after = random.randint(sc["long_every_min"], sc["long_every_max"])
+    sort_referer = (
+        f"https://space.bilibili.com/{cfg['mid']}/favlist"
+        f"?fid={media_id}&ftype=create"
+    )
 
     for i, m in enumerate(moves, 1):
         pred_type = types.get(m["pred"], 2)
@@ -442,14 +482,18 @@ def cmd_run(cfg, args):
         }
         print(f"[run] {i}/{total} {titles[str(m['moved'])]['title'][:30]} "
               f"-> after {m['pred']}", end="  ", flush=True)
-        resp = post_form(cfg, API["sort"].format(base=BASE), body)
+        resp = post_form(
+            cfg, API["sort"].format(base=BASE), body, referer=sort_referer
+        )
         code = resp.get("code")
         print(f"code={code}")
         if code != 0:
             wait = sc["retry_wait"]
             print(f"[run] 返回非 0，冷却 {wait}s 后重试一次")
             time.sleep(wait)
-            resp = post_form(cfg, API["sort"].format(base=BASE), body)
+            resp = post_form(
+            cfg, API["sort"].format(base=BASE), body, referer=sort_referer
+        )
             code = resp.get("code")
             print(f"[run] retry code={code}")
             if code != 0:

tests/test_favsort_http.py

@@ -0,0 +1,91 @@
+# -*- coding: utf-8 -*-
+"""favsort HTTP 层测试（httpx MockTransport，不联网）:
+uv run python tests/test_favsort_http.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")
+
+import favsort  # noqa: E402
+
+
+def with_client(handler):
+    client = httpx.Client(transport=httpx.MockTransport(handler))
+    original = favsort._HTTP_CLIENT
+    favsort._HTTP_CLIENT = client
+    return client, original
+
+
+def test_get_json_and_post_form_headers():
+    calls = []
+
+    def handler(request):
+        calls.append(request)
+        return httpx.Response(200, json={"code": 0, "data": []})
+
+    client, original = with_client(handler)
+    try:
+        cfg = {"cookie": "SESSDATA=x; buvid3=y"}
+        assert favsort.get_json(
+            cfg, "https://api.bilibili.com/x/test", auth=False
+        ) == {"code": 0, "data": []}
+        first = calls[0]
+        assert first.headers["origin"] == "https://space.bilibili.com"
+        assert first.headers["priority"] == "u=1, i"
+        assert "Chrome/155.0.0.0" in first.headers["user-agent"]
+        assert "SESSDATA" not in first.headers["cookie"]
+        assert "buvid3=y" in first.headers["cookie"]
+
+        favsort.post_form(
+            cfg,
+            "https://api.bilibili.com/x/v3/fav/resource/sort",
+            {"a": "b"},
+            referer="https://space.bilibili.com/1/favlist?fid=2",
+        )
+        post = calls[1]
+        assert post.method == "POST"
+        assert post.headers["content-type"] == "application/x-www-form-urlencoded"
+        assert post.headers["referer"] == "https://space.bilibili.com/1/favlist?fid=2"
+        assert post.content == b"a=b"
+        assert "SESSDATA=x" in post.headers["cookie"]
+    finally:
+        favsort._HTTP_CLIENT = original
+        client.close()
+
+
+def test_anonymous_failure_falls_back_to_full_cookie():
+    calls = []
+
+    def handler(request):
+        calls.append(request)
+        if len(calls) == 1:
+            return httpx.Response(200, json={"code": -101, "message": "未登录"})
+        return httpx.Response(200, json={"code": 0, "data": []})
+
+    client, original = with_client(handler)
+    try:
+        cfg = {"cookie": "SESSDATA=x; buvid3=y"}
+        assert favsort.get_json(cfg, "https://api.bilibili.com/x/test") == {
+            "code": 0, "data": [],
+        }
+        assert len(calls) == 2
+        assert "SESSDATA" not in calls[0].headers["cookie"]
+        assert "SESSDATA=x" in calls[1].headers["cookie"]
+    finally:
+        favsort._HTTP_CLIENT = original
+        client.close()
+
+
+def main():
+    test_get_json_and_post_form_headers()
+    test_anonymous_failure_falls_back_to_full_cookie()
+    print("OK: favsort httpx 请求头与匿名回退测试全部通过")
+
+
+if __name__ == "__main__":
+    main()

tests/test_favsort_run.py

@@ -74,7 +74,9 @@ def test_sort_preserves_resource_types():
             return [{"id": aid} for aid in order]
 
         favsort.fetch_ids = fetch_ids
-        favsort.post_form = lambda cfg, url, body: bodies.append(body) or {"code": 0}
+        favsort.post_form = (
+            lambda cfg, url, body, referer=None: bodies.append(body) or {"code": 0}
+        )
         args = Args()
         try:
             code = favsort.cmd_run(config(), args)