宜博 / pixelphone

"""
Real WeChat message send via on-screen UI automation (F3 / S3-07 / S5-01).

Why this module exists
----------------------
The previous `send_wechat_message` in android_mcp.py did two broken things:

  1. It navigated with `am start -n .../LauncherUI --es username '<wxid>'`, which
     WeChat **ignores** — it just lands on the search (FTSMainUI) screen, never
     the chat. (Verified on Pixel 8 Pro / WeChat 8.x: the extra is dropped.)
  2. It returned a hard-coded ``{"status": "sent"}`` regardless of whether the
     message was ever typed, sent, or written — a dangerous "fake success".

This module replaces that with a real UI path and a truthful result that is
gated on the WeChat message DB (``message.isSend = 1``), so a caller can trust
"sent" to mean the message is actually on its way.

The on-device obstacles that shaped the design (all empirically verified)
-------------------------------------------------------------------------
  * **uiautomator dump is blocked inside WeChat.** ``uiautomator dump`` returns
    an empty single-node hierarchy on every WeChat screen (anti-automation), so
    ``find_element`` / coordinate-by-resource-id is impossible here. We navigate
    by tapping screen-fraction coordinates of the standard WeChat 8.x layout and
    *verify* the result out-of-band (foreground activity + the message DB).
  * **`input text` cannot type CJK and gets mangled by composing IMEs.** Tencent
    WeType and Gboard both treat injected keystrokes as pinyin/auto-correct
    composition ("phoneuse" -> "phone色", hyphens -> full-width). So we switch
    the IME to **ADBKeyboard** (com.android.adbkeyboard) for the duration of the
    send and drive it with ``ADB_INPUT_B64`` broadcasts, which commit arbitrary
    UTF-8 verbatim with no composition. The original IME is always restored.
  * **Intent navigation is dead** (see #1), so we open the app normally and use
    WeChat's own search box to locate the contact, then tap the first result.

Routing (on-device Termux vs. off-device host over adb)
-------------------------------------------------------
Tap/screencap/am/settings go through ui_control_tools._run, which already auto-
routes ``su -c`` on-device and ``adb shell`` off-device. Reading the (encrypted)
WeChat DB always needs root, so it goes through _su_sh() which runs ``su -c`` in
both modes (``adb shell su -c`` off-device), with the whole script base64-wrapped
to dodge nested-quoting traps.

Safety / scope
--------------
This module does not enforce a recipient allowlist itself — callers (the MCP
tool wrapper) are responsible for that. It only ever sends the exact text it is
given to the contact the search resolves to, and it reports failure honestly
rather than guessing.
"""

from __future__ import annotations

import base64
import json
import subprocess
import time
from typing import Any, Optional

# Reuse the auto-routing shell runner + adb prefix from the E2 module so this
# module is deployable on-device (Termux) and testable off-device (host) with
# no change, exactly like ui_control_tools itself.
from ui_control_tools import _run, _adb_prefix, _ON_DEVICE

# ── constants ────────────────────────────────────────────────────────────────

ADB_IME = "com.android.adbkeyboard/.AdbIME"
ADB_IME_PKG = "com.android.adbkeyboard"
# A real keyboard to fall back to if we cannot recover the user's original IME
# (e.g. a previous run was killed while ADBKeyboard was active). Gboard ships on
# the device; only ever used as a last resort so the user is never left stuck on
# the invisible ADBKeyboard.
GBOARD_IME = "com.google.android.inputmethod.latin/com.android.inputmethod.latin.LatinIME"
# Persists the user's real IME across runs so an interrupted run can't poison the
# restore target (see _capture_restore_ime / _restore_real_ime).
IME_BACKUP_FILE = "/data/local/tmp/.phoneuse_prev_ime"
SQLCIPHER = "/data/data/com.termux/files/usr/bin/sqlcipher"
WECHAT_MICROMSG = "/data/data/com.tencent.mm/MicroMsg"
WECHAT_KEY_FILE = "/data/data/com.termux/files/home/.hermes/event_bus/wechat_key"
WECHAT_KEY_FALLBACK = "e03ce81"

