fix(weixin): add prefer_platform_transcription config option

gateway/platforms/weixin.py

@@ -956,7 +956,11 @@ def _coerce_bool(value: Any, default: bool = True) -> bool:
     return default
 
 
-def _extract_text(item_list: List[Dict[str, Any]]) -> str:
+def _extract_text(
+    item_list: List[Dict[str, Any]],
+    *,
+    prefer_platform_transcription: bool = False,
+) -> str:
     for item in item_list:
         if item.get("type") == ITEM_TEXT:
             text = str((item.get("text_item") or {}).get("text") or "")
@@ -979,24 +983,34 @@ def _extract_text(item_list: List[Dict[str, Any]]) -> str:
             return text
     for item in item_list:
         if item.get("type") == ITEM_VOICE:
-            # #27300: Tencent Cloud's `voice_item.text` is their STT output,
-            # which is wrong for any non-Chinese audio (the original report
-            # was a Russian voice message that came back as English
-            # gibberish). Return empty so the central STT pipeline in
-            # ``gateway/run.py`` produces the body from the downloaded
-            # audio instead.
             voice_item = item.get("voice_item") or {}
-            if not (voice_item.get("media") or {}):
-                # No raw audio to download — Weixin supplied only its own
+            voice_text = str(voice_item.get("text") or "").strip()
+            has_media = bool(voice_item.get("media") or {})
+
+            if prefer_platform_transcription and voice_text:
+                # User opted in to trust Tencent Cloud's STT (good for
+                # Chinese; avoids needing local STT entirely).
+                return (
+                    "[Voice transcription provided by Weixin]\n"
+                    f"{voice_text}"
+                )
+
+            if not has_media:
+                # No raw audio to download - Weixin supplied only its own
                 # speech-to-text result. Use it, but preserve the voice
                 # origin so the agent can distinguish this from text the
                 # user typed (#65022).
-                voice_text = str(voice_item.get("text") or "")
                 if voice_text:
                     return (
                         "[Voice transcription provided by Weixin]\n"
                         f"{voice_text}"
                     )
+            # #27300: Tencent Cloud's `voice_item.text` is their STT output,
+            # which is wrong for any non-Chinese audio (the original report
+            # was a Russian voice message that came back as English
+            # gibberish). Return empty so the central STT pipeline in
+            # ``gateway/run.py`` produces the body from the downloaded
+            # audio instead.
             continue
     return ""
 
@@ -1242,6 +1256,14 @@ def __init__(self, config: PlatformConfig):
             or os.getenv("WEIXIN_SPLIT_MULTILINE_MESSAGES"),
             default=False,
         )
+        # When true, trust Tencent Cloud's voice transcription (voice_item.text)
+        # instead of downloading raw audio + running local STT. Good for
+        # Chinese-heavy users; avoids needing faster-whisper installed.
+        # Config: gateway.platforms.weixin.extra.prefer_platform_transcription
+        self._prefer_platform_transcription = _coerce_bool(
+            extra.get("prefer_platform_transcription"),
+            default=False,
+        )
 
         # Text debounce batching (mirrors Telegram adapter pattern).
         # iLink delivers messages individually, so rapid multi-message
@@ -1448,7 +1470,10 @@ async def _process_message(self, message: Dict[str, Any]) -> None:
 
         # Secondary content-fingerprint dedup for text messages
         item_list = message.get("item_list") or []
-        text = _extract_text(item_list)
+        text = _extract_text(
+            item_list,
+            prefer_platform_transcription=self._prefer_platform_transcription,
+        )
         if text:
             content_key = f"content:{sender_id}:{hashlib.md5(text.encode()).hexdigest()}"
             if self._dedup.is_duplicate(content_key):
