diff --git a/.gitignore b/.gitignore index bcd688c..c5cffbf 100644 --- a/.gitignore +++ b/.gitignore @@ -23,6 +23,7 @@ dist/ *.pem *.key *.har +*.jsonl *cookie*.json *cookie*.txt browser-profile/ diff --git a/docs/2026-09-06-01-22-notification-subscription.md b/docs/2026-09-06-01-22-notification-subscription.md new file mode 100644 index 0000000..a1084bd --- /dev/null +++ b/docs/2026-09-06-01-22-notification-subscription.md @@ -0,0 +1,129 @@ +# 互动通知订阅:调研日志、实现与安装说明 + +## 结论和入口 + +在原 `src/get_notifications.py` 增加 `--sub`,不删除或替换原分页拉取逻辑。订阅仅覆盖互动/站点通知通道,不处理私信。 + +```bash +# 原行为不变:拉取全部,保存 JSON 快照到 src/notifications.json +python3 src/get_notifications.py + +# 原增量拉取行为不变:到指定边界为止,不含边界本身 +python3 src/get_notifications.py --last-id <已保存的通知ID> + +# 订阅:stdout 每条一行 JSON,可直接交给其他进程处理 +python3 src/get_notifications.py --sub + +# 订阅并追加文件,不覆盖历史文件;stdout 仍同步输出 +python3 src/get_notifications.py --sub --output events.jsonl + +# 启动时先补拉一次历史缺口,然后继续订阅 +python3 src/get_notifications.py --sub --last-id <已保存的通知ID> --output events.jsonl +``` + +Ctrl+C 卸载本脚本的监听器并退出,退出码 130。状态/错误写 stderr,通知 JSONL 写 stdout。默认订阅不抓取已有通知,仅从安装监听器后接收推送;`--last-id` 才进行一次启动补拉。补拉期间的推送暂存在队列中。 + +每行包含 `received_at`(处理输出时间,不是服务端推送时间)、`notification`(复用原摘要字段)、`raw`(原始通知详情)。通知 ID 保留字符串。消费方可按 `notification.id` 做持久化处理记录;同一个聚合通知 ID 可能更新内容,不能永远只处理其第一次出现。 + +## 调研证据 + +1. 检查已有拉取脚本:使用 `/aweme/v1/web/notice/`,保留 `notice_group=960`、`is_mark_read=0`、分页游标和账号一致性校验。 +2. 通过 browser-harness 连接已有 CDP,调用现有当前用户接口确认已登录。未尝试登录、未操作关注/点赞/评论按钮、未打开消息面板,也未调用标记已读接口。 +3. 从 `webpackChunkdouyin_web` 读取已加载模块,定位 `NoticeFrontier`。调研时通知模块编号为 `511386`,codec 为 `114718`;**实现不硬编码这些模块编号**,而是按导出相关源码特征定位。 +4. 通知存在独立 Frontier WebSocket;它不是私信 IM 通道。`NoticeFrontier.frontierInstance` 已连接,并提供 `addEventListener/removeEventListener`。 +5. 原网页按二进制帧的 `service` 分流: + - `20313`:中台通知推送,读取 `notices[].notice_id_str`,按 `effect_groups` 包含 960/961 过滤。 + - `20003`:服务通知推送,读取顶层 `notice_id_str`;沿用页面类型集合 45、31、9009、9002、514、9067。 + - 其他 service 不交给通知业务处理。 +6. 推送不是完整业务详情。原页面按 ID 调用 `/aweme/v1/web/notice/detail/`,`id_list` 为 `[{"notice_id_str":"...","type":0}]`;实测返回 `status_code=0`、`notice_list_v2` 与请求 ID 相符。 +7. 读取 Frontier 的事件管理实现,确认添加独立监听器不需要替换站点 `onmessage`;卸载监听器不会关闭原连接。 + +## 实现 + +```text +页面现有通知 WebSocket + → SDK decodedFrame 解码 + → 自有 message 监听器 + 内存事件队列 + → Promise 事件唤醒 → browser-harness → Python + → 按 ID 请求通知详情(is_mark_read=0) + → 校验账号与 ID → 去重 → JSONL,立即 flush +``` + +新增 `src/subscribe_notifications.py` 负责订阅桥接,通过原脚本的 `--sub` 分支调用。复用 `get_current_user.py` 登录校验,以及 `get_notifications.py` 的摘要/补拉函数,不新装 WebSocket/Protobuf/签名库,不提取或保存 Cookie。 + +### 不属于定时通知拉取 + +- 没有新事件时,本实现不周期性请求通知列表或详情。 +- 本地等待每 2 秒返回一次空队列,目的是低于 browser-harness 的 IPC 超时并检查页面状态;这是浏览器到 Python 的本地桥接心跳,不是每 2 秒请求抖音。 +- 事件到达会立即唤醒等待,不必等到心跳到期。Python 正在处理上一批时,新推送进入队列。 +- 页面自己的既有轮询/心跳不受本脚本控制,因此不能保证浏览器整体没有其他通知请求。 +- 详情暂不可见时仅针对当前推送重试,最多 3 次,间隔 0.5/1 秒;不无限重试或定时全量扫描。 + +### 错误与边界 + +- 登录检查失败时停止,交由用户手动登录。页面刷新、账号切换或通知实例销毁后停止并提示重新运行;不自动导航或登录。 +- SDK 自有网络重连仍保留,脚本报告 open/close;**断线期间不保证补发,不承诺零丢失**。需要时用 `--last-id` 重新启动补拉;这仍受原列表范围(960)及服务端保留历史的限制,不能保证补回每次聚合更新或所有其他站点通知。 +- 当前不持久化 ACK/队列,也不自动恢复页面刷新后的监听。进程异常中止或详情失败时,尚未处理的队列可能丢失;消费方应保存处理记录并在恢复时补拉。 +- 队列最多 1000 个事件,超出时明确报错退出;内存去重保留最近 4096 个 ID,相同 ID/相同完整详情不重复输出,内容变化仍输出。跨进程幂等由消费方负责。 +- `--output` 是追加写 JSONL,与拉取模式的 JSON 快照格式不同,请使用不同文件名;每行 flush,但不承诺断电持久化。 +- 保持已登录抖音标签页和浏览器运行。browser-harness 共享当前标签状态;不要并发切换账号/标签。后台标签节流、系统休眠或浏览器卡顿可能造成 IPC 超时并退出,需恢复页面后重启;推荐保持标签页活动。当前依赖浏览器 SDK,不是脱离浏览器的纯 Python WebSocket 客户端。 +- 平台可能不推送某些通知,或自行聚合/延迟;源码表明推送通道存在,不代表每一次点赞/关注/评论都必定一对一即时推送。 +- 模块导出特征、帧格式、服务类型和详情 API 都是站点内部实现;变化时需重新调研。 + +## 验证记录(区分模拟与真实) + +- `python3 src/test_notifications.py`:原分页/摘要离线检查通过。 +- `python3 src/test_subscribe_notifications.py`:通过 ID 精度、分组/类型过滤、异常结构、详情重试、账号变化、追加输出、去重/聚合更新、`--sub` 路由及原拉取分支检查。 +- Node.js 假 Webpack/FWS 检查:二进制 payload、忽略其他 service、事件到达立即唤醒、退出移除监听器而不关闭连接、实例变化报错,全部通过。 +- 真实浏览器安装监听成功,状态为 connected;空闲等待约 3.2 秒(含进程调用开销)返回空事件;卸载后原通知 WebSocket 仍连接。 +- 真实 `python3 src/get_notifications.py --sub` 运行约 6 秒后发送 SIGINT:安装/停止提示正常,退出码 130,剩余自有监听器为 0,原 WebSocket 仍连接。期间没有自然新通知,不以此证明真实互动推送到达。 +- 质量检查:运行时导入、离线测试、真实 CLI 均通过。LSP 对新增同目录模块报两处 `reportMissingImports`,已按运行证据登记为误报,未向源码添加忽略注释或修改导入路径;独立 basedpyright 检查 45 秒超时,不能声称该检查通过。Ruff 在原 `get_notifications.py` 留有已有的 EXE001/TRY004 风格提示,未为订阅功能改动原逻辑。 +- 使用真实 SDK 的 `encodeFrame` 生成**模拟推送帧**,只调用本次安装的自有监听器,未触发站点原监听器、未发送网络通知。实际 `decodedFrame` 解码成功,Python 解析出 ID,再通过真实详情接口读回对应通知。 +- 上述模拟验证不等于已收到外部新点赞/关注/评论的完整端到端测试。仍需由另一账号产生一条真实互动,观察订阅 stdout;本次没有代替其他账号执行互动,也不虚构实时延迟结论。 + +### 调研中修正的问题 + +- 最初 15 秒 JS 等待超过 harness IPC 超时,改为 2 秒本地等待;事件本身仍立即唤醒。 +- 重复使用同一 Webpack 临时 chunk ID 会导致 runtime 回调不再执行,现使用每次唯一标识。 +- 离线 codec 最初返回普通数组,修正为真实解码器使用的 Uint8Array。 +- 真实 Frontier 的监听项是 `{fn,ctx}`,并非函数本身;模拟注入改为只调用自有监听项。 +- 编码器要求大写 `SeqID/LogID`,修正测试帧后真实 codec 链路验证通过;这些字段不需要业务脚本自行构造。 + +## 空白环境安装与复用 + +Python 3.11+、Chrome/Chromium、browser-harness;测试 JS 桥接另需 Node.js 18+。本项目新增代码只使用 Python 标准库。保留整个 `src/`,不要只复制入口文件。 + +```bash +python3 -m venv .venv +. .venv/bin/activate +python -m pip install 'browser-harness==0.1.9' +# Windows 激活方式为 .venv\Scripts\activate + +# 也可通过 uv 独立安装 CLI: +# uv tool install --python 3.12 'browser-harness==0.1.9' + +# 根据自己的浏览器修改端点,不要将 CDP 端口公开到公网 +export BU_CDP_URL=http://localhost:9222 +browser-harness <<'PY' +print(list_tabs()) +PY + +python3 src/get_notifications.py --sub +``` + +若没有启用远程调试的浏览器,使用本机浏览器可执行文件和独立用户目录启动,例如: + +```bash +chromium --remote-debugging-address=127.0.0.1 --remote-debugging-port=9222 \ + --user-data-dir="$HOME/.douyin-browser" +``` + +手动打开抖音并完成登录,再运行脚本;不会自动登录。已有登录浏览器直接复用,不必另建配置。安装参考:。本次在现有环境验证,未声称在空白机器重新安装验证。 + +```bash +# 离线检查;JS 桥接测试要求 node 在 PATH 中 +python3 src/test_notifications.py +python3 src/test_subscribe_notifications.py +``` + +`.gitignore` 已新增 `*.jsonl`,防止通知原文、用户资料等运行输出进入公共仓库。本文不记录真实账号或通知 ID。 diff --git a/src/get_notifications.py b/src/get_notifications.py index 3b5254e..23d965a 100644 --- a/src/get_notifications.py +++ b/src/get_notifications.py @@ -1,5 +1,5 @@ #!/usr/bin/env python3 -"""自动翻页获取互动通知,遇到 last-id 停止;不指定则获取全部,不标记已读。""" +"""拉取互动通知,或用 --sub 订阅新通知;不标记已读,不含私信。""" import argparse import json @@ -132,10 +132,16 @@ def collect_notifications( def main() -> int: parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False) parser.add_argument( - "--last-id", help="从最新开始,遇到此通知 ID 停止且不包含该条;未指定则获取全部" + "--sub", action="store_true", help="订阅互动通知推送;不指定则保持原拉取行为" ) parser.add_argument( - "--output", type=Path, default=Path(__file__).with_name("notifications.json") + "--last-id", + help="历史拉取截至此 ID(不含);--sub 时用于一次启动补拉,不指定则只接收新推送", + ) + parser.add_argument( + "--output", + type=Path, + help="拉取默认保存 src/notifications.json;订阅时可选追加 JSONL 文件(默认仅 stdout)", ) args = parser.parse_args() if args.last_id is not None and ( @@ -143,6 +149,12 @@ def main() -> int: ): parser.error("--last-id 必须为非空的数字通知 ID") try: + if args.sub: + from subscribe_notifications import subscribe + + subscribe(output=args.output, last_id=args.last_id) + return 0 + args.output = args.output or Path(__file__).with_name("notifications.json") # 复用已有登录态校验;失败就退出,不尝试登录。 user = get_user_from_browser() notices, raw_pages, last_id_found = collect_notifications( @@ -168,6 +180,9 @@ def main() -> int: json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8" ) temporary.replace(args.output) + except KeyboardInterrupt: + print("已停止。", file=sys.stderr) + return 130 except ( OSError, ValueError, diff --git a/src/subscribe_notifications.py b/src/subscribe_notifications.py new file mode 100644 index 0000000..b0a1ebd --- /dev/null +++ b/src/subscribe_notifications.py @@ -0,0 +1,255 @@ +"""互动通知推送桥接;入口:python3 src/get_notifications.py --sub。""" + +import json +import subprocess +import sys +import time +import uuid +from collections import OrderedDict +from contextlib import suppress +from datetime import datetime, timezone +from urllib.parse import urlencode + +from get_current_user import get_user_from_browser + +# 不改写站点 onmessage、不新建连接;退出时仅卸载自己的监听器。 +INSTALL = r"""(() => { + if (location.origin !== 'https://www.douyin.com') throw Error('请选择已登录的抖音标签页'); + const key = KEY, uid = UID, chunks = window.webpackChunkdouyin_web; + if (!chunks) throw Error('未找到抖音运行时'); + let require; + chunks.push([['notice-sub-' + key + '-' + Date.now()], {}, r => { require = r; }]); + chunks.pop(); + const entries = Object.entries(require.m); + const entry = entries.find(([, f]) => String(f).includes('NOTICE_PUSH_EVENT_NAMES:function')); + const codec = entries.find(([, f]) => { + const s = String(f); return s.includes('.decodedFrame=') && s.includes('.encodeFrame='); + }); + if (!entry || !codec) throw Error('通知 SDK 已变化,无法订阅'); + const C = require(entry[0]).NoticeFrontier; + const decode = require(codec[0]).decodedFrame; + const f = C.frontierInstance; + if (!f || String(f._options.deviceID) !== uid) throw Error('通知连接未就绪或登录账号已变化'); + const state = {queue: [], error: null, wake: null, C, f, uid}; + const emit = event => { + if (state.queue.length >= 1000) state.error = '订阅队列已满,可能丢失事件'; + else state.queue.push(event); + if (state.wake) state.wake(); + }; + const message = event => { + try { + const frame = decode(new Uint8Array(event.data)); + if (frame.service === 20313 || frame.service === 20003) + emit({kind: 'push', service: frame.service, payload: new TextDecoder().decode(frame.payload)}); + } catch (_) { state.error = '通知帧解码失败'; if (state.wake) state.wake(); } + }; + const open = () => emit({kind: 'open'}); + const close = () => emit({kind: 'close'}); + f.addEventListener('message', message); + f.addEventListener('open', open); + f.addEventListener('close', close); + state.dispose = () => { + f.removeEventListener('message', message); + f.removeEventListener('open', open); + f.removeEventListener('close', close); + if (state.wake) state.wake(); + }; + window[key] = state; + return JSON.stringify({connected: f.readyState === f.OPEN}); +})()""" + +WAIT = r"""(async () => { + const s = window[KEY]; + if (!s || s.C.frontierInstance !== s.f || String(s.f._options.deviceID) !== s.uid) + throw Error('页面刷新、登录失效或账号变化,请确认登录后重新运行'); + if (!s.queue.length && !s.error) await new Promise(resolve => { + // 本地桥接心跳,不请求通知接口;小于 harness 默认 IPC 超时。 + const timer = setTimeout(done, 2000); + function done() { clearTimeout(timer); s.wake = null; resolve(); } + s.wake = done; + }); + if (s.error) throw Error(s.error); + if (s.C.frontierInstance !== s.f) throw Error('通知连接已销毁,请检查登录态'); + return JSON.stringify(s.queue.splice(0)); +})()""" + + +def browser(expression): + expression = ( + "(async()=>{try{return await (" + + expression + + ");}catch(e){return JSON.stringify({bridge_error:String(e.message)});}})()" + ) + result = subprocess.run( + ["browser-harness"], + input=f"print(js({expression!r}))\n", + text=True, + capture_output=True, + timeout=30, + check=False, + ) + if result.returncode: + # 不回显包含账号等信息的完整表达式。 + raise RuntimeError( + "浏览器订阅操作失败:页面刷新、登录失效、SDK 变化或连接超时;请确认后重启" + ) + try: + value = json.loads(result.stdout.strip().splitlines()[-1]) + except (ValueError, IndexError) as exc: + raise RuntimeError("浏览器返回内容无法解析") from exc + if isinstance(value, dict) and "bridge_error" in value: + raise RuntimeError(value["bridge_error"]) + return value + + +def notice_ids(event): + """只处理互动通知服务,保留字符串 ID,避免 64 位精度损失。""" + try: + payload = json.loads(event["payload"]) + except (ValueError, TypeError) as exc: + raise RuntimeError("通知推送内容无法解析") from exc + if not isinstance(payload, dict): + raise TypeError("通知推送格式异常") + if event["service"] == 20313: + notices = payload.get("notices", []) + if not isinstance(notices, list) or any( + not isinstance(n, dict) or not isinstance(n.get("effect_groups", []), list) + for n in notices + ): + raise RuntimeError("通知推送列表格式异常") + items = [ + n + for n in notices + if {str(g) for g in n.get("effect_groups", [])} & {"960", "961"} + ] + elif event["service"] == 20003 and payload.get("notice_type") in ( + 45, + 31, + 9009, + 9002, + 514, + 9067, + ): + items = [payload] + else: + return [] + ids = [n.get("notice_id_str") for n in items] + if any(not isinstance(n, str) or not n.isascii() or not n.isdecimal() for n in ids): + raise RuntimeError("推送通知 ID 格式异常") + return list(dict.fromkeys(ids)) + + +def details(ids, uid): + url = "/aweme/v1/web/notice/detail/?" + urlencode( + { + "device_platform": "webapp", + "aid": 6383, + "is_mark_read": 0, + "id_list": json.dumps([{"notice_id_str": n, "type": 0} for n in ids]), + } + ) + expression = """(async () => { + if (location.origin !== 'https://www.douyin.com') throw Error('抖音页面已关闭'); + const r = await fetch(URL, {credentials: 'include', signal: AbortSignal.timeout(20000)}); + return JSON.stringify({status: r.status, body: await r.text()}); + })()""".replace("URL", json.dumps(url)) + # 推送可能先于详情入库;仅对这次事件重试,不做定时通知扫描。 + for attempt in range(3): + response = browser(expression) + try: + payload = json.loads(response["body"]) + except (ValueError, TypeError) as exc: + raise RuntimeError("通知详情无法解析,请检查登录态") from exc + if not isinstance(payload, dict): + raise TypeError("通知详情格式异常") + if response["status"] != 200 or payload.get("status_code") != 0: + raise RuntimeError("通知详情请求失败,请检查登录态;不会自动登录") + notices = payload.get("notice_list_v2") + if not isinstance(notices, list) or any( + not isinstance(n, dict) for n in notices + ): + raise RuntimeError("通知详情格式异常") + if any(str(n.get("user_id")) != str(uid) for n in notices): + raise RuntimeError("账号变化,停止订阅") + found = {n.get("nid_str") or str(n.get("nid")) for n in notices} + if found == set(ids): + return notices + if attempt < 2: + time.sleep(0.5 * (attempt + 1)) + raise RuntimeError("推送详情未完整返回,未处理通知 ID:" + ",".join(ids)) + + +def subscribe(output=None, last_id=None): + from get_notifications import collect_notifications, summarize + + user = get_user_from_browser() + key = "__douyin_notice_sub_" + uuid.uuid4().hex + seen = OrderedDict() + stream = None + + def emit(notice): + item = summarize(notice) + # 同一 ID 的聚合内容更新也输出;两路推送的相同详情只处理一次。 + signature = json.dumps(notice, sort_keys=True, ensure_ascii=False) + if seen.get(item["id"]) == signature: + return + record = { + "received_at": datetime.now(timezone.utc).isoformat(), + "notification": item, + "raw": notice, + } + line = json.dumps(record, ensure_ascii=False) + if stream: + stream.write(line + "\n") + stream.flush() + print(line, flush=True) + seen[item["id"]] = signature + seen.move_to_end(item["id"]) + # ponytail: 仅内存保留最近 4096 个 ID;跨重启幂等由消费方持久化处理。 + if len(seen) > 4096: + seen.popitem(last=False) + + try: + if output: + stream = output.open("a", encoding="utf-8") + state = browser( + INSTALL.replace("KEY", json.dumps(key)).replace( + "UID", json.dumps(str(user["uid"])) + ) + ) + print( + "订阅已安装,连接" + + ("已建立" if state["connected"] else "建立中") + + ";Ctrl+C 退出。stdout 为通知 JSONL,不自动标记已读。", + file=sys.stderr, + ) + if last_id: + notices, _, found = collect_notifications(user["uid"], last_id) + for notice in reversed(notices): + emit(notice) + if not found: + print("未找到 last-id,已补拉接口可获取的全部通知。", file=sys.stderr) + while True: + for event in browser(WAIT.replace("KEY", json.dumps(key))): + if event["kind"] != "push": + print( + "通知连接事件:" + + event["kind"] + + ";断线期间可能漏消息,可用 --last-id 补拉。", + file=sys.stderr, + ) + continue + ids = notice_ids(event) + for offset in range(0, len(ids), 3): + for notice in details(ids[offset : offset + 3], user["uid"]): + emit(notice) + finally: + # 页面已关闭时监听器也已销毁,不关闭站点自己的连接。 + with suppress(OSError, ValueError, RuntimeError, subprocess.SubprocessError): + browser( + "(() => {const k=" + + json.dumps(key) + + '; if(window[k]) {window[k].dispose(); delete window[k];} return "null";})()' + ) + if stream: + stream.close() diff --git a/src/test_subscribe_notifications.py b/src/test_subscribe_notifications.py new file mode 100644 index 0000000..eb4292a --- /dev/null +++ b/src/test_subscribe_notifications.py @@ -0,0 +1,211 @@ +"""离线检查(Python 标准库 + Node.js):python3 src/test_subscribe_notifications.py。""" + +import io +import json +import subprocess +from contextlib import redirect_stdout +from pathlib import Path +from tempfile import TemporaryDirectory +from unittest.mock import patch + +import subscribe_notifications as sub +from get_notifications import main + + +def push(service=20313, **payload): + return {"kind": "push", "service": service, "payload": json.dumps(payload)} + + +def check_raises(call): + try: + call() + except (RuntimeError, TypeError): + return + raise AssertionError("必须拒绝异常数据") + + +def test_ids(): + nid = "7680568855210951729" + item = {"notice_id_str": nid, "effect_groups": [960]} + assert sub.notice_ids(push(notices=[item, item])) == [nid] + assert sub.notice_ids(push(notices=[{**item, "effect_groups": ["961"]}])) == [nid] + assert sub.notice_ids(push(notices=[{**item, "effect_groups": [100]}])) == [] + assert sub.notice_ids(push(20003, notice_type=45, notice_id_str=nid)) == [nid] + assert sub.notice_ids(push(20003, notice_type=999, notice_id_str=nid)) == [] + assert sub.notice_ids(push(999)) == [] + for invalid in [int(nid), None, "12", "abc"]: + check_raises( + lambda invalid=invalid: sub.notice_ids( + push(notices=[{**item, "notice_id_str": invalid}]) + ) + ) + check_raises(lambda: sub.notice_ids({"payload": "invalid"})) + check_raises(lambda: sub.notice_ids({"payload": "[]"})) + check_raises(lambda: sub.notice_ids(push(notices=[None]))) + check_raises(lambda: sub.notice_ids(push(notices=[{"effect_groups": None}]))) + + +def test_details(): + notice = {"nid_str": "42", "user_id": "u"} + good = { + "status": 200, + "body": json.dumps({"status_code": 0, "notice_list_v2": [notice]}), + } + empty = { + "status": 200, + "body": json.dumps({"status_code": 0, "notice_list_v2": []}), + } + with ( + patch.object(sub, "browser", side_effect=[empty, good]) as call, + patch.object(sub.time, "sleep"), + ): + assert sub.details(["42"], "u") == [notice] + assert call.call_count == 2 + expression = call.call_args.args[0] + assert "/notice/detail/" in expression and "is_mark_read=0" in expression + assert "id_list=" in expression and "/notice/?" not in expression + with patch.object(sub, "browser", return_value=good): + check_raises(lambda: sub.details(["42"], "other")) + with ( + patch.object(sub, "browser", return_value=empty) as call, + patch.object(sub.time, "sleep"), + ): + check_raises(lambda: sub.details(["42"], "u")) + assert call.call_count == 3 + with patch.object(sub, "browser", return_value={"status": 403, "body": "{}"}): + check_raises(lambda: sub.details(["42"], "u")) + with patch.object(sub, "browser", return_value={"status": 200, "body": "[]"}): + check_raises(lambda: sub.details(["42"], "u")) + + +def test_subscription_and_cli(): + event = push(notices=[{"notice_id_str": "42", "effect_groups": [960]}]) + notice = {"nid_str": "42", "user_id": "u", "create_time": 1788271791} + calls = [] + waits = iter([[], [event, event], [event], KeyboardInterrupt()]) + + def browser(expression): + calls.append(expression) + if "const entries" in expression: + return {"connected": True} + if "s.queue.splice" in expression: + value = next(waits) + if isinstance(value, BaseException): + raise value + return value + return None + + with TemporaryDirectory() as folder: + output = Path(folder) / "events.jsonl" + output.write_text('{"existing":true}\n', encoding="utf-8") + with ( + patch.object(sub, "browser", side_effect=browser), + patch.object(sub, "get_user_from_browser", return_value={"uid": "u"}), + patch.object( + sub, + "details", + side_effect=[[notice], [notice], [{**notice, "type": 2}]], + ), + patch("sys.argv", ["get_notifications", "--sub", "--output", str(output)]), + patch("get_notifications.collect_notifications") as collect, + redirect_stdout(io.StringIO()) as stdout, + ): + assert main() == 130 + collect.assert_not_called() + lines = [json.loads(line) for line in stdout.getvalue().splitlines()] + assert len(lines) == 2 # 同内容重复丢弃,同 ID 内容变化保留。 + assert lines[0]["notification"]["id"] == "42" + assert len(output.read_text(encoding="utf-8").splitlines()) == 3 + assert ".dispose()" in calls[-1] + # 不带 --sub 仍走原分页拉取和 JSON 快照保存路径。 + with TemporaryDirectory() as folder: + output = Path(folder) / "snapshot.json" + with ( + patch("sys.argv", ["get_notifications", "--output", str(output)]), + patch("get_notifications.get_user_from_browser", return_value={"uid": "u"}), + patch( + "get_notifications.collect_notifications", + return_value=([notice], [{"has_more": 0}], False), + ), + patch.object(sub, "subscribe") as subscribe, + redirect_stdout(io.StringIO()), + ): + assert main() == 0 + subscribe.assert_not_called() + assert json.loads(output.read_text(encoding="utf-8"))["count"] == 1 + with ( + patch.object(sub, "subscribe") as subscribe, + patch("sys.argv", ["get_notifications", "--sub"]), + ): + assert main() == 0 + subscribe.assert_called_once_with(output=None, last_id=None) + with ( + patch.object(sub, "subscribe") as subscribe, + patch("sys.argv", ["get_notifications", "--sub", "--last-id", "42"]), + ): + assert main() == 0 + subscribe.assert_called_once_with(output=None, last_id="42") + + +def test_js_bridge(): + # 假 Webpack/FWS;只测自有监听器,不向真实账号发送通知。 + script = r""" +const assert = require('node:assert/strict'); +const input = JSON.parse(require('node:fs').readFileSync(0, 'utf8')); +const listeners = {}; +const f = {_options:{deviceID:'u'}, readyState:1, OPEN:1, + addEventListener(type, callback) { (listeners[type] ||= new Set()).add(callback); }, + removeEventListener(type, callback) { listeners[type].delete(callback); } +}; +const C = {frontierInstance:f}; +const decode = data => {const frame=JSON.parse(new TextDecoder().decode(data)); frame.payload=new Uint8Array(frame.payload); return frame;}; +const r = id => id === '1' ? {NoticeFrontier:C} : {decodedFrame:decode}; +r.m = {'1':function(){/* NOTICE_PUSH_EVENT_NAMES:function */}, + '2':function(){/* exports.decodedFrame= exports.encodeFrame= */}}; +global.window = {webpackChunkdouyin_web:[]}; +window.webpackChunkdouyin_web.push = function(chunk) { chunk[2](r); return Array.prototype.push.call(this, chunk); }; +global.location = {origin:'https://www.douyin.com'}; +(async () => { + assert.equal(JSON.parse(eval(input.install)).connected, true); + const send = frame => {for(const callback of listeners.message) callback({data:new TextEncoder().encode(JSON.stringify(frame))});}; + send({service:999,payload:[]}); + assert.equal(window.probe.queue.length,0); + const text = JSON.stringify({notices:[{notice_id_str:'7680568855210951729',effect_groups:[960]}],text:'测试'}); + send({service:20313,payload:[...new TextEncoder().encode(text)]}); + const events = JSON.parse(await eval(input.wait)); + assert.equal(events.length,1); + assert.equal(events[0].payload,text); + const start=Date.now(); + const pending=eval(input.wait); + setTimeout(()=>send({service:20003,payload:[...new TextEncoder().encode('{}')]}),20); + assert.equal(JSON.parse(await pending)[0].service,20003); + assert.ok(Date.now()-start<1000); // 收到事件立即唤醒,不等本地心跳。 + window.probe.dispose(); + assert.equal(listeners.message.size,0); + assert.equal(listeners.open.size,0); + assert.equal(listeners.close.size,0); + assert.equal(f.readyState,1); + C.frontierInstance=null; + await assert.rejects(eval(input.wait),/登录失效/); + console.log('JS bridge checks passed'); +})().catch(e=>{console.error(e);process.exitCode=1;}); +""" + data = { + "install": sub.INSTALL.replace("KEY", '"probe"').replace("UID", '"u"'), + "wait": sub.WAIT.replace("KEY", '"probe"'), + } + subprocess.run( + ["node", "-e", script], + input=json.dumps(data), + text=True, + check=True, + timeout=15, + ) + + +if __name__ == "__main__": + test_ids() + test_details() + test_subscription_and_cli() + test_js_bridge() + print("subscription checks passed")