宜博 / pixelphone

Commit 58baa21db898c02f7cc530cc3ccb9fc7fc31f8f2

Authored by yibo
1 parent a393337d

feat(wechat-auto-reply): 环节1 通知触发 — inotify 事件驱动的新消息监听器 

wechat_reply_watcher.py：纯 ctypes inotify 监听当前账号 EnMicroMsg.db 目录
（SELinux 放行、零轮询），失败退回 stat mtime；解密增量读 message(isSend=0
& msgId>last)→ 分类(私聊/群@我/群引用我/群其它)→ 写 incoming.jsonl 队列。
账号运行时按 MD5("mm"+uin) 自适应，首跑建基线不重放历史。真机验证 inotify
触发读取、多账号切换自适应、解密/分类均通过。含 --selftest/--once/--tail。

配套需求文档 md/wechat-auto-reply-requirements.md 标注环节1 完成。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Inline Side-by-side
Showing 2 changed files with 533 additions and 0 deletions
md/wechat-auto-reply-requirements.md 0 → 100644
  View file @58baa21
  1 +# 微信自动回复 · 需求文档
  2 +
  3 +> 日期：2026-07-09
  4 +> 设备：Pixel（`59171FDCG0009N`），微信 8.0.74，LSPosed + Termux + Hermes Agent
  5 +> 定位：Pixel 上的 Hermes Agent 自动处理"别人发给小火子的微信"——私聊 / 群里 @ / 群里引用——判断要不要回、怎么回、自动发还是人工审核。
  6 +
  7 +---
  8 +
  9 +## 0. 目标
  10 +
  11 +有人给小火子发微信（私聊、群 @我、群引用我）时，Hermes 能：
  12 +1. 接收到这条消息
  13 +2. 读取会话上下文
  14 +3. 基于上下文 + SOUL.md 生成回复
  15 +4. 判断"直接发"还是"进人工审核队列"
  16 +5. 发送回复（含微信原生引用气泡）
  17 +
  18 +约束：尽量低电量 / 低资源；纯端上（Hermes 在设备 Termux 内跑，模型走 Kimi 或本地 llama-swap）。
  19 +
  20 +---
  21 +
  22 +## 1. 设备现状（建这套的地基，2026-07 核对过）
  23 +
  24 +| 能力 | 现状 | 备注 |
  25 +|---|---|---|
  26 +| 微信 DB 解密读 | ✅ 可用 | 主账号目录 `ec4f423b316d41e9af5e8398155cf105`，**key `1986554`**（`MD5("1234567890ABCDEF"+uin257966415)[:7]`）；参数 `cipher_compatibility=1`+`page_size=1024`。skill `~/.hermes/skills/wechat-db-decrypt/` 已就位 |
  27 +| 发送文本 | ✅ 可用 | LSPosed 内部 API（`am broadcast com.hermes.wechat.SEND_MSG`）；`WeChatHook.java` 已打通文本/语音/图片 |
  28 +| 发送带原生引用气泡 | ❌ 未实现 | 见环节 5 |
  29 +| 通知接收 | 待搭 | NotificationListenerService / dumpsys，见环节 1 |
  30 +| lifelog 人物库 | 明文 SQLite | `com.lemon.lifelog/databases/lifelog.db`（4.8MB，`Person` 表），二期打通 |
  31 +| SOUL.md | ✅ 存在 | `~/.hermes/SOUL.md`（Hermes 人格） |
  32 +
  33 +> ⚠️ 过时提醒：旧文档 `md/wechat-auto-reply-status.md`（2026-05-23）里的 key `e03ce81` / 目录 `653cdc…` 是**次账号**，已不是当前主账号；旧文档主推的 Frida daemon 发送已废弃（常驻注入微信触发反调试崩溃）。以本文件为准。
  34 +
  35 +---
  36 +
  37 +## 2. 五环节需求
  38 +
  39 +### 环节 1 · 接收通知 —— ✅ **已实现并真机验证（2026-07-09）**
  40 +落地：`wechat_plain/scripts/auto-reply/wechat_reply_watcher.py`（设备 `~/.hermes/wechat_reply/`）。
  41 +- **触发**：纯 ctypes inotify 监听当前账号 DB 目录（SELinux 放行，零轮询）；建 watch 失败自动退回 stat mtime 轮询。真机验证 `[inotify] +N` 事件驱动读取成立。
  42 +- **账号自适应**：运行时 `MD5("mm"+uin)` 定位当前登录账号（验证时账号已从 火星人 切到 小火子，daemon 自动跟上，无硬编码）。
  43 +- **增量+去重+基线**：持久化 `last_msgId`；只 `isSend=0 且 msgId>last`；首跑基线=当前最大 msgId，不重放历史。
  44 +- **分类**：private / group_at_me / group_quote_me / group_other；群消息剥 `wxid:\n` 前缀取 sender+body；自身 wxid/昵称从 userinfo 自动检测。
  45 +- **输出**：`incoming.jsonl` 事件队列（供环节 2-5 消费）。自检 `--selftest`、单读 `--once`、`--tail N`。
  46 +- 待接：常驻自启（Termux:Boot / Hermes cron），env4 里再过滤 gh_ 公众号与系统消息（撤回/拍一拍）。
  47 +
  48 +原始设计记录：
  49 +
  50 +### （原）环节 1 · 接收通知 —— **事件驱动，非定时轮询**
  51 +根因：费电的不是 cron 循环，是"每次轮询都 su 拷 90MB 的 EnMicroMsg.db"。定 5s 紧循环拷 90MB → 电量扛不住。
  52 +
  53 +- **主触发**：常驻 NotificationListenerService。系统主动 push，空闲近零成本；微信来消息才被唤醒 → 这时才读库。没消息 = 不动。
  54 +- **兜底**：Hermes cron 每几分钟扫一遍，捞免打扰 / 漏掉的消息。**不要跑紧循环。**
  55 +- **读库**：优先只读 in-place 打开（immutable）；锁住再拷。因为改成事件驱动，拷库只在真有消息时发生，频率极低，可接受。
  56 +- **消息源**：主账号 `ec4f…/EnMicroMsg.db` 的 `message` 表，`WHERE isSend=0 AND msgId>last_processed_id`。
  57 +- **分流（本环节难点）**：
  58 + - 私聊：直接是一条
  59 + - 群 @我：群消息 content 前缀 `wxid:\n`，需匹配我的昵称判 @
  60 + - 群引用我：看 refermsg 是否指向我的 svrid
  61 +
  62 +### 环节 2 · 读取上下文
  63 +- 从 `message` 表取该会话最近 N 条 + 发送者身份（`rcontact` 昵称/备注）。
  64 +- **二期加分**：回链 lifelog `Person` 库做人物增强（"这个人是谁、跟小火子什么关系"）。一期不做。
  65 +
  66 +### 环节 3 · 生成回复
  67 +- Hermes（Kimi / 本地 llama-swap）：**上下文 + SOUL.md → 回复**。
  68 +- 建议一次结构化调用同时出 `{should_reply, confidence, draft, risk_flags}`，别拆两次。
  69 +- ⚠️ 若回复要以"小火子本人"口吻发出，SOUL.md 里可能需加一段"小火子语气"约定（否则像 agent 替他说话）。
  70 +
  71 +### 环节 4 · 自动 vs 审核 —— **白名单直发，其余进 viewer 待回复**
  72 +- **白名单**（联系人 / 群 / 群里某发言人）+ 生成成功 → **直接发送**。
  73 +- **非白名单** → 写入 **Pixel 本地 wechat viewer 的"待回复列表"**，人工点击"发送"才真正发出。
  74 +- 落地新增：①「待回复队列」存储（sqlite/json）；② wechat viewer 的"发送"按钮接到发送链路。
  75 +- 安全红线（贯穿）：涉钱 / 承诺 / 约时间 / 低置信 → 即使白名单也建议进审核。
  76 +
  77 +### 环节 5 · 发送回复带引用（原生气泡）
  78 +需求 = 点原文→引用→发送后显示引用气泡。两条实现路：
  79 +- **(a) A11y UI 自动化（一期推荐）**：长按目标气泡 → 点"引用" → 输入 → 发送。驱动微信自己 UI → **产出货真价实的原生引用气泡，零逆向**。代价：需微信前台、脚本略脆、比 API 慢。已有 A11y 发送服务底子。
  80 +- **(b) 内部 API 构造 refermsg（二期优化）**：微信引用 = type 49 appmsg + refermsg(type 57) 指向被引 svrid。一整块新逆向，量级同当初啃发图，重。留作提速。
  81 +
  82 +---
  83 +
  84 +## 3. 贯穿层（不做会出事）
  85 +- **防循环/去重**：只处理 `isSend=0`；持久化 per-talker `last_processed_msgId`；绝不回自己发的、绝不重放。
  86 +- **限流**：每人/每群回复冷却，防抖防刷屏。
  87 +- **安全红线**：陌生人 / 涉钱 / 承诺 / 约时间 → 永不自动发，硬编码兜底。
  88 +
  89 +---
  90 +
  91 +## 4. 一期最小可跑（MVP）
  92 +
  93 +```
  94 +通知触发(NotificationListener)
  95 + └→ 读库(主账号 ec4f/1986554, message 表, isSend=0 & msgId>last)
  96 + └→ 分流(私聊/群@我/群引用我) + 取上下文(近 N 条 + rcontact 身份)
  97 + └→ Hermes 生成(上下文 + SOUL.md → {should_reply,confidence,draft})
  98 + ├→ 白名单 → A11y 发送(纯文本+@；引用气泡)
  99 + └→ 非白名单 → 写入 wechat viewer 待回复队列 → 人工点发送
  100 +```
  101 +
  102 +**二期**：lifelog `Person` 人物增强；内部 API refermsg 引用提速；cron 兜底调优。
  103 +
  104 +---
  105 +
  106 +## 5. 待决策 / 待确认
  107 +- [ ] 白名单如何维护（配置文件 / viewer 里管理 / 从 lifelog 拉）
  108 +- [ ] 待回复队列 + wechat viewer 的具体形态（现有 `wechat-web-viewer` skill 扩展？）
  109 +- [ ] SOUL.md 是否加"小火子语气"段
  110 +- [ ] 群 @我 / 引用我的判定用昵称还是 wxid（昵称易误判，wxid 更准）
  111 +- [ ] in-place 只读打开能否稳定绕过 WCDB 锁（否则回退拷库）
  112 +- [ ] 环节 2 lifelog join key：`Person` 里存的微信标识字段（二期确认）
