宜博 / pixelphone

#!/usr/bin/env python3
"""环节1 · 通知触发 —— 微信新消息事件监听器

inotify 监听当前账号 EnMicroMsg.db 的写入(纯 ctypes,零额外 pip 包)→ 文件变化时
解密增量读 message 表(isSend=0 且 msgId>last)→ 分流(私聊/群@我/群引用我/群其它)
→ 追加到 incoming.jsonl 队列,供后续环节(读上下文/生成/发送)消费。

作为 root 在 Termux 内跑。inotify 建 watch 失败(SELinux/环境)时自动退回 stat mtime 轮询。

设备事实(2026-07 验证):
  账号目录 = MD5("mm"+uin) ；key = MD5("1234567890ABCDEF"+uin)[:7]
  SQLCipher: cipher_compatibility=1 + page_size=1024
  自己 wxid=userinfo[2]，昵称=userinfo[4]
群消息 content 形如 "wxid_sender:\\n<正文>"，需剥前缀；引用是 type=49 的 refermsg XML。

用法:
  python wechat_reply_watcher.py --selftest   # 自检:推导 key/解密/inotify 是否都 OK
  python wechat_reply_watcher.py --once        # 单次:读一遍新消息、打印、入队(测试用)
  python wechat_reply_watcher.py --tail 10     # 打印最近 N 条 inbound(不改状态)
  python wechat_reply_watcher.py               # 常驻 daemon(inotify 事件驱动)

ponytail: 读库走「拷贝整库→sqlcipher」。整库现 ~96MB,拷一次 ~2-3s;靠事件驱动+
MIN_READ_INTERVAL 合并突发把频率压到很低。上限=活跃群里瞬时多条→仍是一次读(去抖合并)。
升级路径:sqlcipher in-place immutable 只读打开省掉拷贝(WCDB 并发写会 database is locked,
需 immutable=1 URI 验证),或换真 NotificationListenerService 做 push。
"""
import ctypes
import glob
import hashlib
import json
import os
import re
import select
import subprocess
import sys
import time

MM_BASE = "/data/user/0/com.tencent.mm"
TERMUX_BIN = "/data/data/com.termux/files/usr/bin"
SQLCIPHER = f"{TERMUX_BIN}/sqlcipher"
HERMES_HOME = "/data/data/com.termux/files/home/.hermes"
WORK_DIR = f"{HERMES_HOME}/wechat_reply"
STATE_FILE = f"{WORK_DIR}/state.json"
CONFIG_FILE = f"{WORK_DIR}/config.json"
QUEUE_FILE = f"{WORK_DIR}/incoming.jsonl"
LOG_FILE = f"{WORK_DIR}/watcher.log"
IMEI = "1234567890ABCDEF"

# inotify 常量
IN_MODIFY = 0x00000002
IN_CLOSE_WRITE = 0x00000008
IN_MOVED_TO = 0x00000080
IN_CREATE = 0x00000100

MIN_READ_INTERVAL = 2.0   # 两次实际读库最小间隔(秒),合并突发
DEBOUNCE = 0.4            # 收到事件后等一小会儿让突发写落定
MTIME_POLL = 4.0         # inotify 不可用时的兜底轮询间隔


def log(msg):
    line = f"{time.strftime('%Y-%m-%d %H:%M:%S')} {msg}"
    print(line, flush=True)
    try:
        with open(LOG_FILE, "a") as f:
            f.write(line + "\n")
    except OSError:
        pass


def md5hex(s: str) -> str:
    return hashlib.md5(s.encode()).hexdigest()


# ── 账号 / key 推导 ────────────────────────────────────────────────────────
def read_uin() -> str:
    for xml in glob.glob(f"{MM_BASE}/shared_prefs/*.xml"):
        try:
            with open(xml, "r", errors="ignore") as f:
                m = re.search(r'last_login_uin">(-?\d+)', f.read())
                if m:
                    return m.group(1)
        except OSError:
            continue
    raise RuntimeError("没读到微信 last_login_uin(未登录?没 root?)")