# Screen-fraction tap targets for the standard WeChat 8.x layout. Verified on a
# Pixel 8 Pro (1008x2244). They are fractions of the live screen size so they
# scale across resolutions; the post-send DB check is the real correctness gate,
# so a slightly-off tap surfaces as an honest failure, never a fake success.
_FRAC = {
    "wx_tab":        (0.125, 0.965),   # bottom-nav "微信" tab (ensure search bar present)
    "search_icon":   (0.856, 0.0535),  # top-right magnifier on the chat list
    "search_field":  (0.5,   0.089),   # the search input box on FTSMainUI
    "search_clear":  (0.905, 0.088),   # the "x" that clears the search box
    "first_result":  (0.268, 0.207),   # first contact result row under "功能"
    "input_field":   (0.4,   0.952),   # chat input box in its UNFOCUSED resting position
                                        # (must tap here to focus it; the row then rises).
    "send_button":   (0.911, 0.910),   # green "发送" button. Once ADBKeyboard is active and
                                        # the field focused, it shows a thin "{ON}" strip and
                                        # the whole input row lifts to ~y0.910 — higher than
                                        # the unfocused input_field above. (Verified Pixel 8 Pro.)
}

# Aliases that all mean the local "File Transfer Helper" self-chat.
_FILEHELPER_ALIASES = {
    "filehelper", "文件传输助手", "文件传输", "file transfer", "filetransfer",
    "file transfer helper", "files", "transfer",
}


# ── low-level shell helpers ──────────────────────────────────────────────────

def _su_sh(inner: str, timeout: int = 25) -> subprocess.CompletedProcess:
    """Run an arbitrary /bin/sh script *inner* as root, both on- and off-device.

    The script is base64-wrapped so neither adb's argument flattening nor the
    nested ``su -c`` re-splitting can corrupt embedded quotes/pipes/UTF-8.
    """
    b64 = base64.b64encode(inner.encode()).decode()
    payload = f"echo {b64} | base64 -d | sh"
    if _ON_DEVICE:
        argv = ["su", "-c", payload]
    else:
        # Single arg to adb shell; b64 is quote-safe so '...' wrapping is safe.
        argv = _adb_prefix() + [f"su -c '{payload}'"]
    return subprocess.run(argv, capture_output=True, text=True, timeout=timeout)


def _screen_size() -> tuple[int, int]:
    """Return (width, height) in pixels; fall back to the reference device."""
    try:
        r = _run("wm size", timeout=6)
        out = (r.stdout or "")
        # "Physical size: 1008x2244" (Override size wins if present)
        for line in out.splitlines():
            if "size:" in line and "x" in line:
                wh = line.split(":", 1)[1].strip()
                w, h = wh.lower().split("x")
                return int(w), int(h)
    except Exception:
        pass
    return 1008, 2244


def _tap_frac(frac: tuple[float, float], w: int, h: int) -> None:
    x, y = int(frac[0] * w), int(frac[1] * h)
    _run(f"input tap {x} {y}", timeout=8)


def _foreground() -> str:
    """Current foreground window string (lowercased)."""
    try:
        r = _run("dumpsys window 2>/dev/null | grep -i mCurrentFocus", timeout=8)
        return (r.stdout or "").lower()
    except Exception:
        return ""


def _in_chat() -> bool:
    return "chattingui" in _foreground()


# ── ADBKeyboard (deterministic text entry) ───────────────────────────────────

def _adbkb_installed() -> bool:
    try:
        r = _run(f"pm list packages {ADB_IME_PKG}", timeout=8)
        return ADB_IME_PKG in (r.stdout or "")
    except Exception:
        return False


def _get_default_ime() -> str:
    try:
        r = _run("settings get secure default_input_method", timeout=6)
        return (r.stdout or "").strip()
    except Exception:
        return ""


