feat: add push subscription mode for interaction notifications

.gitignore

@@ -23,6 +23,7 @@ dist/
 *.pem
 *.key
 *.har
+*.jsonl
 *cookie*.json
 *cookie*.txt
 browser-profile/

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"
+```
+
+手动打开抖音并完成登录，再运行脚本；不会自动登录。已有登录浏览器直接复用，不必另建配置。安装参考：<https://github.com/browser-use/browser-harness/blob/main/install.md>。本次在现有环境验证，未声称在空白机器重新安装验证。
+
+```bash
+# 离线检查；JS 桥接测试要求 node 在 PATH 中
+python3 src/test_notifications.py
+python3 src/test_subscribe_notifications.py
+```
+
+`.gitignore` 已新增 `*.jsonl`，防止通知原文、用户资料等运行输出进入公共仓库。本文不记录真实账号或通知 ID。

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,

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()

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, "１２", "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")