def account_paths():
    uin = read_uin()
    acc_dir = md5hex("mm" + uin)              # = MD5("mm"+uin)，当前登录账号目录
    db = f"{MM_BASE}/MicroMsg/{acc_dir}/EnMicroMsg.db"
    if not os.path.exists(db):                # 兜底:定位不到就取任一个
        cands = glob.glob(f"{MM_BASE}/MicroMsg/*/EnMicroMsg.db")
        if not cands:
            raise RuntimeError("找不到 EnMicroMsg.db")
        db = cands[0]
    key = md5hex(IMEI + uin)[:7]
    return uin, db, key


# ── SQLCipher 读 ───────────────────────────────────────────────────────────
def sqlcipher_query(db_copy: str, key: str, sql: str):
    """返回 list[list[str]]。content 一律用 hex() 传出,躲开分隔符/换行问题。"""
    # PRAGMA key/compatibility 会往 stdout 各吐一行 "ok"，用 sentinel 只取其后的真结果。
    sentinel = "<<<ROWS>>>"
    script = (
        f"PRAGMA key='{key}';\n"
        "PRAGMA cipher_compatibility=1;\n"
        "PRAGMA cipher_page_size=1024;\n"
        ".mode list\n"
        ".separator \"\\t\"\n"
        ".headers off\n"
        f".print {sentinel}\n"
        f"{sql}\n"
    )
    p = subprocess.run([SQLCIPHER, db_copy], input=script,
                       capture_output=True, text=True, timeout=60)
    if p.returncode != 0 or "file is not a database" in p.stderr:
        raise RuntimeError(f"sqlcipher 失败: {p.stderr.strip()[:120]}")
    rows = []
    started = False
    for line in p.stdout.splitlines():
        if not started:
            if line == sentinel:
                started = True
            continue
        if line == "":
            continue
        rows.append(line.split("\t"))
    return rows


def copy_db(db: str, subdir: str = "_stage") -> str:
    """拷当前库(含 journal)到独立暂存目录,返回副本路径。root 直接文件拷,不经 /sdcard。
    subdir 隔离并发方(watcher=_stage / orchestrator=_stage_orch),避免共用文件互相删/覆盖
    导致 'no such table' 竞态。"""
    os.makedirs(f"{WORK_DIR}/{subdir}", exist_ok=True)
    dst = f"{WORK_DIR}/{subdir}/en.db"
    subprocess.run(["cp", db, dst], check=True, timeout=60)
    for ext in ("-journal", "-wal", "-shm"):
        if os.path.exists(db + ext):
            subprocess.run(["cp", db + ext, dst + ext], timeout=60)
    return dst


# ── 分类 / 解析 ────────────────────────────────────────────────────────────
def parse_group_content(raw: str):
    """群消息 content = 'wxid:\\n正文' → (sender_wxid, body)。私聊无前缀 → (None, raw)。"""
    m = re.match(r"^([A-Za-z0-9_@.\-]+):\n(.*)$", raw, re.S)
    if m:
        return m.group(1), m.group(2)
    return None, raw


def classify(msg_type: int, talker: str, raw: str, self_wxid: str, my_names):
    is_group = talker.endswith("@chatroom")
    if not is_group:
        return {"kind": "private", "sender": talker, "body": raw}
    sender, body = parse_group_content(raw)
    kind = "group_other"
    # 引用我:type 49 的 refermsg 里被引消息作者=我
    if msg_type == 49 and re.search(
            rf"<chatusr>\s*{re.escape(self_wxid)}\s*</chatusr>", raw):
        kind = "group_quote_me"
    else:
        # @我:正文里 '@昵称'(微信 @name 后跟 U+2005 窄空格,子串匹配即可)
        for nm in my_names:
            if nm and ("@" + nm) in body:
                kind = "group_at_me"
                break
    return {"kind": kind, "sender": sender, "body": body}


# ── 状态 / 配置 ────────────────────────────────────────────────────────────
def load_json(path, default):
    try:
        with open(path) as f:
            return json.load(f)
    except (OSError, ValueError):
        return default


def hermes_owner():
    try:
        st = os.stat(HERMES_HOME)
        return st.st_uid, st.st_gid
    except OSError:
        return None


def chown_to_hermes(path):
    o = hermes_owner()
    if o:
        try:
            os.chown(path, o[0], o[1])
        except OSError:
            pass


def save_state(state):
    with open(STATE_FILE, "w") as f:
        json.dump(state, f)
    chown_to_hermes(STATE_FILE)