def _set_ime(ime: str) -> None:
    if not ime or ime.lower() == "null":
        return
    _run(f"ime enable {ime}", timeout=8)
    _run(f"ime set {ime}", timeout=8)


def _capture_restore_ime() -> str:
    """Decide what IME to restore to *after* we switch to ADBKeyboard.

    Robust against a previous interrupted run having left ADBKeyboard as the
    system default: in that case the live default is ADBKeyboard itself, which
    must never become the restore target. We persist the real IME to a file
    whenever we see one, and recover from it (or Gboard) otherwise.
    """
    cur = _get_default_ime()
    if cur and ADB_IME_PKG not in cur and cur.lower() != "null":
        # Healthy real IME — remember it for future interrupted runs and use it.
        b64 = base64.b64encode(cur.encode()).decode()
        _run(f"sh -c 'echo {b64} | base64 -d > {IME_BACKUP_FILE}'", timeout=6)
        return cur
    # Current default is ADBKeyboard (a prior run was killed) or unknown.
    try:
        r = _run(f"cat {IME_BACKUP_FILE} 2>/dev/null", timeout=6)
        prev = (r.stdout or "").strip().splitlines()
        prev = prev[-1].strip() if prev else ""
        if prev and ADB_IME_PKG not in prev:
            return prev
    except Exception:
        pass
    return GBOARD_IME


def _wake_and_unlock() -> None:
    """Wake the screen and dismiss a non-secure keyguard before automating.

    Without this the whole tap/search/send flow runs against a black or locked
    screen and every tap is silently dropped, so the send fails even though the
    layout coordinates are correct. ``wm dismiss-keyguard`` only clears a
    *non-secure* lock; a PIN/pattern lock still blocks and the send then fails
    honestly (navigated=False) rather than faking success.
    """
    _run("input keyevent KEYCODE_WAKEUP", timeout=8)
    time.sleep(0.6)
    _run("wm dismiss-keyguard", timeout=8)
    time.sleep(0.8)


def _adbkb_clear() -> None:
    _run("am broadcast -a ADB_CLEAR_TEXT", timeout=8)


def _adbkb_type(text: str) -> None:
    """Commit *text* verbatim via ADBKeyboard's base64 broadcast (UTF-8 safe)."""
    b64 = base64.b64encode(text.encode("utf-8")).decode()
    _run(f"am broadcast -a ADB_INPUT_B64 --es msg '{b64}'", timeout=10)


# ── WeChat DB access (truth source) ──────────────────────────────────────────

def _find_db() -> Optional[str]:
    try:
        r = _su_sh(f"find {WECHAT_MICROMSG} -name EnMicroMsg.db 2>/dev/null | head -1")
        path = (r.stdout or "").strip().splitlines()
        path = path[-1].strip() if path else ""
        return path or None
    except Exception:
        return None


def _read_key() -> str:
    try:
        r = _su_sh(f"cat {WECHAT_KEY_FILE} 2>/dev/null")
        k = (r.stdout or "").strip().splitlines()
        k = k[-1].strip() if k else ""
        if k:
            return k
    except Exception:
        pass
    return WECHAT_KEY_FALLBACK


def _sqlcipher_query(db: str, key: str, sql: str) -> dict[str, Any]:
    """Run a read-only SQL on the encrypted WeChat DB, return {rows,count} or {error}."""
    script = (
        f'PRAGMA key="{key}";\n'
        f'PRAGMA cipher_compatibility=1;\n'
        f'PRAGMA cipher_page_size=1024;\n'
        f'.mode json\n'
        f'{sql.rstrip(";").strip()};\n'
    )
    script_b64 = base64.b64encode(script.encode()).decode()
    inner = f'echo {script_b64} | base64 -d | {SQLCIPHER} "{db}"'
    try:
        r = _su_sh(inner)
    except subprocess.TimeoutExpired:
        return {"error": "sqlcipher query timed out"}
    if r.returncode != 0 and not (r.stdout or "").strip():
        return {"error": (r.stderr or "sqlcipher error").strip()[:200]}
    raw = (r.stdout or "").strip()
    if not raw:
        return {"rows": [], "count": 0}
    # .mode json takes effect after the PRAGMAs print "ok" lines — skip to JSON.
    starts = [i for i in (raw.find("["), raw.find("{")) if i != -1]
    if not starts:
        return {"rows": [], "count": 0}
    try:
        rows = json.loads(raw[min(starts):])
        if not isinstance(rows, list):
            rows = [rows]
        return {"rows": rows, "count": len(rows)}
    except json.JSONDecodeError:
        return {"error": "could not parse sqlcipher output", "raw": raw[:200]}


