diff --git a/.gitignore b/.gitignore index e65e66d..b12dbda 100644 --- a/.gitignore +++ b/.gitignore @@ -398,9 +398,10 @@ find_all_keys_macos decoded_voices/ voice_transcriptions.json data/ -export/ -build/ -dist/ +export/ +export_plan*.csv +build/ +dist/ decrypted/ all_keys.json wxwork_decrypted/ diff --git a/README.md b/README.md index 3a4b1b4..88d6c42 100644 --- a/README.md +++ b/README.md @@ -218,10 +218,23 @@ py -m pip install --user -r requirements.txt | 解密全部数据库 | `python decrypt_db.py` | | 启动 Web UI（实时消息） | `python main.py` | | 批量导出聊天记录 | `python export_all_chats.py` | +| 生成导出计划 CSV（黑名单，默认） | `python export_all_chats.py --write-plan-csv export_plan.csv` | +| 生成导出计划 CSV（白名单） | `python export_all_chats.py --write-plan-csv export_plan.csv --plan-mode whitelist` | +| 按计划 CSV 导出（黑名单，默认） | `python export_all_chats.py output_dir --from-plan-csv export_plan.csv` | +| 按计划 CSV 导出（白名单） | `python export_all_chats.py output_dir --from-plan-csv export_plan.csv --plan-mode whitelist` | | 批量导出 + 语音转录 | `python export_all_chats.py --with-transcriptions` | | 转录单个文件语音 | `python transcribe_chat.py input.json [output.json]` | | 注册 MCP Server（Claude） | `claude mcp add wechat -- python /path/to/mcp_server.py` | +批量导出会在输出目录自动维护 `_export_index.json`，用稳定的 `username` +追踪当前 JSON 文件。再次导出时如果联系人备注或群名变化，会先把旧文件 +重命名为新的可读文件名；如果同名文件属于另一个 `username`，会追加 +`__` 后缀避免覆盖。 + +导出计划 CSV 支持两种模式：默认 `blacklist` 模式下只有 `export=0` +的行会被跳过，空值、`1` 或没有 `export` 列都会导出；`whitelist` +模式下只有明确 `export=1` 的行会导出。 + ### Web UI `python main.py` 启动后打开 http://localhost:5678 查看实时消息流。 @@ -385,7 +398,7 @@ make help # 列出所有命令 | 文件 | 说明 | |---|---| -| `export_all_chats.py` | 批量导出全部聊天为 JSON (含 `-t` 转录 / `-i` 增量 / 日期范围 / `--dry-run`) | +| `export_all_chats.py` | 批量导出全部聊天为 JSON (含 CSV 计划选择、`-t` 转录、`-i` 增量、日期范围、`--dry-run`) | | `export_chat.py` | 单会话 JSON 导出 (供 `export_all_chats` 调用) | | `chat_export_helpers.py` | JSON 导出共享格式化函数 (避免漂移) | | `export_messages.py` | CSV / HTML / JSON 三种格式导出, 图片可内联 (PR #107) | diff --git a/decrypt_db.py b/decrypt_db.py index 308acbb..876199b 100644 --- a/decrypt_db.py +++ b/decrypt_db.py @@ -106,7 +106,7 @@ def decrypt_database(db_path, out_path, enc_key): return True -def main(): +def main(argv=None): parser = argparse.ArgumentParser( description="WeChat 4.0 数据库解密器" ) @@ -120,7 +120,7 @@ def main(): action="store_true", help="预览模式：显示将要解密的数据库列表", ) - args = parser.parse_args() + args = parser.parse_args(argv) print("=" * 60) print(" WeChat 4.0 数据库解密器") diff --git a/docs/chat_export_format.md b/docs/chat_export_format.md index 03a30f5..f4b5403 100644 --- a/docs/chat_export_format.md +++ b/docs/chat_export_format.md @@ -14,6 +14,12 @@ 为语音消息填充转录文本。`transcribe_chat.py` 可重复运行 —— 已转录的 消息会被跳过。 +`export_all_chats.py` 批量导出时会在输出目录维护 `_export_index.json`。 +该索引用稳定的 `username` 记录当前 JSON 文件名；联系人备注或群名变化后， +下一次导出会先把旧文件重命名为新的可读文件名，再继续增量合并。若新的 +显示名文件已经属于另一个 `username`，导出文件会追加 `__` 后缀， +避免覆盖同名联系人或同名群。 + ## 顶层结构 ```json @@ -21,6 +27,12 @@ "chat": "", "username": "", "exported_at": "YYYY-MM-DD HH:MM:SS", + "date_first_msg": "YYYY-MM-DD HH:MM:SS", + "date_last_msg": "YYYY-MM-DD HH:MM:SS", + "contact_remark": "", + "contact_nick_name": "", + "contact_tags": [""], + "contact_memo": "", "is_group": true, "messages": [ ... ] } @@ -30,6 +42,9 @@ - `username` —— 稳定的 WeChat 用户名（1-on-1 聊天为 `wxid_*`，群聊为 `*@chatroom`）。 `transcribe_chat.py` 会优先读取本字段而非基于 `chat` 再次模糊匹配，避免同名联系人漂移。 - `exported_at` —— 本地时间字符串，仅作溯源用途。 +- `date_first_msg` / `date_last_msg` —— 本次导出结果中第一条 / 最后一条消息的本地时间。 +- `contact_remark`、`contact_nick_name`、`contact_tags`、`contact_memo` —— + 单聊联系人 metadata；群聊中省略。 - `is_group` —— **仅**群聊出现且为 `true`；1-on-1 聊天时省略。 - `messages` —— 消息数组，跨所有 DB 分片按时间由旧到新排序。 diff --git a/export_all_chats.py b/export_all_chats.py index 866ac19..d99460b 100644 --- a/export_all_chats.py +++ b/export_all_chats.py @@ -10,6 +10,8 @@ transcription_backend 为 whisper_cpp / openai / local）。未启用 backend 用法: python3 export_all_chats.py # 全量导出所有会话 + python3 export_all_chats.py --write-plan-csv export_plan.csv + python3 export_all_chats.py output_dir --from-plan-csv export_plan.csv python3 export_all_chats.py --with-transcriptions # 全量导出 + 转录语音 python3 export_all_chats.py -i # 增量（只导出最新消息） python3 export_all_chats.py --start 2025-01-01 # 按日期范围 @@ -18,6 +20,8 @@ transcription_backend 为 whisper_cpp / openai / local）。未启用 backend """ import argparse +import csv +import hashlib import json import os import re @@ -38,6 +42,28 @@ except ImportError: from chat_export_helpers import _extract_content, _msg_type_str, _resolve_sender +PLAN_CSV_FIELDS = [ + "export", + "index", + "username", + "chat_name", + "chat_type", + "message_count", + "first_time", + "last_time", + "attachment_estimated_bytes", + "attachment_scanned_bytes", + "total_estimated_bytes", + "size_status", +] + +EXPORT_INDEX_FILE = "_export_index.json" +EXPORT_INDEX_VERSION = 1 +PLAN_MODE_BLACKLIST = "blacklist" +PLAN_MODE_WHITELIST = "whitelist" +_UNSAFE_FILENAME_RE = re.compile(r'[\\/:*?"<>|]') + + def _parse_timestamp(ts_str): """解析时间字符串返回 unix timestamp。 支持格式: '2025-01-01', '2025-01-01 14:30', '2025-01-01T14:30:00' @@ -77,6 +103,912 @@ def _get_existing_messages(json_path): return [] +def _export_index_path(output_dir): + return os.path.join(output_dir, EXPORT_INDEX_FILE) + + +def _empty_export_index(): + return {"version": EXPORT_INDEX_VERSION, "chats": {}} + + +def _safe_export_filename_part(value): + cleaned = _UNSAFE_FILENAME_RE.sub("_", str(value or "")).strip() + return cleaned or "unknown" + + +def _export_filename(display_name, is_group, username=None): + prefix = "group" if is_group else "single" + label = display_name or username or "unknown" + return f"{_safe_export_filename_part(f'{prefix}_{label}')}.json" + + +def _collision_export_filename(filename, username, suffix=None): + stem, ext = os.path.splitext(filename) + user_part = _safe_export_filename_part(username) + extra = f"__{suffix}" if suffix else "" + return f"{stem}__{user_part}{extra}{ext or '.json'}" + + +def _safe_index_filename(filename): + filename = str(filename or "") + if not filename: + return "" + if filename != os.path.basename(filename): + return "" + if filename == EXPORT_INDEX_FILE: + return "" + return filename + + +def _read_json_string_field(prefix, field): + pattern = rf'"{re.escape(field)}"\s*:\s*("(?:(?:\\.)|[^"\\])*")' + match = re.search(pattern, prefix) + if not match: + return "" + try: + value = json.loads(match.group(1)) + except json.JSONDecodeError: + return "" + return value if isinstance(value, str) else "" + + +def _read_export_file_identity(path): + """只读取 JSON 文件头部，避免为建索引加载巨大 messages 数组。""" + try: + with open(path, encoding="utf-8") as f: + prefix = f.read(256 * 1024) + except OSError: + return {} + + username = _read_json_string_field(prefix, "username") + if not username: + return {} + return { + "username": username, + "chat": _read_json_string_field(prefix, "chat"), + "is_group": bool(re.search(r'"is_group"\s*:\s*true', prefix)), + "exported_at": _read_json_string_field(prefix, "exported_at"), + "date_first_msg": _read_json_string_field(prefix, "date_first_msg"), + "date_last_msg": _read_json_string_field(prefix, "date_last_msg"), + } + + +def _file_mtime(path): + try: + return os.path.getmtime(path) + except OSError: + return -1 + + +def _index_entry_from_identity(filename, identity): + return { + "username": identity["username"], + "is_group": bool(identity.get("is_group")), + "current_chat_name": identity.get("chat", ""), + "current_file": filename, + "previous_files": [], + "last_exported_at": identity.get("exported_at", ""), + "date_first_msg": identity.get("date_first_msg", ""), + "date_last_msg": identity.get("date_last_msg", ""), + } + + +def _bootstrap_export_index(output_dir): + index = _empty_export_index() + if not os.path.isdir(output_dir): + return index + + for filename in os.listdir(output_dir): + safe_filename = _safe_index_filename(filename) + if not safe_filename or not safe_filename.lower().endswith(".json"): + continue + path = os.path.join(output_dir, safe_filename) + if not os.path.isfile(path): + continue + identity = _read_export_file_identity(path) + username = identity.get("username") + if not username: + continue + + chats = index["chats"] + entry = chats.get(username) + if entry is None: + chats[username] = _index_entry_from_identity(safe_filename, identity) + continue + + previous = set(entry.get("previous_files") or []) + current_file = entry.get("current_file") + current_path = os.path.join(output_dir, current_file) + if _file_mtime(path) >= _file_mtime(current_path): + if current_file: + previous.add(current_file) + entry.update(_index_entry_from_identity(safe_filename, identity)) + else: + previous.add(safe_filename) + entry["previous_files"] = sorted( + f for f in previous + if _safe_index_filename(f) and f != entry.get("current_file") + ) + + return index + + +def _normalize_export_index(data): + if not isinstance(data, dict) or not isinstance(data.get("chats"), dict): + return None + + index = _empty_export_index() + for username, entry in data["chats"].items(): + if not username or not isinstance(entry, dict): + continue + current_file = _safe_index_filename(entry.get("current_file")) + if not current_file: + continue + previous = [] + for item in entry.get("previous_files") or []: + filename = _safe_index_filename(item) + if filename and filename != current_file and filename not in previous: + previous.append(filename) + index["chats"][str(username)] = { + "username": str(username), + "is_group": bool(entry.get("is_group")), + "current_chat_name": entry.get("current_chat_name", ""), + "current_file": current_file, + "previous_files": previous, + "last_exported_at": entry.get("last_exported_at", ""), + "date_first_msg": entry.get("date_first_msg", ""), + "date_last_msg": entry.get("date_last_msg", ""), + } + return index + + +def _load_export_index(output_dir): + path = _export_index_path(output_dir) + if not os.path.isfile(path): + return _bootstrap_export_index(output_dir) + try: + with open(path, encoding="utf-8") as f: + data = json.load(f) + except (OSError, json.JSONDecodeError): + return _bootstrap_export_index(output_dir) + return _normalize_export_index(data) or _bootstrap_export_index(output_dir) + + +def _write_export_index(output_dir, index): + os.makedirs(output_dir or ".", exist_ok=True) + path = _export_index_path(output_dir) + tmp_path = f"{path}.tmp" + with open(tmp_path, "w", encoding="utf-8") as f: + json.dump(index, f, ensure_ascii=False, indent=2) + os.replace(tmp_path, path) + + +def _choose_export_filename(output_dir, desired_filename, username): + desired_path = os.path.join(output_dir, desired_filename) + if not os.path.exists(desired_path): + return desired_filename + if _read_export_file_identity(desired_path).get("username") == username: + return desired_filename + + for counter in range(1, 1000): + suffix = None if counter == 1 else str(counter) + candidate = _collision_export_filename(desired_filename, username, suffix) + candidate_path = os.path.join(output_dir, candidate) + if not os.path.exists(candidate_path): + return candidate + if _read_export_file_identity(candidate_path).get("username") == username: + return candidate + raise RuntimeError(f"无法为 {username} 生成不冲突的导出文件名") + + +def _resolve_indexed_export_path(output_dir, username, display_name, is_group): + os.makedirs(output_dir or ".", exist_ok=True) + index = _load_export_index(output_dir) + chats = index.setdefault("chats", {}) + desired_filename = _export_filename(display_name, is_group, username) + entry = chats.get(username) + previous = set(entry.get("previous_files") or []) if entry else set() + + current_file = _safe_index_filename(entry.get("current_file")) if entry else "" + if current_file: + current_path = os.path.join(output_dir, current_file) + if os.path.isfile(current_path): + target_file = _choose_export_filename( + output_dir, desired_filename, username + ) + if target_file != current_file: + target_path = os.path.join(output_dir, target_file) + if not os.path.exists(target_path): + os.replace(current_path, target_path) + previous.add(current_file) + elif _read_export_file_identity(target_path).get("username") == username: + previous.add(current_file) + current_file = target_file + else: + current_file = "" + + if not current_file: + current_file = _choose_export_filename(output_dir, desired_filename, username) + + previous = { + f for f in previous + if _safe_index_filename(f) and f != current_file + } + chats[username] = { + "username": username, + "is_group": bool(is_group), + "current_chat_name": display_name, + "current_file": current_file, + "previous_files": sorted(previous), + "last_exported_at": (entry or {}).get("last_exported_at", ""), + "date_first_msg": (entry or {}).get("date_first_msg", ""), + "date_last_msg": (entry or {}).get("date_last_msg", ""), + } + return os.path.join(output_dir, current_file), index + + +def _update_export_index(output_dir, index, username, display_name, is_group, + out_path, output): + filename = os.path.basename(out_path) + chats = index.setdefault("chats", {}) + entry = chats.get(username, {}) + previous = [] + for item in entry.get("previous_files") or []: + safe = _safe_index_filename(item) + if safe and safe != filename and safe not in previous: + previous.append(safe) + + chats[username] = { + "username": username, + "is_group": bool(is_group), + "current_chat_name": display_name, + "current_file": filename, + "previous_files": previous, + "last_exported_at": output.get("exported_at", ""), + "date_first_msg": output.get("date_first_msg", ""), + "date_last_msg": output.get("date_last_msg", ""), + } + _write_export_index(output_dir, index) + + +def _load_session_usernames(session_db): + """读取 SessionTable 中的会话 username，保持数据库原始顺序。""" + with closing(sqlite3.connect(session_db)) as conn: + return [ + u for u, _ in conn.execute( + "SELECT username, type FROM SessionTable" + ) + ] + + +def _build_chat_rows(sessions, names, contact_full=None): + """构建可展示、可选择的会话行。""" + contact_meta = { + item.get("username"): item + for item in (contact_full or []) + if item.get("username") + } + rows = [] + for index, username in enumerate(sessions, 1): + display_name = names.get(username, username) + kind = "group" if str(username).endswith("@chatroom") else "single" + meta = contact_meta.get(username, {}) + rows.append({ + "index": index, + "username": username, + "display_name": display_name, + "kind": kind, + "remark": meta.get("remark", ""), + "nick_name": meta.get("nick_name", ""), + }) + return rows + + +def _format_plan_time(ts): + if not ts: + return "" + return datetime.fromtimestamp(int(ts)).strftime("%Y-%m-%d %H:%M:%S") + + +def _date_from_message_ts(ts): + if not ts: + return "" + return datetime.fromtimestamp(int(ts)).strftime("%Y-%m-%d %H:%M:%S") + + +def _contact_metadata_for_export(username, is_group=False): + if is_group: + return {} + contact = {} + for item in mcp_server.get_contact_full(): + if item.get("username") == username: + contact = item + break + try: + tag_map = mcp_server.get_contact_tag_names_by_username() + except Exception: + tag_map = {} + return { + "contact_remark": contact.get("remark", ""), + "contact_nick_name": contact.get("nick_name", ""), + "contact_tags": tag_map.get(username, []), + "contact_memo": contact.get("description", ""), + } + + +def _where_for_time_range(start_ts=None, end_ts=None, column="create_time"): + clauses = [] + params = [] + if start_ts is not None: + clauses.append(f"{column} >= ?") + params.append(start_ts) + if end_ts is not None: + clauses.append(f"{column} <= ?") + params.append(end_ts) + where_sql = "WHERE " + " AND ".join(clauses) if clauses else "" + return where_sql, params + + +def _query_message_table_plan_stats(db_path, table_name, start_ts=None, end_ts=None): + if not mcp_server._is_safe_msg_table_name(table_name): + raise ValueError(f"非法消息表名: {table_name}") + where_sql, params = _where_for_time_range(start_ts, end_ts) + sql = f""" + SELECT COUNT(*), MIN(create_time), MAX(create_time), + COALESCE(SUM( + COALESCE(length(message_content), 0) + + COALESCE(length(compress_content), 0) + + COALESCE(length(packed_info_data), 0) + ), 0) + FROM [{table_name}] + {where_sql} + """ + with closing(sqlite3.connect(db_path)) as conn: + return conn.execute(sql, params).fetchone() + + +def _get_message_resource_db_path(): + try: + path = mcp_server._cache.get("message/message_resource.db") + except Exception: + path = None + candidates = [ + path, + os.path.join(mcp_server.DECRYPTED_DIR, "message", "message_resource.db"), + os.path.join( + mcp_server.DECRYPTED_DIR, + "_monitor_cache", + "message_message_resource.db", + ), + ] + for candidate in candidates: + if candidate and os.path.exists(candidate): + return candidate + return None + + +def _query_resource_estimated_bytes(username, start_ts=None, end_ts=None): + resource_db = _get_message_resource_db_path() + if not resource_db: + return 0, "resource_missing" + with closing(sqlite3.connect(resource_db)) as conn: + chat_row = conn.execute( + "SELECT rowid FROM ChatName2Id WHERE user_name = ?", + (username,), + ).fetchone() + if not chat_row: + return 0, None + clauses = ["i.chat_id = ?"] + params = [chat_row[0]] + if start_ts is not None: + clauses.append("i.message_create_time >= ?") + params.append(start_ts) + if end_ts is not None: + clauses.append("i.message_create_time <= ?") + params.append(end_ts) + where_sql = " AND ".join(clauses) + row = conn.execute(f""" + SELECT COALESCE(SUM(COALESCE(d.size, 0)), 0) + FROM MessageResourceInfo i + LEFT JOIN MessageResourceDetail d ON d.message_id = i.message_id + WHERE {where_sql} + """, params).fetchone() + return int(row[0] or 0), None + + +def _query_voice_estimated_bytes(username, start_ts=None, end_ts=None): + try: + media_paths = list(mcp_server._iter_media_db_paths()) + except Exception: + return 0, "media_error" + if not media_paths: + return 0, "media_missing" + + total = 0 + for media_db in media_paths: + try: + with closing(sqlite3.connect(media_db)) as conn: + chat_name_id = mcp_server._get_chat_name_id(conn, username) + if chat_name_id is None: + continue + clauses = ["chat_name_id = ?"] + params = [chat_name_id] + if start_ts is not None: + clauses.append("create_time >= ?") + params.append(start_ts) + if end_ts is not None: + clauses.append("create_time <= ?") + params.append(end_ts) + row = conn.execute(f""" + SELECT COALESCE(SUM(COALESCE(length(voice_data), 0)), 0) + FROM VoiceInfo + WHERE {" AND ".join(clauses)} + """, params).fetchone() + total += int(row[0] or 0) + except sqlite3.Error: + return total, "media_error" + return total, None + + +def _scan_dir_bytes(path): + total = 0 + if not os.path.isdir(path): + return 0 + for root, _, files in os.walk(path): + for name in files: + full = os.path.join(root, name) + try: + total += os.path.getsize(full) + except OSError: + pass + return total + + +def _scan_local_attachment_bytes(username): + base = getattr(mcp_server, "WECHAT_BASE_DIR", "") + if not base: + return 0, "scan_base_missing" + + username_hash = hashlib.md5(username.encode()).hexdigest() + msg_dir = os.path.join(base, "msg") + roots = [ + os.path.join(msg_dir, "attach", username_hash), + os.path.join(msg_dir, "file", username_hash), + os.path.join(msg_dir, "video", username_hash), + ] + total = sum(_scan_dir_bytes(path) for path in roots) + + global_roots = [ + os.path.join(msg_dir, "file"), + os.path.join(msg_dir, "video"), + ] + has_unattributed_global_dirs = any( + os.path.isdir(path) and not os.path.isdir(os.path.join(path, username_hash)) + for path in global_roots + ) + if has_unattributed_global_dirs: + return total, "scan_limited" + return total, None + + +def _collect_chat_plan_stats(username, message_tables, start_ts=None, end_ts=None, + size_mode="estimate"): + status = [] + message_count = 0 + message_body_bytes = 0 + first_ts = None + last_ts = None + + if not message_tables: + status.append("no_message_table") + + for table in message_tables: + try: + count, min_ts, max_ts, body_bytes = _query_message_table_plan_stats( + table["db_path"], + table["table_name"], + start_ts=start_ts, + end_ts=end_ts, + ) + except (sqlite3.Error, ValueError): + status.append("message_error") + continue + message_count += int(count or 0) + message_body_bytes += int(body_bytes or 0) + if min_ts: + first_ts = min_ts if first_ts is None else min(first_ts, min_ts) + if max_ts: + last_ts = max_ts if last_ts is None else max(last_ts, max_ts) + + try: + resource_bytes, resource_status = _query_resource_estimated_bytes( + username, start_ts=start_ts, end_ts=end_ts + ) + except sqlite3.Error: + resource_bytes, resource_status = 0, "resource_error" + if resource_status: + status.append(resource_status) + + voice_bytes, voice_status = _query_voice_estimated_bytes( + username, start_ts=start_ts, end_ts=end_ts + ) + if voice_status: + status.append(voice_status) + + scanned_bytes = "" + if size_mode == "scan": + scanned_bytes, scan_status = _scan_local_attachment_bytes(username) + if scan_status: + status.append(scan_status) + + attachment_estimated = int(resource_bytes or 0) + int(voice_bytes or 0) + size_status = "ok" if not status else "partial:" + ",".join(sorted(set(status))) + return { + "message_count": message_count, + "message_body_bytes": message_body_bytes, + "first_time": _format_plan_time(first_ts), + "last_time": _format_plan_time(last_ts), + "attachment_estimated_bytes": attachment_estimated, + "attachment_scanned_bytes": scanned_bytes, + "total_estimated_bytes": message_body_bytes + attachment_estimated, + "size_status": size_status, + } + + +def _new_plan_accumulator(): + return { + "message_count": 0, + "message_body_bytes": 0, + "first_ts": None, + "last_ts": None, + "attachment_estimated_bytes": 0, + "attachment_scanned_bytes": "", + "statuses": set(), + } + + +def _message_table_name_for_username(username): + table_hash = hashlib.md5(username.encode()).hexdigest() + return f"Msg_{table_hash}" + + +def _iter_message_db_paths(): + for rel_key in getattr(mcp_server, "MSG_DB_KEYS", []): + try: + path = mcp_server._cache.get(rel_key) + except Exception: + path = None + if path: + yield path + + +def _fetch_existing_message_tables(conn, table_names): + if not table_names: + return set() + existing = set() + batch_size = 500 + table_names = list(table_names) + for i in range(0, len(table_names), batch_size): + batch = table_names[i:i + batch_size] + placeholders = ",".join("?" for _ in batch) + rows = conn.execute( + "SELECT name FROM sqlite_master " + f"WHERE type='table' AND name IN ({placeholders})", + batch, + ).fetchall() + existing.update(row[0] for row in rows) + return existing + + +def _query_message_table_plan_stats_conn(conn, table_name, start_ts=None, end_ts=None): + if not mcp_server._is_safe_msg_table_name(table_name): + raise ValueError(f"非法消息表名: {table_name}") + where_sql, params = _where_for_time_range(start_ts, end_ts) + sql = f""" + SELECT COUNT(*), MIN(create_time), MAX(create_time), + COALESCE(SUM( + COALESCE(length(message_content), 0) + + COALESCE(length(compress_content), 0) + + COALESCE(length(packed_info_data), 0) + ), 0) + FROM [{table_name}] + {where_sql} + """ + return conn.execute(sql, params).fetchone() + + +def _collect_message_stats_batch(usernames, start_ts=None, end_ts=None): + table_to_username = { + _message_table_name_for_username(username): username + for username in usernames + } + stats = {username: _new_plan_accumulator() for username in usernames} + found = set() + db_paths = list(_iter_message_db_paths()) + if not db_paths: + for username in usernames: + stats[username]["statuses"].add("message_db_missing") + return stats + + for db_path in db_paths: + try: + with closing(sqlite3.connect(db_path)) as conn: + existing = _fetch_existing_message_tables(conn, table_to_username) + for table_name in existing: + username = table_to_username[table_name] + try: + count, min_ts, max_ts, body_bytes = ( + _query_message_table_plan_stats_conn( + conn, + table_name, + start_ts=start_ts, + end_ts=end_ts, + ) + ) + except (sqlite3.Error, ValueError): + stats[username]["statuses"].add("message_error") + continue + found.add(username) + stats[username]["message_count"] += int(count or 0) + stats[username]["message_body_bytes"] += int(body_bytes or 0) + if min_ts: + current = stats[username]["first_ts"] + stats[username]["first_ts"] = ( + min_ts if current is None else min(current, min_ts) + ) + if max_ts: + current = stats[username]["last_ts"] + stats[username]["last_ts"] = ( + max_ts if current is None else max(current, max_ts) + ) + except sqlite3.Error: + for username in usernames: + stats[username]["statuses"].add("message_error") + + for username in usernames: + if username not in found: + stats[username]["statuses"].add("no_message_table") + return stats + + +def _collect_resource_estimates_batch(usernames, start_ts=None, end_ts=None): + values = {username: 0 for username in usernames} + statuses = {username: set() for username in usernames} + resource_db = _get_message_resource_db_path() + if not resource_db: + for username in usernames: + statuses[username].add("resource_missing") + return values, statuses + + try: + with closing(sqlite3.connect(resource_db)) as conn: + usernames = list(usernames) + batch_size = 500 + for i in range(0, len(usernames), batch_size): + batch = usernames[i:i + batch_size] + placeholders = ",".join("?" for _ in batch) + params = list(batch) + clauses = [f"c.user_name IN ({placeholders})"] + if start_ts is not None: + clauses.append("i.message_create_time >= ?") + params.append(start_ts) + if end_ts is not None: + clauses.append("i.message_create_time <= ?") + params.append(end_ts) + where_sql = " AND ".join(clauses) + rows = conn.execute(f""" + SELECT c.user_name, COALESCE(SUM(COALESCE(d.size, 0)), 0) + FROM ChatName2Id c + JOIN MessageResourceInfo i ON i.chat_id = c.rowid + LEFT JOIN MessageResourceDetail d ON d.message_id = i.message_id + WHERE {where_sql} + GROUP BY c.user_name + """, params).fetchall() + for username, size in rows: + values[username] = int(size or 0) + except sqlite3.Error: + for username in usernames: + statuses[username].add("resource_error") + return values, statuses + + +def _collect_voice_estimates_batch(usernames, start_ts=None, end_ts=None): + values = {username: 0 for username in usernames} + statuses = {username: set() for username in usernames} + try: + media_paths = list(mcp_server._iter_media_db_paths()) + except Exception: + for username in usernames: + statuses[username].add("media_error") + return values, statuses + + if not media_paths: + for username in usernames: + statuses[username].add("media_missing") + return values, statuses + + usernames = list(usernames) + try: + for media_db in media_paths: + with closing(sqlite3.connect(media_db)) as conn: + batch_size = 500 + for i in range(0, len(usernames), batch_size): + batch = usernames[i:i + batch_size] + placeholders = ",".join("?" for _ in batch) + params = list(batch) + clauses = [f"n.user_name IN ({placeholders})"] + if start_ts is not None: + clauses.append("v.create_time >= ?") + params.append(start_ts) + if end_ts is not None: + clauses.append("v.create_time <= ?") + params.append(end_ts) + where_sql = " AND ".join(clauses) + rows = conn.execute(f""" + SELECT n.user_name, + COALESCE(SUM(COALESCE(length(v.voice_data), 0)), 0) + FROM Name2Id n + JOIN VoiceInfo v ON v.chat_name_id = n.rowid + WHERE {where_sql} + GROUP BY n.user_name + """, params).fetchall() + for username, size in rows: + values[username] += int(size or 0) + except sqlite3.Error: + for username in usernames: + statuses[username].add("media_error") + return values, statuses + + +def _finalize_plan_stats(acc): + status = sorted(acc["statuses"]) + size_status = "ok" if not status else "partial:" + ",".join(status) + return { + "message_count": acc["message_count"], + "message_body_bytes": acc["message_body_bytes"], + "first_time": _format_plan_time(acc["first_ts"]), + "last_time": _format_plan_time(acc["last_ts"]), + "attachment_estimated_bytes": acc["attachment_estimated_bytes"], + "attachment_scanned_bytes": acc["attachment_scanned_bytes"], + "total_estimated_bytes": ( + acc["message_body_bytes"] + acc["attachment_estimated_bytes"] + ), + "size_status": size_status, + } + + +def _collect_all_plan_stats(chat_rows, start_ts=None, end_ts=None, + size_mode="estimate"): + usernames = [row["username"] for row in chat_rows] + print("[*] 批量统计消息表...", flush=True) + stats = _collect_message_stats_batch( + usernames, + start_ts=start_ts, + end_ts=end_ts, + ) + + print("[*] 批量统计资源附件...", flush=True) + resource_values, resource_statuses = _collect_resource_estimates_batch( + usernames, + start_ts=start_ts, + end_ts=end_ts, + ) + for username in usernames: + stats[username]["attachment_estimated_bytes"] += resource_values[username] + stats[username]["statuses"].update(resource_statuses[username]) + + print("[*] 批量统计语音数据...", flush=True) + voice_values, voice_statuses = _collect_voice_estimates_batch( + usernames, + start_ts=start_ts, + end_ts=end_ts, + ) + for username in usernames: + stats[username]["attachment_estimated_bytes"] += voice_values[username] + stats[username]["statuses"].update(voice_statuses[username]) + + if size_mode == "scan": + print("[*] 扫描本地附件目录...", flush=True) + iterable = _tqdm(usernames, desc="扫描附件") if _tqdm else usernames + for username in iterable: + scanned_bytes, scan_status = _scan_local_attachment_bytes(username) + stats[username]["attachment_scanned_bytes"] = scanned_bytes + if scan_status: + stats[username]["statuses"].add(scan_status) + + return { + username: _finalize_plan_stats(stats[username]) + for username in usernames + } + + +def _build_plan_csv_rows(chat_rows, start_ts=None, end_ts=None, size_mode="estimate"): + stats_by_username = _collect_all_plan_stats( + chat_rows, + start_ts=start_ts, + end_ts=end_ts, + size_mode=size_mode, + ) + rows = [] + iterable = _tqdm(chat_rows, desc="统计会话") if _tqdm else chat_rows + for row in iterable: + username = row["username"] + stats = stats_by_username[username] + rows.append({ + "export": "0", + "index": row["index"], + "username": username, + "chat_name": row["display_name"], + "chat_type": row["kind"], + "message_count": stats["message_count"], + "first_time": stats["first_time"], + "last_time": stats["last_time"], + "attachment_estimated_bytes": stats["attachment_estimated_bytes"], + "attachment_scanned_bytes": stats["attachment_scanned_bytes"], + "total_estimated_bytes": stats["total_estimated_bytes"], + "size_status": stats["size_status"], + }) + return rows + + +def _validate_plan_mode(plan_mode): + if plan_mode not in (PLAN_MODE_BLACKLIST, PLAN_MODE_WHITELIST): + raise ValueError(f"未知导出计划模式: {plan_mode}") + return plan_mode + + +def _write_plan_csv(path, rows, plan_mode=PLAN_MODE_BLACKLIST): + plan_mode = _validate_plan_mode(plan_mode) + out_dir = os.path.dirname(os.path.abspath(path)) + if out_dir: + os.makedirs(out_dir, exist_ok=True) + with open(path, "w", newline="", encoding="utf-8-sig") as f: + writer = csv.DictWriter(f, fieldnames=PLAN_CSV_FIELDS, extrasaction="ignore") + writer.writeheader() + for row in rows: + full = {field: "" for field in PLAN_CSV_FIELDS} + full.update(row) + if full["export"] == "": + full["export"] = ( + "0" if plan_mode == PLAN_MODE_WHITELIST else "1" + ) + writer.writerow(full) + + +def _load_selected_usernames_from_plan_csv( + path, valid_usernames, plan_mode=PLAN_MODE_BLACKLIST +): + plan_mode = _validate_plan_mode(plan_mode) + selected = [] + seen = {} + with open(path, newline="", encoding="utf-8-sig") as f: + reader = csv.DictReader(f) + if not reader.fieldnames or "username" not in reader.fieldnames: + raise ValueError("CSV 缺少 username 列") + for line_no, row in enumerate(reader, 2): + username = (row.get("username") or "").strip() + if not username: + raise ValueError(f"第 {line_no} 行缺少 username") + if username in seen: + raise ValueError( + f"CSV 中 username 重复: {username} " + f"(第 {seen[username]} 行和第 {line_no} 行)" + ) + seen[username] = line_no + + flag = (row.get("export") or "").strip() + if plan_mode == PLAN_MODE_WHITELIST: + should_export = flag == "1" + else: + should_export = flag != "0" + if not should_export: + continue + if username not in valid_usernames: + raise ValueError(f"第 {line_no} 行 username 当前不存在: {username}") + selected.append(username) + return selected + + def export_one(username, output_dir, names, transcribe=False, start_ts=None, end_ts=None, incremental=False): """ @@ -99,10 +1031,13 @@ def export_one(username, output_dir, names, transcribe=False, if not message_tables: return False, 0, 0, "no tables" - # 构造输出路径 - 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") + # 输出文件名可随备注变化；索引用 username 维持稳定匹配。 + try: + out_path, export_index = _resolve_indexed_export_path( + output_dir, username, display_name, ctx["is_group"] + ) + except Exception as e: + return False, 0, 0, f"export index error: {e}" # 增量模式：读取已有消息和最后时间戳 existing_msgs = [] @@ -226,14 +1161,22 @@ def export_one(username, output_dir, names, transcribe=False, "chat": display_name, "username": username, "exported_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), - "messages": messages, + "date_first_msg": _date_from_message_ts(messages[0].get("timestamp")), + "date_last_msg": _date_from_message_ts(messages[-1].get("timestamp")), } if ctx["is_group"]: output["is_group"] = True + else: + output.update(_contact_metadata_for_export(username, ctx["is_group"])) + output["messages"] = messages 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) + _update_export_index( + output_dir, export_index, username, display_name, ctx["is_group"], + out_path, output + ) return True, len(messages), new_count, None @@ -252,13 +1195,18 @@ def _resolve_backend(): return _BACKEND_CACHE -def main(): +def main(argv=None): parser = argparse.ArgumentParser( description="批量导出所有微信聊天记录为 JSON 文件，可选附带语音转录", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" 示例: python3 export_all_chats.py 全量导出所有会话 + python3 export_all_chats.py --write-plan-csv export_plan.csv + python3 export_all_chats.py --write-plan-csv export_plan.csv --plan-mode whitelist + python3 export_all_chats.py --write-plan-csv export_plan.csv --size-mode scan + python3 export_all_chats.py output_dir --from-plan-csv export_plan.csv + python3 export_all_chats.py output_dir --from-plan-csv export_plan.csv --plan-mode whitelist python3 export_all_chats.py -t 全量导出 + 转录语音 python3 export_all_chats.py -i 增量（追加新消息） python3 export_all_chats.py --start 2025-01-01 按日期范围导出 @@ -278,6 +1226,31 @@ def main(): action="store_true", help="导出时一并转录语音消息（依赖 config.json 配置的 backend）", ) + parser.add_argument( + "--write-plan-csv", + default=None, + help="生成可人工编辑的导出计划 CSV，不导出聊天", + ) + parser.add_argument( + "--from-plan-csv", + default=None, + help="读取导出计划 CSV，按 --plan-mode 判断哪些 username 需要导出", + ) + parser.add_argument( + "--plan-mode", + choices=(PLAN_MODE_BLACKLIST, PLAN_MODE_WHITELIST), + default=PLAN_MODE_BLACKLIST, + help=( + "导出计划模式：blacklist=只有 export=0 跳过；" + "whitelist=只有 export=1 导出" + ), + ) + parser.add_argument( + "--size-mode", + choices=("estimate", "scan"), + default="estimate", + help="生成计划 CSV 时的大小统计方式：estimate=快估，scan=尝试扫描本地附件", + ) parser.add_argument( "-i", "--incremental", @@ -297,7 +1270,7 @@ def main(): parser.add_argument( "--dry-run", action="store_true", - help="预览模式：显示将导出的会话数和新消息数，不实际写入", + help="预览模式：显示将导出的会话，不实际写入", ) parser.add_argument( "--users", @@ -305,7 +1278,10 @@ def main(): help="只导出指定 username 的会话, 逗号分隔 (如 wxid_xxx,12345@chatroom). " "为空时导出全部 (旧行为). 也可用 env WECHAT_EXPORT_USERS", ) - args = parser.parse_args() + args = parser.parse_args(argv) + + if args.write_plan_csv and args.from_plan_csv: + parser.error("--write-plan-csv 和 --from-plan-csv 只能使用一个") script_dir = os.path.dirname(os.path.abspath(__file__)) output_dir = args.output_dir or os.path.join(script_dir, "exported_chats") @@ -332,14 +1308,10 @@ def main(): if not os.path.exists(mcp_server.DECRYPTED_DIR): print(f"错误: 解密目录不存在: {mcp_server.DECRYPTED_DIR}", file=sys.stderr) sys.exit(1) - os.makedirs(output_dir, exist_ok=True) 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 = _load_session_usernames(session_db) except sqlite3.Error as e: print(f"会话数据库查询失败: {e}", file=sys.stderr) sys.exit(1) @@ -357,6 +1329,8 @@ def main(): sys.exit(1) names = mcp_server.get_contact_names() + contact_full = mcp_server.get_contact_full() + chat_rows = _build_chat_rows(sessions, names, contact_full) # 显示模式信息 mode = "" @@ -379,13 +1353,62 @@ def main(): print(f"模式: {mode}") print("=" * 60) + if args.write_plan_csv: + rows = _build_plan_csv_rows( + chat_rows, + start_ts=start_ts, + end_ts=end_ts, + size_mode=args.size_mode, + ) + try: + _write_plan_csv(args.write_plan_csv, rows, plan_mode=args.plan_mode) + except OSError as e: + print( + f"写入导出计划 CSV 失败: {e}\n" + "请确认目标文件没有被 Excel/WPS 打开，或换一个输出文件名。", + file=sys.stderr, + ) + sys.exit(1) + print(f"已生成导出计划 CSV: {args.write_plan_csv}") + if args.plan_mode == PLAN_MODE_WHITELIST: + print("白名单模式：请在 export 列填写 1 后，使用 --from-plan-csv 导出。") + else: + print("黑名单模式：请将不导出的行改为 export=0，再使用 --from-plan-csv 导出。") + return + + if args.from_plan_csv: + try: + sessions = _load_selected_usernames_from_plan_csv( + args.from_plan_csv, + {row["username"] for row in chat_rows}, + plan_mode=args.plan_mode, + ) + except (OSError, ValueError) as e: + print(f"读取导出计划 CSV 失败: {e}", file=sys.stderr) + sys.exit(1) + if not sessions: + print("未选择任何会话，已取消。") + return + + if args.from_plan_csv: + print(f"本次选择: {len(sessions)} 个会话") + print("=" * 60) + + if args.dry_run: + print("预览模式：未写入任何导出文件。") + return + + os.makedirs(output_dir, exist_ok=True) + 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): + total_sessions = len(sessions) + for i, username in enumerate(sessions, 1): display = names.get(username, username) + chat_t0 = time.time() + print(f"[{i}/{total_sessions}] 开始导出: {display} ({username})", flush=True) success, total_msgs, new_msgs, reason = export_one( username, output_dir, names, transcribe=args.with_transcriptions, @@ -401,26 +1424,35 @@ def main(): 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 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} - {label}" - + (f" ETA {eta/60:.0f}分" if i > 1 else "") - ) + elapsed = time.time() - t0 + chat_elapsed = time.time() - chat_t0 + eta = (elapsed / i) * (total_sessions - i) if i > 0 else 0 + print( + f"[{i}/{total_sessions}] 完成导出: {display} - {label} " + f"(本会话 {chat_elapsed:.1f}s, ETA {eta/60:.1f}分)", + flush=True, + ) else: if "no tables" in str(reason) or "empty" in str(reason): skip += 1 - if not _tqdm: - if i <= 10 or i % 50 == 0: - print(f"[{i}/{len(sessions)}] {display} - 跳过({reason})") + elapsed = time.time() - t0 + chat_elapsed = time.time() - chat_t0 + eta = (elapsed / i) * (total_sessions - i) if i > 0 else 0 + print( + f"[{i}/{total_sessions}] 跳过: {display} ({reason}) " + f"(本会话 {chat_elapsed:.1f}s, ETA {eta/60:.1f}分)", + flush=True, + ) else: err += 1 - if not _tqdm: - print(f"[{i}/{len(sessions)}] {display} - 失败: {reason}") - elif _tqdm: - _tqdm.write(f"失败: {display} - {reason}") + elapsed = time.time() - t0 + chat_elapsed = time.time() - chat_t0 + eta = (elapsed / i) * (total_sessions - i) if i > 0 else 0 + print( + f"[{i}/{total_sessions}] 失败: {display} - {reason} " + f"(本会话 {chat_elapsed:.1f}s, ETA {eta/60:.1f}分)", + flush=True, + ) elapsed = time.time() - t0 print() diff --git a/main.py b/main.py index 5163fe7..54d943f 100644 --- a/main.py +++ b/main.py @@ -315,19 +315,20 @@ def main(): print("[*] 开始解密全部数据库...") print() from decrypt_db import main as decrypt_all - _call_with_argv(decrypt_all, ["decrypt_db.py", *sys.argv[2:]]) + decrypt_all(sys.argv[2:]) elif cmd in ("export", "all"): print("[*] 开始解密全部数据库...") print() from decrypt_db import main as decrypt_all - _call_with_argv(decrypt_all, ["decrypt_db.py"]) + decrypt_all([]) print() print("[*] 开始批量导出聊天记录...") print() from export_all_chats import main as export_all try: - _call_with_argv(export_all, ["export_all_chats.py"]) + export_args = sys.argv[2:] if cmd == "export" else [] + export_all(export_args) except SystemExit: pass diff --git a/mcp_server.py b/mcp_server.py index bc360ba..6ddff11 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -222,7 +222,7 @@ atexit.register(_cache.cleanup) # ============ 联系人缓存 ============ _contact_names = None # {username: display_name} -_contact_full = None # [{username, nick_name, remark}] +_contact_full = None # [{username, nick_name, remark, alias, description, phone}] _contact_tags = None # {label_id: {name, sort_order, members: [{username, display_name}]}} _self_username = None _contact_db_mtime = 0 # mtime of the decrypted contact.db when caches were last populated @@ -240,19 +240,55 @@ _QUERY_LIMIT_MAX = 500 _HISTORY_QUERY_BATCH_SIZE = 500 -def _load_contacts_from(db_path): - names = {} - full = [] - conn = sqlite3.connect(db_path) - try: - for r in conn.execute("SELECT username, nick_name, remark FROM contact").fetchall(): - uname, nick, remark = r - display = remark if remark else nick if nick else uname - names[uname] = display - full.append({'username': uname, 'nick_name': nick or '', 'remark': remark or ''}) - finally: - conn.close() - return names, full +def _load_contacts_from(db_path): + names = {} + full = [] + conn = sqlite3.connect(db_path) + try: + columns = { + row[1] for row in conn.execute("PRAGMA table_info(contact)").fetchall() + } + optional_columns = { + "alias": "", + "description": "", + "phone": "", + "phone_number": "", + "mobile": "", + "mobile_phone": "", + "telephone": "", + } + select_columns = ["username", "nick_name", "remark"] + select_columns.extend( + col for col in optional_columns + if col in columns and col not in select_columns + ) + rows = conn.execute( + "SELECT " + ", ".join(f"[{col}]" for col in select_columns) + + " FROM contact" + ).fetchall() + for r in rows: + data = dict(zip(select_columns, r)) + uname = data.get("username") + nick = data.get("nick_name") + remark = data.get("remark") + display = remark if remark else nick if nick else uname + names[uname] = display + phone = "" + for col in ("phone", "phone_number", "mobile", "mobile_phone", "telephone"): + if data.get(col): + phone = data.get(col) or "" + break + full.append({ + 'username': uname, + 'nick_name': nick or '', + 'remark': remark or '', + 'alias': data.get("alias") or '', + 'description': data.get("description") or '', + 'phone': phone, + }) + finally: + conn.close() + return names, full def _get_contact_db_path(): @@ -301,9 +337,23 @@ def get_contact_names(): return {} -def get_contact_full(): - get_contact_names() - return _contact_full or [] +def get_contact_full(): + get_contact_names() + return _contact_full or [] + + +def get_contact_tag_names_by_username(): + tags = _load_contact_tags() + by_username = {} + for tag in tags.values(): + name = tag.get('name') or '' + if not name: + continue + for member in tag.get('members', []): + username = member.get('username') + if username: + by_username.setdefault(username, []).append(name) + return by_username def _extract_pb_field_30(data): diff --git a/tests/test_export_all_chats_selection.py b/tests/test_export_all_chats_selection.py new file mode 100644 index 0000000..1a0acbb --- /dev/null +++ b/tests/test_export_all_chats_selection.py @@ -0,0 +1,717 @@ +import csv +import hashlib +import io +import json +import os +import sqlite3 +import tempfile +import unittest +from contextlib import closing, redirect_stderr, redirect_stdout +from unittest.mock import patch + +import export_all_chats + + +class ChatRowsTests(unittest.TestCase): + def setUp(self): + self.rows = export_all_chats._build_chat_rows( + ["wxid_alice", "12345@chatroom", "wxid_bob"], + { + "wxid_alice": "Alice", + "12345@chatroom": "Project Group", + "wxid_bob": "Bob", + }, + [ + { + "username": "wxid_alice", + "remark": "Alice Remark", + "nick_name": "Alice Nick", + }, + ], + ) + + def test_chat_rows_include_contact_remark_and_nickname(self): + self.assertEqual(self.rows[0]["remark"], "Alice Remark") + self.assertEqual(self.rows[0]["nick_name"], "Alice Nick") + self.assertEqual(self.rows[1]["remark"], "") + self.assertEqual(self.rows[1]["nick_name"], "") + + +class ExportAllChatsCliCsvOnlyTests(unittest.TestCase): + def _run_main(self, argv): + with patch.object(export_all_chats.mcp_server, "DECRYPTED_DIR", "decrypted"), \ + patch.object(export_all_chats.os.path, "exists", return_value=True), \ + patch.object(export_all_chats, "_load_session_usernames", + return_value=["wxid_alice", "12345@chatroom"]), \ + patch.object(export_all_chats.mcp_server, "get_contact_names", + return_value={ + "wxid_alice": "Alice", + "12345@chatroom": "Project Group", + }), \ + patch.object(export_all_chats.mcp_server, "get_contact_full", + return_value=[]), \ + patch.object(export_all_chats, "export_one", + return_value=(True, 1, 1, None)) as export_one: + out = io.StringIO() + with redirect_stdout(out): + export_all_chats.main(argv) + return out.getvalue(), export_one + + def test_direct_selection_args_are_not_supported(self): + for argv in (["--list-chats"], ["--select"], ["--chats", "Alice"]): + with self.subTest(argv=argv), \ + patch.object(export_all_chats.mcp_server, "DECRYPTED_DIR", "decrypted"), \ + patch.object(export_all_chats.os.path, "exists", return_value=True), \ + patch.object(export_all_chats, "_load_session_usernames", + return_value=["wxid_alice", "12345@chatroom"]), \ + patch.object(export_all_chats.mcp_server, "get_contact_names", + return_value={ + "wxid_alice": "Alice", + "12345@chatroom": "Project Group", + }), \ + patch.object(export_all_chats.mcp_server, "get_contact_full", + return_value=[]), \ + patch("builtins.input", return_value=""), \ + redirect_stdout(io.StringIO()), \ + redirect_stderr(io.StringIO()): + with self.assertRaises(SystemExit) as cm: + export_all_chats.main(argv) + self.assertEqual(cm.exception.code, 2) + + def test_dry_run_does_not_export(self): + output, export_one = self._run_main(["--dry-run", "--users", "wxid_alice"]) + + self.assertIn("预览", output) + self.assertIn("会话总数: 1", output) + export_one.assert_not_called() + + +class ExportPlanCsvTests(unittest.TestCase): + def test_writes_utf8_sig_csv_without_contact_remark_or_nickname(self): + row = { + "index": 1, + "username": "wxid_alice", + "chat_name": '张三, "A"', + "chat_type": "single", + "message_count": 3, + "first_time": "2026-05-01 00:00:00", + "last_time": "2026-05-02 00:00:00", + "attachment_estimated_bytes": 12, + "attachment_scanned_bytes": "", + "total_estimated_bytes": 20, + "size_status": "ok", + } + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "plan.csv") + + export_all_chats._write_plan_csv(path, [row]) + + with open(path, "rb") as f: + self.assertTrue(f.read(3).startswith(b"\xef\xbb\xbf")) + with open(path, newline="", encoding="utf-8-sig") as f: + rows = list(csv.DictReader(f)) + self.assertEqual(export_all_chats.PLAN_CSV_FIELDS, list(rows[0].keys())) + self.assertEqual(rows[0]["export"], "1") + self.assertEqual(rows[0]["chat_name"], '张三, "A"') + self.assertNotIn("contact_remark", rows[0]) + self.assertNotIn("contact_nick_name", rows[0]) + + def test_whitelist_plan_csv_defaults_to_export_zero(self): + row = { + "index": 1, + "username": "wxid_alice", + "chat_name": "Alice", + "chat_type": "single", + "message_count": 0, + "first_time": "", + "last_time": "", + "attachment_estimated_bytes": 0, + "attachment_scanned_bytes": "", + "total_estimated_bytes": 0, + "size_status": "ok", + } + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "plan.csv") + + export_all_chats._write_plan_csv( + path, [row], plan_mode="whitelist" + ) + + with open(path, newline="", encoding="utf-8-sig") as f: + rows = list(csv.DictReader(f)) + + self.assertEqual(rows[0]["export"], "0") + + def test_loads_all_rows_except_explicit_export_zero(self): + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "plan.csv") + _write_csv(path, [ + {"export": "1", "username": "wxid_alice", "chat_name": "Alice"}, + {"export": "0", "username": "wxid_bob", "chat_name": "Bob"}, + {"export": "", "username": "12345@chatroom", "chat_name": "Group"}, + {"export": "yes", "username": "wxid_carl", "chat_name": "Carl"}, + ]) + + selected = export_all_chats._load_selected_usernames_from_plan_csv( + path, { + "wxid_alice", "wxid_bob", "12345@chatroom", "wxid_carl" + } + ) + + self.assertEqual(selected, ["wxid_alice", "12345@chatroom", "wxid_carl"]) + + def test_whitelist_plan_mode_loads_only_explicit_export_one(self): + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "plan.csv") + no_export = os.path.join(tmp, "no_export.csv") + _write_csv(path, [ + {"export": "1", "username": "wxid_alice", "chat_name": "Alice"}, + {"export": "0", "username": "wxid_bob", "chat_name": "Bob"}, + {"export": "", "username": "12345@chatroom", "chat_name": "Group"}, + {"export": "yes", "username": "wxid_carl", "chat_name": "Carl"}, + ]) + _write_custom_csv(no_export, ["username", "chat_name"], [ + {"username": "wxid_alice", "chat_name": "Alice"}, + ]) + + selected = export_all_chats._load_selected_usernames_from_plan_csv( + path, + {"wxid_alice", "wxid_bob", "12345@chatroom", "wxid_carl"}, + plan_mode="whitelist", + ) + selected_without_export = ( + export_all_chats._load_selected_usernames_from_plan_csv( + no_export, {"wxid_alice"}, plan_mode="whitelist" + ) + ) + + self.assertEqual(selected, ["wxid_alice"]) + self.assertEqual(selected_without_export, []) + + def test_missing_export_column_exports_all_rows(self): + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "plan.csv") + _write_custom_csv(path, ["username", "chat_name"], [ + {"username": "wxid_alice", "chat_name": "Alice"}, + {"username": "12345@chatroom", "chat_name": "Group"}, + ]) + + selected = export_all_chats._load_selected_usernames_from_plan_csv( + path, {"wxid_alice", "12345@chatroom"} + ) + + self.assertEqual(selected, ["wxid_alice", "12345@chatroom"]) + + def test_rejects_duplicate_and_missing_chat(self): + with tempfile.TemporaryDirectory() as tmp: + duplicate = os.path.join(tmp, "duplicate.csv") + missing = os.path.join(tmp, "missing.csv") + _write_csv(duplicate, [ + {"export": "0", "username": "wxid_alice"}, + {"export": "1", "username": "wxid_alice"}, + ]) + _write_csv(missing, [{"export": "1", "username": "wxid_missing"}]) + + with self.assertRaisesRegex(ValueError, "重复"): + export_all_chats._load_selected_usernames_from_plan_csv( + duplicate, {"wxid_alice"} + ) + with self.assertRaisesRegex(ValueError, "不存在"): + export_all_chats._load_selected_usernames_from_plan_csv( + missing, {"wxid_alice"} + ) + + +class ExportPlanStatsTests(unittest.TestCase): + def test_collects_message_resource_and_voice_estimates_with_date_filter(self): + username = "wxid_alice" + table_name = "Msg_" + hashlib.md5(username.encode()).hexdigest() + with tempfile.TemporaryDirectory() as tmp: + msg_db = os.path.join(tmp, "message_0.db") + resource_db = os.path.join(tmp, "message_resource.db") + media_db = os.path.join(tmp, "media_0.db") + _create_message_db(msg_db, table_name) + _create_resource_db(resource_db, username) + _create_media_db(media_db, username) + + with patch.object(export_all_chats, "_get_message_resource_db_path", + return_value=resource_db), \ + patch.object(export_all_chats.mcp_server, "_iter_media_db_paths", + return_value=[media_db]): + stats = export_all_chats._collect_chat_plan_stats( + username, + [{"db_path": msg_db, "table_name": table_name}], + start_ts=150, + end_ts=None, + size_mode="estimate", + ) + + self.assertEqual(stats["message_count"], 1) + self.assertEqual(stats["message_body_bytes"], 6) + self.assertEqual(stats["attachment_estimated_bytes"], 13) + self.assertEqual(stats["attachment_scanned_bytes"], "") + self.assertEqual(stats["total_estimated_bytes"], 19) + self.assertEqual(stats["first_time"], "1970-01-01 08:03:20") + self.assertEqual(stats["last_time"], "1970-01-01 08:03:20") + self.assertEqual(stats["size_status"], "ok") + + def test_build_plan_rows_uses_batched_stats_without_per_chat_table_lookup(self): + chat_rows = [ + { + "index": 1, + "username": "wxid_alice", + "display_name": "Alice", + "kind": "single", + "remark": "Alice Remark", + "nick_name": "Alice Nick", + } + ] + stats = { + "wxid_alice": { + "message_count": 2, + "message_body_bytes": 11, + "first_time": "2026-01-01 00:00:00", + "last_time": "2026-01-02 00:00:00", + "attachment_estimated_bytes": 5, + "attachment_scanned_bytes": "", + "total_estimated_bytes": 16, + "size_status": "ok", + } + } + + with patch.object(export_all_chats, "_collect_all_plan_stats", + return_value=stats), \ + patch.object(export_all_chats.mcp_server, "_find_msg_tables_for_user", + side_effect=AssertionError("per-chat lookup too slow")): + rows = export_all_chats._build_plan_csv_rows(chat_rows) + + self.assertEqual(rows[0]["username"], "wxid_alice") + self.assertNotIn("contact_remark", rows[0]) + self.assertNotIn("contact_nick_name", rows[0]) + self.assertEqual(rows[0]["message_count"], 2) + self.assertEqual(rows[0]["total_estimated_bytes"], 16) + + +class ExportAllChatsCliPlanCsvTests(unittest.TestCase): + def _base_patches(self): + return ( + patch.object(export_all_chats.mcp_server, "DECRYPTED_DIR", "decrypted"), + patch.object(export_all_chats.os.path, "exists", return_value=True), + patch.object(export_all_chats, "_load_session_usernames", + return_value=["wxid_alice", "12345@chatroom"]), + patch.object(export_all_chats.mcp_server, "get_contact_names", + return_value={ + "wxid_alice": "Alice", + "12345@chatroom": "Project Group", + }), + patch.object(export_all_chats.mcp_server, "get_contact_full", + return_value=[]), + ) + + def test_write_plan_csv_does_not_export(self): + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "plan.csv") + patches = self._base_patches() + with patches[0], patches[1], patches[2], patches[3], \ + patch.object(export_all_chats, "_build_plan_csv_rows", + return_value=[{ + "index": 1, + "username": "wxid_alice", + "chat_name": "Alice", + "chat_type": "single", + "message_count": 0, + "first_time": "", + "last_time": "", + "attachment_estimated_bytes": 0, + "attachment_scanned_bytes": "", + "total_estimated_bytes": 0, + "size_status": "ok", + }]), \ + patch.object(export_all_chats, "export_one") as export_one: + out = io.StringIO() + with redirect_stdout(out): + export_all_chats.main(["--write-plan-csv", path]) + + export_one.assert_not_called() + self.assertTrue(os.path.exists(path)) + + def test_write_plan_csv_passes_whitelist_plan_mode(self): + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "plan.csv") + patches = self._base_patches() + with patches[0], patches[1], patches[2], patches[3], \ + patch.object(export_all_chats, "_build_plan_csv_rows", + return_value=[]), \ + patch.object(export_all_chats, "_write_plan_csv") as write_csv: + out = io.StringIO() + with redirect_stdout(out): + export_all_chats.main([ + "--write-plan-csv", path, "--plan-mode", "whitelist" + ]) + + write_csv.assert_called_once() + self.assertEqual(write_csv.call_args.kwargs["plan_mode"], "whitelist") + self.assertIn("白名单模式", out.getvalue()) + + def test_write_plan_csv_reports_write_error_without_traceback(self): + patches = self._base_patches() + with patches[0], patches[1], patches[2], patches[3], \ + patch.object(export_all_chats, "_build_plan_csv_rows", + return_value=[]), \ + patch.object(export_all_chats, "_write_plan_csv", + side_effect=PermissionError("locked")): + err = io.StringIO() + with redirect_stderr(err): + with self.assertRaises(SystemExit) as cm: + export_all_chats.main(["--write-plan-csv", "plan.csv"]) + + self.assertEqual(cm.exception.code, 1) + self.assertIn("写入导出计划 CSV 失败", err.getvalue()) + + def test_from_plan_csv_exports_only_selected_username(self): + with tempfile.TemporaryDirectory() as tmp: + plan = os.path.join(tmp, "plan.csv") + _write_csv(plan, [ + {"export": "1", "username": "wxid_alice", "chat_name": "Alice"}, + {"export": "0", "username": "12345@chatroom", "chat_name": "Group"}, + ]) + patches = self._base_patches() + with patches[0], patches[1], patches[2], patches[3], \ + patch.object(export_all_chats, "export_one", + return_value=(True, 1, 1, None)) as export_one: + out = io.StringIO() + with redirect_stdout(out): + export_all_chats.main([tmp, "--from-plan-csv", plan]) + + export_one.assert_called_once() + self.assertEqual(export_one.call_args.args[0], "wxid_alice") + output = out.getvalue() + self.assertIn("开始导出", output) + self.assertIn("完成导出", output) + + def test_from_plan_csv_whitelist_exports_only_explicit_one(self): + with tempfile.TemporaryDirectory() as tmp: + plan = os.path.join(tmp, "plan.csv") + _write_csv(plan, [ + {"export": "1", "username": "wxid_alice", "chat_name": "Alice"}, + {"export": "", "username": "12345@chatroom", "chat_name": "Group"}, + ]) + patches = self._base_patches() + with patches[0], patches[1], patches[2], patches[3], \ + patch.object(export_all_chats, "export_one", + return_value=(True, 1, 1, None)) as export_one: + out = io.StringIO() + with redirect_stdout(out): + export_all_chats.main([ + tmp, "--from-plan-csv", plan, + "--plan-mode", "whitelist", + ]) + + export_one.assert_called_once() + self.assertEqual(export_one.call_args.args[0], "wxid_alice") + self.assertIn("本次选择: 1 个会话", out.getvalue()) + + def test_from_plan_csv_dry_run_does_not_export(self): + with tempfile.TemporaryDirectory() as tmp: + plan = os.path.join(tmp, "plan.csv") + _write_csv(plan, [ + {"export": "1", "username": "wxid_alice", "chat_name": "Alice"}, + ]) + patches = self._base_patches() + with patches[0], patches[1], patches[2], patches[3], \ + patch.object(export_all_chats, "export_one") as export_one: + out = io.StringIO() + with redirect_stdout(out): + export_all_chats.main([ + tmp, "--from-plan-csv", plan, "--dry-run" + ]) + + self.assertIn("本次选择: 1 个会话", out.getvalue()) + export_one.assert_not_called() + + +class ExportOneMetadataTests(unittest.TestCase): + def test_single_chat_json_includes_contact_metadata_and_message_dates(self): + username = "wxid_zhangsan" + table_name = "Msg_" + hashlib.md5(username.encode()).hexdigest() + with tempfile.TemporaryDirectory() as tmp: + db_path = os.path.join(tmp, "message_0.db") + _create_export_message_db(db_path, table_name) + ctx = { + "username": username, + "display_name": "张三", + "message_tables": [{"db_path": db_path, "table_name": table_name}], + "is_group": False, + } + with patch.object(export_all_chats.mcp_server, "_resolve_chat_context", + return_value=ctx), \ + patch.object(export_all_chats, "_resolve_sender", + side_effect=["me", "张三备注"]), \ + patch.object(export_all_chats, "_extract_content", + side_effect=[("你好", None), ("收到", None)]), \ + patch.object(export_all_chats.mcp_server, "get_contact_full", + return_value=[{ + "username": username, + "remark": "张三备注", + "nick_name": "张三昵称", + "phone": "13800000000", + "description": "重要客户", + }]), \ + patch.object(export_all_chats.mcp_server, + "get_contact_tag_names_by_username", + return_value={username: ["客户", "北京"]}): + ok, total, new_count, reason = export_all_chats.export_one( + username, tmp, {"wxid_zhangsan": "张三"} + ) + + self.assertTrue(ok, reason) + self.assertEqual(total, 2) + self.assertEqual(new_count, 2) + out_path = os.path.join(tmp, "single_张三.json") + with open(out_path, encoding="utf-8") as f: + data = json.load(f) + + self.assertEqual(data["contact_remark"], "张三备注") + self.assertEqual(data["contact_nick_name"], "张三昵称") + self.assertNotIn("contact_phone", data) + self.assertEqual(data["contact_tags"], ["客户", "北京"]) + self.assertEqual(data["contact_memo"], "重要客户") + self.assertEqual(data["date_first_msg"], "2026-05-01 08:00:00") + self.assertEqual(data["date_last_msg"], "2026-05-01 08:01:00") + self.assertEqual( + list(data.keys())[:9], + [ + "chat", + "username", + "exported_at", + "date_first_msg", + "date_last_msg", + "contact_remark", + "contact_nick_name", + "contact_tags", + "contact_memo", + ], + ) + self.assertLess( + list(data.keys()).index("contact_memo"), + list(data.keys()).index("messages"), + ) + + +class ExportIndexTests(unittest.TestCase): + def _ctx_for(self, username, display_name, db_path, table_name): + return { + "username": username, + "display_name": display_name, + "message_tables": [{"db_path": db_path, "table_name": table_name}], + "is_group": False, + } + + def test_incremental_export_renames_existing_file_when_remark_changes(self): + username = "wxid_zhangsan" + table_name = "Msg_" + hashlib.md5(username.encode()).hexdigest() + with tempfile.TemporaryDirectory() as tmp: + db_path = os.path.join(tmp, "message_0.db") + _create_export_message_db(db_path, table_name) + old_path = os.path.join(tmp, "single_张三.json") + with open(old_path, "w", encoding="utf-8") as f: + json.dump({ + "chat": "张三", + "username": username, + "messages": [ + {"local_id": 1, "timestamp": 1777593600, + "sender": "me", "content": "你好"} + ], + }, f, ensure_ascii=False) + + with patch.object(export_all_chats.mcp_server, "_resolve_chat_context", + return_value=self._ctx_for( + username, "张三-客户", db_path, table_name + )), \ + patch.object(export_all_chats, "_resolve_sender", + return_value="张三备注"), \ + patch.object(export_all_chats, "_extract_content", + return_value=("收到", None)), \ + patch.object(export_all_chats, "_contact_metadata_for_export", + return_value={}): + ok, total, new_count, reason = export_all_chats.export_one( + username, tmp, {username: "张三-客户"}, incremental=True + ) + + new_path = os.path.join(tmp, "single_张三-客户.json") + index_path = os.path.join(tmp, "_export_index.json") + + self.assertTrue(ok, reason) + self.assertFalse(os.path.exists(old_path)) + self.assertTrue(os.path.exists(new_path)) + self.assertEqual(total, 2) + self.assertEqual(new_count, 1) + with open(new_path, encoding="utf-8") as f: + data = json.load(f) + self.assertEqual([m["local_id"] for m in data["messages"]], [1, 2]) + with open(index_path, encoding="utf-8") as f: + index = json.load(f) + entry = index["chats"][username] + self.assertEqual(entry["current_file"], "single_张三-客户.json") + self.assertIn("single_张三.json", entry["previous_files"]) + + def test_export_uses_username_suffix_when_display_name_file_belongs_to_other_chat(self): + username = "wxid_zhangsan" + table_name = "Msg_" + hashlib.md5(username.encode()).hexdigest() + with tempfile.TemporaryDirectory() as tmp: + db_path = os.path.join(tmp, "message_0.db") + _create_export_message_db(db_path, table_name) + occupied_path = os.path.join(tmp, "single_张三.json") + with open(occupied_path, "w", encoding="utf-8") as f: + json.dump({ + "chat": "张三", + "username": "wxid_other", + "messages": [], + }, f, ensure_ascii=False) + + with patch.object(export_all_chats.mcp_server, "_resolve_chat_context", + return_value=self._ctx_for( + username, "张三", db_path, table_name + )), \ + patch.object(export_all_chats, "_resolve_sender", + side_effect=["me", "张三备注"]), \ + patch.object(export_all_chats, "_extract_content", + side_effect=[("你好", None), ("收到", None)]), \ + patch.object(export_all_chats, "_contact_metadata_for_export", + return_value={}): + ok, total, new_count, reason = export_all_chats.export_one( + username, tmp, {username: "张三"} + ) + + collision_path = os.path.join(tmp, "single_张三__wxid_zhangsan.json") + + self.assertTrue(ok, reason) + self.assertEqual(total, 2) + self.assertEqual(new_count, 2) + self.assertTrue(os.path.exists(occupied_path)) + self.assertTrue(os.path.exists(collision_path)) + with open(occupied_path, encoding="utf-8") as f: + occupied = json.load(f) + with open(collision_path, encoding="utf-8") as f: + exported = json.load(f) + self.assertEqual(occupied["username"], "wxid_other") + self.assertEqual(exported["username"], username) + + +def _write_csv(path, rows): + fields = export_all_chats.PLAN_CSV_FIELDS + with open(path, "w", newline="", encoding="utf-8-sig") as f: + writer = csv.DictWriter(f, fieldnames=fields) + writer.writeheader() + for row in rows: + full = {field: "" for field in fields} + full.update(row) + writer.writerow(full) + + +def _write_custom_csv(path, fields, rows): + with open(path, "w", newline="", encoding="utf-8-sig") as f: + writer = csv.DictWriter(f, fieldnames=fields) + writer.writeheader() + for row in rows: + writer.writerow(row) + + +def _create_message_db(path, table_name): + with closing(sqlite3.connect(path)) as conn: + conn.execute(f""" + CREATE TABLE [{table_name}] ( + create_time INTEGER, + message_content TEXT, + compress_content TEXT, + packed_info_data BLOB + ) + """) + conn.execute( + f"INSERT INTO [{table_name}] VALUES (?, ?, ?, ?)", + (100, "abc", "zz", b"1234"), + ) + conn.execute( + f"INSERT INTO [{table_name}] VALUES (?, ?, ?, ?)", + (200, "hello", None, b"1"), + ) + conn.commit() + + +def _create_resource_db(path, username): + with closing(sqlite3.connect(path)) as conn: + conn.execute("CREATE TABLE ChatName2Id (user_name TEXT)") + conn.execute(""" + CREATE TABLE MessageResourceInfo ( + message_id INTEGER, + chat_id INTEGER, + message_create_time INTEGER + ) + """) + conn.execute(""" + CREATE TABLE MessageResourceDetail ( + message_id INTEGER, + size INTEGER + ) + """) + conn.execute("INSERT INTO ChatName2Id(rowid, user_name) VALUES (?, ?)", + (1, username)) + conn.execute("INSERT INTO MessageResourceInfo VALUES (?, ?, ?)", + (10, 1, 100)) + conn.execute("INSERT INTO MessageResourceInfo VALUES (?, ?, ?)", + (11, 1, 200)) + conn.execute("INSERT INTO MessageResourceDetail VALUES (?, ?)", (10, 7)) + conn.execute("INSERT INTO MessageResourceDetail VALUES (?, ?)", (11, 8)) + conn.commit() + + +def _create_media_db(path, username): + with closing(sqlite3.connect(path)) as conn: + conn.execute("CREATE TABLE Name2Id (user_name TEXT)") + conn.execute(""" + CREATE TABLE VoiceInfo ( + chat_name_id INTEGER, + local_id INTEGER, + create_time INTEGER, + voice_data BLOB + ) + """) + conn.execute("INSERT INTO Name2Id(rowid, user_name) VALUES (?, ?)", + (1, username)) + conn.execute("INSERT INTO VoiceInfo VALUES (?, ?, ?, ?)", + (1, 1, 100, b"abc")) + conn.execute("INSERT INTO VoiceInfo VALUES (?, ?, ?, ?)", + (1, 2, 200, b"abcde")) + conn.commit() + + +def _create_export_message_db(path, table_name): + with closing(sqlite3.connect(path)) as conn: + conn.execute("CREATE TABLE Name2Id (user_name TEXT)") + conn.execute(f""" + CREATE TABLE [{table_name}] ( + local_id INTEGER, + local_type INTEGER, + create_time INTEGER, + real_sender_id INTEGER, + message_content TEXT, + WCDB_CT_message_content INTEGER + ) + """) + conn.execute("INSERT INTO Name2Id(rowid, user_name) VALUES (?, ?)", + (1, "self_wxid")) + conn.execute("INSERT INTO Name2Id(rowid, user_name) VALUES (?, ?)", + (2, "wxid_zhangsan")) + conn.execute( + f"INSERT INTO [{table_name}] VALUES (?, ?, ?, ?, ?, ?)", + (1, 1, 1777593600, 1, "你好", 0), + ) + conn.execute( + f"INSERT INTO [{table_name}] VALUES (?, ?, ?, ?, ?, ?)", + (2, 1, 1777593660, 2, "收到", 0), + ) + conn.commit() + + +if __name__ == "__main__": + unittest.main()