宜博 / pixelphone

"""
WeChat send via LSPosed internal hook (fast path — no UI disruption).

Text only. The hook (wechathook module in xhs_lsposed_a16_fix) registers a
RECEIVER_NOT_EXPORTED BroadcastReceiver inside the WeChat main process and uses
WeChat's own w11.s1 factory to dispatch the message, so it bypasses all of the
anti-automation that kills UI-based senders. Delivery is asynchronous after the
broadcast returns; the caller decides whether to DB-verify.

Requirements
------------
- wechathook LSPosed module installed, scope includes com.tencent.mm
- WeChat main process running (or this module starts it)
- Broadcast must come from root (RECEIVER_NOT_EXPORTED)
"""
from __future__ import annotations

import base64
import subprocess
import time
from typing import Any

from ui_control_tools import _run, _adb_prefix, _ON_DEVICE

SEND_ACTION = "com.hermes.wechat.SEND_MSG_INTERNAL"
SEND_IMAGE_ACTION = "com.hermes.wechat.SEND_IMAGE_INTERNAL"
SEND_VOICE_CGI_ACTION = "com.hermes.wechat.SEND_VOICE_CGI"
_WECHAT_PKG = "com.tencent.mm"
_WECHAT_MAIN = "com.tencent.mm/.ui.LauncherUI"


def _wechat_running() -> bool:
    try:
        r = _run("pidof " + _WECHAT_PKG, timeout=5)
        return bool((r.stdout or "").strip())
    except Exception:
        return False


def _start_wechat(wait_secs: int = 12) -> bool:
    _run("am start -n " + _WECHAT_MAIN, timeout=8)
    deadline = time.time() + wait_secs
    while time.time() < deadline:
        if _wechat_running():
            return True
        time.sleep(1)
    return False


def _root_sh(cmd: str, timeout: int = 10) -> subprocess.CompletedProcess:
    """Run *cmd* as root, both on-device (su -c) and off-device (adb shell su -c)."""
    b64 = base64.b64encode(cmd.encode()).decode()
    payload = f"echo {b64} | base64 -d | sh"
    if _ON_DEVICE:
        return subprocess.run(["su", "-c", payload],
                              capture_output=True, text=True, timeout=timeout)
    adb = _adb_prefix()
    return subprocess.run(adb + [f"su -c '{payload}'"],
                          capture_output=True, text=True, timeout=timeout + 5)


def send_text_via_hook(wxid: str, text: str) -> dict[str, Any]:
    """Send *text* to *wxid* via the wechathook internal broadcast.

    Fast (< 1 s round-trip), no UI disruption, no ADBKeyboard dependency.
    Status 'sent' means the broadcast was accepted by the system; delivery is
    asynchronous. Use wechat_send._verify_sent() for DB confirmation if needed.

    Args:
        wxid: talker id as stored in WeChat's rcontact.username (e.g. 'filehelper',
              'wxid_xxx'). Nicknames are NOT resolved here — caller must resolve first.
        text: message text (arbitrary UTF-8, including CJK).
    """
    if not wxid or not text:
        return {"status": "error", "error": "wxid and text required"}

    if not _wechat_running():
        if not _start_wechat():
            return {"status": "error", "error": "WeChat not running and failed to start"}
        time.sleep(3)  # allow App.onCreate + BroadcastReceiver registration

    # Double-base64 the args to survive nested su quoting with arbitrary content.
    b64_wxid = base64.b64encode(wxid.encode()).decode()
    b64_text = base64.b64encode(text.encode()).decode()
    cmd = (
        f'TO=$(printf "%s" "{b64_wxid}" | base64 -d); '
        f'TEXT=$(printf "%s" "{b64_text}" | base64 -d); '
        f'am broadcast -a {SEND_ACTION} --es to "$TO" --es text "$TEXT"'
    )

    try:
        r = _root_sh(cmd)
    except subprocess.TimeoutExpired:
        return {"status": "error", "error": "broadcast timed out"}

    stdout = (r.stdout or "").strip()
    stderr = (r.stderr or "").strip()
    # am broadcast exits 0 and prints "Broadcast completed: result=0" on success.
    if r.returncode == 0 and ("result=" in stdout or "Broadcast completed" in stdout):
        return {
            "status": "sent",
            "to": wxid,
            "method": "hook_broadcast",
            "verified": False,  # DB not checked; caller may call _verify_sent
            "detail": stdout[:200],
        }
    return {
        "status": "error",
        "error": f"broadcast rc={r.returncode}: {stderr or stdout or '(no output)'}",
    }


