""" 企业微信聊天记录导出 - 多账号版 支持每个用户目录使用独立密钥解密, 然后统一导出 CSV 用法: python wxwork_export_multi.py [--output <输出目录>] """ import csv import json import os import sqlite3 import sys from datetime import datetime BASE_DIR = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, BASE_DIR) from wxwork_crypto import ( decrypt_wxwork_database, is_plain_sqlite_page, is_wxsqlite3_aes128_page1, verify_sqlite_file, verify_wxsqlite3_aes128_key, ) from wxwork_export import parse_content, format_timestamp, get_message_type_name DEFAULT_DB_BASE = os.path.join(os.path.expanduser("~"), "Documents", "WXWork") DEFAULT_OUTPUT = os.path.join(BASE_DIR, "wxwork_export") KEYS_FILE = os.path.join(BASE_DIR, "wxwork_keys.json") def load_keys(): """从密钥文件加载密钥映射 {user_dir: key_hex}""" if not os.path.exists(KEYS_FILE): return {} with open(KEYS_FILE) as f: data = json.load(f) keys = data.get("keys", {}) # 兼容旧格式 if "global_key" in data and "global_key" not in keys: keys["*"] = data["global_key"] return keys def decrypt_with_keys(db_base, out_dir, keys_map): """使用密钥映射解密所有数据库, 返回 (解密文件列表, 解密统计)""" decrypted_dbs = [] stats = {"total_dbs": 0, "decrypted": 0, "failed": 0, "skipped": 0} os.makedirs(out_dir, exist_ok=True) # 收集所有已知密钥 all_keys = set() for k in keys_map.values(): try: all_keys.add(bytes.fromhex(k.replace("x'", "").replace("'", ""))) except ValueError: pass if not os.path.isdir(db_base): print(f"[-] 数据库目录不存在: {db_base}") return decrypted_dbs, stats for user_dir in sorted(os.listdir(db_base)): data_dir = os.path.join(db_base, user_dir, "Data") if not os.path.isdir(data_dir): continue db_files = [f for f in os.listdir(data_dir) if f.endswith(".db") and not f.endswith("-wal") and not f.endswith("-shm")] if not db_files: continue user_out = os.path.join(out_dir, user_dir) os.makedirs(user_out, exist_ok=True) # 该目录的密钥: 优先精确匹配, 否则尝试所有密钥 dir_key = None if user_dir in keys_map: try: dir_key = bytes.fromhex(keys_map[user_dir]) except ValueError: dir_key = None print(f"\n[+] 用户目录: {user_dir} ({len(db_files)} 个数据库文件)") for db_name in sorted(db_files): db_path = os.path.join(data_dir, db_name) out_path = os.path.join(user_out, db_name) stats["total_dbs"] += 1 with open(db_path, "rb") as f: page1 = f.read(4096) # 未加密数据库直接复制 if is_plain_sqlite_page(page1): with open(db_path, "rb") as fin, open(out_path, "wb") as fout: fout.write(fin.read()) print(f" 复制: {db_name} (未加密)") decrypted_dbs.append((out_path, db_name, user_dir)) stats["decrypted"] += 1 continue if not is_wxsqlite3_aes128_page1(page1): print(f" 跳过: {db_name} (未知加密格式)") stats["skipped"] += 1 continue # 用目录密钥验证 if dir_key and verify_wxsqlite3_aes128_key(dir_key, page1): try: decrypt_wxwork_database(db_path, out_path, dir_key) verify_sqlite_file(out_path) print(f" 解密: {db_name} (目录密钥)") decrypted_dbs.append((out_path, db_name, user_dir)) stats["decrypted"] += 1 continue except Exception as e: print(f" 失败: {db_name} ({e})") stats["failed"] += 1 continue # 目录密钥无效或缺失, 尝试所有已知密钥 matched = False for key in all_keys: if verify_wxsqlite3_aes128_key(key, page1): try: decrypt_wxwork_database(db_path, out_path, key) verify_sqlite_file(out_path) print(f" 解密: {db_name} (通用密钥 {key.hex()[:8]}...)") decrypted_dbs.append((out_path, db_name, user_dir)) stats["decrypted"] += 1 matched = True break except Exception as e: print(f" 失败: {db_name} ({e})") stats["failed"] += 1 matched = True # 密钥匹配但解密失败 break if not matched: print(f" 失败: {db_name} (无可用密钥)") stats["failed"] += 1 return decrypted_dbs, stats def export_messages(decrypted_dbs, out_dir): """从解密后的数据库提取消息并导出 CSV (多账号版)""" os.makedirs(out_dir, exist_ok=True) all_messages = [] user_cache = {} group_cache = {} session_cache = {} # 阶段 1: 收集元数据 (用户/群/会话) for db_path, db_name, user_dir in decrypted_dbs: try: conn = sqlite3.connect(db_path) cursor = conn.cursor() if db_name in ("user.db", "im.db", "userinfo.db"): try: # 尝试多种可能的字段 cursor.execute("PRAGMA table_info(userinfo)") cols = [r[1] for r in cursor.fetchall()] id_col = "user_id" if "user_id" in cols else (cols[0] if cols else None) if id_col: cursor.execute(f'SELECT "{id_col}", nick_name, remark FROM userinfo') for uid, nick, remark in cursor.fetchall(): if uid: user_cache[(user_dir, uid)] = remark or nick or uid except sqlite3.Error: pass if db_name in ("session.db", "im.db"): try: cursor.execute("PRAGMA table_info(session)") cols = [r[1] for r in cursor.fetchall()] if cols: # 找 sid 和名称字段 sid_col = "sid" if "sid" in cols else cols[0] name_col = None for c in ["group_name", "name", "title", "remark"]: if c in cols: name_col = c break if name_col: cursor.execute(f'SELECT "{sid_col}", "{name_col}" FROM session') for sid, name in cursor.fetchall(): if sid and name: session_cache[(user_dir, sid)] = name except sqlite3.Error: pass conn.close() except sqlite3.Error as e: print(f" [警告] 元数据读取失败 {db_path}: {e}") # 阶段 2: 导出消息 for db_path, db_name, user_dir in decrypted_dbs: if not (db_name.startswith("message") or db_name.startswith("msg")): continue try: conn = sqlite3.connect(db_path) cursor = conn.cursor() cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='message'") if not cursor.fetchone(): conn.close() continue cursor.execute("PRAGMA table_info(message)") columns = [row[1] for row in cursor.fetchall()] print(f"\n[+] 导出消息: {user_dir}/{db_name}") print(f" 字段: {', '.join(columns)}") cursor.execute("SELECT * FROM message ORDER BY time ASC") rows = cursor.fetchall() print(f" 共 {len(rows)} 条消息") for row in rows: msg_dict = dict(zip(columns, row)) msg_id = msg_dict.get("id", "") sid = msg_dict.get("sid", "") msg_type = msg_dict.get("type", "") ts = msg_dict.get("time") or msg_dict.get("client_time") or 0 is_self = msg_dict.get("is_self", 0) sender = msg_dict.get("sender", "") content = parse_content(msg_dict.get("content")) element_desc = msg_dict.get("element_descriptions", "") or "" seq = msg_dict.get("seq", "") status = msg_dict.get("status", "") # 会话名称 (优先取本账号的缓存) session_name = session_cache.get((user_dir, sid), str(sid) if sid else "未知会话") # 发送者名称 sender_name = "" if sender: sender_name = user_cache.get((user_dir, sender), sender) elif not is_self: sender_name = "其他" full_content = content if element_desc and element_desc not in content: full_content = f"{content} | {element_desc}" direction = "发送" if is_self else "接收" all_messages.append({ "账号目录": user_dir, "会话ID": sid, "会话名称": session_name, "时间": format_timestamp(ts), "方向": direction, "发送者ID": sender, "发送者": sender_name, "消息类型": get_message_type_name(msg_type), "内容": full_content, "消息序号": seq, "状态": status, "数据库": f"{user_dir}/{db_name}", }) conn.close() except sqlite3.Error as e: print(f" [警告] 导出消息失败 {db_path}: {e}") if not all_messages: print("\n[-] 未找到任何消息数据") return None # 按时间排序 all_messages.sort(key=lambda m: m["时间"]) # 导出主 CSV csv_path = os.path.join(out_dir, "企业微信聊天记录.csv") fieldnames = ["账号目录", "会话ID", "会话名称", "时间", "方向", "发送者ID", "发送者", "消息类型", "内容", "消息序号", "状态", "数据库"] with open(csv_path, "w", newline="", encoding="utf-8-sig") as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() writer.writerows(all_messages) print(f"\n[+] 导出完成!") print(f" 总消息数: {len(all_messages)}") print(f" 输出文件: {csv_path}") # 导出按会话汇总 session_stats = {} for m in all_messages: key = (m["账号目录"], m["会话名称"]) if key not in session_stats: session_stats[key] = {"账号": m["账号目录"], "会话名称": m["会话名称"], "会话ID": m["会话ID"], "消息数": 0, "最后时间": ""} session_stats[key]["消息数"] += 1 if m["时间"] > session_stats[key]["最后时间"]: session_stats[key]["最后时间"] = m["时间"] summary_path = os.path.join(out_dir, "会话汇总.csv") with open(summary_path, "w", newline="", encoding="utf-8-sig") as f: writer = csv.writer(f) writer.writerow(["账号", "会话名称", "会话ID", "消息数", "最后消息时间"]) for info in sorted(session_stats.values(), key=lambda x: -x["消息数"]): writer.writerow([info["账号"], info["会话名称"], info["会话ID"], info["消息数"], info["最后时间"]]) print(f" 会话汇总: {summary_path}") print(f" 共 {len(session_stats)} 个会话") # 按账号导出单独文件 by_account = {} for m in all_messages: acc = m["账号目录"] if acc not in by_account: by_account[acc] = [] by_account[acc].append(m) for acc, msgs in by_account.items(): acc_path = os.path.join(out_dir, f"聊天记录_{acc}.csv") with open(acc_path, "w", newline="", encoding="utf-8-sig") as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() writer.writerows(msgs) print(f" 账号 {acc}: {len(msgs)} 条 -> {acc_path}") return csv_path def main(): import argparse parser = argparse.ArgumentParser(description="企业微信聊天记录导出 (多账号)") parser.add_argument("--output", default=DEFAULT_OUTPUT, help="输出目录") parser.add_argument("--db-dir", default=DEFAULT_DB_BASE, help="企业微信数据库根目录") args = parser.parse_args() print("=" * 60) print(" 企业微信聊天记录导出工具 (多账号版)") print("=" * 60) # 加载密钥 keys_map = load_keys() if not keys_map: print("[-] 未找到密钥文件, 请先运行密钥提取") return 1 print(f"[+] 已加载 {len(keys_map)} 个账号密钥") # 解密 print(f"\n[*] 扫描数据库目录: {args.db_dir}") decrypted_dbs, stats = decrypt_with_keys( args.db_dir, os.path.join(args.output, "decrypted"), keys_map) print(f"\n[*] 解密统计: 共 {stats['total_dbs']} 个数据库, " f"成功 {stats['decrypted']}, 失败 {stats['failed']}, 跳过 {stats['skipped']}") if not decrypted_dbs: print("[-] 没有成功解密的数据库") return 1 # 导出 print(f"\n[*] 开始导出聊天记录...") csv_path = export_messages(decrypted_dbs, args.output) if csv_path: print(f"\n[+] 全部完成! 主文件: {csv_path}") print(f" 输出目录: {args.output}") return 0 if __name__ == "__main__": sys.exit(main())