def detect_self(db_copy, key):
    rows = sqlcipher_query(db_copy, key,
                           "SELECT id, value FROM userinfo WHERE id IN (2,4);")
    self_wxid, nickname = "", ""
    for r in rows:
        if r[0] == "2":
            self_wxid = r[1]
        elif r[0] == "4":
            nickname = r[1]
    return self_wxid, nickname


# ── 核心:读增量 → 入队 ─────────────────────────────────────────────────────
def read_new(emit=True):
    _, db, key = account_paths()
    stage = copy_db(db)
    try:
        state = load_json(STATE_FILE, {})
        cfg = load_json(CONFIG_FILE, {})
        if "last_msgId" not in state:
            # 首跑:基线 = 当前最大 msgId，不重放历史(否则把 2000+ 条旧消息当新消息)
            mx = sqlcipher_query(stage, key, "SELECT COALESCE(MAX(msgId),0) FROM message;")
            base = int(mx[0][0]) if mx and mx[0] else 0
            state["last_msgId"] = base
            save_state(state)
            log(f"首跑建立基线 last_msgId={base}(历史不重放)")
            return []
        last_id = int(state["last_msgId"])
        self_wxid = cfg.get("my_wxid") or ""
        nickname = ""
        if not self_wxid:
            self_wxid, nickname = detect_self(stage, key)
        my_names = list(cfg.get("my_names", []))
        if nickname and nickname not in my_names:
            my_names.append(nickname)

        rows = sqlcipher_query(stage, key,
            "SELECT msgId, type, talker, createTime, hex(content) "
            f"FROM message WHERE isSend=0 AND msgId>{last_id} "
            "ORDER BY msgId ASC LIMIT 200;")
        events = []
        max_id = last_id
        for r in rows:
            if len(r) < 5:
                continue
            msg_id, mtype, talker, ctime, chex = r
            msg_id = int(msg_id)
            max_id = max(max_id, msg_id)
            try:
                raw = bytes.fromhex(chex).decode("utf-8", "ignore")
            except ValueError:
                raw = ""
            info = classify(int(mtype), talker, raw, self_wxid, my_names)
            ev = {
                "msgId": msg_id,
                "type": int(mtype),
                "talker": talker,
                "createTime": int(ctime) if ctime.isdigit() else ctime,
                "isGroup": talker.endswith("@chatroom"),
                **info,
            }
            events.append(ev)

        if emit and events:
            with open(QUEUE_FILE, "a") as f:
                for ev in events:
                    f.write(json.dumps(ev, ensure_ascii=False) + "\n")
            chown_to_hermes(QUEUE_FILE)
        if max_id > last_id:
            state["last_msgId"] = max_id
            state["self_wxid"] = self_wxid
            save_state(state)
        return events
    finally:
        # 删明文副本
        for p in glob.glob(f"{WORK_DIR}/_stage/en.db*"):
            try:
                os.remove(p)
            except OSError:
                pass


# ── inotify ────────────────────────────────────────────────────────────────
def make_inotify(watch_dir):
    libc = ctypes.CDLL(None, use_errno=True)
    fd = libc.inotify_init()
    if fd < 0:
        return None, None
    mask = IN_MODIFY | IN_CLOSE_WRITE | IN_CREATE | IN_MOVED_TO
    wd = libc.inotify_add_watch(fd, watch_dir.encode(), mask)
    if wd < 0:
        os.close(fd)
        return None, None
    return libc, fd


def daemon():
    _, db, _ = account_paths()
    watch_dir = os.path.dirname(db)
    os.makedirs(WORK_DIR, exist_ok=True)
    chown_to_hermes(WORK_DIR)
    libc, fd = make_inotify(watch_dir)
    if fd is not None:
        log(f"inotify 监听 {watch_dir}")
        _watch_loop_inotify(fd)
    else:
        log(f"inotify 不可用,退回 mtime 轮询 {db}")
        _watch_loop_mtime(db)


def _trigger_orchestrator():
    """读到新消息后 fire-and-forget 触发编排器 drain（生成/策略/发送在独立进程，
    不阻塞 watcher；编排器自身有单实例锁）。缺文件/异常都不影响 watcher。"""
    orch = os.path.join(os.path.dirname(os.path.abspath(__file__)),
                        "wechat_reply_orchestrator.py")
    if not os.path.exists(orch):
        return
    try:
        import sys
        subprocess.Popen([sys.executable, orch, "--drain"],
                         stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
                         start_new_session=True)
    except Exception as e:  # noqa: BLE001
        log(f"触发编排器失败: {e}")


