431 lines
14 KiB
Python
431 lines
14 KiB
Python
"""
|
|
企业微信聊天记录导出工具
|
|
解密数据库 → 解析消息 → 导出 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())
|