宜博 / pixelphone

#!/usr/bin/env python3
"""
Hermes WeChat 本地守护进程 — Pixel Termux 闭环运行

架构（HTTP API 模式）：
  1. 监控 Termux 通知 / 定期轮询
  2. 复制+解密微信 DB
  3. 读取新消息
  4. 通过 HTTP POST 调用 hermes /api/master/chat（SSE）
  5. 提取 hermes 返回的回复文本
  6. 通过 am broadcast 发送回复

Usage:
    python3 hermes_wechat_daemon.py              # 前台运行
    python3 hermes_wechat_daemon.py --daemon     # 后台运行 (fork)
    python3 hermes_wechat_daemon.py --once       # 单次轮询（测试用）
    python3 hermes_wechat_daemon.py --config custom.json
"""
import subprocess, os, sys, json, time, sqlite3, logging, argparse, urllib.request, urllib.error
from datetime import datetime
from pathlib import Path

# ── Configuration ─────────────────────────────────────────────────────────────
DEFAULT_CONFIG = {
    "db_src_path": "",
    "db_copy_dir": "/data/data/com.termux/files/home/dbs",
    "msg_log_file": "/data/data/com.termux/files/home/wechat_msgs.jsonl",
    "sqlcipher_key": "e03ce81",
    "cipher_compatibility": 1,
    "cipher_page_size": 1024,
    "poll_interval_sec": 10,
    "max_msgs_per_poll": 50,
    "hermes": {
        "api_url": "http://192.168.110.15:3010/api/master/chat",
        "timeout_sec": 120,
        "default_session_prefix": "wx-",
        "system_hint": "你是一个微信智能助手。根据收到的消息，给出简短、自然、得体的中文回复。群聊中被@时要友好回应。"
    },
    "auto_reply": {
        "enabled": False,
        "contacts": ["文件传输助手"],
        "exclude_types": [10000],
        "cooldown_sec": 5
    },
    "send": {
        "method": "am_broadcast",
        "broadcast_action": "com.hermes.wechat.SEND_MSG",
        "am_path": "/system/bin/am"
    },
    "contact_display_names": {
        "文件传输助手": "文件传输助手"
    },
    "logging": {
        "level": "INFO",
        "file": "/data/data/com.termux/files/home/daemon.log"
    }
}

# ── Globals ───────────────────────────────────────────────────────────────────
config = {}
last_msg_id = 0
last_reply_time = {}  # talker -> timestamp
urgent_poll_needed = False
known_notif_keys = set()

# ── Logging ───────────────────────────────────────────────────────────────────
def setup_logging(cfg):
    level = getattr(logging, cfg.get("level", "INFO").upper(), logging.INFO)
    log_file = cfg.get("file", "/sdcard/hermes/daemon.log")
    os.makedirs(os.path.dirname(log_file), exist_ok=True)

    handlers = [logging.StreamHandler(sys.stdout)]
    try:
        handlers.append(logging.FileHandler(log_file))
    except Exception as e:
        print(f"[WARN] Cannot open log file {log_file}: {e}")

    logging.basicConfig(
        level=level,
        format="%(asctime)s [%(levelname)s] %(message)s",
        handlers=handlers
    )

# ── Config ────────────────────────────────────────────────────────────────────
def load_config(path="config.json"):
    global config
    cfg = DEFAULT_CONFIG.copy()
    if os.path.exists(path):
        try:
            with open(path) as f:
                user_cfg = json.load(f)
            # Deep merge for nested dicts
            for key, val in user_cfg.items():
                if isinstance(val, dict) and key in cfg and isinstance(cfg[key], dict):
                    cfg[key].update(val)
                else:
                    cfg[key] = val
            logging.info(f"Config loaded from {path}")
        except Exception as e:
            logging.error(f"Failed to load config: {e}")
    config = cfg
    return cfg

