374 lines
14 KiB
Python
374 lines
14 KiB
Python
"""
|
|
企业微信聊天记录导出 - 多账号版
|
|
支持每个用户目录使用独立密钥解密, 然后统一导出 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())
|