def _resolve_contact(to: str) -> dict[str, Any]:
    """Map *to* to a navigation query + the expected ``message.talker`` value.

    filehelper aliases resolve deterministically. Anything else is looked up in
    rcontact (F2's schema); unresolved names still get a best-effort query so the
    failure path (no chat opened) is exercised honestly rather than short-circuited.
    """
    t = to.strip()
    if t.lower() in _FILEHELPER_ALIASES or t in _FILEHELPER_ALIASES:
        return {"query": "文件传输助手", "talker": "filehelper", "resolved": True}

    db, key = _find_db(), _read_key()
    if db:
        esc = t.replace("'", "''")
        res = _sqlcipher_query(
            db, key,
            "SELECT username, nickname, conRemark FROM rcontact "
            f"WHERE type != 0 AND (username='{esc}' OR nickname LIKE '%{esc}%' "
            f"OR conRemark LIKE '%{esc}%') LIMIT 1",
        )
        rows = res.get("rows") or []
        if rows:
            row = rows[0]
            display = row.get("conRemark") or row.get("nickname") or t
            return {"query": display, "talker": row.get("username", ""), "resolved": True}
    # Unresolved: search by the raw string; talker unknown.
    return {"query": t, "talker": "", "resolved": False}


# ── public entry point ───────────────────────────────────────────────────────