... ...
wechat_plain/scripts/auto-reply/wechat_reply_watcher.py 0 → 100644
  View file @58baa21
  1 +#!/usr/bin/env python3
  2 +"""环节1 · 通知触发 —— 微信新消息事件监听器
  3 +
  4 +inotify 监听当前账号 EnMicroMsg.db 的写入(纯 ctypes,零额外 pip 包)→ 文件变化时
  5 +解密增量读 message 表(isSend=0 且 msgId>last)→ 分流(私聊/群@我/群引用我/群其它)
  6 +→ 追加到 incoming.jsonl 队列,供后续环节(读上下文/生成/发送)消费。
  7 +
  8 +作为 root 在 Termux 内跑。inotify 建 watch 失败(SELinux/环境)时自动退回 stat mtime 轮询。
  9 +
  10 +设备事实(2026-07 验证):
  11 + 账号目录 = MD5("mm"+uin) ；key = MD5("1234567890ABCDEF"+uin)[:7]
  12 + SQLCipher: cipher_compatibility=1 + page_size=1024
  13 + 自己 wxid=userinfo[2]，昵称=userinfo[4]
  14 +群消息 content 形如 "wxid_sender:\\n<正文>"，需剥前缀；引用是 type=49 的 refermsg XML。
  15 +
  16 +用法:
  17 + python wechat_reply_watcher.py --selftest # 自检:推导 key/解密/inotify 是否都 OK
  18 + python wechat_reply_watcher.py --once # 单次:读一遍新消息、打印、入队(测试用)
  19 + python wechat_reply_watcher.py --tail 10 # 打印最近 N 条 inbound(不改状态)
  20 + python wechat_reply_watcher.py # 常驻 daemon(inotify 事件驱动)
  21 +
  22 +ponytail: 读库走「拷贝整库→sqlcipher」。整库现 ~96MB,拷一次 ~2-3s;靠事件驱动+
  23 +MIN_READ_INTERVAL 合并突发把频率压到很低。上限=活跃群里瞬时多条→仍是一次读(去抖合并)。
  24 +升级路径:sqlcipher in-place immutable 只读打开省掉拷贝(WCDB 并发写会 database is locked,
  25 +需 immutable=1 URI 验证),或换真 NotificationListenerService 做 push。
  26 +"""
  27 +import ctypes
  28 +import glob
  29 +import hashlib
  30 +import json
  31 +import os
  32 +import re
  33 +import select
  34 +import subprocess
  35 +import sys
  36 +import time
  37 +
  38 +MM_BASE = "/data/user/0/com.tencent.mm"
  39 +TERMUX_BIN = "/data/data/com.termux/files/usr/bin"
  40 +SQLCIPHER = f"{TERMUX_BIN}/sqlcipher"
  41 +HERMES_HOME = "/data/data/com.termux/files/home/.hermes"
  42 +WORK_DIR = f"{HERMES_HOME}/wechat_reply"
  43 +STATE_FILE = f"{WORK_DIR}/state.json"
  44 +CONFIG_FILE = f"{WORK_DIR}/config.json"
  45 +QUEUE_FILE = f"{WORK_DIR}/incoming.jsonl"
  46 +LOG_FILE = f"{WORK_DIR}/watcher.log"
  47 +IMEI = "1234567890ABCDEF"
  48 +
  49 +# inotify 常量
  50 +IN_MODIFY = 0x00000002
  51 +IN_CLOSE_WRITE = 0x00000008
  52 +IN_MOVED_TO = 0x00000080
  53 +IN_CREATE = 0x00000100
  54 +
  55 +MIN_READ_INTERVAL = 2.0 # 两次实际读库最小间隔(秒),合并突发
  56 +DEBOUNCE = 0.4 # 收到事件后等一小会儿让突发写落定
  57 +MTIME_POLL = 4.0 # inotify 不可用时的兜底轮询间隔
  58 +
  59 +
  60 +def log(msg):
  61 + line = f"{time.strftime('%Y-%m-%d %H:%M:%S')} {msg}"
  62 + print(line, flush=True)
  63 + try:
  64 + with open(LOG_FILE, "a") as f:
  65 + f.write(line + "\n")
  66 + except OSError:
  67 + pass
  68 +
  69 +
  70 +def md5hex(s: str) -> str:
  71 + return hashlib.md5(s.encode()).hexdigest()
  72 +
  73 +
  74 +# ── 账号 / key 推导 ────────────────────────────────────────────────────────
  75 +def read_uin() -> str:
  76 + for xml in glob.glob(f"{MM_BASE}/shared_prefs/*.xml"):
  77 + try:
  78 + with open(xml, "r", errors="ignore") as f:
  79 + m = re.search(r'last_login_uin">(-?\d+)', f.read())
  80 + if m:
  81 + return m.group(1)
  82 + except OSError:
  83 + continue
  84 + raise RuntimeError("没读到微信 last_login_uin(未登录?没 root?)")
  85 +
  86 +
  87 +def account_paths():
  88 + uin = read_uin()
  89 + acc_dir = md5hex("mm" + uin) # = MD5("mm"+uin)，当前登录账号目录
  90 + db = f"{MM_BASE}/MicroMsg/{acc_dir}/EnMicroMsg.db"
  91 + if not os.path.exists(db): # 兜底:定位不到就取任一个
  92 + cands = glob.glob(f"{MM_BASE}/MicroMsg/*/EnMicroMsg.db")
  93 + if not cands:
  94 + raise RuntimeError("找不到 EnMicroMsg.db")
  95 + db = cands[0]
  96 + key = md5hex(IMEI + uin)[:7]
  97 + return uin, db, key
  98 +
  99 +
  100 +# ── SQLCipher 读 ───────────────────────────────────────────────────────────
  101 +def sqlcipher_query(db_copy: str, key: str, sql: str):
  102 + """返回 list[list[str]]。content 一律用 hex() 传出,躲开分隔符/换行问题。"""
  103 + # PRAGMA key/compatibility 会往 stdout 各吐一行 "ok"，用 sentinel 只取其后的真结果。
  104 + sentinel = "<<<ROWS>>>"
  105 + script = (
  106 + f"PRAGMA key='{key}';\n"
  107 + "PRAGMA cipher_compatibility=1;\n"
  108 + "PRAGMA cipher_page_size=1024;\n"
  109 + ".mode list\n"
  110 + ".separator \"\\t\"\n"
  111 + ".headers off\n"
  112 + f".print {sentinel}\n"
  113 + f"{sql}\n"
  114 + )
  115 + p = subprocess.run([SQLCIPHER, db_copy], input=script,
  116 + capture_output=True, text=True, timeout=60)
  117 + if p.returncode != 0 or "file is not a database" in p.stderr:
  118 + raise RuntimeError(f"sqlcipher 失败: {p.stderr.strip()[:120]}")
  119 + rows = []
  120 + started = False
  121 + for line in p.stdout.splitlines():
  122 + if not started:
  123 + if line == sentinel:
  124 + started = True
  125 + continue
  126 + if line == "":
  127 + continue
  128 + rows.append(line.split("\t"))
  129 + return rows
  130 +
  131 +
  132 +def copy_db(db: str) -> str:
  133 + """拷当前库(含 journal)到 Termux 私有 tmp,返回副本路径。root 直接文件拷,不经 /sdcard。"""
  134 + os.makedirs(f"{WORK_DIR}/_stage", exist_ok=True)
  135 + dst = f"{WORK_DIR}/_stage/en.db"
  136 + subprocess.run(["cp", db, dst], check=True, timeout=60)
  137 + for ext in ("-journal", "-wal", "-shm"):
  138 + if os.path.exists(db + ext):
  139 + subprocess.run(["cp", db + ext, dst + ext], timeout=60)
  140 + return dst
  141 +
  142 +
  143 +# ── 分类 / 解析 ────────────────────────────────────────────────────────────
  144 +def parse_group_content(raw: str):
  145 + """群消息 content = 'wxid:\\n正文' → (sender_wxid, body)。私聊无前缀 → (None, raw)。"""
  146 + m = re.match(r"^([A-Za-z0-9_@.\-]+):\n(.*)$", raw, re.S)
  147 + if m:
  148 + return m.group(1), m.group(2)
  149 + return None, raw
  150 +
  151 +
  152 +def classify(msg_type: int, talker: str, raw: str, self_wxid: str, my_names):
  153 + is_group = talker.endswith("@chatroom")
  154 + if not is_group:
  155 + return {"kind": "private", "sender": talker, "body": raw}
  156 + sender, body = parse_group_content(raw)
  157 + kind = "group_other"
  158 + # 引用我:type 49 的 refermsg 里被引消息作者=我
  159 + if msg_type == 49 and re.search(
  160 + rf"<chatusr>\s*{re.escape(self_wxid)}\s*</chatusr>", raw):
  161 + kind = "group_quote_me"
  162 + else:
  163 + # @我:正文里 '@昵称'(微信 @name 后跟 U+2005 窄空格,子串匹配即可)
  164 + for nm in my_names:
  165 + if nm and ("@" + nm) in body:
  166 + kind = "group_at_me"
  167 + break
  168 + return {"kind": kind, "sender": sender, "body": body}
  169 +
  170 +
  171 +# ── 状态 / 配置 ────────────────────────────────────────────────────────────
  172 +def load_json(path, default):
  173 + try:
  174 + with open(path) as f:
  175 + return json.load(f)
  176 + except (OSError, ValueError):
  177 + return default
  178 +
  179 +
  180 +def hermes_owner():
  181 + try:
  182 + st = os.stat(HERMES_HOME)
  183 + return st.st_uid, st.st_gid
  184 + except OSError:
  185 + return None
  186 +
  187 +
  188 +def chown_to_hermes(path):
  189 + o = hermes_owner()
  190 + if o:
  191 + try:
  192 + os.chown(path, o[0], o[1])
  193 + except OSError:
  194 + pass
  195 +
  196 +
  197 +def save_state(state):
  198 + with open(STATE_FILE, "w") as f:
  199 + json.dump(state, f)
  200 + chown_to_hermes(STATE_FILE)
  201 +
  202 +
  203 +def detect_self(db_copy, key):
  204 + rows = sqlcipher_query(db_copy, key,
  205 + "SELECT id, value FROM userinfo WHERE id IN (2,4);")
  206 + self_wxid, nickname = "", ""
  207 + for r in rows:
  208 + if r[0] == "2":
  209 + self_wxid = r[1]
  210 + elif r[0] == "4":
  211 + nickname = r[1]
  212 + return self_wxid, nickname
  213 +
  214 +
  215 +# ── 核心:读增量 → 入队 ─────────────────────────────────────────────────────
  216 +def read_new(emit=True):
  217 + _, db, key = account_paths()
  218 + stage = copy_db(db)
  219 + try:
  220 + state = load_json(STATE_FILE, {})
  221 + cfg = load_json(CONFIG_FILE, {})
  222 + if "last_msgId" not in state:
  223 + # 首跑:基线 = 当前最大 msgId，不重放历史(否则把 2000+ 条旧消息当新消息)
  224 + mx = sqlcipher_query(stage, key, "SELECT COALESCE(MAX(msgId),0) FROM message;")
  225 + base = int(mx[0][0]) if mx and mx[0] else 0
  226 + state["last_msgId"] = base
  227 + save_state(state)
  228 + log(f"首跑建立基线 last_msgId={base}(历史不重放)")
  229 + return []
  230 + last_id = int(state["last_msgId"])
  231 + self_wxid = cfg.get("my_wxid") or ""
  232 + nickname = ""
  233 + if not self_wxid:
  234 + self_wxid, nickname = detect_self(stage, key)
  235 + my_names = list(cfg.get("my_names", []))
  236 + if nickname and nickname not in my_names:
  237 + my_names.append(nickname)
  238 +
  239 + rows = sqlcipher_query(stage, key,
  240 + "SELECT msgId, type, talker, createTime, hex(content) "
  241 + f"FROM message WHERE isSend=0 AND msgId>{last_id} "
  242 + "ORDER BY msgId ASC LIMIT 200;")
  243 + events = []
  244 + max_id = last_id
  245 + for r in rows:
  246 + if len(r) < 5:
  247 + continue
  248 + msg_id, mtype, talker, ctime, chex = r
  249 + msg_id = int(msg_id)
  250 + max_id = max(max_id, msg_id)
  251 + try:
  252 + raw = bytes.fromhex(chex).decode("utf-8", "ignore")
  253 + except ValueError:
  254 + raw = ""
  255 + info = classify(int(mtype), talker, raw, self_wxid, my_names)
  256 + ev = {
  257 + "msgId": msg_id,
  258 + "type": int(mtype),
  259 + "talker": talker,
  260 + "createTime": int(ctime) if ctime.isdigit() else ctime,
  261 + "isGroup": talker.endswith("@chatroom"),
  262 + **info,
  263 + }
  264 + events.append(ev)
  265 +
  266 + if emit and events:
  267 + with open(QUEUE_FILE, "a") as f:
  268 + for ev in events:
  269 + f.write(json.dumps(ev, ensure_ascii=False) + "\n")
  270 + chown_to_hermes(QUEUE_FILE)
  271 + if max_id > last_id:
  272 + state["last_msgId"] = max_id
  273 + state["self_wxid"] = self_wxid
  274 + save_state(state)
  275 + return events
  276 + finally:
  277 + # 删明文副本
  278 + for p in glob.glob(f"{WORK_DIR}/_stage/en.db*"):
  279 + try:
  280 + os.remove(p)
  281 + except OSError:
  282 + pass
  283 +
  284 +
  285 +# ── inotify ────────────────────────────────────────────────────────────────
  286 +def make_inotify(watch_dir):
  287 + libc = ctypes.CDLL(None, use_errno=True)
  288 + fd = libc.inotify_init()
  289 + if fd < 0:
  290 + return None, None
  291 + mask = IN_MODIFY | IN_CLOSE_WRITE | IN_CREATE | IN_MOVED_TO
  292 + wd = libc.inotify_add_watch(fd, watch_dir.encode(), mask)
  293 + if wd < 0:
  294 + os.close(fd)
  295 + return None, None
  296 + return libc, fd
  297 +
  298 +
  299 +def daemon():
  300 + _, db, _ = account_paths()
  301 + watch_dir = os.path.dirname(db)
  302 + os.makedirs(WORK_DIR, exist_ok=True)
  303 + chown_to_hermes(WORK_DIR)
  304 + libc, fd = make_inotify(watch_dir)
  305 + if fd is not None:
  306 + log(f"inotify 监听 {watch_dir}")
  307 + _watch_loop_inotify(fd)
  308 + else:
  309 + log(f"inotify 不可用,退回 mtime 轮询 {db}")
  310 + _watch_loop_mtime(db)
  311 +
  312 +
  313 +def _process(reason):
  314 + try:
  315 + evs = read_new()
  316 + if evs:
  317 + kinds = ",".join(e["kind"] for e in evs)
  318 + log(f"[{reason}] +{len(evs)} 条 → {kinds} (至 msgId {evs[-1]['msgId']})")
  319 + except Exception as e: # noqa: BLE001 — daemon 不能因单次读失败退出
  320 + log(f"[{reason}] 读失败: {e}")
  321 +
  322 +
  323 +def _watch_loop_inotify(fd):
  324 + _process("startup")
  325 + last_read = time.time()
  326 + while True:
  327 + r, _, _ = select.select([fd], [], [], None)
  328 + if not r:
  329 + continue
  330 + os.read(fd, 8192) # 清空事件(内容不重要,只当触发信号)
  331 + time.sleep(DEBOUNCE) # 让突发写落定
  332 + try: # 榨干去抖期内堆积的事件
  333 + while True:
  334 + rr, _, _ = select.select([fd], [], [], 0)
  335 + if not rr:
  336 + break
  337 + os.read(fd, 8192)
  338 + except OSError:
  339 + pass
  340 + wait = MIN_READ_INTERVAL - (time.time() - last_read)
  341 + if wait > 0:
  342 + time.sleep(wait)
  343 + _process("inotify")
  344 + last_read = time.time()
  345 +
  346 +
  347 +def _watch_loop_mtime(db):
  348 + _process("startup")
  349 + last_mtime = 0.0
  350 + while True:
  351 + try:
  352 + m = os.stat(db).st_mtime
  353 + except OSError:
  354 + m = last_mtime
  355 + if m != last_mtime:
  356 + last_mtime = m
  357 + _process("mtime")
  358 + time.sleep(MTIME_POLL)
  359 +
  360 +
  361 +# ── 自检 / CLI ─────────────────────────────────────────────────────────────
  362 +def selftest():
  363 + ok = True
  364 + try:
  365 + uin, db, key = account_paths()
  366 + print(f"[OK] uin={uin} key={key}\n db={db}")
  367 + except Exception as e: # noqa: BLE001
  368 + print(f"[FAIL] 账号/key 推导: {e}")
  369 + return False
  370 + try:
  371 + stage = copy_db(db)
  372 + n = sqlcipher_query(stage, key, "SELECT count(*) FROM message;")[0][0]
  373 + sw, nick = detect_self(stage, key)
  374 + print(f"[OK] 解密成功 message={n} 条; self_wxid={sw} 昵称={nick}")
  375 + for p in glob.glob(f"{WORK_DIR}/_stage/en.db*"):
  376 + os.remove(p)
  377 + except Exception as e: # noqa: BLE001
  378 + print(f"[FAIL] 解密: {e}")
  379 + ok = False
  380 + try:
  381 + libc, fd = make_inotify(os.path.dirname(db))
  382 + if fd is not None:
  383 + print("[OK] inotify add_watch 成功(事件驱动可用)")
  384 + os.close(fd)
  385 + else:
  386 + print("[WARN] inotify 不可用 → 会退回 mtime 轮询(仍能工作)")
  387 + except Exception as e: # noqa: BLE001
  388 + print(f"[WARN] inotify 探测异常: {e} → 退回 mtime")
  389 + return ok
  390 +
  391 +
  392 +def main():
  393 + os.makedirs(WORK_DIR, exist_ok=True)
  394 + args = sys.argv[1:]
  395 + if "--selftest" in args:
  396 + sys.exit(0 if selftest() else 1)
  397 + if "--tail" in args:
  398 + n = int(args[args.index("--tail") + 1]) if len(args) > args.index("--tail") + 1 else 10
  399 + _, db, key = account_paths()
  400 + stage = copy_db(db)
  401 + rows = sqlcipher_query(stage, key,
  402 + "SELECT msgId, type, talker, hex(content) FROM message "
  403 + f"WHERE isSend=0 ORDER BY msgId DESC LIMIT {n};")
  404 + for r in rows:
  405 + body = bytes.fromhex(r[3]).decode("utf-8", "ignore").replace("\n", " ")[:60]
  406 + print(f" #{r[0]} type={r[1]} {r[2]}: {body}")
  407 + for p in glob.glob(f"{WORK_DIR}/_stage/en.db*"):
  408 + os.remove(p)
  409 + return
  410 + if "--once" in args:
  411 + evs = read_new()
  412 + print(f"新消息 {len(evs)} 条:")
  413 + for e in evs:
  414 + print(f" [{e['kind']}] #{e['msgId']} {e.get('sender')}: "
  415 + f"{str(e.get('body',''))[:60]!r}")
  416 + return
  417 + daemon()
  418 +
  419 +
  420 +if __name__ == "__main__":
  421 + main()
... ...