diff --git a/DEVELOPMENT.md b/DEVELOPMENT.md index 540409e..3498e6c 100644 --- a/DEVELOPMENT.md +++ b/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：本地看板 diff --git a/README.md b/README.md index 9102993..734a4a5 100644 --- a/README.md +++ b/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 # 单写队列、事务与并发 diff --git a/favsort.py b/favsort.py index 9c78cd2..c7afb79 100644 --- a/favsort.py +++ b/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): - 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")) +_HTTP_CLIENT: httpx.Client | None = None -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): - headers = make_headers(cfg, auth=True) +def post_form(cfg, url, data, referer=None): + headers = make_headers(cfg, auth=True, referer=referer) headers["content-type"] = "application/x-www-form-urlencoded" - req = urllib.request.Request( - url, data=urllib.parse.urlencode(data).encode(), headers=headers, method="POST" + response = _http_client().post( + url, headers=headers, data=data, timeout=30 ) - with urllib.request.urlopen(req, timeout=30) as r: - return json.loads(r.read().decode("utf-8", "replace")) + response.raise_for_status() + 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: diff --git a/tests/test_favsort_http.py b/tests/test_favsort_http.py new file mode 100644 index 0000000..ef6b44d --- /dev/null +++ b/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() diff --git a/tests/test_favsort_run.py b/tests/test_favsort_run.py index 00f62e2..e28520b 100644 --- a/tests/test_favsort_run.py +++ b/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)