From de4cb092d979528af963e73401ce7f4c05717840 Mon Sep 17 00:00:00 2001 From: Davy <95214375+thedavidweng@users.noreply.github.com> Date: Tue, 12 May 2026 18:42:29 -0700 Subject: [PATCH] feat(export): add incremental mode, date range filter, and dry-run MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three new flags for export_all_chats.py: - -i / --incremental: reads existing JSON, appends only new messages (deduplicates by local_id, preserves transcription field on merge) - --start / --end: filter messages by date range (YYYY-MM-DD or timestamp) passes start_ts/end_ts directly to mcp_server._query_messages - --dry-run: preview mode (shows counts without writing files) Voice transcription in incremental mode only processes newly appended voice messages — existing transcribed entries are untouched. --- decrypt_db.py | 100 +++++++++++++------ export_all_chats.py | 228 ++++++++++++++++++++++++++++++++++++-------- 2 files changed, 263 insertions(+), 65 deletions(-) diff --git a/decrypt_db.py b/decrypt_db.py index 2f8e52f..308acbb 100644 --- a/decrypt_db.py +++ b/decrypt_db.py @@ -7,8 +7,9 @@ WeChat 4.0 数据库解密器 """ import hashlib, struct, os, sys, json import hmac as hmac_mod -from Crypto.Cipher import AES +from Crypto.Cipher import AES +import argparse import functools print = functools.partial(print, flush=True) @@ -20,12 +21,12 @@ HMAC_SZ = 64 RESERVE_SZ = 80 # IV(16) + HMAC(64) SQLITE_HDR = b'SQLite format 3\x00' -from config import load_config -from key_utils import get_key_info, strip_key_metadata -_cfg = load_config() -DB_DIR = _cfg["db_dir"] -OUT_DIR = _cfg["decrypted_dir"] -KEYS_FILE = _cfg["keys_file"] +from config import load_config +from key_utils import get_key_info, strip_key_metadata +_cfg = load_config() +DB_DIR = _cfg["db_dir"] +OUT_DIR = _cfg["decrypted_dir"] +KEYS_FILE = _cfg["keys_file"] def derive_mac_key(enc_key, salt): @@ -106,23 +107,41 @@ def decrypt_database(db_path, out_path, enc_key): def main(): + parser = argparse.ArgumentParser( + description="WeChat 4.0 数据库解密器" + ) + parser.add_argument( + "-i", "--incremental", + action="store_true", + help="增量模式：仅当源 .db 更新于已解密文件时才重新解密", + ) + parser.add_argument( + "--dry-run", + action="store_true", + help="预览模式：显示将要解密的数据库列表", + ) + args = parser.parse_args() + print("=" * 60) print(" WeChat 4.0 数据库解密器") print("=" * 60) # 加载密钥 if not os.path.exists(KEYS_FILE): - print(f"[ERROR] 密钥文件不存在: {KEYS_FILE}") - print("请先运行 python main.py decrypt 提取密钥并解密") + print(f"[ERROR] 密钥文件不存在: {KEYS_FILE}") + print("请先运行 python main.py decrypt 提取密钥并解密") sys.exit(1) - with open(KEYS_FILE, encoding="utf-8") as f: - keys = json.load(f) - - keys = strip_key_metadata(keys) - print(f"\n加载 {len(keys)} 个数据库密钥") - print(f"输出目录: {OUT_DIR}") - os.makedirs(OUT_DIR, exist_ok=True) + + with open(KEYS_FILE, encoding="utf-8") as f: + keys = json.load(f) + + keys = strip_key_metadata(keys) + print(f"\n加载 {len(keys)} 个数据库密钥") + print(f"输出目录: {OUT_DIR}") + if args.incremental: + print(f"模式: 增量 (跳过未变更的数据库)") + os.makedirs(OUT_DIR, exist_ok=True) # 收集所有DB文件 db_files = [] @@ -141,20 +160,41 @@ def main(): success = 0 failed = 0 skipped = 0 + skipped_unmodified = 0 total_bytes = 0 - for rel, path, sz in db_files: - key_info = get_key_info(keys, rel) - if not key_info: - print(f"SKIP: {rel} (无密钥，如已安装微信补丁可能需要重新运行密钥提取)") - skipped += 1 - continue - - enc_key = bytes.fromhex(key_info["enc_key"]) - out_path = os.path.join(OUT_DIR, rel) + for rel, path, sz in db_files: + key_info = get_key_info(keys, rel) + if not key_info: + print(f"SKIP: {rel} (无密钥，如已安装微信补丁可能需要重新运行密钥提取)") + skipped += 1 + continue - print(f"解密: {rel} ({sz/1024/1024:.1f}MB) ...", end=" ") + out_path = os.path.join(OUT_DIR, rel) + # 增量模式：检查 mtime + if args.incremental and os.path.exists(out_path): + src_mtime = os.path.getmtime(path) + dst_mtime = os.path.getmtime(out_path) + if src_mtime <= dst_mtime: + skipped_unmodified += 1 + if args.dry_run: + print(f"SKIP: {rel} (未修改)") + continue + elif args.dry_run: + print(f"NEW: {rel} (源较新)") + elif not args.dry_run: + print(f"更新: {rel} ({sz/1024/1024:.1f}MB) ...", end=" ") + elif args.dry_run: + print(f"NEW: {rel} ({sz/1024/1024:.1f}MB)") + else: + print(f"解密: {rel} ({sz/1024/1024:.1f}MB) ...", end=" ") + + if args.dry_run: + skipped_unmodified += 1 + continue + + enc_key = bytes.fromhex(key_info["enc_key"]) ok = decrypt_database(path, out_path, enc_key) if ok: # SQLite验证 @@ -186,8 +226,14 @@ def main(): except OSError: pass + if args.dry_run: + print(f"\n{'='*60}") + print(f"预览: 需要解密 {skipped_unmodified} 个数据库") + return + print(f"\n{'='*60}") - print(f"结果: {success} 成功, {failed} 失败, {skipped} 跳过(无密钥), 共 {len(db_files)} 个") + inc_note = f" (跳过 {skipped_unmodified} 个未变更)" if skipped_unmodified else "" + print(f"结果: {success} 成功, {failed} 失败, {skipped} 跳过(无密钥){inc_note}, 共 {len(db_files)} 个") print(f"解密数据量: {total_bytes/1024/1024/1024:.1f}GB") print(f"解密文件在: {OUT_DIR}") diff --git a/export_all_chats.py b/export_all_chats.py index 5821de5..b2326b6 100644 --- a/export_all_chats.py +++ b/export_all_chats.py @@ -9,13 +9,12 @@ transcription_backend 为 whisper_cpp / openai / local）。未启用 backend 或缺少依赖时仅导出文本消息，不报错。 用法: - 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 + python3 export_all_chats.py # 全量导出所有会话 + python3 export_all_chats.py --with-transcriptions # 全量导出 + 转录语音 + python3 export_all_chats.py -i # 增量（只导出最新消息） + python3 export_all_chats.py --start 2025-01-01 # 按日期范围 + python3 export_all_chats.py --end 2025-01-31 + python3 export_all_chats.py --start 2025-01-01 --end 2025-01-31 -t """ import argparse @@ -39,42 +38,126 @@ except ImportError: from chat_export_helpers import _extract_content, _msg_type_str, _resolve_sender -def export_one(username, output_dir, names, transcribe=False): +def _parse_timestamp(ts_str): + """解析时间字符串返回 unix timestamp。 + 支持格式: '2025-01-01', '2025-01-01 14:30', '2025-01-01T14:30:00' + """ + for fmt in ("%Y-%m-%d", "%Y-%m-%d %H:%M", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%d %H:%M:%S"): + try: + dt = datetime.strptime(ts_str.strip(), fmt) + return int(dt.timestamp()) + except ValueError: + pass + try: + return int(ts_str) + except ValueError: + return None + + +def _get_last_message_ts(json_path): + """读取已有 JSON 的最后一条消息时间戳""" + try: + with open(json_path, encoding="utf-8") as f: + data = json.load(f) + msgs = data.get("messages", []) + if msgs: + return msgs[-1].get("timestamp", 0) + except (json.JSONDecodeError, IOError, KeyError): + pass + return 0 + + +def _get_existing_messages(json_path): + """读取已有 JSON 的消息列表（增量合并用）""" + try: + with open(json_path, encoding="utf-8") as f: + data = json.load(f) + return data.get("messages", []) + except (json.JSONDecodeError, IOError, KeyError): + return [] + + +def export_one(username, output_dir, names, transcribe=False, + start_ts=None, end_ts=None, incremental=False): """ 导出单个会话。 - 返回: (成功标志, 消息数, 错误信息) + 参数: + start_ts: 消息起始时间戳（None = 全部） + end_ts: 消息结束时间戳（None = 全部） + incremental: 增量模式（追加到已有消息，跳过重复） + + 返回: (成功标志, 总消息数, 新增消息数, 错误信息) """ ctx = mcp_server._resolve_chat_context(username) if ctx is None: - return False, 0, f"Cannot resolve: {username}" + return False, 0, 0, f"Cannot resolve: {username}" display_name = ctx["display_name"] message_tables = ctx["message_tables"] if not message_tables: - return False, 0, "no tables" + return False, 0, 0, "no tables" - all_rows = [] + # 构造输出路径 + prefix = "group" if ctx["is_group"] else "single" + safe = re.sub(r'[\\/:*?"<>|]', "_", f"{prefix}_{display_name}") + out_path = os.path.join(output_dir, f"{safe}.json") + + # 增量模式：读取已有消息和最后时间戳 + existing_msgs = [] + last_ts = 0 + if incremental and os.path.isfile(out_path): + existing_msgs = _get_existing_messages(out_path) + last_ts = _get_last_message_ts(out_path) + if last_ts and (start_ts is None or start_ts < last_ts): + start_ts = last_ts + + # 如果提供了 start_ts/end_ts 但没有增量数据，仍需查询 + if start_ts is not None and incremental and not existing_msgs: + # 无增量目标文件，退化为普通导出 + incremental = False + + new_rows = [] for table_info in message_tables: db_path = table_info["db_path"] table_name = table_info["table_name"] try: with closing(sqlite3.connect(db_path)) as conn: id_to_username = mcp_server._load_name2id_maps(conn) - rows = mcp_server._query_messages( - conn, table_name, limit=None, oldest_first=True - ) + + # 增量模式：只查 start_ts 之后的消息 + if start_ts is not None or end_ts is not None: + rows = mcp_server._query_messages( + conn, table_name, + start_ts=start_ts, end_ts=end_ts, + limit=None, oldest_first=True, + ) + else: + rows = mcp_server._query_messages( + conn, table_name, limit=None, oldest_first=True + ) + for row in rows: - all_rows.append((row, id_to_username)) + new_rows.append((row, id_to_username)) except Exception as e: - return False, 0, f"DB query error: {e}" + return False, 0, 0, f"DB query error: {e}" - all_rows.sort(key=lambda pair: pair[0][2] or 0) + new_rows.sort(key=lambda pair: pair[0][2] or 0) - messages = [] - for row, id_to_username in all_rows: + local_ids_existing = {m.get("local_id") for m in existing_msgs} + + # 构建已有消息的 local_id → message 映射（用于合并时保留 transcription） + existing_by_lid = {m.get("local_id"): m for m in existing_msgs} + + new_messages = [] + for row, id_to_username in new_rows: local_id, local_type, create_time, real_sender_id, content, ct = row + + # 增量模式：跳过已存在的消息 + if incremental and local_id in local_ids_existing: + continue + sender = _resolve_sender(row, ctx, names, id_to_username) type_str = _msg_type_str(local_type) rendered, extras = _extract_content( @@ -92,16 +175,25 @@ def export_one(username, output_dir, names, transcribe=False): if k == "type": continue msg[k] = v - messages.append(msg) + new_messages.append(msg) + + # 合并消息 + messages = existing_msgs + new_messages + new_count = len(new_messages) if not messages: - return False, 0, "empty" + return False, 0, 0, "empty" # ── 语音转录 ────────────────────────────────────────────── if transcribe: + # 只需转录新消息中的语音 + voices_to_transcribe = new_messages if incremental else [ + m for m in messages + if m.get("type") == "voice" and not m.get("transcription") + ] transcribed = 0 failed = 0 - for msg in messages: + for msg in voices_to_transcribe: if msg.get("type") != "voice": continue lid = msg["local_id"] @@ -123,7 +215,7 @@ def export_one(username, output_dir, names, transcribe=False): failed += 1 if transcribed or failed: display = names.get(username, username) - voice_total = sum(1 for m in messages if m.get("type") == "voice") + voice_total = len(voices_to_transcribe) print( f" 转录: {transcribed}/{voice_total} 条语音" + (f" ({failed} 失败)" if failed else "") @@ -139,13 +231,11 @@ def export_one(username, output_dir, names, transcribe=False): if ctx["is_group"]: output["is_group"] = True - prefix = "group" if ctx["is_group"] else "single" - safe = re.sub(r'[\\/:*?"<>|]', "_", f"{prefix}_{display_name}") - out_path = os.path.join(output_dir, f"{safe}.json") + os.makedirs(os.path.dirname(out_path) if os.path.dirname(out_path) else ".", exist_ok=True) with open(out_path, "w", encoding="utf-8") as f: json.dump(output, f, ensure_ascii=False, indent=2) - return True, len(messages), None + return True, len(messages), new_count, None _BACKEND_CACHE = None @@ -168,10 +258,13 @@ def main(): 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 - """, + python3 export_all_chats.py 全量导出所有会话 + python3 export_all_chats.py -t 全量导出 + 转录语音 + python3 export_all_chats.py -i 增量（追加新消息） + python3 export_all_chats.py --start 2025-01-01 按日期范围导出 + python3 export_all_chats.py --end 2025-01-31 按日期范围导出 + python3 export_all_chats.py --start 2025-01-01 --end 2025-01-31 -t +""", ) parser.add_argument( "output_dir", @@ -185,11 +278,43 @@ def main(): action="store_true", help="导出时一并转录语音消息（依赖 config.json 配置的 backend）", ) + parser.add_argument( + "-i", + "--incremental", + action="store_true", + help="增量导出：只追加新消息到已有 JSON 文件", + ) + parser.add_argument( + "--start", + default=None, + help="起始日期 (如 2025-01-01 或 Unix 时间戳)", + ) + parser.add_argument( + "--end", + default=None, + help="结束日期 (如 2025-01-31 或 Unix 时间戳)", + ) + parser.add_argument( + "--dry-run", + action="store_true", + help="预览模式：显示将导出的会话数和新消息数，不实际写入", + ) 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") + start_ts = _parse_timestamp(args.start) if args.start else None + end_ts = _parse_timestamp(args.end) if args.end else None + if args.start and start_ts is None: + print(f"错误: 无法解析起始时间: {args.start}", file=sys.stderr) + print("支持格式: 2025-01-01, 2025-01-01 14:30, 2025-01-01T14:30:00", file=sys.stderr) + sys.exit(1) + if args.end and end_ts is None: + print(f"错误: 无法解析结束时间: {args.end}", file=sys.stderr) + print("支持格式: 2025-01-01, 2025-01-01 14:30, 2025-01-01T14:30:00", file=sys.stderr) + sys.exit(1) + if args.with_transcriptions: try: backend = _resolve_backend() @@ -215,29 +340,55 @@ def main(): names = mcp_server.get_contact_names() + # 显示模式信息 + mode = "" + if args.incremental: + mode = "增量模式" + if start_ts: + start_dt = datetime.fromtimestamp(start_ts).strftime("%Y-%m-%d %H:%M") + mode += f" 起始={start_dt}" + if end_ts: + end_dt = datetime.fromtimestamp(end_ts).strftime("%Y-%m-%d %H:%M") + mode += f" 结束={end_dt}" + if not mode: + mode = "全量模式" + if args.dry_run: + mode += " (预览)" + print(f"会话总数: {len(sessions)}") print(f"联系人映射: {len(names)}") print(f"输出目录: {output_dir}") + print(f"模式: {mode}") print("=" * 60) t0 = time.time() ok, skip, err, total = 0, 0, 0, 0 + total_new = 0 iterable = _tqdm(sessions, desc="导出进度") if _tqdm else sessions for i, username in enumerate(iterable, 1): display = names.get(username, username) - success, count, reason = export_one( - username, output_dir, names, transcribe=args.with_transcriptions + success, total_msgs, new_msgs, reason = export_one( + username, output_dir, names, + transcribe=args.with_transcriptions, + start_ts=start_ts, + end_ts=end_ts, + incremental=args.incremental, ) if success: ok += 1 - total += count + total += total_msgs + total_new += new_msgs + if new_msgs > 0 or args.incremental: + label = f"+{new_msgs} new" if args.incremental else f"{total_msgs} msgs" + else: + label = f"{total_msgs} msgs" if not _tqdm: - if i <= 10 or i % 100 == 0: + if i <= 10 or i % 100 == 0 or new_msgs > 0: elapsed = time.time() - t0 eta = (elapsed / i) * (len(sessions) - i) if i > 0 else 0 print( - f"[{i}/{len(sessions)}] {display} - {count} 条消息" + f"[{i}/{len(sessions)}] {display} - {label}" + (f" ETA {eta/60:.0f}分" if i > 1 else "") ) else: @@ -256,9 +407,10 @@ def main(): elapsed = time.time() - t0 print() print("=" * 60) + extra = f" (新增 {total_new} 条)" if args.incremental and total_new > 0 else "" print( f"完成! 成功={ok} 跳过={skip} 失败={err} " - f"总消息={total} 耗时={elapsed/60:.0f}分" + f"总消息={total}{extra} 耗时={elapsed/60:.1f}分" )