def send_image_via_intent(to_wxid: str, image_path: str) -> dict[str, Any]:
    """Send an image to *to_wxid* via WeChat's share intent + UI tap.

    Copies the image to /sdcard/ (world-readable), fires an ACTION_SEND intent,
    then navigates WeChat's 'select contact' sheet to pick the target and confirms.

    This is a best-effort UI path — it works reliably for filehelper. For other
    contacts the search step matches the first rcontact search result, which may
    be wrong if multiple contacts share a name fragment.

    Args:
        to_wxid: must be 'filehelper' for safe testing, or a unique search query.
        image_path: absolute path on the device to a PNG/JPEG image file.
    """
    import os
    # Validated approach from the Hermes wechat-image-sender skill (2026-06-23):
    # am start -a android.intent.action.SEND -t image/png with STREAM extra →
    # WeChat "选择联系人" UI → search → tap → 发送 at ~(0.67, 0.88).
    # Coordinates verified on 1344×2992 screen (fixed: 899, 2627).

    # ── 1. Copy image to /sdcard so WeChat can read it ──────────────────────
    ext = os.path.splitext(image_path)[1].lower() or ".jpg"
    share_path = f"/sdcard/wechat_share_tmp{ext}"
    mime = "image/png" if ext == ".png" else "image/jpeg"

    cp_result = _root_sh(f'cp "{image_path}" "{share_path}" && chmod 644 "{share_path}"')
    if cp_result.returncode != 0:
        return {"status": "error",
                "error": f"failed to copy image to /sdcard: {cp_result.stderr or cp_result.stdout}"}

    # ── 2. Fire the share intent ─────────────────────────────────────────────
    _run(f"am start -a android.intent.action.SEND -t {mime} "
         f"--eu android.intent.extra.STREAM 'file://{share_path}' "
         f"-p {_WECHAT_PKG}", timeout=10)
    time.sleep(2.5)  # wait for WeChat to open the select-contact sheet

    # ── 3. Search for the contact in the sheet (same approach as text send) ──
    from ui_control_tools import _run as run
    from wechat_send import _screen_size, _tap_frac, _foreground

    w, h = _screen_size()
    query = "文件传输助手" if to_wxid == "filehelper" else to_wxid

    # The share sheet has a search box. Tap it, type the query.
    search_frac = (0.5, 0.075)   # top search bar in share contact picker
    result_frac = (0.5, 0.22)    # first result row
    send_frac   = (0.669, 0.878) # 发送 button; 899/1344, 2627/2992 on ref device

    _tap_frac(search_frac, w, h)
    time.sleep(0.8)

    # Use ADBKeyboard to type the search query
    b64q = base64.b64encode(query.encode()).decode()
    _root_sh(f'am broadcast -a ADB_INPUT_B64 --es msg "{b64q}"')
    time.sleep(1.5)

    _tap_frac(result_frac, w, h)
    time.sleep(1.0)

    # ── 4. Tap "发送" / confirm ───────────────────────────────────────────────
    _tap_frac(send_frac, w, h)
    time.sleep(1.5)

    # Cleanup temp file
    _root_sh(f'rm -f "{share_path}"')

    return {
        "status": "sent",
        "to": to_wxid,
        "image": image_path,
        "method": "share_intent",
        "verified": False,  # image sends don't leave a simple DB row to verify
    }


