878 lines
35 KiB
Python
878 lines
35 KiB
Python
"""
|
|
企业微信聊天记录导出 - 最终版
|
|
解密数据库 → 解析 message_table → 提取 protobuf 文本 → 导出 CSV
|
|
|
|
用法:
|
|
python wxwork_export_final.py # 默认: 只导出今天
|
|
python wxwork_export_final.py --days 3 # 导出最近 3 天 (含今天)
|
|
python wxwork_export_final.py --date 2026-08-19 # 导出指定日期
|
|
python wxwork_export_final.py --all # 全量导出所有历史
|
|
"""
|
|
|
|
import csv
|
|
import json
|
|
import os
|
|
import re
|
|
import sqlite3
|
|
import sys
|
|
from datetime import datetime, timedelta
|
|
|
|
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_media import export_media
|
|
from wxwork_voice2text import load_voice2text, asr_missing
|
|
|
|
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 detect_wxwork_dir(candidates=None):
|
|
"""自动检测企业微信数据目录 (WXWork)
|
|
|
|
依次尝试:
|
|
1. 用户配置的候选目录 (GUI 手动设置的值)
|
|
2. 当前用户 Documents\\WXWork (默认安装位置)
|
|
3. 注册表 "文档" 位置下的 WXWork (用户改过"我的文档"路径时)
|
|
4. 扫描 C:\\Users\\*\\Documents\\WXWork (企微装在别的登录用户下)
|
|
|
|
返回: 第一个包含账号 Data/message.db 的有效目录, 找不到返回 None
|
|
"""
|
|
def _valid(base):
|
|
if not base or not os.path.isdir(base):
|
|
return False
|
|
try:
|
|
for d in os.listdir(base):
|
|
if os.path.exists(os.path.join(base, d, "Data", "message.db")):
|
|
return True
|
|
except OSError:
|
|
pass
|
|
return False
|
|
|
|
# 1. 用户配置的候选目录
|
|
for c in (candidates or []):
|
|
if _valid(c):
|
|
return c
|
|
|
|
# 2. 当前用户默认位置
|
|
default = os.path.join(os.path.expanduser("~"), "Documents", "WXWork")
|
|
if _valid(default):
|
|
return default
|
|
|
|
# 3. 注册表 "文档" 位置 (用户改过系统"文档"文件夹)
|
|
try:
|
|
import winreg
|
|
with winreg.OpenKey(winreg.HKEY_CURRENT_USER,
|
|
r"Software\Microsoft\Windows\CurrentVersion\Explorer\Shell Folders") as k:
|
|
personal, _ = winreg.QueryValueEx(k, "Personal")
|
|
if personal:
|
|
reg_path = os.path.join(personal, "WXWork")
|
|
if _valid(reg_path):
|
|
return reg_path
|
|
except Exception:
|
|
pass
|
|
|
|
# 4. 扫描其他登录用户的 Documents\WXWork
|
|
users_root = os.path.join(os.environ.get("SystemDrive", "C:"), os.sep, "Users")
|
|
if os.path.isdir(users_root):
|
|
try:
|
|
for u in sorted(os.listdir(users_root)):
|
|
p = os.path.join(users_root, u, "Documents", "WXWork")
|
|
if _valid(p):
|
|
return p
|
|
except OSError:
|
|
pass
|
|
|
|
return None
|
|
|
|
# ---- 内容解析 ----
|
|
|
|
# 消息类型映射 (企业微信 content_type)
|
|
MSG_TYPE_MAP = {
|
|
0: "文本", 1: "文本", 2: "文本",
|
|
3: "图片", 4: "语音", 5: "表情",
|
|
6: "链接", 7: "文件", 8: "视频",
|
|
9: "位置", 10: "名片", 11: "系统",
|
|
12: "引用", 13: "红包", 14: "图片",
|
|
15: "转账", 16: "语音", 17: "视频号",
|
|
20: "合并转发", 21: "日程", 22: "接龙",
|
|
23: "文件回复", 26: "位置共享", 29: "视频通话",
|
|
31: "图文链接", 38: "系统通知", 40: "待办",
|
|
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: "群主转让",
|
|
101: "系统", 111: "文件", 123: "截图",
|
|
132: "系统",
|
|
503: "应用消息", 529: "朋友圈", 561: "应用消息",
|
|
565: "应用消息", 573: "应用消息",
|
|
1002: "安全通知", 1011: "系统消息", 1012: "系统消息",
|
|
1017: "系统消息", 1022: "系统消息", 1025: "系统消息",
|
|
1043: "系统消息",
|
|
1988: "系统消息",
|
|
}
|
|
|
|
|
|
def get_msg_type_name(ct):
|
|
"""消息类型名"""
|
|
try:
|
|
ct = int(ct)
|
|
except (ValueError, TypeError):
|
|
return f"类型{ct}"
|
|
return MSG_TYPE_MAP.get(ct, f"类型{ct}")
|
|
|
|
|
|
def is_personal_chat(conv_id):
|
|
"""判断会话是否为人与人之间的单聊
|
|
|
|
保留 (True):
|
|
M:xxx 微信单聊 (对方是微信用户)
|
|
S:对方UID_本账号UID 企微单聊 (两个数字 UID 用下划线连接)
|
|
|
|
过滤 (False, 即"无关紧要/群组"数据):
|
|
R:xxx 群聊 (Room)
|
|
Y:xxx 企业微信应用消息 (工作台/审批/汇报等)
|
|
O:xxx 第三方应用/服务号会话 (SCRM、加粉工具等)
|
|
MAIL/APPROVAL/... 系统虚拟会话 (企业邮箱/审批/公告等)
|
|
"""
|
|
c = str(conv_id or "")
|
|
if c.startswith("M:"):
|
|
return True
|
|
if c.startswith("S:"):
|
|
parts = c[2:].split("_")
|
|
if len(parts) == 2 and parts[0].isdigit() and parts[1].isdigit():
|
|
return True
|
|
return False
|
|
return False
|
|
|
|
|
|
# 按会话名称过滤的官方/系统账号 (命中任一关键词即过滤该会话)
|
|
DEFAULT_BLOCKED_CONV_NAMES = ("企业微信团队", "微信团队", "微信支付", "腾讯客服", "腾讯新闻")
|
|
|
|
|
|
def is_blocked_conv_name(conv_name, blocked=None):
|
|
"""按会话名称判断是否命中过滤名单 (官方/系统账号)
|
|
|
|
blocked: 关键词元组/列表; None 使用默认名单
|
|
"""
|
|
if not conv_name:
|
|
return False
|
|
blocked = DEFAULT_BLOCKED_CONV_NAMES if blocked is None else tuple(blocked)
|
|
return any(kw and kw in str(conv_name) for kw in blocked)
|
|
|
|
|
|
def clean_text(text):
|
|
"""清洗提取的文本: 去除控制字符和 protobuf 残留"""
|
|
if not text:
|
|
return ""
|
|
# 移除控制字符 (保留 \n \t)
|
|
cleaned = ''.join(c for c in text if c.isprintable() or c in '\n\t')
|
|
# 移除开头/结尾的空格和换行
|
|
cleaned = cleaned.strip()
|
|
# 压缩多余空白
|
|
cleaned = re.sub(r'[ \t]+', ' ', cleaned)
|
|
cleaned = re.sub(r'\n{3,}', '\n\n', cleaned)
|
|
return cleaned
|
|
|
|
|
|
def extract_protobuf_texts(data):
|
|
"""
|
|
从 protobuf 编码的 content 中递归提取可读文本
|
|
返回去重后的文本列表
|
|
"""
|
|
if not data or not isinstance(data, bytes):
|
|
return []
|
|
|
|
texts = []
|
|
seen = set()
|
|
|
|
def add_text(t):
|
|
t = clean_text(t)
|
|
if len(t) >= 2 and t not in seen:
|
|
# 过滤: 必须包含中文字符 或 足够多的字母
|
|
has_chinese = any('\u4e00' <= c <= '\u9fff' for c in t)
|
|
alpha_count = sum(1 for c in t if c.isalpha())
|
|
if has_chinese or alpha_count >= 6:
|
|
seen.add(t)
|
|
texts.append(t)
|
|
|
|
def parse(buf, depth):
|
|
if depth > 6 or not buf:
|
|
return
|
|
i = 0
|
|
while i < len(buf):
|
|
try:
|
|
# 读 tag (varint)
|
|
tag = 0
|
|
shift = 0
|
|
while i < len(buf) and shift < 70:
|
|
b = buf[i]
|
|
tag |= (b & 0x7f) << shift
|
|
i += 1
|
|
if not (b & 0x80):
|
|
break
|
|
shift += 7
|
|
if shift >= 70:
|
|
break
|
|
field_num = tag >> 3
|
|
wire_type = tag & 7
|
|
if field_num == 0:
|
|
break
|
|
|
|
if wire_type == 0: # varint
|
|
while i < len(buf):
|
|
b = buf[i]
|
|
i += 1
|
|
if not (b & 0x80):
|
|
break
|
|
elif wire_type == 1: # 64-bit
|
|
i += 8
|
|
elif wire_type == 2: # length-delimited
|
|
length = 0
|
|
shift = 0
|
|
while i < len(buf) and shift < 70:
|
|
b = buf[i]
|
|
length |= (b & 0x7f) << shift
|
|
i += 1
|
|
if not (b & 0x80):
|
|
break
|
|
shift += 7
|
|
if shift >= 70:
|
|
break
|
|
chunk = buf[i:i + length]
|
|
i += length
|
|
|
|
# 尝试 UTF-8 解码
|
|
try:
|
|
text = chunk.decode('utf-8')
|
|
# 无控制字符且可打印比例高
|
|
if all(ord(c) >= 32 or c in '\n\t\r' for c in text):
|
|
add_text(text)
|
|
except (UnicodeDecodeError, ValueError):
|
|
pass
|
|
|
|
# 递归解析子结构
|
|
if length < 8192:
|
|
parse(chunk, depth + 1)
|
|
elif wire_type == 5: # 32-bit
|
|
i += 4
|
|
else:
|
|
break
|
|
except Exception:
|
|
break
|
|
|
|
parse(data, 0)
|
|
|
|
# 如果 protobuf 解析没有提取到中文, 直接扫描 UTF-8 中文序列
|
|
if not any('\u4e00' <= c <= '\u9fff' for t in texts for c in t[:50]):
|
|
try:
|
|
text = data.decode('utf-8', errors='ignore')
|
|
readable = re.findall(
|
|
r'[\u4e00-\u9fff\w\s，。！？、；：“”‘’（）《》【】\-—.,;:!?()\[\]/\\+*=<>@#$%^&~`]{2,}',
|
|
text)
|
|
for r in readable:
|
|
add_text(r.strip())
|
|
except Exception:
|
|
pass
|
|
|
|
return texts
|
|
|
|
|
|
def parse_content(content):
|
|
"""解析消息内容, 返回可读文本"""
|
|
if content is None:
|
|
return ""
|
|
|
|
if not isinstance(content, bytes):
|
|
content = str(content).encode('utf-8', errors='replace')
|
|
|
|
if not content:
|
|
return ""
|
|
|
|
# 情况 1: 直接是 UTF-8 文本 (无控制字符)
|
|
try:
|
|
text = content.decode('utf-8')
|
|
if text and all(ord(c) >= 32 or c in '\n\r\t' for c in text[:500]):
|
|
return clean_text(text)
|
|
except (UnicodeDecodeError, ValueError):
|
|
pass
|
|
|
|
# 情况 2: protobuf 编码
|
|
texts = extract_protobuf_texts(content)
|
|
if texts:
|
|
# 按长度排序, 取最完整的信息
|
|
texts.sort(key=len, reverse=True)
|
|
# 过滤掉 JSON 字符串和 URL
|
|
meaningful = [t for t in texts
|
|
if not (t.startswith('{') and '}' in t)
|
|
and not t.startswith('http')
|
|
and len(t) > 1]
|
|
if meaningful:
|
|
# 清洗 protobuf 残留 (内部 tag/length 字节被误并入文本)
|
|
cleaned_msgs = []
|
|
for t in meaningful:
|
|
t = t.strip()
|
|
# 模式 A: 'T 残留 T' 重复 (如 "?点击...公交z?点击...公交"), 保留后半完整文本
|
|
m = re.match(r'^(.{6,}?)[?zxa-zA-Z]{1,4}\1$', t)
|
|
if m:
|
|
t = m.group(1)
|
|
# 模式 B: 前导 ASCII 残留后紧跟中文/中文标点 (如 "?这个是域名..." / "?，刚刚更新...")
|
|
t = re.sub(r'^[?zxa-zA-Z]{1,4}(?=[\u4e00-\u9fff\u3000-\u303f\uff00-\uffef])', '', t)
|
|
# 模式 C: 单数字残留后跟中文, 数字后前 3 个汉字内无常见量词才剥离
|
|
# (保护 "3月13日" / "7天无理由" / "5个苹果" 等合法文本)
|
|
t = re.sub(
|
|
r'^[0-9](?=[\u4e00-\u9fff])(?![\u4e00-\u9fff]{0,2}(?:月|日|年|个|天|点|号|楼|人|次|分|秒|周|元|块|台|家|辆|条|件|张|本|层|岁|万|亿|美元|室|房|折))',
|
|
'', t)
|
|
t = t.strip()
|
|
# 子串去重: 短文本是长文本的子串时跳过 (protobuf 递归解析的冗余)
|
|
if len(t) >= 3:
|
|
dup = False
|
|
for c in cleaned_msgs:
|
|
if len(c) >= 4 and (t in c or c in t):
|
|
dup = True
|
|
break
|
|
if dup:
|
|
continue
|
|
if t and t not in cleaned_msgs:
|
|
cleaned_msgs.append(t)
|
|
if cleaned_msgs:
|
|
return ' | '.join(cleaned_msgs[:2])
|
|
return texts[0]
|
|
|
|
# 情况 3: JSON
|
|
try:
|
|
obj = json.loads(content.decode('utf-8', errors='replace'))
|
|
return json.dumps(obj, ensure_ascii=False)[:2000]
|
|
except Exception:
|
|
pass
|
|
|
|
# 情况 4: 二进制数据
|
|
if len(content) <= 64:
|
|
return content.hex()
|
|
return f"[二进制数据 {len(content)} 字节]"
|
|
|
|
|
|
def format_timestamp(ts):
|
|
"""Unix 时间戳转可读格式"""
|
|
if ts is None or ts == "":
|
|
return ""
|
|
try:
|
|
ts = int(ts)
|
|
if ts > 10**12:
|
|
ts = ts // 1000
|
|
if ts < 0:
|
|
return str(ts)
|
|
return datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M:%S")
|
|
except (ValueError, OSError, OverflowError):
|
|
return str(ts)
|
|
|
|
|
|
# ---- 数据库解密 ----
|
|
|
|
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:
|
|
keys.setdefault("*", data["global_key"])
|
|
return keys
|
|
|
|
|
|
def decrypt_with_keys(db_base, out_dir, keys_map, use_cache=True):
|
|
"""解密所有数据库
|
|
|
|
use_cache=True 时启用增量缓存: 若解密副本的修改时间不早于源库文件,
|
|
且大小正常, 则直接复用, 避免每天重复解密 (日常导出秒开).
|
|
"""
|
|
decrypted_dbs = []
|
|
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
|
|
|
|
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
|
|
|
|
# 只处理与聊天相关的数据库
|
|
relevant = [f for f in db_files if f in
|
|
("message.db", "session.db", "user.db", "company.db",
|
|
"message_lookup.db", "user_extend.db")]
|
|
for db_name in relevant:
|
|
db_path = os.path.join(data_dir, db_name)
|
|
out_path = os.path.join(user_out, db_name)
|
|
|
|
# 增量缓存: 解密副本存在且不比源文件旧则直接复用
|
|
# 同时检查 -wal 文件: 若 WAL 比解密副本新, 说明有未 checkpoint 的新消息
|
|
wal_path = db_path + "-wal"
|
|
wal_mtime = 0
|
|
try:
|
|
if os.path.exists(wal_path) and os.path.getsize(wal_path) > 0:
|
|
wal_mtime = os.path.getmtime(wal_path)
|
|
except OSError:
|
|
wal_mtime = 0
|
|
src_mtime = max(os.path.getmtime(db_path), wal_mtime)
|
|
|
|
if use_cache and os.path.exists(out_path):
|
|
try:
|
|
if (os.path.getmtime(out_path) >= src_mtime
|
|
and os.path.getsize(out_path) > 4096):
|
|
decrypted_dbs.append((out_path, db_name, user_dir))
|
|
continue
|
|
except OSError:
|
|
pass
|
|
|
|
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())
|
|
decrypted_dbs.append((out_path, db_name, user_dir))
|
|
continue
|
|
|
|
if not is_wxsqlite3_aes128_page1(page1):
|
|
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)
|
|
decrypted_dbs.append((out_path, db_name, user_dir))
|
|
continue
|
|
except Exception:
|
|
pass
|
|
|
|
# 尝试所有密钥
|
|
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)
|
|
decrypted_dbs.append((out_path, db_name, user_dir))
|
|
break
|
|
except Exception:
|
|
break
|
|
|
|
return decrypted_dbs
|
|
|
|
|
|
# ---- 导出 ----
|
|
|
|
def connect_sqlite(path):
|
|
conn = sqlite3.connect(path)
|
|
conn.text_factory = lambda b: b.decode('utf-8', errors='replace')
|
|
return conn
|
|
|
|
|
|
def export_messages(decrypted_dbs, out_dir, date_from=None, date_to=None,
|
|
personal_only=True, blocked_conv_names=None,
|
|
voice2text_map=None, voice_asr=False):
|
|
"""导出消息到 CSV
|
|
|
|
date_from / date_to: 'YYYY-MM-DD' 日期范围 (含两端)
|
|
None + None -> 全量导出到 out_dir 根目录 (企业微信聊天记录.csv + 会话汇总.csv + 分账号)
|
|
单天 from==to -> out_dir/按天/YYYY-MM-DD_聊天记录.csv
|
|
多天 from<to -> out_dir/按天/YYYY-MM-DD_至_YYYY-MM-DD_聊天记录.csv
|
|
personal_only: True=只导出单聊 (过滤群聊/应用消息/第三方应用, 默认); False=导出全部会话
|
|
blocked_conv_names: 会话名称关键词黑名单 (官方/系统账号, 如"企业微信团队"),
|
|
命中即过滤整个会话; None 使用 DEFAULT_BLOCKED_CONV_NAMES, 传空元组 () 关闭
|
|
voice2text_map: 语音转写文本 {(user_dir, str(server_id)): text}, None 自动读取本地缓存
|
|
voice_asr: True=对无本地缓存的语音用本地 AI 识别 (需装 pilk+faster-whisper)
|
|
返回值: 主 CSV 路径 (若 date_from/date_to 为 None 则同时输出分账号/汇总 CSV)
|
|
"""
|
|
os.makedirs(out_dir, exist_ok=True)
|
|
|
|
all_messages = []
|
|
user_cache = {} # (user_dir, uid) -> name
|
|
conv_cache = {} # (user_dir, conv_id) -> name
|
|
|
|
# 阶段 1: 收集元数据
|
|
for db_path, db_name, user_dir in decrypted_dbs:
|
|
try:
|
|
conn = connect_sqlite(db_path)
|
|
cursor = conn.cursor()
|
|
|
|
if db_name == "user.db":
|
|
try:
|
|
cursor.execute("PRAGMA table_info(user_table)")
|
|
cols = [r[1] for r in cursor.fetchall()]
|
|
if cols:
|
|
cursor.execute("SELECT id, name, real_name, account FROM user_table")
|
|
for uid, name, real_name, account in cursor.fetchall():
|
|
if uid is not None:
|
|
nm = name or real_name or account or str(uid)
|
|
user_cache[(user_dir, str(uid))] = str(nm)
|
|
except sqlite3.Error:
|
|
pass
|
|
|
|
if db_name == "session.db":
|
|
try:
|
|
cursor.execute("PRAGMA table_info(conversation_table)")
|
|
cols = [r[1] for r in cursor.fetchall()]
|
|
if cols:
|
|
cursor.execute("SELECT id, name, roomname_remark, session_id FROM conversation_table")
|
|
for cid, name, remark, sid in cursor.fetchall():
|
|
if cid:
|
|
nm = remark or name or sid
|
|
conv_cache[(user_dir, str(cid))] = str(nm) if nm else ""
|
|
except sqlite3.Error:
|
|
pass
|
|
|
|
conn.close()
|
|
except sqlite3.Error as e:
|
|
print(f" [警告] 元数据失败: {db_path}: {e}")
|
|
|
|
# 阶段 1.5: 按会话名称/ID过滤, 先计算要排除的会话ID集合 (供媒体导出复用)
|
|
blocked_conv_names = DEFAULT_BLOCKED_CONV_NAMES if blocked_conv_names is None else tuple(blocked_conv_names)
|
|
blocked_conv_ids = set()
|
|
for (user_dir, cid), cname in conv_cache.items():
|
|
if is_personal_chat(cid) and (
|
|
is_blocked_conv_name(cname, blocked_conv_names)
|
|
or is_blocked_conv_name(str(cid), blocked_conv_names)):
|
|
blocked_conv_ids.add(str(cid))
|
|
|
|
# 阶段 1.6: 导出媒体文件 (图片/语音/视频/文件) 到 out_dir/媒体文件
|
|
media_map = {}
|
|
media_stats = {}
|
|
try:
|
|
print("\n[*] 导出媒体文件 (图片/语音/视频/文件)...")
|
|
media_map, media_stats = export_media(
|
|
decrypted_dbs, out_dir, date_from=date_from, date_to=date_to,
|
|
log=print, personal_only=personal_only, blocked_conv_ids=blocked_conv_ids)
|
|
print(f"[+] 媒体导出: 复制 {media_stats.get('copied', 0)} 个文件, "
|
|
f"UUID匹配 {media_stats.get('matched_uuid', 0)}, "
|
|
f"文件名匹配 {media_stats.get('matched_filename', 0)}, "
|
|
f"仅URL {media_stats.get('url_only', 0)}")
|
|
except Exception as e:
|
|
print(f" [警告] 媒体导出失败, Excel 将不显示媒体路径: {e}")
|
|
|
|
# 阶段 1.7: 语音转文字 (本地缓存 + 可选本地 ASR 兜底)
|
|
if voice2text_map is None:
|
|
print("\n[*] 读取语音转写缓存 (msg_voice2text)...")
|
|
voice2text_map = load_voice2text(decrypted_dbs, log=print)
|
|
if voice_asr and media_map:
|
|
print("\n[*] 本地识别无缓存语音...")
|
|
voice2text_map = asr_missing(voice2text_map, media_map, log=print)
|
|
|
|
def _get_voice_text(user_dir, server_id):
|
|
"""查语音转写文本 (兼容账号维度和全局 server_id 两种 key)"""
|
|
if not voice2text_map:
|
|
return ""
|
|
sid = str(server_id)
|
|
return voice2text_map.get((user_dir, sid)) or voice2text_map.get(("", sid)) or ""
|
|
|
|
# 阶段 2: 导出消息
|
|
for db_path, db_name, user_dir in decrypted_dbs:
|
|
if db_name != "message.db":
|
|
continue
|
|
try:
|
|
conn = connect_sqlite(db_path)
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='message_table'")
|
|
if not cursor.fetchone():
|
|
conn.close()
|
|
continue
|
|
|
|
cursor.execute("PRAGMA table_info(message_table)")
|
|
columns = [row[1] for row in cursor.fetchall()]
|
|
print(f"\n[+] 导出消息: {user_dir}/message.db")
|
|
print(f" 字段: {', '.join(columns)}")
|
|
|
|
cursor.execute("SELECT * FROM message_table ORDER BY send_time ASC")
|
|
rows = cursor.fetchall()
|
|
print(f" 共 {len(rows)} 条消息")
|
|
|
|
exported = 0
|
|
skipped_non_personal = 0
|
|
skipped_blocked_name = 0
|
|
for row in rows:
|
|
msg_dict = dict(zip(columns, row))
|
|
|
|
conv_id = msg_dict.get("conversation_id", "")
|
|
sender_id = msg_dict.get("sender_id", "")
|
|
ct = msg_dict.get("content_type", "")
|
|
ts = msg_dict.get("send_time") or 0
|
|
time_str = format_timestamp(ts)
|
|
# 日期范围过滤 (含两端)
|
|
day = time_str[:10]
|
|
if date_from and day < date_from:
|
|
continue
|
|
if date_to and day > date_to:
|
|
continue
|
|
# 只导出单聊 (过滤群聊 R:/ 应用 Y:/ 第三方应用 O:/ 系统会话)
|
|
if personal_only and not is_personal_chat(conv_id):
|
|
skipped_non_personal += 1
|
|
continue
|
|
content = parse_content(msg_dict.get("content"))
|
|
extra = parse_content(msg_dict.get("extra_content"))
|
|
client_id = msg_dict.get("client_id", "")
|
|
sequence = msg_dict.get("sequence", "")
|
|
message_id = msg_dict.get("message_id", "")
|
|
server_id = msg_dict.get("server_id", "")
|
|
|
|
# extra_content 仅在 content 完全无文本时兜底 (短文本如 "在不" 是有效内容, 不能被覆盖)
|
|
if extra and not content.strip():
|
|
if not (extra.startswith('{') and '}' in extra) and 'http' not in extra[:8]:
|
|
content = extra
|
|
|
|
# 会话名称
|
|
conv_name = conv_cache.get((user_dir, str(conv_id)), "")
|
|
if not conv_name:
|
|
# 单聊: M:xxx -> 查找用户
|
|
if str(conv_id).startswith("M:"):
|
|
uid = str(conv_id)[2:]
|
|
conv_name = user_cache.get((user_dir, uid), uid)
|
|
elif str(conv_id).startswith("S:"):
|
|
# S:会话为单聊, 两个 UID (对方/本账号) 顺序不固定,
|
|
# 取不是本账号(user_dir)的那个作为对方
|
|
parts = str(conv_id)[2:].split("_")
|
|
peer_uid = ""
|
|
for p in parts:
|
|
if p != user_dir:
|
|
peer_uid = p
|
|
break
|
|
if not peer_uid:
|
|
peer_uid = parts[0] if parts else ""
|
|
peer_name = user_cache.get((user_dir, peer_uid), "")
|
|
conv_name = peer_name if peer_name else ("单聊 " + str(conv_id))
|
|
elif str(conv_id).startswith("O:"):
|
|
conv_name = str(conv_id)[2:]
|
|
elif str(conv_id).startswith("Y:"):
|
|
conv_name = "应用 " + str(conv_id)[2:]
|
|
else:
|
|
conv_name = str(conv_id)
|
|
|
|
# 按会话名称/ID过滤 (官方/系统账号, 如"企业微信团队"; 也可填会话ID如 13102691405879689)
|
|
if is_blocked_conv_name(conv_name, blocked_conv_names) \
|
|
or is_blocked_conv_name(str(conv_id), blocked_conv_names):
|
|
skipped_blocked_name += 1
|
|
continue
|
|
|
|
# 发送者名称
|
|
sender_name = user_cache.get((user_dir, str(sender_id)), "")
|
|
if not sender_name:
|
|
sender_name = str(sender_id) if sender_id else "系统"
|
|
|
|
# 媒体文件路径/URL (从阶段 1.6 的 media_map 中按 server_id 匹配)
|
|
media_info = media_map.get(str(server_id), {})
|
|
media_file_path = media_info.get("path", "")
|
|
media_url = media_info.get("url", "")
|
|
|
|
# 语音转文字: 语音消息 (.silk/.amr) 优先显示转写文本
|
|
voice_text = _get_voice_text(user_dir, server_id)
|
|
is_voice = str(media_file_path).lower().endswith((".silk", ".amr")) or \
|
|
(str(media_url or "").endswith((".silk", ".amr")))
|
|
if is_voice and voice_text:
|
|
content = voice_text
|
|
|
|
all_messages.append({
|
|
"账号": user_dir,
|
|
"会话ID": conv_id,
|
|
"会话名称": conv_name,
|
|
"时间": time_str,
|
|
"发送者ID": sender_id,
|
|
"发送者": sender_name,
|
|
"消息类型": get_msg_type_name(ct),
|
|
"内容": content,
|
|
"语音转文字": voice_text,
|
|
"媒体文件路径": media_file_path,
|
|
"媒体URL": media_url,
|
|
"消息序号": sequence,
|
|
"消息ID": message_id,
|
|
"服务器ID": server_id,
|
|
"客户端ID": client_id,
|
|
})
|
|
exported += 1
|
|
|
|
conn.close()
|
|
if skipped_non_personal or skipped_blocked_name:
|
|
print(f" 已导出 {exported} 条 (过滤非单聊 {skipped_non_personal} 条, "
|
|
f"过滤官方/系统账号 {skipped_blocked_name} 条)")
|
|
else:
|
|
print(f" 已导出 {exported} 条")
|
|
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["时间"])
|
|
|
|
fieldnames = ["账号", "会话ID", "会话名称", "时间", "发送者ID", "发送者",
|
|
"消息类型", "内容", "语音转文字", "媒体文件路径", "媒体URL",
|
|
"消息序号", "消息ID", "服务器ID", "客户端ID"]
|
|
|
|
# 按天/范围模式: 输出到 按天/ 目录
|
|
if date_from or date_to:
|
|
day_dir = os.path.join(out_dir, "按天")
|
|
os.makedirs(day_dir, exist_ok=True)
|
|
if date_from and date_from == date_to:
|
|
csv_path = os.path.join(day_dir, f"{date_from}_聊天记录.csv")
|
|
summary_path = os.path.join(day_dir, f"{date_from}_会话汇总.csv")
|
|
else:
|
|
frm = date_from or "2000-01-01"
|
|
to = date_to or datetime.now().strftime("%Y-%m-%d")
|
|
csv_path = os.path.join(day_dir, f"{frm}_至_{to}_聊天记录.csv")
|
|
summary_path = os.path.join(day_dir, f"{frm}_至_{to}_会话汇总.csv")
|
|
else:
|
|
csv_path = os.path.join(out_dir, "企业微信聊天记录.csv")
|
|
summary_path = os.path.join(out_dir, "会话汇总.csv")
|
|
|
|
# 主 CSV
|
|
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[+] 主文件: {csv_path}")
|
|
print(f" 总消息数: {len(all_messages)}")
|
|
|
|
# 会话汇总
|
|
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, "最早": m["时间"], "最晚": m["时间"]}
|
|
session_stats[key]["消息数"] += 1
|
|
if m["时间"] and (not session_stats[key]["最早"] or m["时间"] < session_stats[key]["最早"]):
|
|
session_stats[key]["最早"] = m["时间"]
|
|
if m["时间"] and m["时间"] > session_stats[key]["最晚"]:
|
|
session_stats[key]["最晚"] = m["时间"]
|
|
|
|
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["最早"], info["最晚"]])
|
|
print(f" 会话汇总: {summary_path} ({len(session_stats)} 个会话)")
|
|
|
|
# 按账号分文件 (仅全量模式)
|
|
if not date_from and not date_to:
|
|
by_account = {}
|
|
for m in all_messages:
|
|
by_account.setdefault(m["账号"], []).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="企业微信数据库根目录")
|
|
parser.add_argument("--date", default=None, help="导出指定日期 (YYYY-MM-DD)")
|
|
parser.add_argument("--days", type=int, default=None,
|
|
help="导出最近 N 天 (含今天), 如 --days 1 只导今天; 默认 1")
|
|
parser.add_argument("--all", action="store_true", help="全量导出所有历史记录")
|
|
parser.add_argument("--with-groups", action="store_true",
|
|
help="同时导出群聊/应用消息 (默认只导出单聊)")
|
|
args = parser.parse_args()
|
|
|
|
print("=" * 60)
|
|
print(" 企业微信聊天记录导出工具 v2.3 (按天/范围导出, 仅单聊)")
|
|
print("=" * 60)
|
|
|
|
keys_map = load_keys()
|
|
if not keys_map:
|
|
print("[-] 未找到密钥文件")
|
|
return 1
|
|
print(f"[+] 已加载 {len(keys_map)} 个账号密钥")
|
|
if args.with_groups:
|
|
print("[*] 范围: 全部会话 (单聊 + 群聊 + 应用)")
|
|
else:
|
|
print("[*] 范围: 仅单聊 (群聊/应用消息/第三方应用已过滤)")
|
|
|
|
# 导出模式: --all > --date > --days > 默认今天
|
|
date_from = date_to = None
|
|
if args.all:
|
|
print("[*] 模式: 全量导出")
|
|
elif args.date:
|
|
date_from = date_to = args.date
|
|
print(f"[*] 模式: 导出指定日期 {args.date}")
|
|
else:
|
|
days = args.days if args.days and args.days > 0 else 1
|
|
date_to = datetime.now().strftime("%Y-%m-%d")
|
|
date_from = (datetime.now() - timedelta(days=days - 1)).strftime("%Y-%m-%d")
|
|
if days == 1:
|
|
print(f"[*] 模式: 只导出今天 ({date_to})")
|
|
else:
|
|
print(f"[*] 模式: 导出最近 {days} 天 ({date_from} ~ {date_to})")
|
|
|
|
print(f"\n[*] 解密数据库...")
|
|
decrypted_dbs = decrypt_with_keys(
|
|
args.db_dir, os.path.join(args.output, "decrypted"), keys_map, use_cache=True)
|
|
print(f"[+] 成功解密 {len(decrypted_dbs)} 个数据库")
|
|
|
|
if not decrypted_dbs:
|
|
print("[-] 没有可用的数据库")
|
|
return 1
|
|
|
|
print(f"\n[*] 导出聊天记录...")
|
|
csv_path = export_messages(decrypted_dbs, args.output,
|
|
date_from=date_from, date_to=date_to,
|
|
personal_only=not args.with_groups)
|
|
|
|
if csv_path:
|
|
print(f"\n{'='*60}")
|
|
print(f" 导出完成! 主文件: {csv_path}")
|
|
print(f" 输出目录: {args.output}")
|
|
print(f"{'='*60}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|