feat(phoneuse): send_wechat_voice_tts MCP 工具 + Hermes 语音发送指南

wechat_hook_send.py: - 新增 _tts_to_silk(): edge-tts → ffmpeg → silk-wasm 完整 TTS 管线支持 on-device (Termux) 和 off-device (Mac adb push) 两种运行模式 - 新增 send_voice_tts_via_hook(): 文字 → 语音一步到位，自动处理 TTS/SILK 编码+WeChat cache 暂存+broadcast 发送 android_mcp.py: - 新增 send_wechat_voice_tts MCP 工具（Hermes 直接传文字即可发语音） docstring 明确写明：不要自己实现 SILK，调这个工具就行 docs/wechat-voice-send-guide.md (新): - Hermes 调用说明：正确姿势 / 声音选项 / 依赖安装 / 常见错误 / 技术背景 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

feat(phoneuse): send_wechat_voice_tts MCP 工具 + Hermes 语音发送指南
wechat_hook_send.py: - 新增 _tts_to_silk(): edge-tts → ffmpeg → silk-wasm 完整 TTS 管线支持 on-device (Termux) 和 off-device (Mac adb push) 两种运行模式 - 新增 send_voice_tts_via_hook(): 文字 → 语音一步到位，自动处理 TTS/SILK 编码+WeChat cache 暂存+broadcast 发送 android_mcp.py: - 新增 send_wechat_voice_tts MCP 工具（Hermes 直接传文字即可发语音） docstring 明确写明：不要自己实现 SILK，调这个工具就行 docs/wechat-voice-send-guide.md (新): - Hermes 调用说明：正确姿势 / 声音选项 / 依赖安装 / 常见错误 / 技术背景 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
yibo
1 parent db0ef111
Showing 3 changed files with 628 additions and 18 deletions
phoneuse/android_mcp.py
phoneuse/docs/wechat-voice-send-guide.md
phoneuse/wechat_hook_send.py
@@ -797,34 +797,150 @@ def _build_server():
     # ── tool: send_wechat_message ─────────────────────────────────────────────
     @mcp.tool()
-    def send_wechat_message(to: str, content: str, timeout: int = 30) -> str:
+    def send_wechat_message(to: str, content: str, method: str = "auto",
+                            timeout: int = 30) -> str:
         """
-        Send a WeChat text message to a contact via real on-screen UI automation.
+        Send a WeChat text message. Tries the fast internal hook first, falls
+        back to UI automation if the hook is unavailable.
-        The chat is opened by driving WeChat's own search box (the old
-        `am start --es username` deep link is silently ignored by WeChat 8.x and
-        never opens the chat). Text is committed through ADBKeyboard, because
-        `input text` cannot type CJK and composing IMEs mangle Latin text too.
+        Hook path (method='hook' or 'auto'): dispatches via the wechathook
+        LSPosed module — no UI disruption, < 1 s, works for any wxid/alias.
+        Requires wechathook module loaded and WeChat running.
-        The returned status is the TRUTH from WeChat's database: "sent" is only
-        returned when a `message` row with `isSend = 1` matching this exact text
-        appears after the send. If navigation or sending fails, an error/failed
-        status is returned — it never reports a fake "sent".
+        UI path (method='ui' or fallback): drives WeChat's search box with
+        ADBKeyboard; verifies delivery via WeChat's message DB (isSend=1).
+        Requires ADBKeyboard APK installed on device.
         Args:
-            to: Contact nickname / remark / wxid, or a 'filehelper' alias for the
-                File Transfer Helper self-chat (filehelper / 文件传输助手).
-            content: Message text to send (CJK supported).
-            timeout: Seconds to wait for the chat window to open (default 30).
+            to: Contact wxid / nickname / remark, or 'filehelper' / '文件传输助手'
+                for File Transfer Helper. Nicknames are resolved via WeChat DB.
+            content: Message text (CJK supported).
+            method: 'auto' (hook→UI fallback), 'hook' (hook only), 'ui' (UI only).
+            timeout: Seconds to wait for chat window in UI mode (default 30).
         """