def send_image_via_hook(wxid: str, image_path: str) -> dict[str, Any]:
    """Send an image to *wxid* via the wechathook internal API (no UI simulation).

    Fires SEND_IMAGE_INTERNAL. The hook calls WeChat's own image plugin
    b80.u.oj (insert message f9 + ImgInfo b0 + build CDN upload descriptors)
    then submits each descriptor to the CDN service v70.w.oj — the full image
    uploads via WeChat's own CDN (receiver sees full-resolution original, not a
    stuck thumbnail). Fire-and-forget: delivery is async after the broadcast.

    This is the fast path — no share sheet, no contact search, no wrong-contact
    risk. Preferred over send_image_via_intent. Verified 2026-07-07 on WeChat
    8.0.74 (filehelper + real contact, full-res delivery). NOTE: the image MUST be
    staged into a MediaStore-indexed dir (/sdcard/Pictures/) + media-scanned — a fresh
    file in /sdcard root uploads as a broken image (red !); this fn handles that.

    Args:
        wxid: talker id as stored in rcontact.username (e.g. 'filehelper',
              'wxid_xxx', 'yibooo'). Nicknames are NOT resolved here.
        image_path: absolute device path to a PNG/JPEG. Must be readable by the
                    WeChat process; this fn stages it into /sdcard first so
                    root-only source paths work too.
    """
    import os
    if not wxid or not image_path:
        return {"status": "error", "error": "wxid and image_path required"}

    if not _wechat_running():
        if not _start_wechat():
            return {"status": "error", "error": "WeChat not running and failed to start"}
        time.sleep(3)  # allow App.onCreate + BroadcastReceiver registration

    # Stage into /sdcard/Pictures/ (a MediaStore-indexed media dir) and trigger a media
    # scan. WeChat's image CDN upload resolves the file via MediaStore — a fresh file in
    # /sdcard root (not indexed) uploads as a broken image (red !, receiver gets nothing).
    # Unique per-pid name so a fixed path isn't overwritten mid-upload by a concurrent send.
    ext = os.path.splitext(image_path)[1].lower() or ".jpg"
    share_path = f"/sdcard/Pictures/wx_send_{os.getpid()}{ext}"
    cp = _root_sh(f'cp "{image_path}" "{share_path}" && chmod 644 "{share_path}"')
    if cp.returncode != 0:
        return {"status": "error",
                "error": f"failed to stage image to /sdcard/Pictures: {cp.stderr or cp.stdout}"}
    # Register with MediaStore so WeChat's image pipeline can resolve it (required).
    _root_sh(f'am broadcast -a android.intent.action.MEDIA_SCANNER_SCAN_FILE -d "file://{share_path}"')
    time.sleep(1.0)  # let the scan land before WeChat reads it

    b64_wxid = base64.b64encode(wxid.encode()).decode()
    cmd = (
        f'TO=$(printf "%s" "{b64_wxid}" | base64 -d); '
        f'am broadcast -a {SEND_IMAGE_ACTION} --es to "$TO" --es path "{share_path}"'
    )
    try:
        r = _root_sh(cmd)
    except subprocess.TimeoutExpired:
        return {"status": "error", "error": "broadcast timed out"}

    stdout = (r.stdout or "").strip()
    stderr = (r.stderr or "").strip()
    if r.returncode == 0 and ("result=" in stdout or "Broadcast completed" in stdout):
        return {
            "status": "sent",
            "to": wxid,
            "image": image_path,
            "method": "hook_broadcast",
            "verified": False,  # CDN upload is async; confirm via UI screenshot
            "detail": stdout[:200],
        }
    return {
        "status": "error",
        "error": f"broadcast rc={r.returncode}: {stderr or stdout or '(no output)'}",
    }