# ── DB Discovery ──────────────────────────────────────────────────────────────
def find_enmicromsg_db():
    """Auto-discover EnMicroMsg.db path."""
    result = subprocess.run(
        ["su", "-c", "find /data/data/com.tencent.mm/MicroMsg -name 'EnMicroMsg.db' -type f 2>/dev/null"],
        capture_output=True, text=True
    )
    paths = [p.strip() for p in result.stdout.strip().split("\n") if p.strip()]
    return paths[0] if paths else None

# ── DB Copy ───────────────────────────────────────────────────────────────────
def copy_db(src, dst_dir):
    """Copy DB + WAL + SHM. Returns list of copied file paths."""
    os.makedirs(dst_dir, exist_ok=True)
    files = [src, src + "-wal", src + "-shm"]
    copied = []
    for f in files:
        dst = os.path.join(dst_dir, os.path.basename(f))
        subprocess.run(["su", "-c", f"cp '{f}' '{dst}' 2>/dev/null || true"],
                       capture_output=True)
        if os.path.exists(dst):
            copied.append(dst)
    return copied

# ── DB Decrypt ────────────────────────────────────────────────────────────────
def decrypt_db(encrypted_path, output_path, key, page_size=1024, compat=1):
    """Decrypt SQLCipher DB. Returns True/False."""
    if os.path.exists(output_path):
        os.remove(output_path)

    sql = f"""
PRAGMA key = "{key}";
PRAGMA cipher_compatibility = {compat};
PRAGMA cipher_page_size = {page_size};
PRAGMA wal_checkpoint(TRUNCATE);
ATTACH DATABASE '{output_path}' AS plain KEY '';
SELECT sqlcipher_export('plain');
DETACH DATABASE plain;
.quit
"""
    # Use full path to sqlcipher from Termux
    sqlcipher_path = "/data/data/com.termux/files/usr/bin/sqlcipher"
    if not os.path.exists(sqlcipher_path):
        sqlcipher_path = "sqlcipher"
    r = subprocess.run(
        [sqlcipher_path, encrypted_path],
        input=sql, capture_output=True, text=True, timeout=60
    )
    ok = r.returncode == 0 and os.path.exists(output_path) and os.path.getsize(output_path) > 4096
    print(f"[DECRYPT] rc={r.returncode} stderr={r.stderr[:80]!r} out_exists={os.path.exists(output_path)} out_size={os.path.getsize(output_path) if os.path.exists(output_path) else 0} ok={ok}", flush=True)
    if not ok:
        # Fallback: reuse existing decrypted DB from sdcard
        fallback = "/data/data/com.termux/files/home/EnMicroMsg.db.decrypted.db"
        if os.path.exists(fallback):
            try:
                import shutil
                shutil.copy2(fallback, output_path)
                print(f"[DECRYPT] Fallback to existing decrypted DB: {fallback} -> {output_path}", flush=True)
                return True
            except Exception as e:
                print(f"[DECRYPT] Fallback copy failed: {e}", flush=True)
    return ok

# ── Message Query ─────────────────────────────────────────────────────────────
def query_new_messages(db_path, after_msg_id, limit=50):
    """Query messages with msgId > after_msg_id, newest first."""
    try:
        conn = sqlite3.connect(db_path)
        conn.row_factory = sqlite3.Row
        rows = conn.execute(
            """SELECT msgId, talker, content, type, isSend, createTime
               FROM message WHERE msgId > ? ORDER BY msgId DESC LIMIT ?""",
            (after_msg_id, limit)
        ).fetchall()
        conn.close()
        # Reverse to process in chronological order
        return list(reversed([dict(r) for r in rows]))
    except Exception as e:
        logging.error(f"DB query failed: {e}")
        return []

def get_max_msg_id(db_path):
    """Get current max msgId."""
    try:
        conn = sqlite3.connect(db_path)
        row = conn.execute("SELECT MAX(msgId) FROM message").fetchone()
        conn.close()
        return row[0] or 0
    except Exception:
        return 0