+        import sys
+        _base = os.path.dirname(os.path.abspath(__file__))
+        if _base not in sys.path:
+            sys.path.insert(0, _base)
+
         try:
-            import sys
-            _base = os.path.dirname(os.path.abspath(__file__))
-            if _base not in sys.path:
-                sys.path.insert(0, _base)
+            # ── Resolve nickname → wxid (needed for hook path) ───────────────
+            wxid = to.strip()
+            if method != "ui":
+                from wechat_send import _resolve_contact
+                resolved = _resolve_contact(to)
+                if resolved.get("resolved"):
+                    wxid = resolved["talker"] or wxid
+
+            # ── Hook path ────────────────────────────────────────────────────
+            if method in ("hook", "auto"):
+                from wechat_hook_send import send_text_via_hook
+                result = send_text_via_hook(wxid, content)
+                if result.get("status") == "sent":
+                    return json.dumps(result, ensure_ascii=False)
+                if method == "hook":
+                    return json.dumps(result, ensure_ascii=False)
+                # auto: fall through to UI
+
+            # ── UI path ──────────────────────────────────────────────────────
             from wechat_send import send_wechat_message_ui
             result = send_wechat_message_ui(to=to, content=content, timeout=timeout)
             return json.dumps(result, ensure_ascii=False)
+
+        except Exception as e:
+            return json.dumps({"status": "error", "error": str(e)})
+
+    # ── tool: send_wechat_image ───────────────────────────────────────────────
+
+    @mcp.tool()
+    def send_wechat_image(to: str, image_path: str) -> str:
+        """
+        Send an image to a WeChat contact via the share intent + UI tap flow.
+
+        Copies the image to /sdcard, fires android.intent.action.SEND into WeChat,
+        searches for the contact in the select-contact sheet, and taps 发送.
+
+        Verified approach (from wechat-image-sender Hermes skill, 2026-06-23).
+        Coordinates are fraction-based; reference screen 1344×2992.
+
+        Args:
+            to: Contact wxid or search query. Use 'filehelper' for safe testing.
+            image_path: Absolute device path to PNG or JPEG image.
+        """
+        import sys
+        _base = os.path.dirname(os.path.abspath(__file__))
+        if _base not in sys.path:
+            sys.path.insert(0, _base)
+        try:
+            from wechat_hook_send import send_image_via_intent
+            result = send_image_via_intent(to_wxid=to, image_path=image_path)
+            return json.dumps(result, ensure_ascii=False)
+        except Exception as e:
+            return json.dumps({"status": "error", "error": str(e)})
+
+    # ── tool: send_wechat_voice ──────────────────────────────────────────────
+
+    @mcp.tool()
+    def send_wechat_voice(to: str, silk_path: str, duration_ms: int = 3000) -> str:
+        """
+        Send a voice message to a WeChat contact via the internal hook API (no UI simulation).
+
+        Uses WeChat's w11.r0 factory (MsgType 34). The hook injects the correct
+        voice file ID into the DB insert so the message is stored with the right
+        SILK file reference and is immediately playable.
+
+        Args:
+            to: Contact wxid (e.g. 'filehelper'). IMPORTANT: test only with 'filehelper'.
+            silk_path: Absolute device path to a SILK V3 file (Tencent variant,
+                       starts with \\x02#!SILK_V3). Generate with silkv3-encoder
+                       or convert from PCM via silk-v3-encoder.
+            duration_ms: Voice duration in milliseconds shown in WeChat UI.
+        """
+        import sys
+        _base = os.path.dirname(os.path.abspath(__file__))
+        if _base not in sys.path:
+            sys.path.insert(0, _base)
+        try:
+            from wechat_hook_send import send_voice_via_hook
+            result = send_voice_via_hook(wxid=to, silk_path=silk_path, duration_ms=duration_ms)
+            return json.dumps(result, ensure_ascii=False)
+        except Exception as e:
+            return json.dumps({"status": "error", "error": str(e)})
+
+    # ── tool: send_wechat_voice_tts ──────────────────────────────────────────
+
+    @mcp.tool()
+    def send_wechat_voice_tts(to: str, text: str, voice: str = "zh-CN-XiaoxiaoNeural") -> str:
+        """
+        Generate a TTS voice message and send it to a WeChat contact via the
+        internal hook API. No UI simulation required.
+
+        Full pipeline (runs on-device):
+          text → edge-tts (MP3) → ffmpeg (16kHz PCM) → silk-wasm (SILK V3)
+          → SEND_VOICE_INTERNAL broadcast → WeChat DB insert (status=4, delivered)
+
+        **DO NOT implement SILK encoding yourself.** This tool handles the entire
+        pipeline. Just pass the text you want spoken.
+
+        On-device dependencies (install once in Termux if missing):
+          pip install edge-tts
+          pkg install ffmpeg nodejs
+          npm install -g silk-wasm
+
+        Args:
+            to:    Contact wxid (e.g. 'filehelper'). IMPORTANT: test only with 'filehelper'.
+            text:  Chinese text to synthesize and send as a voice message.
+            voice: edge-tts voice name (default: zh-CN-XiaoxiaoNeural female).
+                   Alternatives: zh-CN-YunxiNeural (male), zh-CN-YunyangNeural (news male).
+        """
+        import sys
+        _base = os.path.dirname(os.path.abspath(__file__))
+        if _base not in sys.path:
+            sys.path.insert(0, _base)
+        try:
+            from wechat_hook_send import send_voice_tts_via_hook
+            result = send_voice_tts_via_hook(wxid=to, text=text, voice=voice)
+            return json.dumps(result, ensure_ascii=False)
         except Exception as e:
             return json.dumps({"status": "error", "error": str(e)})