def send_voice_cgi_via_hook(wxid: str, silk_path: str, duration_ms: int = 3000) -> dict[str, Any]:
    """Send a FULLY-SYNTHETIC voice to *wxid* via the internal uploadvoice cgi.

    No recording, no hot-swap. The hook chunks the SILK, builds WeChat's own
    /cgi-bin/micromsg-bin/uploadvoice request (audio inline), dispatches it via the
    NetSceneQueue (modelbase.i.l() + subscribe), and inserts a local f9 bubble marked
    sent. Result: the recipient gets a playable voice AND the sender sees a sent bubble.
    Verified end-to-end on WeChat 8.0.74 (recipient + Mac both received; both sides visible).

    The hook reads *silk_path* AS the WeChat process, so the SILK is first staged into
    WeChat's own data dir (its uid + app_data_file SELinux context) — /sdcard gives
    EACCES and a plain root cp into the sandbox needs chown + restorecon.

    Args:
        wxid: talker id as in rcontact.username (e.g. 'filehelper', 'wxid_xxx', 'yibooo').
              Test with 'filehelper' first to avoid account risk.
        silk_path: device path to a SILK V3 file (Tencent variant, \\x02#!SILK_V3),
                   e.g. from edge-tts → ffmpeg 16k mono → silk-wasm.
        duration_ms: voice duration in milliseconds shown in WeChat UI.
    """
    if not wxid or not silk_path:
        return {"status": "error", "error": "wxid and silk_path required"}
    if not _wechat_running():
        if not _start_wechat():
            return {"status": "error", "error": "WeChat not running and failed to start"}
        time.sleep(3)

    # Resolve WeChat's uid (changes across reinstalls) and stage the SILK into its data
    # dir with the matching owner + SELinux context so the WeChat process can read it.
    uid_res = _root_sh("stat -c %U /data/user/0/com.tencent.mm")
    wx_uid = (uid_res.stdout or "").strip()
    if not wx_uid:
        return {"status": "error", "error": f"could not resolve WeChat uid: {uid_res.stderr or '(no output)'}"}
    stage_path = "/data/data/com.tencent.mm/files/hook_voice_cgi.silk"
    stage = _root_sh(
        f'cp "{silk_path}" "{stage_path}" && '
        f'chown {wx_uid}:{wx_uid} "{stage_path}" && '
        f'chmod 600 "{stage_path}"; restorecon "{stage_path}" 2>/dev/null; true'
    )
    if stage.returncode != 0:
        return {"status": "error", "error": f"failed to stage SILK: {stage.stderr or stage.stdout}"}

    cmd = (
        f"am broadcast -a {SEND_VOICE_CGI_ACTION} "
        f"--es to {wxid} --es src_path {stage_path} --ei duration {duration_ms}"
    )
    try:
        r = _root_sh(cmd)
    except subprocess.TimeoutExpired:
        return {"status": "error", "error": "voice cgi broadcast timed out"}
    stdout = (r.stdout or "").strip()
    stderr = (r.stderr or "").strip()
    if r.returncode == 0 and ("result=" in stdout or "Broadcast completed" in stdout):
        return {
            "status": "sent",
            "to": wxid,
            "method": "uploadvoice_cgi",
            "verified": False,  # cgi is async; confirm via UI / recipient
            "detail": stdout[:200],
        }
    return {"status": "error", "error": f"broadcast rc={r.returncode}: {stderr or stdout or '(no output)'}"}