def get_recent_context(db_path, talker, before_msg_id, limit=5):
    """Get recent context messages for a talker before given msgId."""
    try:
        conn = sqlite3.connect(db_path)
        conn.row_factory = sqlite3.Row
        rows = conn.execute(
            """SELECT msgId, talker, content, type, isSend, createTime
               FROM message
               WHERE talker = ? AND msgId < ?
               ORDER BY msgId DESC LIMIT ?""",
            (talker, before_msg_id, limit)
        ).fetchall()
        conn.close()
        return list(reversed([dict(r) for r in rows]))
    except Exception as e:
        logging.debug(f"Context query failed: {e}")
        return []

# ── LLM API Call ──────────────────────────────────────────────────────────────
def call_hermes_api(talker, content, context_msgs=None):
    """
    Call configured LLM API (Anthropic / OpenAI / Azure / local proxy).
    Returns reply text or None.
    """
    hermes_cfg = config.get("hermes", {})
    api_url = hermes_cfg.get("api_url", "")
    api_key = hermes_cfg.get("api_key", "")
    provider = hermes_cfg.get("provider", "anthropic")
    model = hermes_cfg.get("model", "")
    timeout = hermes_cfg.get("timeout_sec", 120)
    prefix = hermes_cfg.get("default_session_prefix", "wx-")
    system_hint = hermes_cfg.get("system_hint", "")

    session_id = f"{prefix}{talker}"

    # Build messages
    messages = []
    if system_hint:
        messages.append({"role": "system", "content": system_hint})
    if context_msgs:
        ctx_lines = []
        for m in context_msgs:
            who = "对方" if m.get("isSend") == 0 else "我"
            ctx_lines.append(f"{who}: {m.get('content', '')[:100]}")
        messages.append({"role": "system", "content": "最近对话上下文：\n" + "\n".join(ctx_lines)})
    messages.append({"role": "user", "content": f"收到新消息（来自 {talker}）：{content}\n请直接给出回复内容，不要加任何解释或格式标记。"})

    # Provider-specific request building
    headers = {"Content-Type": "application/json"}
    if provider == "anthropic":
        if not api_url:
            api_url = "https://api.anthropic.com/v1/messages"
        if not model:
            model = "claude-sonnet-4-6-20251001"
        headers["x-api-key"] = api_key
        headers["anthropic-version"] = "2023-06-01"
        payload = json.dumps({
            "model": model,
            "max_tokens": 1024,
            "messages": messages,
            "metadata": {"session_id": session_id}
        }).encode("utf-8")
    elif provider in ("openai", "azure", "doubao"):
        headers["Authorization"] = f"Bearer {api_key}"
        payload = json.dumps({
            "model": model,
            "messages": messages,
            "stream": False,
            "temperature": 0.7,
        }).encode("utf-8")
    else:
        logging.error(f"[LLM] Unknown provider: {provider}")
        return None

    req = urllib.request.Request(api_url, data=payload, headers=headers, method="POST")

    try:
        print(f"[LLM] Calling {provider} API for {talker}...", flush=True)
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            print(f"[LLM] API connected (status={resp.status})", flush=True)
            data = json.loads(resp.read().decode("utf-8"))

        # Parse response
        result_text = None
        if provider == "anthropic":
            result_text = data.get("content", [{}])[0].get("text", "")
        elif provider in ("openai", "azure", "doubao"):
            result_text = data.get("choices", [{}])[0].get("message", {}).get("content", "")

        if result_text:
            print(f"[LLM] Reply for {talker}: {result_text[:80]}", flush=True)
            logging.info(f"[LLM] Reply for {talker}: {result_text[:60]}")
        else:
            print(f"[LLM] Empty reply for {talker}", flush=True)
        return result_text if result_text else None
    except urllib.error.HTTPError as e:
        body = e.read().decode("utf-8", errors="replace")[:200]
        print(f"[LLM] HTTP {e.code}: {body}", flush=True)
        logging.error(f"[LLM] HTTP {e.code}: {body}")
        return None
    except urllib.error.URLError as e:
        print(f"[LLM] Connection failed: {e}", flush=True)
        logging.error(f"[LLM] Connection failed: {e}")
        return None
    except Exception as e:
        print(f"[LLM] API call failed: {e}", flush=True)
        logging.error(f"[LLM] API call failed: {e}")
        return None

