宜博 / pixelphone

#!/usr/bin/env python3
"""环节 1→5 编排器 —— 把整条自动回复链串起来

消费 watcher 的 incoming.jsonl（cursor 增量）→ 对候选事件（私聊/群@我/群引用我）：
  env2 查上下文（发送者身份 + 最近历史）→ env3 hermes -z 生成草稿 →
  env4 policy.decide()（白名单读 wechat-viewer 的 circle.db）→
  env5 落地：auto 直接 send_text_via_hook + 入队(auto_sent) / review 入队(pending，在 /approvals 等人工)

作为 root 在 Termux 跑（同 watcher 上下文）。由 watcher 每次读到新消息后触发 --drain。

安全：policy 默认 auto_send_enabled=False → 一切进 review（不自动发）。生成超时/失败 →
无草稿 → policy skip → 不发（失败即静默，绝不误发）。

ponytail: 每次 drain 拷一次整库(~96MB)供该批所有事件查上下文；事件稀疏可接受。
升级：与 watcher 共享一次拷贝；incoming.jsonl 增长需按 offset 游标+轮转。
"""
import json
import os
import sqlite3
import subprocess
import sys
import time

_HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, _HERE)
import wechat_reply_watcher as w   # 复用解密/查询/账号推导
import wechat_reply_policy as policy

WORK_DIR = w.WORK_DIR
QUEUE_FILE = f"{WORK_DIR}/incoming.jsonl"
CURSOR_FILE = f"{WORK_DIR}/orch_state.json"
ORCH_LOG = f"{WORK_DIR}/orchestrator.log"
CIRCLE_DB = "/data/data/com.termux/files/home/wechat-viewer/circle.db"
EVENT_BUS = "/data/data/com.termux/files/home/.hermes/event_bus"
HERMES = "/data/data/com.termux/files/usr/bin/hermes"

CANDIDATE_KINDS = {"private", "group_at_me", "group_quote_me"}
HISTORY_N = 12
GEN_TIMEOUT = 160


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


# ── env2：查上下文 ─────────────────────────────────────
def _display_name(stage, key, wxid):
    if not wxid:
        return ""
    rows = w.sqlcipher_query(
        stage, key,
        f"SELECT conRemark, nickname, alias FROM rcontact WHERE username='{wxid}';")
    if rows and rows[0]:
        r = rows[0] + ["", "", ""]
        return r[0] or r[1] or r[2] or wxid
    return wxid


def _placeholder(mtype, text):
    if mtype == 1:
        return text
    return {3: "[图片]", 34: "[语音]", 43: "[视频]", 47: "[表情]",
            42: "[名片]", 49: "[链接/文件]", 48: "[位置]"}.get(mtype, text[:40] if text else "[消息]")


def build_context(stage, key, event, self_wxid):
    talker = event.get("talker", "")
    is_group = event.get("isGroup") or talker.endswith("@chatroom")
    sender_wxid = event.get("sender") or talker
    sender_disp = _display_name(stage, key, sender_wxid)
    lines = [f"对方：{sender_disp}"]
    if is_group:
        gname = _display_name(stage, key, talker)
        lines.insert(0, f"[群] {gname}")
    # 最近历史
    rows = w.sqlcipher_query(
        stage, key,
        "SELECT isSend, type, createTime, hex(content) FROM message "
        f"WHERE talker='{talker}' ORDER BY msgId DESC LIMIT {HISTORY_N};")
    hist = []
    for r in reversed(rows):
        if len(r) < 4:
            continue
        is_send, mtype, _, chex = r
        try:
            raw = bytes.fromhex(chex).decode("utf-8", "ignore")
        except ValueError:
            raw = ""
        if is_send == "1":
            who, body = "我", raw
        else:
            s, body = w.parse_group_content(raw) if is_group else (sender_wxid, raw)
            who = _display_name(stage, key, s) if is_group and s else sender_disp
        txt = _placeholder(int(mtype) if mtype.isdigit() else 0, body).replace("\n", " ").strip()
        if txt:
            hist.append(f"  {who}: {txt}")
    ctx = "\n".join(lines)
    if hist:
        ctx += "\n最近对话：\n" + "\n".join(hist[-HISTORY_N:])
    ctx += f"\n本条：{sender_disp}: {(event.get('body') or '').strip()[:200]}"
    return ctx, sender_disp


# ── env3：hermes 生成 ─────────────────────────────────
def generate(context_text):
    prompt = (
        "【任务】下面是别人发到微信来的一条消息，请以你本人（火星人）的口吻，写一条你会真的发出去的"
        "微信回复。只输出这条回复本身，不要解释、不要引号、不要用任何工具、不要真的发送。\n\n"
        "【上下文】\n" + context_text)
    env = dict(os.environ)
    env.update({
        "HOME": "/data/data/com.termux/files/home",
        "PREFIX": "/data/data/com.termux/files/usr",
        "PATH": "/data/data/com.termux/files/usr/bin:/system/bin",
        "LD_LIBRARY_PATH": "/data/data/com.termux/files/usr/lib",
    })
    try:
        p = subprocess.run([HERMES, "-z", prompt], capture_output=True, text=True,
                           timeout=GEN_TIMEOUT, env=env)
        out = (p.stdout or "").strip()
        # 取最后一段非空文本（躲开可能的日志行）
        return out.split("\n")[-1].strip() if out else ""
    except subprocess.TimeoutExpired:
        log("生成超时")
        return ""
    except Exception as e:  # noqa: BLE001
        log(f"生成异常: {e}")
        return ""