def _process(reason):
    try:
        evs = read_new()
        if evs:
            kinds = ",".join(e["kind"] for e in evs)
            log(f"[{reason}] +{len(evs)} 条 → {kinds} (至 msgId {evs[-1]['msgId']})")
            _trigger_orchestrator()
    except Exception as e:  # noqa: BLE001 — daemon 不能因单次读失败退出
        log(f"[{reason}] 读失败: {e}")


def _watch_loop_inotify(fd):
    _process("startup")
    last_read = time.time()
    while True:
        r, _, _ = select.select([fd], [], [], None)
        if not r:
            continue
        os.read(fd, 8192)                 # 清空事件(内容不重要,只当触发信号)
        time.sleep(DEBOUNCE)              # 让突发写落定
        try:                             # 榨干去抖期内堆积的事件
            while True:
                rr, _, _ = select.select([fd], [], [], 0)
                if not rr:
                    break
                os.read(fd, 8192)
        except OSError:
            pass
        wait = MIN_READ_INTERVAL - (time.time() - last_read)
        if wait > 0:
            time.sleep(wait)
        _process("inotify")
        last_read = time.time()


def _watch_loop_mtime(db):
    _process("startup")
    last_mtime = 0.0
    while True:
        try:
            m = os.stat(db).st_mtime
        except OSError:
            m = last_mtime
        if m != last_mtime:
            last_mtime = m
            _process("mtime")
        time.sleep(MTIME_POLL)


# ── 自检 / CLI ─────────────────────────────────────────────────────────────
def selftest():
    ok = True
    try:
        uin, db, key = account_paths()
        print(f"[OK] uin={uin} key={key}\n     db={db}")
    except Exception as e:  # noqa: BLE001
        print(f"[FAIL] 账号/key 推导: {e}")
        return False
    try:
        stage = copy_db(db)
        n = sqlcipher_query(stage, key, "SELECT count(*) FROM message;")[0][0]
        sw, nick = detect_self(stage, key)
        print(f"[OK] 解密成功 message={n} 条; self_wxid={sw} 昵称={nick}")
        for p in glob.glob(f"{WORK_DIR}/_stage/en.db*"):
            os.remove(p)
    except Exception as e:  # noqa: BLE001
        print(f"[FAIL] 解密: {e}")
        ok = False
    try:
        libc, fd = make_inotify(os.path.dirname(db))
        if fd is not None:
            print("[OK] inotify add_watch 成功(事件驱动可用)")
            os.close(fd)
        else:
            print("[WARN] inotify 不可用 → 会退回 mtime 轮询(仍能工作)")
    except Exception as e:  # noqa: BLE001
        print(f"[WARN] inotify 探测异常: {e} → 退回 mtime")
    return ok


def main():
    os.makedirs(WORK_DIR, exist_ok=True)
    args = sys.argv[1:]
    if "--selftest" in args:
        sys.exit(0 if selftest() else 1)
    if "--tail" in args:
        n = int(args[args.index("--tail") + 1]) if len(args) > args.index("--tail") + 1 else 10
        _, db, key = account_paths()
        stage = copy_db(db)
        rows = sqlcipher_query(stage, key,
            "SELECT msgId, type, talker, hex(content) FROM message "
            f"WHERE isSend=0 ORDER BY msgId DESC LIMIT {n};")
        for r in rows:
            body = bytes.fromhex(r[3]).decode("utf-8", "ignore").replace("\n", " ")[:60]
            print(f"  #{r[0]} type={r[1]} {r[2]}: {body}")
        for p in glob.glob(f"{WORK_DIR}/_stage/en.db*"):
            os.remove(p)
        return
    if "--once" in args:
        evs = read_new()
        print(f"新消息 {len(evs)} 条:")
        for e in evs:
            print(f"  [{e['kind']}] #{e['msgId']} {e.get('sender')}: "
                  f"{str(e.get('body',''))[:60]!r}")
        return
    daemon()


if __name__ == "__main__":
    main()