# ── Send Message ──────────────────────────────────────────────────────────────
def send_message(to_user, content, to_user_display_name=None):
    """Send text message via am broadcast."""
    send_cfg = config.get("send", {})
    method = send_cfg.get("method", "am_broadcast")
    am_path = send_cfg.get("am_path", "/system/bin/am")
    action = send_cfg.get("broadcast_action", "com.hermes.wechat.SEND_MSG")

    if to_user_display_name is None:
        to_user_display_name = to_user

    # Ensure WeChat is in foreground before sending
    if not is_wechat_foreground():
        wake_wechat()
        time.sleep(2)

    if method == "am_broadcast":
        cmd = [
            am_path, "broadcast",
            "-a", action,
            "--es", "toUser", to_user,
            "--es", "toUserDisplayName", to_user_display_name,
            "--es", "content", content
        ]
    else:
        logging.error(f"Unknown send method: {method}")
        return False

    try:
        r = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
        ok = r.returncode == 0
        if ok:
            print(f"[SEND] -> {to_user}: {content[:60]}", flush=True)
            logging.info(f"[SEND] -> {to_user}: {content[:40]}")
        else:
            print(f"[SEND FAIL] rc={r.returncode} stderr={r.stderr[:100]}", flush=True)
            logging.error(f"[SEND FAIL] rc={r.returncode} stderr={r.stderr[:100]}")
        return ok
    except Exception as e:
        print(f"[SEND ERROR] {e}", flush=True)
        logging.error(f"[SEND ERROR] {e}")
        return False

# ── WeChat State ──────────────────────────────────────────────────────────────
def is_wechat_foreground():
    """Check if WeChat is in foreground."""
    try:
        r = subprocess.run(
            ["dumpsys", "activity", "activities"],
            capture_output=True, text=True, timeout=3
        )
        return "com.tencent.mm/.ui.LauncherUI" in r.stdout
    except Exception:
        return False

def wake_wechat():
    """Bring WeChat to foreground so it syncs messages to DB."""
    try:
        subprocess.run(
            ["/system/bin/am", "start", "-n", "com.tencent.mm/.ui.LauncherUI"],
            capture_output=True, timeout=5
        )
        print("[WECHAT] Woken up", flush=True)
        return True
    except Exception as e:
        print(f"[WECHAT] Failed to wake: {e}", flush=True)
        return False

