""" 企业微信聊天记录导出 - 最终版 解密数据库 → 解析 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 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())