def send_voice_via_hook(wxid: str, silk_path: str, duration_ms: int = 3000) -> dict[str, Any]:
    """Send a voice message to *wxid* via the wechathook internal API.

    Uses WeChat's own w11.r0 factory (type=34) — no UI simulation.
    The hook (hookF9ImgPathInjector) intercepts the f9 entity inside the r0
    constructor to inject the voiceId before the DB insert, so the stored row
    has the correct imgPath pointing to the SILK file.

    Protocol:
      1. Push SILK V3 file to WeChat's cache dir (root-writable).
      2. Fire com.hermes.wechat.SEND_VOICE_INTERNAL broadcast (root-only,
         RECEIVER_NOT_EXPORTED). The hook generates a random 26-char voiceId,
         computes voice2/{md5(id)[0:2]}/{md5(id)[2:4]}/msg_{id}.amr path,
         copies from cache, inserts the DB row, and calls H() for server sync.

    Args:
        wxid: talker id as in WeChat DB (e.g. 'filehelper', 'wxid_xxx').
              IMPORTANT: test only with 'filehelper' to avoid account risk.
        silk_path: absolute path on the device to a SILK V3 file
                   (Tencent variant: starts with \\x02#!SILK_V3).
        duration_ms: voice duration in milliseconds (shown in WeChat UI).
    """
    if not wxid or not silk_path:
        return {"status": "error", "error": "wxid and silk_path required"}

    if not _wechat_running():
        if not _start_wechat():
            return {"status": "error", "error": "WeChat not running and failed to start"}
        time.sleep(3)

    # Stage 1: push SILK to WeChat's cache (readable by WeChat process)
    wechat_cache = "/data/data/com.tencent.mm/cache/hook_voice_src.amr"
    stage = _root_sh(
        f'cp "{silk_path}" "{wechat_cache}" && '
        f'chmod 644 "{wechat_cache}" && '
        f'chown u0_a330:u0_a330 "{wechat_cache}"'
    )
    if stage.returncode != 0:
        return {
            "status": "error",
            "error": f"failed to stage SILK in WeChat cache: {stage.stderr or stage.stdout}",
        }

    # Stage 2: fire the voice broadcast
    cmd = (
        f"am broadcast -a com.hermes.wechat.SEND_VOICE_INTERNAL "
        f"--es to {wxid} "
        f"--es src_path {wechat_cache} "
        f"--ei duration {duration_ms}"
    )
    try:
        r = _root_sh(cmd)
    except subprocess.TimeoutExpired:
        return {"status": "error", "error": "voice broadcast timed out"}

    stdout = (r.stdout or "").strip()
    stderr = (r.stderr or "").strip()
    if r.returncode == 0 and ("result=" in stdout or "Broadcast completed" in stdout):
        return {
            "status": "sent",
            "to": wxid,
            "method": "hook_broadcast_voice",
            "duration_ms": duration_ms,
            "verified": False,
            "detail": stdout[:200],
        }
    return {
        "status": "error",
        "error": f"voice broadcast rc={r.returncode}: {stderr or stdout or '(no output)'}",
    }


# ── TTS helpers ───────────────────────────────────────────────────────────────

