From c3a6427fa516628bcd0bee40aab53ea8c003872f Mon Sep 17 00:00:00 2001 From: yibo Date: Sun, 5 Jul 2026 23:12:00 +0800 Subject: [PATCH] feat(phoneuse): send_wechat_voice_tts MCP 工具 + Hermes 语音发送指南 --- phoneuse/android_mcp.py | 152 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++------------------ phoneuse/docs/wechat-voice-send-guide.md | 106 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ phoneuse/wechat_hook_send.py | 388 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 628 insertions(+), 18 deletions(-) create mode 100644 phoneuse/docs/wechat-voice-send-guide.md create mode 100644 phoneuse/wechat_hook_send.py diff --git a/phoneuse/android_mcp.py b/phoneuse/android_mcp.py index 25727da..5476cc8 100644 --- a/phoneuse/android_mcp.py +++ b/phoneuse/android_mcp.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)}) diff --git a/phoneuse/docs/wechat-voice-send-guide.md b/phoneuse/docs/wechat-voice-send-guide.md new file mode 100644 index 0000000..75a8a5f --- /dev/null +++ b/phoneuse/docs/wechat-voice-send-guide.md @@ -0,0 +1,106 @@ +# 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 反自动化检测 diff --git a/phoneuse/wechat_hook_send.py b/phoneuse/wechat_hook_send.py new file mode 100644 index 0000000..2398e05 --- /dev/null +++ b/phoneuse/wechat_hook_send.py @@ -0,0 +1,388 @@ +""" +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 -- libgit2 0.21.4