feat: transcribe_voice 新增 OpenAI Whisper API 后端 (#66)

默认 local，零行为变化。opt-in 双因素：transcription_backend=openai
且 openai_api_key 都齐才生效；任一缺失静默回退 local + stderr 一行警告。
首次进入云路径会 stderr 警告"语音将上传至 OpenAI 服务器"。

新增 config.json 字段：
- transcription_backend: "local" (默认) | "openai"
- local_whisper_model: "base" (替换 mcp_server.py 里硬编码 DEFAULT_WHISPER_MODEL)
- openai_api_key: "" (默认空；openai 包为 optional，按需 pip install)

关键技术选择：
- _transcribe(wav, backend) 单一 if/else 分发，不引入插件/工厂层
  (Rule of Three —— 只有一个云后端时不值得抽象)
- 文件 > 25MB 在 OpenAI() 实例化之前提前拒绝，避免无谓上传
- 错误分类清晰: 缺 key / 缺 openai 包 / 401 / 429 / APIError 各自的提示
- PR #58 缓存 schema 自然扩展: 条目加 backend 字段，命中需 backend+model_size 都匹配
- 旧条目缺 backend 字段视为 "local"，向前兼容 PR #58 已落盘的所有数据
- transcribe_chat.py 批量 CLI 与 MCP 工具共享同一份配置，保持一致

新增 2 个测试 (tests/test_openai_backend.py)，只覆盖回归风险最高的两条：
- 文件 > 25MB 必须在 SDK 实例化前拒绝（隐私契约的防线）
- backend 不匹配的旧条目不命中（避免切后端时返回错后端结果）

其余路径要么琐碎（默认值读取）、要么坏掉时声音很大（SDK 错误、ImportError），
要么已被 PR #58 现有测试隐式覆盖（缺 backend 字段的旧条目），不再单独写测试。

顺手把 README 里 PR #53 漏掉的 voice 三件套（get_voice_messages /
decode_voice / transcribe_voice）补进 MCP 工具表，并新增"⚠️ 语音转录隐私"
章节说清数据流向、成本（约 \$0.006/分钟）、25MB 上限、回退行为。

Closes ylytdeng/wechat-decrypt#59

README.md

@@ -196,11 +196,35 @@ claude mcp add wechat -- python C:\Users\你的用户名\wechat-decrypt\mcp_serv
 | `get_contact_tags()` | 列出所有联系人标签及成员数量 |
 | `get_tag_members(tag_name)` | 获取指定标签下的所有联系人，支持模糊匹配 |
 | `get_new_messages()` | 获取自上次调用以来的新消息 |
+| `get_voice_messages(chat_name)` | 列出某会话所有语音消息（local_id、时长、时间戳） |
+| `decode_voice(chat_name, local_id)` | 解码 SILK 语音为本地 WAV 文件 |
+| `transcribe_voice(chat_name, local_id)` | 转录语音为文字（自动检测语言） |
 
 前置条件：需要先运行 `python main.py` 或 `python find_all_keys.py` 完成密钥提取。
 
 说明：`search_messages` 的 `limit` 最大为 `500`；`get_chat_history` 支持更大的 `limit`，但消息很多时仍建议配合 `offset` 分页读取。
 
+#### ⚠️ 语音转录隐私
+
+`transcribe_voice` 默认使用本地 Whisper（CPU），数据全程留在本机。`transcribe_chat.py` 批量 CLI 共享同一份配置。
+
+如需切换到 OpenAI Whisper API（更快、Mandarin 精度更高），在 `config.json` 中：
+
+```json
+{
+    "transcription_backend": "openai",
+    "openai_api_key": "sk-..."
+}
+```
+
+启用后**语音文件会上传至 OpenAI 服务器**进行转录。需 `pip install openai`。
+
+- 成本：约 $0.006 / 分钟（OpenAI 计价）
+- 文件 > 25MB 在上传前被拒绝（OpenAI 上限）
+- 首次启用云后端时 stderr 会打一行警告
+- `transcription_backend` 或 `openai_api_key` 任一缺失时静默回退 local
+- 切换后端后，旧缓存条目（backend 不匹配）会自动重新转录
+
 **[查看使用案例 →](USAGE.md)**
 
 ### 图片解密 (V2 格式)

config.py

@@ -29,6 +29,11 @@ _DEFAULT = {
     "decrypted_dir": "decrypted",
     "decoded_image_dir": "decoded_images",
     "wechat_process": _DEFAULT_PROCESS,
+    # 语音转录后端: "local" (默认, 本地 Whisper) 或 "openai" (OpenAI API)
+    # 切到 openai 时语音将上传至 OpenAI 服务器, 详见 README "语音转录隐私" 章节
+    "transcription_backend": "local",
+    "local_whisper_model": "base",
+    "openai_api_key": "",
 }
 
 

mcp_server.py

@@ -1971,28 +1971,141 @@ def _save_voice_transcription_cache():
                 pass
 
 
-DEFAULT_WHISPER_MODEL = "base"
+# ============ 语音转录后端 ============
+#
+# 默认 local: 完全保留原有行为，CPU 上跑本地 Whisper。
+# opt-in openai: 需要 transcription_backend="openai" 且 openai_api_key 都齐
+# 才会上云；任一缺失静默回退 local + stderr 一行警告（用户感知到误配置但不阻塞）。
+# 详见 README "语音转录隐私" 章节。
+
+TRANSCRIPTION_BACKEND = _cfg.get("transcription_backend", "local")
+LOCAL_WHISPER_MODEL = _cfg.get("local_whisper_model", "base")
+OPENAI_API_KEY = _cfg.get("openai_api_key", "")
+
+OPENAI_WHISPER_MODEL = "whisper-1"           # OpenAI 当前唯一型号
+OPENAI_AUDIO_LIMIT_BYTES = 25 * 1024 * 1024  # OpenAI 25MB 上限
 
 _whisper_model = None
+_openai_client = None
+_openai_warning_emitted = False
+_fallback_warning_emitted = False
 
-def _get_whisper_model(model_size=DEFAULT_WHISPER_MODEL):
+
+def _resolve_active_backend():
+    """两因素 opt-in：openai 需要 flag + key 都齐才生效。"""
+    global _fallback_warning_emitted
+    if TRANSCRIPTION_BACKEND == "openai":
+        if not OPENAI_API_KEY:
+            if not _fallback_warning_emitted:
+                print(
+                    "[whisper] transcription_backend=openai 但未配置 openai_api_key，"
+                    "回退到本地模型",
+                    file=sys.stderr, flush=True,
+                )
+                _fallback_warning_emitted = True
+            return "local"
+        return "openai"
+    return "local"
+
+
+def _cache_signature():
+    """当前生效后端 + 模型，用作缓存命中判定 + 落盘字段。"""
+    backend = _resolve_active_backend()
+    if backend == "openai":
+        return {"backend": "openai", "model_size": OPENAI_WHISPER_MODEL}
+    return {"backend": "local", "model_size": LOCAL_WHISPER_MODEL}
+
+
+def _get_whisper_model(model_size=None):
     global _whisper_model
+    if model_size is None:
+        model_size = LOCAL_WHISPER_MODEL
     if _whisper_model is None:
         import whisper
         _whisper_model = whisper.load_model(model_size)
     return _whisper_model
 
 
+def _transcribe_local(wav_path):
+    model = _get_whisper_model()
+    result = model.transcribe(wav_path)
+    return {
+        "language": result.get("language", "unknown"),
+        "text": result.get("text", "").strip(),
+    }
+
+
+def _transcribe_openai(wav_path):
+    """通过 OpenAI Whisper API 转录。失败抛 RuntimeError，调用方负责面向用户的提示。"""
+    global _openai_client, _openai_warning_emitted
+
+    # 尺寸预检：放在 SDK 导入和实例化之前，确保超限文件绝不上传
+    size = os.path.getsize(wav_path)
+    if size > OPENAI_AUDIO_LIMIT_BYTES:
+        raise RuntimeError(
+            f"音频 {size / 1024 / 1024:.1f}MB 超过 OpenAI 25MB 上限，"
+            "提前拒绝以避免无谓上传"
+        )
+
+    try:
+        from openai import OpenAI
+        from openai import AuthenticationError, RateLimitError, APIError
+    except ImportError:
+        raise RuntimeError("缺少依赖: pip install openai")
+
+    if not _openai_warning_emitted:
+        print(
+            "[whisper] 已启用 OpenAI Whisper API，"
+            "语音将上传至 OpenAI 服务器进行转录",
+            file=sys.stderr, flush=True,
+        )
+        _openai_warning_emitted = True
+
+    if _openai_client is None:
+        _openai_client = OpenAI(api_key=OPENAI_API_KEY)
+
+    try:
+        with open(wav_path, "rb") as f:
+            result = _openai_client.audio.transcriptions.create(
+                model=OPENAI_WHISPER_MODEL,
+                file=f,
+                response_format="verbose_json",
+            )
+    except AuthenticationError:
+        raise RuntimeError("OpenAI 鉴权失败 (401)：检查 openai_api_key")
+    except RateLimitError:
+        raise RuntimeError("OpenAI 限流 (429)：稍后重试")
+    except APIError as e:
+        raise RuntimeError(f"OpenAI API 错误: {e}")
+
+    return {
+        "language": getattr(result, "language", "unknown"),
+        "text": (getattr(result, "text", "") or "").strip(),
+    }
+
+
+def _transcribe(wav_path, backend):
+    if backend == "openai":
+        return _transcribe_openai(wav_path)
+    return _transcribe_local(wav_path)
+
+
 @mcp.tool()
 def transcribe_voice(chat_name: str, local_id: int) -> str:
     """将微信语音消息转录为文字（自动检测语言，保留原语言）。
 
     首次转录会先解码 SILK 语音为 WAV，再用 Whisper 转录；结果缓存到
     voice_transcriptions.json，重复调用直接返回缓存（跳过 SILK 解码
-    和 Whisper 推理）。若 Whisper 默认模型升级（如 base → small），
+    和 Whisper 推理）。后端切换或本地模型升级（如 base → small）后，
-    旧条目自动视为失效并重新转录。首次运行会下载 Whisper 模型（约 145MB）。
+    旧条目自动视为失效并重新转录。首次运行本地模型会下载约 145MB 权重。
 
-    依赖: pip install silk-python openai-whisper (silk-python 的 import 名为 pysilk)
+    后端由 config.json 中 transcription_backend 字段控制（local/openai）。
+    详见 README "语音转录隐私" 章节。
+
+    依赖:
+      - 本地后端: pip install silk-python openai-whisper
+        (silk-python 的 import 名为 pysilk)
+      - OpenAI 后端: pip install silk-python openai
 
     Args:
         chat_name: 聊天对象的名字、备注名或wxid
@@ -2002,15 +2115,20 @@ def transcribe_voice(chat_name: str, local_id: int) -> str:
     if not username:
         return f"找不到聊天对象: {chat_name}"
 
+    sig = _cache_signature()
     cache_key = _voice_transcription_cache_key(username, local_id)
     cache = _load_voice_transcription_cache()
     entry = cache.get(cache_key)
+    # 命中要求 backend + model_size 都匹配。
+    # 旧条目 (PR #58 schema) 缺 backend 字段，回填默认 "local" 保持向前兼容
+    # —— 那时唯一存在的后端就是 local，语义上等价。
     if (
         isinstance(entry, dict)
         and "text" in entry
-        and entry.get("model_size") == DEFAULT_WHISPER_MODEL
+        and entry.get("backend", "local") == sig["backend"]
+        and entry.get("model_size") == sig["model_size"]
     ):
-        # 命中缓存：跳过 DB 查询、SILK 解码、Whisper 推理。
+        # 命中缓存：跳过 DB 查询、SILK 解码、转录。
         # 条目里存了 create_time，即使源 DB 中消息已被清理仍能返回历史转录。
         lang = entry.get("language", "unknown")
         cached_ts = entry.get("create_time")
@@ -2020,11 +2138,13 @@ def transcribe_voice(chat_name: str, local_id: int) -> str:
             time_label = "-"
         return f"[{time_label}] ({lang})\n{entry['text']}"
 
-    # 未命中：只有这条路径才需要 whisper / pysilk。
+    # 未命中：本地后端才需要 whisper 包，云后端在 _transcribe_openai 内单独检查
-    try:
+    if sig["backend"] == "local":
-        import whisper  # noqa: F401
+        try:
-    except ImportError:
+            import whisper  # noqa: F401
-        return "缺少依赖: pip install openai-whisper"
+        except ImportError:
+            return "缺少依赖: pip install openai-whisper"
+    # SILK 解码两条路径都需要
     try:
         import pysilk  # noqa: F401
     except ImportError:
@@ -2037,18 +2157,21 @@ def transcribe_voice(chat_name: str, local_id: int) -> str:
     voice_data, create_time = row
     wav_path, _ = _silk_to_wav(voice_data, create_time, username, local_id)
 
-    model = _get_whisper_model()
+    try:
-    result = model.transcribe(wav_path)
+        result = _transcribe(wav_path, sig["backend"])
-    lang = result.get("language", "unknown")
+    except RuntimeError as e:
-    text = result.get("text", "").strip()
+        return str(e)
+    text = result["text"]
+    lang = result["language"]
 
     # 写缓存：即使 text 为空也缓存（Whisper 偶尔对静音/极短片段返回空），
-    # 配合 model_size 字段，升级模型后会自动重转，避免永久钉死空结果。
+    # 配合 backend + model_size 字段，切换后端或升级模型后会自动重转。
     cache[cache_key] = {
         "text": text,
         "language": lang,
         "create_time": int(create_time),
-        "model_size": DEFAULT_WHISPER_MODEL,
+        "backend": sig["backend"],
+        "model_size": sig["model_size"],
     }
     _save_voice_transcription_cache()
 

tests/test_openai_backend.py

@@ -0,0 +1,116 @@
+"""
+issue #59: opt-in OpenAI Whisper API 后端的两条关键回归测试。
+
+只测两件事：
+1. 隐私契约: 文件 > 25MB 在调用 OpenAI SDK 之前就被拒绝（保证不会无意上传）
+2. 缓存正确性: backend 不匹配的旧条目不会被命中（避免切后端时返回错后端结果）
+
+其余路径要么琐碎（默认值读取）、要么坏掉时声音很大（SDK 错误、ImportError），
+不再单独覆盖。
+"""
+import os
+import sys
+import tempfile
+import unittest
+from unittest.mock import MagicMock, patch
+
+import mcp_server
+
+
+class _CacheIsolationMixin:
+    """与 test_voice_transcription_cache.py 同款隔离：避免污染 module-level 缓存状态。"""
+
+    def setUp(self):
+        self._saved_cache = mcp_server._voice_transcription_cache
+        self._saved_path = mcp_server.VOICE_TRANSCRIPTION_CACHE_FILE
+        self._saved_warned = mcp_server._voice_transcription_save_warned
+
+        mcp_server._voice_transcription_cache = None
+        mcp_server._voice_transcription_save_warned = False
+
+        self._tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self._tmp.cleanup)
+        mcp_server.VOICE_TRANSCRIPTION_CACHE_FILE = os.path.join(
+            self._tmp.name, "voice_transcriptions.json"
+        )
+
+    def tearDown(self):
+        mcp_server._voice_transcription_cache = self._saved_cache
+        mcp_server.VOICE_TRANSCRIPTION_CACHE_FILE = self._saved_path
+        mcp_server._voice_transcription_save_warned = self._saved_warned
+
+
+class OpenAIBackendPrivacyTests(unittest.TestCase):
+    """隐私契约：超限文件必须在 OpenAI SDK 实例化之前就被拒绝。
+
+    若有人把 size check 移到 OpenAI(api_key=...) 之后（即便仍在 upload 前），
+    本测试会失败 —— 这层防御边界值得守住。
+    """
+
+    def test_oversize_audio_rejected_before_sdk_call(self):
+        # 写一个 26MB 临时 WAV (用稀疏写法快速生成)
+        with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
+            f.seek(26 * 1024 * 1024)
+            f.write(b"\0")
+            big_path = f.name
+        self.addCleanup(os.unlink, big_path)
+
+        # 注入一个假的 openai 模块，保证 import 成功；OpenAI 构造函数若被调用即测试失败
+        fake_openai = MagicMock()
+        fake_openai.OpenAI = MagicMock(
+            side_effect=AssertionError("OpenAI() must not be instantiated for oversize files")
+        )
+        fake_openai.AuthenticationError = type("AuthenticationError", (Exception,), {})
+        fake_openai.RateLimitError = type("RateLimitError", (Exception,), {})
+        fake_openai.APIError = type("APIError", (Exception,), {})
+
+        with patch.dict(sys.modules, {"openai": fake_openai}):
+            with self.assertRaises(RuntimeError) as ctx:
+                mcp_server._transcribe_openai(big_path)
+
+        self.assertIn("25MB", str(ctx.exception))
+        fake_openai.OpenAI.assert_not_called()
+
+
+class CacheBackendMatchTests(_CacheIsolationMixin, unittest.TestCase):
+    """缓存正确性：backend 不匹配 → 视为 miss，避免切后端时返回错后端结果。"""
+
+    def test_cache_hit_requires_backend_match(self):
+        # 种入一条 openai 后端的缓存条目
+        key = mcp_server._voice_transcription_cache_key("wxid_test", 42)
+        cache = mcp_server._load_voice_transcription_cache()
+        cache[key] = {
+            "text": "openai-result",
+            "language": "zh",
+            "create_time": 1700000000,
+            "backend": "openai",
+            "model_size": "whisper-1",
+        }
+        mcp_server._save_voice_transcription_cache()
+
+        # 当前后端是 local，应当 miss → 走转录流程而非返回 "openai-result"
+        with patch.object(mcp_server, "TRANSCRIPTION_BACKEND", "local"), \
+             patch.object(mcp_server, "OPENAI_API_KEY", ""), \
+             patch.object(mcp_server, "resolve_username", return_value="wxid_test"), \
+             patch.object(mcp_server, "_fetch_voice_row",
+                          return_value=(b"\x02fake-silk-blob", 1700000001)), \
+             patch.object(mcp_server, "_silk_to_wav",
+                          return_value=("/tmp/fake.wav", 24000 * 2)), \
+             patch.object(mcp_server, "_transcribe_local",
+                          return_value={"text": "local-result", "language": "zh"}), \
+             patch.dict(sys.modules, {"whisper": MagicMock(), "pysilk": MagicMock()}):
+            result = mcp_server.transcribe_voice("test_contact", 42)
+
+        # 没返回旧 openai 缓存，而是走了 local 转录流程
+        self.assertNotIn("openai-result", result)
+        self.assertIn("local-result", result)
+
+        # 落盘的新条目应记录当前后端
+        mcp_server._voice_transcription_cache = None
+        reloaded = mcp_server._load_voice_transcription_cache()
+        self.assertEqual(reloaded[key]["backend"], "local")
+        self.assertEqual(reloaded[key]["text"], "local-result")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_voice_transcription_cache.py

@@ -194,7 +194,7 @@ class TranscribeVoiceCacheHitTests(_CacheIsolationMixin, unittest.TestCase):
             "text": "缓存命中文本",
             "language": "zh",
             "create_time": 1700000000,
-            "model_size": mcp_server.DEFAULT_WHISPER_MODEL,
+            "model_size": mcp_server.LOCAL_WHISPER_MODEL,
         })
 
         with patch.object(mcp_server, "resolve_username", return_value="wxid_test") as mock_resolve, \
@@ -216,7 +216,7 @@ class TranscribeVoiceCacheHitTests(_CacheIsolationMixin, unittest.TestCase):
         self._seed(key, {
             "text": "历史条目",
             "language": "zh",
-            "model_size": mcp_server.DEFAULT_WHISPER_MODEL,
+            "model_size": mcp_server.LOCAL_WHISPER_MODEL,
         })
 
         with patch.object(mcp_server, "resolve_username", return_value="wxid_test"), \
@@ -233,7 +233,7 @@ class TranscribeVoiceCacheHitTests(_CacheIsolationMixin, unittest.TestCase):
             "text": "",
             "language": "zh",
             "create_time": 1700000000,
-            "model_size": mcp_server.DEFAULT_WHISPER_MODEL,
+            "model_size": mcp_server.LOCAL_WHISPER_MODEL,
         })
 
         with patch.object(mcp_server, "resolve_username", return_value="wxid_test"), \
@@ -244,7 +244,7 @@ class TranscribeVoiceCacheHitTests(_CacheIsolationMixin, unittest.TestCase):
         self.assertIn("(zh)", result)
 
     def test_model_mismatch_is_treated_as_miss(self):
-        # 缓存条目的 model_size 和当前 DEFAULT_WHISPER_MODEL 不一致时，
+        # 缓存条目的 model_size 和当前 LOCAL_WHISPER_MODEL 不一致时，
         # 不应命中；进入 miss 路径（这里无 whisper 依赖，应落到"缺少依赖"分支）。
         key = mcp_server._voice_transcription_cache_key("wxid_test", 10)
         self._seed(key, {

transcribe_chat.py

@@ -13,10 +13,12 @@
     .venv/bin/python3 transcribe_chat.py /tmp/chat.json /tmp/chat_transcribed.json
 
 行为说明:
-    - 使用 OpenAI Whisper (CPU，单线程) 对每条语音消息转录。
+    - 后端由 config.json 中 transcription_backend 字段控制 (local/openai)，
+      与 MCP transcribe_voice 工具共享配置。详见 README "语音转录隐私" 章节。
+    - 默认 local: 使用本地 Whisper (CPU，单线程)，首次运行下载 ~145 MB 权重。
+    - 切到 openai: 语音上传至 OpenAI 服务器转录 (~$0.006/分钟)。
     - 幂等: 已有 "transcription" 字段的消息会被跳过，因此崩溃/中断后可安全重跑。
     - 崩溃安全: 每处理完一条即整体重写输出 JSON，进程中断最多丢失当前一条。
-    - 首次运行会下载 Whisper 模型 (~145 MB) 并缓存。
 
 需要 WeChat DB 仍然在线/已解密 —— 语音 blob 是从 DB 现场按 local_id 读取的，
 不从 JSON 读。
@@ -29,7 +31,7 @@ from datetime import datetime
 import mcp_server
 
 
-def _transcribe_local_id(username, local_id):
+def _transcribe_local_id(username, local_id, backend):
     row = mcp_server._fetch_voice_row(username, local_id)
     if row is None:
         return "[not found]"
@@ -41,9 +43,8 @@ def _transcribe_local_id(username, local_id):
         return f"[decode error: {e}]"
 
     try:
-        model = mcp_server._get_whisper_model()
+        result = mcp_server._transcribe(wav_path, backend)
-        result = model.transcribe(wav_path)
+        return result["text"]
-        return result.get("text", "").strip()
     except Exception as e:
         return f"[transcribe error: {e}]"
 
@@ -70,17 +71,22 @@ def transcribe_export(input_path, output_path):
         print("No voice messages to transcribe.")
         return
 
+    backend = mcp_server._resolve_active_backend()
     print(f"Found {total} voice messages to transcribe.")
-    print("Loading Whisper model (first run downloads ~145MB)...")
+    print(f"Backend: {backend}")
-    mcp_server._get_whisper_model()
+    if backend == "local":
-    print("Model ready.\n")
+        print("Loading Whisper model (first run downloads ~145MB)...")
+        mcp_server._get_whisper_model()
+        print("Model ready.\n")
+    else:
+        print("")
 
     for i, msg in enumerate(pending, 1):
         local_id = msg["local_id"]
         ts = msg["timestamp"]
         ts_str = datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M:%S") if isinstance(ts, (int, float)) else ts
         print(f"[{i}/{total}] local_id={local_id} ({ts_str}) ... ", end="", flush=True)
-        result = _transcribe_local_id(username, local_id)
+        result = _transcribe_local_id(username, local_id, backend)
         msg["transcription"] = result
         print(repr(result[:60]) if result else '""')