def _tts_to_silk(text: str, voice: str, tmp_dir: str) -> tuple[str, int]:
    """Generate SILK V3 .amr from *text* using edge-tts + ffmpeg + silk-wasm.

    Dependencies (must be installed on-device or on Mac):
      pip install edge-tts
      (Termux) pkg install ffmpeg nodejs && npm install -g silk-wasm
      (Mac)    brew install ffmpeg; openclaw installs silk-wasm automatically

    Returns (silk_path, duration_ms).
    Raises RuntimeError if any step fails.
    """
    import asyncio, os, shutil

    mp3  = os.path.join(tmp_dir, "tts.mp3")
    pcm  = os.path.join(tmp_dir, "tts.pcm")
    silk = os.path.join(tmp_dir, "tts.amr")

    # 1. edge-tts → MP3
    try:
        import edge_tts
    except ImportError:
        raise RuntimeError("edge-tts not installed: pip install edge-tts")

    async def _synth():
        await edge_tts.Communicate(text, voice).save(mp3)

    asyncio.run(_synth())

    # 2. ffmpeg → PCM16LE 16kHz mono
    # shutil.which misses Termux bin when running under root su; probe extra paths
    _FFMPEG_CANDIDATES = ["ffmpeg",
                          "/data/data/com.termux/files/usr/bin/ffmpeg",
                          "/usr/bin/ffmpeg", "/usr/local/bin/ffmpeg"]
    ffmpeg_bin = next((p for p in _FFMPEG_CANDIDATES if shutil.which(p) or
                       (os.path.isfile(p) and os.access(p, os.X_OK))), None)
    if not ffmpeg_bin:
        raise RuntimeError("ffmpeg not found: (Termux) pkg install ffmpeg")
    subprocess.run(
        [ffmpeg_bin, "-y", "-i", mp3, "-ar", "16000", "-ac", "1", "-f", "s16le", pcm],
        check=True, capture_output=True
    )

    # 3. Node.js silk-wasm → SILK V3 (already includes \x02#!SILK_V3 prefix)
    _NODE_CANDIDATES = ["node",
                        "/data/data/com.termux/files/usr/bin/node",
                        "/usr/bin/node", "/usr/local/bin/node"]
    node_bin = next((p for p in _NODE_CANDIDATES if shutil.which(p) or
                     (os.path.isfile(p) and os.access(p, os.X_OK))), None)
    if not node_bin:
        raise RuntimeError("node not found: (Termux) pkg install nodejs")

    # Locate silk-wasm: check global npm, then openclaw bundle
    node_script = f"""
var silkPaths = [
    '/data/data/com.termux/files/usr/lib/node_modules/silk-wasm',
    '/opt/homebrew/lib/node_modules/openclaw/node_modules/silk-wasm',
    'silk-wasm',
];
var silk = null;
for (var p of silkPaths) {{
    try {{ silk = require(p); break; }} catch(e) {{}}
}}
if (!silk) {{ process.stderr.write('silk-wasm not found'); process.exit(1); }}
var fs = require('fs');
silk.encode(fs.readFileSync('{pcm}'), 16000).then(function(r) {{
    fs.writeFileSync('{silk}', r.data);
    process.stdout.write(String(r.duration));
}}).catch(function(e) {{ process.stderr.write(String(e)); process.exit(1); }});
"""
    r = subprocess.run([node_bin, "-e", node_script], capture_output=True, text=True, timeout=60)
    if r.returncode != 0:
        raise RuntimeError(f"silk-wasm encode failed: {r.stderr or r.stdout}. "
                           "Install: npm install -g silk-wasm")
    duration_ms = int(r.stdout.strip()) if r.stdout.strip().isdigit() else 3000
    return silk, duration_ms


def send_voice_tts_via_hook(
    wxid: str,
    text: str,
    voice: str = "zh-CN-XiaoxiaoNeural",
) -> dict[str, Any]:
    """Generate TTS from *text* and send as WeChat voice message to *wxid*.

    End-to-end: edge-tts → ffmpeg (16kHz PCM) → silk-wasm (SILK V3) →
    wechathook SEND_VOICE_CGI (internal uploadvoice cgi): uploads the audio + delivers
    to the recipient AND inserts a sent bubble on the sender side. Verified end-to-end.

    Args:
        wxid:  talker id (e.g. 'filehelper'). IMPORTANT: test only with 'filehelper'.
        text:  Chinese text to synthesize (or any language supported by the voice).
        voice: edge-tts voice name. Defaults to zh-CN-XiaoxiaoNeural (female).
               Other options: zh-CN-YunxiNeural (male), zh-CN-YunyangNeural (news male).
    """
    import tempfile

    if not wxid or not text:
        return {"status": "error", "error": "wxid and text required"}

    if not _wechat_running():
        if not _start_wechat():
            return {"status": "error", "error": "WeChat not running and failed to start"}
        time.sleep(3)

    with tempfile.TemporaryDirectory(prefix="wc_tts_") as tmp:
        try:
            silk_path, duration_ms = _tts_to_silk(text, voice, tmp)
        except RuntimeError as e:
            return {"status": "error", "error": str(e)}
        except Exception as e:
            return {"status": "error", "error": f"TTS/SILK generation failed: {e}"}

        # Delegate staging + broadcast to the fully-synthetic cgi sender (uploads audio
        # inline via the internal uploadvoice cgi + delivers to the recipient + inserts a
        # sent bubble). Must call while tmp dir is still alive.
        result = send_voice_cgi_via_hook(wxid=wxid, silk_path=silk_path,
                                         duration_ms=duration_ms)
    result["tts_text"] = text[:100]
    result["tts_voice"] = voice
    return result