# ── env4→5：落地 ─────────────────────────────────────
def _load_whitelist():
    contacts, groups = [], []
    try:
        con = sqlite3.connect(CIRCLE_DB)
        for kind, wxid in con.execute("SELECT kind, wxid FROM reply_whitelist"):
            (groups if kind == "group" else contacts).append(wxid)
        con.close()
    except Exception:
        pass
    return contacts, groups


def _queue_insert(event, draft, sender_disp, action, reason, status):
    con = sqlite3.connect(CIRCLE_DB)
    con.execute(
        """INSERT INTO reply_queue
           (msg_id, talker, sender, sender_name, kind, incoming, draft, action, reason, status, sent_at)
           VALUES (?,?,?,?,?,?,?,?,?,?, CASE WHEN ?='auto_sent' THEN datetime('now','localtime') ELSE NULL END)""",
        (event.get("msgId"), event.get("talker"), event.get("sender"), sender_disp,
         event.get("kind"), event.get("body"), draft, action, reason, status, status))
    con.commit()
    con.close()


def _send_text(to_wxid, text):
    sys.path.insert(0, EVENT_BUS)
    from wechat_hook_send import send_text_via_hook
    return send_text_via_hook(to_wxid, text)


def process_event(event, stage, key, self_wxid, cfg):
    ctx, sender_disp = build_context(stage, key, event, self_wxid)
    draft = generate(ctx)
    d = policy.decide(event, draft, self_wxid=self_wxid, cfg=cfg)
    action, reason = d["action"], d["reason"]
    if action == "skip":
        log(f"skip #{event.get('msgId')} {sender_disp}: {reason}")
        return
    if action == "auto":
        res = _send_text(event.get("talker"), draft)
        ok = bool(res.get("ok") or res.get("status") == "sent")
        _queue_insert(event, draft, sender_disp, "auto",
                      reason, "auto_sent" if ok else "failed")
        log(f"auto {'OK' if ok else 'FAIL'} #{event.get('msgId')} {sender_disp} → {draft!r}")
    else:  # review
        _queue_insert(event, draft, sender_disp, "review", reason, "pending")
        log(f"review #{event.get('msgId')} {sender_disp}: {reason} → 草稿 {draft!r}")


# ── drain：消费 incoming.jsonl ─────────────────────────
def _read_events():
    evs = []
    try:
        with open(QUEUE_FILE) as f:
            for line in f:
                line = line.strip()
                if line:
                    try:
                        evs.append(json.loads(line))
                    except ValueError:
                        pass
    except OSError:
        pass
    return evs


def drain():
    """非阻塞单实例：已有 drain 在跑就跳过（后续消息的 inotify 会再触发）。"""
    lock = f"{WORK_DIR}/orch.lock"
    try:
        fd = os.open(lock, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
        os.close(fd)
    except FileExistsError:
        try:  # 5min 以上视为残留锁，抢占
            if time.time() - os.path.getmtime(lock) > 300:
                os.remove(lock)
                os.close(os.open(lock, os.O_CREAT | os.O_EXCL | os.O_WRONLY))
            else:
                return
        except OSError:
            return
    try:
        _drain_inner()
    finally:
        try:
            os.remove(lock)
        except OSError:
            pass


def _drain_inner():
    evs = _read_events()
    if not evs:
        return
    try:
        with open(CURSOR_FILE) as f:
            cursor = json.load(f).get("last_msgId", 0)
    except (OSError, ValueError):
        cursor = None
    max_id = max((int(e.get("msgId", 0)) for e in evs), default=0)
    if cursor is None:  # 首跑：基线，不回放
        with open(CURSOR_FILE, "w") as f:
            json.dump({"last_msgId": max_id}, f)
        w.chown_to_hermes(CURSOR_FILE)
        log(f"编排器首跑基线 last_msgId={max_id}")
        return
    todo = [e for e in evs if int(e.get("msgId", 0)) > cursor
            and e.get("kind") in CANDIDATE_KINDS]
    if todo:
        _, db, key = w.account_paths()
        stage = w.copy_db(db)
        try:
            self_wxid, _ = w.detect_self(stage, key)
            cfg = policy.load_config()
            cfg["whitelist_contacts"], cfg["whitelist_groups"] = _load_whitelist()
            log(f"drain：{len(todo)} 条候选（auto_send={cfg['auto_send_enabled']}）")
            for e in todo:
                try:
                    process_event(e, stage, key, self_wxid, cfg)
                except Exception as ex:  # noqa: BLE001
                    log(f"处理 #{e.get('msgId')} 失败: {ex}")
        finally:
            for p in __import__("glob").glob(f"{WORK_DIR}/_stage/en.db*"):
                try:
                    os.remove(p)
                except OSError:
                    pass
    with open(CURSOR_FILE, "w") as f:
        json.dump({"last_msgId": max_id}, f)
    w.chown_to_hermes(CURSOR_FILE)


def main():
    args = sys.argv[1:]
    if "--drain" in args:
        drain()
    else:
        print("用法: wechat_reply_orchestrator.py --drain")


if __name__ == "__main__":
    main()