def send_wechat_message_ui(to: str, content: str, timeout: int = 30) -> dict[str, Any]:
    """Send *content* to WeChat contact *to* via real UI automation.

    Returns a dict whose ``status`` is "sent" **only** when a matching
    ``message`` row with ``isSend = 1`` is found in WeChat's DB after sending;
    otherwise ``status`` is "failed"/"error" with a reason. Never fakes success.
    """
    if not content:
        return {"status": "error", "error": "empty content"}
    if not _adbkb_installed():
        return {"status": "error", "error": (
            "ADBKeyboard (com.android.adbkeyboard) is not installed; it is "
            "required for reliable text entry. Install ADBKeyboard.apk and retry.")}

    target = _resolve_contact(to)
    query, expected_talker = target["query"], target["talker"]

    db, key = _find_db(), _read_key()
    if not db:
        return {"status": "error", "error": "WeChat DB not found (need root)."}

    # Wake + dismiss a non-secure keyguard first, else every tap below lands on
    # a black/locked screen and the send fails despite correct coordinates.
    _wake_and_unlock()

    w, h = _screen_size()
    start_ms = int(time.time() * 1000)
    # Capture the IME to restore to *before* switching — robust against a prior
    # interrupted run having left ADBKeyboard as the system default.
    saved_ime = _capture_restore_ime()
    navigated = False

    try:
        _set_ime(ADB_IME)

        # 1) Open WeChat on its chat-list root. WeChat preserves its task back
        #    stack, so a bare `am start LauncherUI` merely *resumes* whatever
        #    screen was last open (e.g. a chat / chat-info page) instead of the
        #    list — leaving the search icon unreachable and navigation dead. A
        #    force-stop guarantees LauncherUI starts fresh on the chat list with
        #    the top-right search magnifier present. (This is normal app launch,
        #    not the dead `--es username` deep link, and never fakes a result.)
        _run("am force-stop com.tencent.mm", timeout=10)
        time.sleep(1.0)
        _run("am start -n com.tencent.mm/.ui.LauncherUI", timeout=10)
        time.sleep(3.0)
        _tap_frac(_FRAC["wx_tab"], w, h)
        time.sleep(0.6)

        # 2) Open search, clear any stale query, type the contact name.
        _tap_frac(_FRAC["search_icon"], w, h)
        time.sleep(1.5)
        _tap_frac(_FRAC["search_field"], w, h)
        time.sleep(0.6)
        _tap_frac(_FRAC["search_clear"], w, h)  # no-op if already empty
        time.sleep(0.3)
        _tap_frac(_FRAC["search_field"], w, h)
        time.sleep(0.4)
        _adbkb_clear()
        time.sleep(0.3)
        _adbkb_type(query)
        time.sleep(1.8)

        # 3) Tap the first result and confirm we actually entered a chat.
        _tap_frac(_FRAC["first_result"], w, h)
        deadline = time.time() + max(6, min(timeout, 30))
        while time.time() < deadline:
            if _in_chat():
                navigated = True
                break
            time.sleep(0.8)
        if not navigated:
            return {"status": "failed", "to": to, "query": query,
                    "error": f"could not open a chat for {to!r} "
                             "(no matching contact / unexpected screen)"}

        # 4) Focus the input box, clear it, type the message.
        _tap_frac(_FRAC["input_field"], w, h)
        time.sleep(0.8)
        _adbkb_clear()
        time.sleep(0.3)
        _adbkb_type(content)
        time.sleep(0.8)

        # 5) Send.
        _tap_frac(_FRAC["send_button"], w, h)
        time.sleep(2.0)

        # A best-effort pass for a system permission dialog that can appear on
        # first send (e.g. storage). Never disables any setting; just allows.
        _maybe_allow_permission()

    finally:
        _set_ime(saved_ime)

    # 6) Truth check: did the message actually land as a sent row?
    verify = _verify_sent(db, key, content, expected_talker, start_ms)
    if verify.get("ok"):
        row = verify["row"]
        return {
            "status": "sent",
            "to": to,
            "talker": row.get("talker"),
            "content": content,
            "msgId": row.get("msgId"),
            "createTime": row.get("createTime"),
            "method": "ui_automation+adbkeyboard",
            "verified": "message.isSend=1",
        }
    return {
        "status": "failed",
        "to": to,
        "content": content,
        "error": "message was not found as a sent row in WeChat DB after the "
                 "UI flow (isSend=1 match failed)",
        "navigated_to_chat": navigated,
        "verify_detail": verify.get("error"),
    }


def _maybe_allow_permission() -> None:
    fg = _foreground()
    if "grantpermission" in fg or "permissioncontroller" in fg:
        # System permission dialogs are not WeChat, so uiautomator works on them;
        # but to stay dependency-free we tap the common "allow" position. This is
        # best-effort and additive — it never toggles a security setting off.
        w, h = _screen_size()
        for frac in ((0.5, 0.62), (0.5, 0.70), (0.78, 0.62)):
            _tap_frac(frac, w, h)
            time.sleep(0.5)
            if "grantpermission" not in _foreground():
                break


def _verify_sent(db: str, key: str, content: str, talker: str,
                 start_ms: int) -> dict[str, Any]:
    """Look for a freshly-sent row matching *content* (and *talker* if known)."""
    esc = content.replace("'", "''")
    where = (f"content = '{esc}' AND isSend = 1 "
             f"AND createTime >= {start_ms - 8000}")
    if talker:
        where += f" AND talker = '{talker}'"
    res = _sqlcipher_query(
        db, key,
        f"SELECT msgId, isSend, talker, content, createTime FROM message "
        f"WHERE {where} ORDER BY createTime DESC LIMIT 1",
    )
    if "error" in res:
        return {"ok": False, "error": res["error"]}
    rows = res.get("rows") or []
    if rows:
        return {"ok": True, "row": rows[0]}
    return {"ok": False, "error": "no matching isSend=1 row"}