@@ -1680,11 +1705,20 @@ async def _download_file(self, item: Dict[str, Any]) -> Tuple[Optional[str], str
 
     async def _download_voice(self, item: Dict[str, Any]) -> Optional[str]:
         voice_item = item.get("voice_item") or {}
+
+        # When the user opts in to platform transcription, skip downloading
+        # the raw audio - _extract_text() already used voice_item.text as
+        # the message body, so there's no need for the audio file.
+        if self._prefer_platform_transcription:
+            voice_text = str(voice_item.get("text") or "").strip()
+            if voice_text:
+                return None
+
         media = voice_item.get("media") or {}
         # #27300: previously short-circuited when ``voice_item.text`` was set
         # on the assumption that Tencent Cloud's STT was good enough.
         # For non-Chinese audio that text is garbage (e.g. a Russian
-        # message comes back as English phonemes) — we must always
+        # message comes back as English phonemes) - we must always
         # download the raw audio so ``gateway/run.py``'s central STT
         # pipeline can re-transcribe with the user's configured
         # mlx-whisper / whisper.cpp / faster-whisper backend.

tests/gateway/test_weixin.py

@@ -824,7 +824,155 @@ async def _capture(event):
         # The text field must be empty (Tencent text dropped) so the runner
         # has no pre-filled body and routes the audio to STT.
         assert event.text != tencent_text, (
-            "VOICE event body leaked Tencent's STT text — runner would trust "
+            "VOICE event body leaked Tencent's STT text - runner would trust "
             "the wrong transcript instead of re-transcribing (#27300)."
         )
 
+
+class TestWeixinPreferPlatformTranscription:
+    """Tests for ``prefer_platform_transcription`` config option.
+
+    When enabled (``gateway.platforms.weixin.extra.prefer_platform_transcription:
+    true``), the adapter trusts Tencent Cloud's ``voice_item.text`` and skips
+    downloading / re-transcribing the raw audio. This is useful for
+    Chinese-heavy users where Tencent's STT quality is high and local STT
+    (faster-whisper etc.) may not be installed.
+
+    When disabled (default), behaviour is unchanged from #27300: always
+    download audio, ignore Tencent's text.
+    """
+
+    def _make_voice_item(self, text: str = "", with_media: bool = True) -> dict:
+        item: dict = {"type": weixin.ITEM_VOICE, "voice_item": {}}
+        if text:
+            item["voice_item"]["text"] = text
+        if with_media:
+            item["voice_item"]["media"] = {
+                "encrypt_query_param": "q",
+                "aes_key": "a" * 32,
+                "full_url": "https://example.invalid/voice.silk",
+            }
+        return item
+
+    def test_extract_text_returns_tencent_text_when_enabled(self):
+        """With prefer_platform_transcription=True, Tencent's text is used."""
+        item_list = [self._make_voice_item(text="帮我查一下今天天气")]
+        result = weixin._extract_text(
+            item_list, prefer_platform_transcription=True
+        )
+        assert result == (
+            "[Voice transcription provided by Weixin]\n"
+            "帮我查一下今天天气"
+        )
+
+    def test_extract_text_ignores_tencent_text_when_disabled(self):
+        """With prefer_platform_transcription=False (default), Tencent's
+        text is ignored when raw audio is available - the central STT
+        pipeline will produce the body instead (#27300)."""
+        item_list = [self._make_voice_item(text="garbled-tencent-transcript")]
+        result = weixin._extract_text(
+            item_list, prefer_platform_transcription=False
+        )
+        assert result == ""
+
+    def test_extract_text_uses_tencent_text_when_no_media(self):
+        """Even with prefer_platform_transcription=False, if Tencent
+        supplied text but no raw audio, the text is used (there's
+        nothing else to transcribe)."""
+        item_list = [self._make_voice_item(text="some text", with_media=False)]
+        result = weixin._extract_text(
+            item_list, prefer_platform_transcription=False
+        )
+        assert result == (
+            "[Voice transcription provided by Weixin]\n"
+            "some text"
+        )
+
+    def test_extract_text_empty_when_no_text_and_no_media(self):
+        """No text and no media -> empty string regardless of setting."""
+        item_list = [self._make_voice_item(text="", with_media=False)]
+        result = weixin._extract_text(
+            item_list, prefer_platform_transcription=True
+        )
+        assert result == ""
+
+    @pytest.mark.asyncio
+    async def test_download_voice_skips_when_enabled(self, tmp_path, monkeypatch):
+        """With prefer_platform_transcription=True, _download_voice returns
+        None when Tencent text is available - no need to download audio."""
+        adapter = _make_adapter()
+        adapter._prefer_platform_transcription = True
+
+        download_called = False
+
+        async def _fake_download(*a, **k):
+            nonlocal download_called
+            download_called = True
+            return b"should-not-reach-here"
+
+        monkeypatch.setattr(weixin, "_download_and_decrypt_media", _fake_download)
+
+        item = self._make_voice_item(text="帮我查一下今天天气")
+        result = await adapter._download_voice(item)
+
+        assert result is None
+        assert not download_called, "Audio was downloaded despite having Tencent text"
+
+    @pytest.mark.asyncio
+    async def test_download_voice_downloads_when_enabled_but_no_text(self, tmp_path, monkeypatch):
+        """With prefer_platform_transcription=True but no Tencent text,
+        audio is still downloaded for local STT."""
+        adapter = _make_adapter()
+        adapter._prefer_platform_transcription = True
+        adapter._cdn_base_url = "https://example.invalid"
+        adapter._poll_session = Mock()
+
+        monkeypatch.setattr(weixin, "cache_audio_from_bytes",
+                            lambda data, ext: str(tmp_path / f"voice.{ext.lstrip('.')}"))
+
+        async def _fake_download(*a, **k):
+            return b"\x00\x01FAKE_SILK"
+
+        monkeypatch.setattr(weixin, "_download_and_decrypt_media", _fake_download)
+
+        item = self._make_voice_item(text="", with_media=True)
+        result = await adapter._download_voice(item)
+
+        assert result is not None
+        assert result.endswith(".silk")
+
+    @pytest.mark.asyncio
+    async def test_download_voice_downloads_when_disabled(self, tmp_path, monkeypatch):
+        """With prefer_platform_transcription=False (default), audio is
+        always downloaded even when Tencent text is present (#27300)."""
+        adapter = _make_adapter()
+        adapter._prefer_platform_transcription = False
+        adapter._cdn_base_url = "https://example.invalid"
+        adapter._poll_session = Mock()
+
+        monkeypatch.setattr(weixin, "cache_audio_from_bytes",
+                            lambda data, ext: str(tmp_path / f"voice.{ext.lstrip('.')}"))
+
+        async def _fake_download(*a, **k):
+            return b"\x00\x01FAKE_SILK"
+
+        monkeypatch.setattr(weixin, "_download_and_decrypt_media", _fake_download)
+
+        item = self._make_voice_item(text="garbled-tencent-transcript")
+        result = await adapter._download_voice(item)
+
+        assert result is not None
+        assert result.endswith(".silk")
+
+    def test_config_option_read_from_extra(self):
+        """The config option is read from platform extra dict."""
+        from gateway.platforms.weixin import _coerce_bool
+
+        extra = {"account_id": "test", "prefer_platform_transcription": True}
+        val = _coerce_bool(extra.get("prefer_platform_transcription"), default=False)
+        assert val is True
+
+        extra2 = {"account_id": "test"}
+        val2 = _coerce_bool(extra2.get("prefer_platform_transcription"), default=False)
+        assert val2 is False
+