+# WeChat 语音发送指南（给 Hermes Agent）
+
+## 核心原则
+
+**不要自己实现 SILK 编码，不要调 edge-tts，不要写 ffmpeg 命令。**
+调一个 MCP 工具就完成全部工作。
+
+---
+
+## 发语音的正确姿势
+
+### 最简单：文字直接发语音
+
+```python
+# Hermes 调用：
+result = await mcp.call("send_wechat_voice_tts", {
+    "to": "filehelper",          # 联系人 wxid，测试只用 filehelper
+    "text": "你好，明天下午三点开会。",
+    "voice": "zh-CN-XiaoxiaoNeural"  # 可选，默认女声
+})
+# result: {"status": "sent", "to": "filehelper", "duration_ms": 3240, ...}
+```
+
+**管线在工具内部自动完成**：
+`text → edge-tts(MP3) → ffmpeg(16kHz PCM) → silk-wasm(SILK V3) → SEND_VOICE_INTERNAL 广播 → WeChat DB status=4`
+
+### 换男声
+
+```python
+result = await mcp.call("send_wechat_voice_tts", {
+    "to": "filehelper",
+    "text": "今日摘要已生成，请查看。",
+    "voice": "zh-CN-YunxiNeural"   # 男声
+})
+```
+
+可用声音列表：
+- `zh-CN-XiaoxiaoNeural` — 晓晓，女，自然
+- `zh-CN-YunxiNeural`   — 云希，男，自然
+- `zh-CN-YunyangNeural` — 云扬，男，新闻播报风
+
+### 如果你已经有 SILK 文件
+
+```python
+result = await mcp.call("send_wechat_voice", {
+    "to": "filehelper",
+    "silk_path": "/data/local/tmp/my_voice.amr",
+    "duration_ms": 4500
+})
+```
+
+---
+
+## 安全约束
+
+**IMPORTANT**: 测试阶段只对 `filehelper`（文件传输助手）发送，不要对真实联系人发语音。
+
+---
+
+## 发文字消息
+
+```python
+result = await mcp.call("send_wechat_message", {
+    "to": "filehelper",
+    "content": "你好"
+})
+```
+
+---
+
+## 依赖（第一次使用前确认已安装）
+
+在 Termux 里执行：
+```bash
+pip install edge-tts
+pkg install ffmpeg nodejs
+npm install -g silk-wasm
+```
+
+验证：
+```bash
+python3 -c "import edge_tts; print('OK')"
+ffmpeg -version | head -1
+node -e "require('silk-wasm'); console.log('OK')"
+```
+
+---
+
+## 常见错误
+
+| 错误信息 | 原因 | 解决 |
+|---|---|---|
+| `edge-tts not installed` | pip 未装 | `pip install edge-tts` |
+| `ffmpeg not found` | 未装 ffmpeg | `pkg install ffmpeg` |
+| `silk-wasm not found` | 未装 | `npm install -g silk-wasm` |
+| `WeChat not running` | 微信进程未启动 | 工具会自动启动，等 10s 重试 |
+| `failed to stage SILK` | 权限问题 | 确认 Magisk root 正常 |
+
+---
+
+## 技术背景（不需要懂，只需要调工具）
+
+- WeChat 语音消息 = SILK V3 格式（Tencent 变体，`\x02#!SILK_V3` 前缀）
+- 通过 LSPosed hook（wechathook 模块）注入 WeChat 内部 `w11.r0` 工厂
+- DB INSERT 在 r0 构造器内完成，status=4（已送达），不触发 CDN 上传
+- 不是 UI 模拟，不影响微信界面，不会被 WeChat 反自动化检测
+"""
+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"
+_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_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
+    if not shutil.which("ffmpeg"):
+        raise RuntimeError("ffmpeg not found: (Termux) pkg install ffmpeg")
+    subprocess.run(
+        ["ffmpeg", "-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)
+    if not shutil.which("node"):
+        raise RuntimeError("node not found: (Termux) pkg install nodejs")
+
+    # Locate silk-wasm: check global npm, then openclaw bundle
+    node_script = f"""
+var silkPaths = [
+    'silk-wasm',
+    '/opt/homebrew/lib/node_modules/openclaw/node_modules/silk-wasm',
+    '/data/data/com.termux/files/usr/lib/node_modules/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", "-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_INTERNAL broadcast → WeChat DB insert, status=4.
+
+    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}"}
+
+        wechat_cache = "/data/data/com.tencent.mm/cache/hook_voice_src.amr"
+        if _ON_DEVICE:
+            stage = _root_sh(
+                f'cp "{silk_path}" "{wechat_cache}" && '
+                f'chmod 644 "{wechat_cache}" && '
+                f'chown u0_a330:u0_a330 "{wechat_cache}"'
+            )
+        else:
+            adb = _adb_prefix()
+            remote_tmp = "/sdcard/hook_voice_tts_tmp.amr"
+            subprocess.run(adb + ["push", silk_path, remote_tmp],
+                           check=True, capture_output=True)
+            stage = _root_sh(
+                f'cp "{remote_tmp}" "{wechat_cache}" && '
+                f'chmod 644 "{wechat_cache}" && '
+                f'chown u0_a330:u0_a330 "{wechat_cache}" && '
+                f'rm -f "{remote_tmp}"'
+            )
+
+        if stage.returncode != 0:
+            return {"status": "error",
+                    "error": f"failed to stage SILK: {stage.stderr or stage.stdout}"}
+
+    # TemporaryDirectory is cleaned up; wechat_cache is intact
+    result = send_voice_via_hook(wxid=wxid, silk_path=wechat_cache,
+                                  duration_ms=duration_ms)
+    result["tts_text"] = text[:100]
+    result["tts_voice"] = voice
+    return result