diff --git a/export_all_chats.py b/export_all_chats.py index cd6e30c..b7f0014 100644 --- a/export_all_chats.py +++ b/export_all_chats.py @@ -1,14 +1,21 @@ #!/usr/bin/env python3 -"""批量导出所有微信聊天记录为 JSON 文件。 +"""批量导出所有微信聊天记录为 JSON 文件，可选附带语音转录。 此脚本将导出所有会话的聊天记录，输出格式与 export_chat.py 完全一致。 支持导出到指定目录，默认输出到 ./exported_chats 目录。 +语音转录通过 mcp_server 的 backend 配置驱动（config.json 中设置 +transcription_backend 为 whisper_cpp / openai / local）。未启用 backend +或缺少依赖时仅导出文本消息，不报错。 + 用法: - python3 export_all_chats.py [output_dir] + python3 export_all_chats.py [output_dir] # 仅导出 + python3 export_all_chats.py --with-transcriptions # 导出 + 语音转录 示例: python3 export_all_chats.py /path/to/output + python3 export_all_chats.py --with-transcriptions + python3 export_all_chats.py --with-transcriptions /path/to/output """ import argparse @@ -17,6 +24,7 @@ import os import re import sqlite3 import sys +import time from contextlib import closing from datetime import datetime @@ -24,7 +32,7 @@ import mcp_server from chat_export_helpers import _extract_content, _msg_type_str, _resolve_sender -def export_one(username, output_dir, names): +def export_one(username, output_dir, names, transcribe=False): """ 导出单个会话。 @@ -82,6 +90,39 @@ def export_one(username, output_dir, names): if not messages: return False, 0, "empty" + # ── 语音转录 ────────────────────────────────────────────── + if transcribe: + transcribed = 0 + failed = 0 + for msg in messages: + if msg.get("type") != "voice": + continue + lid = msg["local_id"] + try: + row = mcp_server._fetch_voice_row(username, lid) + if row is None: + continue + voice_data, create_time = row + wav_path, _ = mcp_server._silk_to_wav( + voice_data, create_time, username, lid + ) + backend = _resolve_backend() + result = mcp_server._transcribe(wav_path, backend) + if result and result.get("text"): + msg["transcription"] = result["text"] + transcribed += 1 + os.unlink(wav_path) + except Exception: + failed += 1 + if transcribed or failed: + display = names.get(username, username) + voice_total = sum(1 for m in messages if m.get("type") == "voice") + print( + f" 转录: {transcribed}/{voice_total} 条语音" + + (f" ({failed} 失败)" if failed else "") + ) + + # ── 写文件 ──────────────────────────────────────────────── output = { "chat": display_name, "username": username, @@ -100,13 +141,29 @@ def export_one(username, output_dir, names): return True, len(messages), None +_BACKEND_CACHE = None + + +def _resolve_backend(): + """解析转录 backend，结果缓存以避免重复检测。""" + global _BACKEND_CACHE + if _BACKEND_CACHE is None: + try: + _BACKEND_CACHE = mcp_server._resolve_active_backend() + except Exception: + _BACKEND_CACHE = "local" + return _BACKEND_CACHE + + def main(): parser = argparse.ArgumentParser( - description="批量导出所有微信聊天记录为 JSON 文件", + description="批量导出所有微信聊天记录为 JSON 文件，可选附带语音转录", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" 示例: python3 export_all_chats.py /path/to/output + python3 export_all_chats.py --with-transcriptions + python3 export_all_chats.py -t /path/to/output """, ) parser.add_argument( @@ -115,11 +172,25 @@ def main(): default=None, help="输出目录路径 (默认: ./exported_chats)", ) + parser.add_argument( + "-t", + "--with-transcriptions", + action="store_true", + help="导出时一并转录语音消息（依赖 config.json 配置的 backend）", + ) args = parser.parse_args() script_dir = os.path.dirname(os.path.abspath(__file__)) output_dir = args.output_dir or os.path.join(script_dir, "exported_chats") + if args.with_transcriptions: + try: + backend = _resolve_backend() + print(f"语音转录: 启用 (backend={backend})") + except Exception as e: + print(f"语音转录: backend 解析失败: {e}", file=sys.stderr) + args.with_transcriptions = False + if not os.path.exists(mcp_server.DECRYPTED_DIR): print(f"错误: 解密目录不存在: {mcp_server.DECRYPTED_DIR}", file=sys.stderr) sys.exit(1) @@ -128,7 +199,9 @@ def main(): session_db = os.path.join(mcp_server.DECRYPTED_DIR, "session", "session.db") try: with closing(sqlite3.connect(session_db)) as conn: - sessions = [u for u, _ in conn.execute("SELECT username, type FROM SessionTable")] + sessions = [u for u, _ in conn.execute( + "SELECT username, type FROM SessionTable" + )] except sqlite3.Error as e: print(f"会话数据库查询失败: {e}", file=sys.stderr) sys.exit(1) @@ -140,15 +213,23 @@ def main(): print(f"输出目录: {output_dir}") print("=" * 60) + t0 = time.time() ok, skip, err, total = 0, 0, 0, 0 for i, username in enumerate(sessions, 1): display = names.get(username, username) - success, count, reason = export_one(username, output_dir, names) + success, count, reason = export_one( + username, output_dir, names, transcribe=args.with_transcriptions + ) if success: ok += 1 total += count if i <= 10 or i % 100 == 0: - print(f"[{i}/{len(sessions)}] {display} - {count} 条消息") + elapsed = time.time() - t0 + eta = (elapsed / i) * (len(sessions) - i) if i > 0 else 0 + print( + f"[{i}/{len(sessions)}] {display} - {count} 条消息" + + (f" ETA {eta/60:.0f}分" if i > 1 else "") + ) else: if "no tables" in str(reason) or "empty" in str(reason): skip += 1 @@ -158,9 +239,13 @@ def main(): err += 1 print(f"[{i}/{len(sessions)}] {display} - 失败: {reason}") + elapsed = time.time() - t0 print() print("=" * 60) - print(f"完成! 成功={ok} 跳过={skip} 失败={err} 总消息={total}") + print( + f"完成! 成功={ok} 跳过={skip} 失败={err} " + f"总消息={total} 耗时={elapsed/60:.0f}分" + ) if __name__ == "__main__":