""" 企业微信聊天记录导出工具 解密数据库 → 解析消息 → 导出 CSV 用法: python wxwork_export.py --key <32位hex密钥> [--output <输出目录>] [--db-dir <数据库目录>] """ import argparse import csv import json import os import sqlite3 import struct import sys from datetime import datetime from wxwork_crypto import ( decrypt_wxwork_database, is_plain_sqlite_page, is_wxsqlite3_aes128_page1, verify_sqlite_file, verify_wxsqlite3_aes128_key, ) # 默认数据库目录 DEFAULT_DB_BASE = os.path.join(os.path.expanduser("~"), "Documents", "WXWork") DEFAULT_OUTPUT = os.path.join(os.getcwd(), "wxwork_export") def parse_content(content_blob): """尝试解析消息内容 BLOB, 返回可读文本""" if content_blob is None: return "" # 尝试多种解析方式 if isinstance(content_blob, bytes): # 方式1: UTF-8 文本 try: text = content_blob.decode('utf-8') if text and all(c.isprintable() or c in '\n\r\t' for c in text[:100]): return text except (UnicodeDecodeError, ValueError): pass # 方式2: JSON 编码 try: obj = json.loads(content_blob.decode('utf-8')) return json.dumps(obj, ensure_ascii=False) except (UnicodeDecodeError, json.JSONDecodeError): pass # 方式3: 16进制 (二进制数据) if len(content_blob) <= 256: return content_blob.hex() return f"[二进制数据 {len(content_blob)} 字节]" return str(content_blob) def format_timestamp(ts): """将 Unix 时间戳转为可读格式""" if not ts: return "" try: # 微信/企微时间戳为秒 ts = int(ts) if ts > 10**12: # 毫秒 ts = ts // 1000 return datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M:%S") except (ValueError, OSError, OverflowError): return str(ts) def get_message_type_name(msg_type): """消息类型映射""" type_map = { 1: "文本", 3: "图片", 4: "语音", 5: "表情", 6: "链接", 7: "文件", 8: "视频", 9: "位置", 10: "名片", 13: "系统消息", 15: "引用", 18: "转账", 19: "红包", 20: "小程序", 21: "视频号", 22: "合并转发", 24: "文件回复", 27: "接龙", 28: "日程", 29: "收集表", 30: "会议", 34: "卡券", 42: "文件预览", 46: "话题", 47: "图文", 49: "待办", 51: "收藏", 53: "订阅通知", 55: "文件分享", 56: "群公告", 57: "投票", 59: "回执", 61: "文件编辑", 62: "文件评论", 63: "收藏合并", 64: "卡片", 65: "群文件", 66: "工作台", 67: "位置共享", 68: "视频通话", 69: "音频通话", 70: "企业微信应用", 73: "微信好友", 74: "回复", 76: "邀请", 77: "移除", 78: "修改群名", 79: "修改群公告", 80: "加入群聊", 81: "退出群聊", 82: "解散群聊", 83: "群主转让", } return type_map.get(msg_type, f"类型{msg_type}") def decrypt_all_databases(db_base, out_dir, raw_key): """解密所有用户目录下的数据库""" decrypted_dbs = [] os.makedirs(out_dir, exist_ok=True) # 扫描所有用户目录 if not os.path.isdir(db_base): print(f"[-] 数据库目录不存在: {db_base}") return decrypted_dbs 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")] if not db_files: continue user_out = os.path.join(out_dir, user_dir) print(f"\n[+] 处理用户目录: {user_dir}") print(f" 数据库文件: {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) with open(db_path, "rb") as f: page1 = f.read(4096) if is_plain_sqlite_page(page1): # 未加密, 直接复制 os.makedirs(os.path.dirname(out_path), exist_ok=True) 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)) continue if not is_wxsqlite3_aes128_page1(page1): print(f" 跳过: {db_name} (未知加密格式)") continue if not verify_wxsqlite3_aes128_key(raw_key, page1): print(f" 失败: {db_name} (密钥验证失败)") continue try: decrypt_wxwork_database(db_path, out_path, raw_key) tables = verify_sqlite_file(out_path) print(f" 解密: {db_name} (表: {', '.join(tables[:6]) or '无'})") decrypted_dbs.append((out_path, db_name, user_dir)) except Exception as e: print(f" 失败: {db_name} ({e})") return decrypted_dbs def export_messages_to_csv(decrypted_dbs, out_dir): """从解密后的数据库中提取消息并导出 CSV""" os.makedirs(out_dir, exist_ok=True) # 收集所有消息 all_messages = [] # 用户信息缓存: user_id -> nick_name/remark user_cache = {} # 群信息缓存: group_id -> group_name group_cache = {} # 会话信息缓存: sid -> session_name session_cache = {} # 遍历所有解密数据库 for db_path, db_name, user_dir in decrypted_dbs: try: conn = sqlite3.connect(db_path) cursor = conn.cursor() # 读取 userinfo 表 if db_name in ("user.db", "im.db"): try: cursor.execute("SELECT user_id, nick_name, remark FROM userinfo") for user_id, nick, remark in cursor.fetchall(): if user_id: user_cache[user_id] = remark or nick or user_id except sqlite3.Error: pass # 读取 groupinfo 表 if db_name in ("session.db", "im.db"): try: cursor.execute("SELECT group_id, group_name FROM groupinfo") for gid, gname in cursor.fetchall(): if gid: group_cache[gid] = gname or gid except sqlite3.Error: pass # 读取 session 表 if db_name in ("session.db", "im.db"): try: cursor.execute( "SELECT sid, type, group_name, active_time FROM session") for sid, stype, gname, atime in cursor.fetchall(): if sid: if gname: session_cache[sid] = gname else: session_cache[sid] = sid except sqlite3.Error: pass conn.close() except sqlite3.Error as e: print(f" [警告] 读取元数据失败 {db_path}: {e}") # 主循环: 从 message.db / msg_*.db 导出消息 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() # 检查 message 表是否存在 cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='message'") if not cursor.fetchone(): continue # 读取 message 表结构 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(sid, sid or "未知会话") # 获取发送者名称 sender_name = "" if sender: sender_name = user_cache.get(sender, sender) if sender == "" and 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({ "会话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 # 按时间排序 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["会话名称"] if key not in session_stats: session_stats[key] = {"会话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 name, info in sorted(session_stats.items(), key=lambda x: -x[1]["消息数"]): writer.writerow([name, info["会话ID"], info["消息数"], info["最后时间"]]) print(f" 会话汇总: {summary_path}") print(f" 共 {len(session_stats)} 个会话") return csv_path def main(): parser = argparse.ArgumentParser(description="企业微信聊天记录导出工具") parser.add_argument("--key", help="16字节密钥 (32位hex)") 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) # 获取密钥 raw_key = None if args.key: raw_key = bytes.fromhex(args.key.replace("x'", "").replace("'", "")) if len(raw_key) != 16: print("[-] 密钥必须是 32 位 hex 字符 (16字节)") return 1 print(f"[+] 使用手动提供的密钥: {raw_key.hex()}") else: # 尝试从密钥文件加载 keys_file = os.path.join(os.getcwd(), "wxwork_keys.json") if os.path.exists(keys_file): with open(keys_file) as f: keys_data = json.load(f) if "global_key" in keys_data: raw_key = bytes.fromhex(keys_data["global_key"]) print(f"[+] 从密钥文件加载密钥: {raw_key.hex()}") elif "enc_key" in keys_data: raw_key = bytes.fromhex(keys_data["enc_key"]) print(f"[+] 从密钥文件加载密钥: {raw_key.hex()}") if raw_key is None: print("[-] 未提供密钥!") print(" 请先运行密钥提取, 或使用 --key 参数传入密钥") print() print(" 提取密钥方法:") print(" 1. 启动并登录企业微信") print(" 2. 运行: python wxwork_find_key.py") print(" 3. 将提取的密钥传入本工具: python wxwork_export.py --key <密钥>") return 1 # 解密所有数据库 print(f"\n[*] 扫描数据库目录: {args.db_dir}") decrypted_dbs = decrypt_all_databases(args.db_dir, os.path.join(args.output, "decrypted"), raw_key) if not decrypted_dbs: print("[-] 没有成功解密的数据库") return 1 # 导出消息 print(f"\n[*] 开始导出聊天记录...") csv_path = export_messages_to_csv(decrypted_dbs, args.output) if csv_path: print(f"\n[+] 导出完成! 文件位置: {csv_path}") return 0 if __name__ == "__main__": sys.exit(main())