# ── Main Poll Cycle ───────────────────────────────────────────────────────────
def poll_cycle():
    """One poll cycle: wake WeChat -> copy -> decrypt -> query -> LLM API -> send."""
    global last_msg_id

    db_src = config.get("db_src_path", "")
    if not db_src:
        db_src = find_enmicromsg_db()
        if db_src:
            config["db_src_path"] = db_src
            try:
                with open("config.json", "w") as f:
                    json.dump(config, f, indent=2, ensure_ascii=False)
            except Exception:
                pass

    if not db_src:
        logging.error("Cannot find EnMicroMsg.db")
        return False

    # 0. Wake WeChat if not in foreground
    if not is_wechat_foreground():
        wake_wechat()
        time.sleep(3)  # Give WeChat time to sync DB
    else:
        print("[WECHAT] Already in foreground", flush=True)

    copy_dir = config.get("db_copy_dir", "/sdcard/hermes/dbs")
    key = config.get("sqlcipher_key", "e03ce81")
    page_size = config.get("cipher_page_size", 1024)
    compat = config.get("cipher_compatibility", 1)

    # 1. Copy DB to Termux home (FUSE /sdcard is readonly for root)
    termux_home = "/data/data/com.termux/files/home"
    work_dir = os.path.join(termux_home, "hermes_work")
    os.makedirs(work_dir, exist_ok=True)
    copied = copy_db(db_src, work_dir)
    print(f"[POLL] copied={copied}", flush=True)
    if not copied:
        logging.error("DB copy failed")
        return False

    enc_path = os.path.join(work_dir, "EnMicroMsg.db")
    plain_path = os.path.join(termux_home, "EnMicroMsg.db.decrypted.db")
    print(f"[POLL] enc_path={enc_path} exists={os.path.exists(enc_path)} size={os.path.getsize(enc_path) if os.path.exists(enc_path) else 0}", flush=True)

    # 2. Decrypt
    if not decrypt_db(enc_path, plain_path, key, page_size, compat):
        fallback_db = "/data/data/com.termux/files/home/EnMicroMsg.db.decrypted.db"
        if os.path.exists(fallback_db):
            import shutil
            shutil.copy2(fallback_db, plain_path)
            print(f"[POLL] Using fallback DB: {fallback_db}", flush=True)
        else:
            logging.error("DB decryption failed — key may have changed.")
            return False

    # 3. Query new messages
    new_msgs = query_new_messages(plain_path, last_msg_id, config.get("max_msgs_per_poll", 50))
    print(f"[POLL] last_msg_id={last_msg_id} new_msgs={len(new_msgs)}", flush=True)
    if not new_msgs:
        return True

    logging.info(f"[POLL] {len(new_msgs)} new messages (last_msg_id was {last_msg_id})")
    last_msg_id = max(m["msgId"] for m in new_msgs)

    # 4. Process via hermes API
    auto_reply = config.get("auto_reply", {})
    if not auto_reply.get("enabled", False):
        return True

    allowed_contacts = auto_reply.get("contacts", [])
    exclude_types = auto_reply.get("exclude_types", [10000])
    cooldown = auto_reply.get("cooldown_sec", 5)

    for msg in new_msgs:
        if msg.get("isSend") == 1:
            continue
        if msg.get("type") in exclude_types:
            continue
        talker = msg.get("talker", "")
        if allowed_contacts and talker not in allowed_contacts:
            continue

        # Cooldown check
        now = time.time()
        if talker in last_reply_time and (now - last_reply_time[talker]) < cooldown:
            logging.info(f"[COOLDOWN] Skip {talker}")
            continue

        content = msg.get("content", "")
        print(f"[MSG] from={talker}: {content[:60]}", flush=True)

        # Get context
        context = get_recent_context(plain_path, talker, msg["msgId"], limit=5)

        # Call hermes
        reply = call_hermes_api(talker, content, context)
        if reply:
            display_names = config.get("contact_display_names", {})
            display_name = display_names.get(talker, talker)
            send_message(talker, reply, display_name)
            last_reply_time[talker] = now
        else:
            logging.warning(f"[HERMES] No reply generated for {talker}")

    return True

# ── Notification Watcher (Termux API) ─────────────────────────────────────────
def ensure_termux_api_service():
    """Ensure Termux:API background service is running."""
    try:
        subprocess.run(
            ["/system/bin/am", "startservice", "-n", "com.termux.api/.TermuxApiService"],
            capture_output=True, timeout=5
        )
        time.sleep(2)
    except Exception as e:
        logging.debug(f"[NOTIF] Failed to start TermuxApiService: {e}")

def poll_notifications():
    """
    Poll notifications via termux-notification-list.
    When a WeChat notification is detected, wake WeChat then trigger urgent DB poll.
    Runs in a separate thread.
    """
    global urgent_poll_needed, known_notif_keys

    termux_prefix = "/data/data/com.termux/files/usr/bin/"
    termux_bin = termux_prefix + "termux-notification-list"

    if not os.path.exists(termux_bin):
        logging.warning("[NOTIF] termux-notification-list not found. Skipping notification watcher.")
        logging.warning("        Install: pkg install termux-api + Termux:API app")
        return

    ensure_termux_api_service()
    logging.info("[NOTIF] Notification watcher started (3s interval)")

    while True:
        try:
            r = subprocess.run(
                [termux_bin],
                capture_output=True, text=True, timeout=10
            )
            if r.returncode == 0:
                try:
                    notifs = json.loads(r.stdout)
                    wechat_notifs = [n for n in notifs if n.get("pkgName") == "com.tencent.mm"]

                    for n in wechat_notifs:
                        title = n.get("title", "")
                        content = n.get("content", "")
                        key = f"{title}:{content}"

                        if key not in known_notif_keys:
                            known_notif_keys.add(key)
                            if len(known_notif_keys) > 200:
                                known_notif_keys = set(list(known_notif_keys)[-100:])

                            print(f"[NOTIF] WeChat: {title}: {content[:60]}", flush=True)
                            logging.info(f"[NOTIF] WeChat notification: {title}: {content[:60]}")

                            # Wake WeChat to sync, then trigger poll
                            wake_wechat()
                            time.sleep(3)  # Give WeChat time to sync DB
                            urgent_poll_needed = True

                except json.JSONDecodeError:
                    pass
        except Exception as e:
            logging.debug(f"[NOTIF] poll error: {e}")

        time.sleep(3)

# ── Daemon Loop ───────────────────────────────────────────────────────────────
def run_daemon():
    """Main daemon loop with notification-triggered urgent polls."""
    global urgent_poll_needed

    interval = config.get("poll_interval_sec", 10)
    use_notifications = config.get("notifications", {}).get("enabled", True)

    logging.info("=" * 50)
    logging.info("Hermes WeChat Daemon started (HTTP API mode)")
    logging.info(f"Poll interval: {interval}s")
    logging.info(f"Auto-reply: {config.get('auto_reply', {}).get('enabled', False)}")
    logging.info(f"Hermes API: {config.get('hermes', {}).get('api_url')}")
    logging.info(f"Notification watcher: {use_notifications}")
    logging.info("=" * 50)

    # Start notification watcher thread
    if use_notifications:
        import threading
        notif_thread = threading.Thread(target=poll_notifications, daemon=True)
        notif_thread.start()

    # Initial poll to establish baseline
    logging.info("[INIT] First poll to get current msgId baseline...")
    poll_cycle()
    logging.info(f"[INIT] Baseline msgId: {last_msg_id}")

    last_regular_poll = time.time()

    while True:
        try:
            now = time.time()
            regular_due = (now - last_regular_poll) >= interval

            if urgent_poll_needed:
                logging.info("[URGENT] Notification triggered DB poll")
                urgent_poll_needed = False
                poll_cycle()
                last_regular_poll = now
            elif regular_due:
                poll_cycle()
                last_regular_poll = now
            else:
                time.sleep(1)
                continue

        except Exception as e:
            logging.error(f"Poll cycle error: {e}")
            time.sleep(1)

# ── Main ──────────────────────────────────────────────────────────────────────
def main():
    parser = argparse.ArgumentParser(description="Hermes WeChat Local Daemon")
    parser.add_argument("--config", default="config.json", help="Config file path")
    parser.add_argument("--once", action="store_true", help="Run one poll cycle and exit")
    parser.add_argument("--daemon", action="store_true", help="Run in background")
    args = parser.parse_args()

    cfg = load_config(args.config)
    setup_logging(cfg.get("logging", {}))

    if args.daemon:
        pid = os.fork()
        if pid > 0:
            print(f"Daemon started (PID={pid}). Log: {cfg.get('logging', {}).get('file')}")
            sys.exit(0)

    if args.once:
        logging.info("Running single poll cycle...")
        ok = poll_cycle()
        sys.exit(0 if ok else 1)

    run_daemon()

if __name__ == "__main__":
    main()