#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 企业微信聊天记录导出助手 - 桌面版 ================================= 功能: 1. 一键导出全部聊天记录 (CSV + Excel) 2. 自动提取企微进程内存密钥 (无需手动操作) 3. 每日定时自动导出 4. 导出完成后自动打开输出目录 打包: pyinstaller --onefile --windowed --name "企业微信导出助手" wxwork_gui.py """ import json import os import queue import sys import threading import time import tkinter as tk from datetime import datetime, timedelta from tkinter import filedialog, messagebox, ttk # ---- 路径处理 (兼容 PyInstaller onefile 模式) ---- APP_DIR = os.path.dirname(os.path.abspath( sys.executable if getattr(sys, 'frozen', False) else __file__)) sys.path.insert(0, APP_DIR) import wxwork_crypto # noqa: E402 import wxwork_export_final as exporter # noqa: E402 import wxwork_export_db as db_exporter # noqa: E402 import wxwork_find_key as keyfinder # noqa: E402 KEYS_FILE = os.path.join(APP_DIR, 'wxwork_keys.json') CONFIG_FILE = os.path.join(APP_DIR, 'wxwork_gui_config.json') DEFAULT_DB_BASE = os.path.join(os.path.expanduser('~'), 'Documents', 'WXWork') DEFAULT_OUT = os.path.join(APP_DIR, 'wxwork_export') FONT = "Microsoft YaHei UI" BG = "#f5f7fa" CARD = "#ffffff" PRIMARY = "#2f6fed" SUCCESS = "#16a34a" WARNING = "#d97706" # 导出范围选项: (显示文本, 天数) days=0 表示全部历史 DAYS_OPTIONS = [1, 3, 7, 30, 0] DAYS_LABELS = ["只导今天 (最近1天)", "最近 3 天", "最近 7 天", "最近 30 天", "全部历史"] # ================ 密钥提取 ================ def extract_all_keys(log, db_dir=None): """扫描企微进程内存, 提取所有已登录账号的密钥, 返回 {user_dir: key_hex} db_dir: 企业微信数据目录 (WXWork), None 使用默认 Documents\\WXWork """ db_dir = (db_dir or DEFAULT_DB_BASE).strip() if not os.path.isdir(db_dir): log('[-] 未找到企业微信数据目录: ' + db_dir) log(' 请在「导出设置」中手动选择或点击「自动检测」') return {} # 1. 收集所有加密的 message.db targets = [] # [(user_dir, page1)] for d in sorted(os.listdir(db_dir)): msg_path = os.path.join(db_dir, d, 'Data', 'message.db') if os.path.exists(msg_path): with open(msg_path, 'rb') as f: page1 = f.read(4096) if page1[:16] != b'SQLite format 3\x00' and \ exporter.is_wxsqlite3_aes128_page1(page1): targets.append((d, page1)) if not targets: log('[-] 未找到任何加密的数据库 (企业微信可能未安装或未登录)') return {} log(f'[+] 发现 {len(targets)} 个账号的加密数据库') # 2. 检查企微进程 pids = keyfinder.get_wxwork_pids() if not pids: log('[-] 未找到运行中的企业微信进程') log(' 请先启动并登录企业微信, 再重新提取密钥') return {} log(f'[+] 找到企业微信进程: {pids}') log('[*] 正在扫描进程内存 (可能需要 1-2 分钟)...') # 3. 收集候选密钥 candidates = [] seen = set() for pid in pids: regions = keyfinder.read_process_memory_regions(pid) log(f' PID {pid}: 读取 {len(regions)} 个内存区域') for base_addr, data in regions: for key_cand, addr, reason in keyfinder.find_key_candidates_in_data(data, base_addr): kb = bytes(key_cand) if len(kb) == 16 and kb not in seen: seen.add(kb) candidates.append(kb) log(f'[+] 内存中候选密钥 {len(candidates)} 个, 开始逐库验证...') # 4. 逐库验证 keys = {} for user_dir, page1 in targets: for kb in candidates: if keyfinder.verify_key(kb, page1): keys[user_dir] = kb.hex() log(f' [OK] 账号 {user_dir} -> {kb.hex()[:8]}...') break else: log(f' [--] 账号 {user_dir}: 未匹配到密钥 (该账号未登录?)') if not keys: log('[-] 未能匹配任何密钥, 请确认企微已登录并重新尝试') return keys # ================ 导出 ================ def do_export(keys_map, out_dir, log, days=1, with_db=True, personal_only=True, blocked_conv_names=None, db_dir=None, voice_text=True, voice_asr=False): """执行完整导出流程 (解密 + 导出 CSV/Excel + 可选 MySQL SQL) days: 0=全部历史, 1=只导今天, N=最近 N 天 (含今天) with_db: 是否同时生成 MySQL 导入 SQL (CRM 绑定用) personal_only: True=只导出单聊 (过滤群聊/应用消息/第三方应用, 默认); False=导出全部会话 blocked_conv_names: 会话名称关键词黑名单 (官方/系统账号, 如"企业微信团队"), None 使用默认名单, 空元组 () 关闭 db_dir: 企业微信数据目录 (WXWork), None 使用默认 Documents\\WXWork voice_text: True=读取企微本地语音转写缓存并写入「语音转文字」列 (默认) voice_asr: True=对无本地缓存的语音用本地 AI 识别 (需装 pilk+faster-whisper, 较慢) """ if not keys_map: log('[-] 没有可用密钥, 请先点击「重新提取密钥」') return None db_dir = (db_dir or DEFAULT_DB_BASE).strip() if not os.path.isdir(db_dir): log('[-] 企业微信数据目录不存在: ' + db_dir) log(' 请在「导出设置」中手动选择或点击「自动检测」') return None # WAL 检测: 企业微信运行时消息先写 -wal, 退出后才 checkpoint 到主库 wal_accounts = [] for acc in sorted(os.listdir(db_dir)): data_dir = os.path.join(db_dir, acc, 'Data') wal = os.path.join(data_dir, 'message.db-wal') try: if os.path.exists(wal) and os.path.getsize(wal) > 0: wal_accounts.append(acc) except OSError: pass if wal_accounts: log('[!] 检测到企业微信仍在运行 (message.db-wal 未同步)') log('[!] 账号 ' + ', '.join(wal_accounts) + ' 的最新消息可能尚未写入本地数据库') log('[!] 建议退出企业微信后重新导出, 以获取刚收到的消息') # 日期范围 date_from = date_to = None if days and days > 0: date_to = datetime.now().strftime('%Y-%m-%d') date_from = (datetime.now() - timedelta(days=days - 1)).strftime('%Y-%m-%d') if days == 1: log(f'[*] 导出范围: 只导今天 ({date_to})') else: log(f'[*] 导出范围: 最近 {days} 天 ({date_from} ~ {date_to})') else: log('[*] 导出范围: 全部历史') if personal_only: log('[*] 会话范围: 仅单聊 (群聊/应用消息/第三方应用已过滤)') else: log('[*] 会话范围: 全部会话 (含群聊/应用消息)') if blocked_conv_names is not None and len(blocked_conv_names) > 0: log(f'[*] 名称过滤: 排除官方/系统账号 ({", ".join(blocked_conv_names)})') log(f'[*] 数据目录: {db_dir}') # 解密 dec_out = os.path.join(out_dir, 'decrypted') log('[*] 步骤 1/4: 解密数据库...') dbs = exporter.decrypt_with_keys(db_dir, dec_out, keys_map) log(f'[+] 解密完成: {len(dbs)} 个数据库') msg_dbs = [d for d in dbs if d[1] == 'message.db'] if not msg_dbs: log('[-] 未找到消息数据库, 导出中止') return None # 语音转文字: 预读本地缓存 (ASR 结果会就地合并, 供 CSV 与 DB 共享, 避免重复识别) voice2text_map = None if voice_text: try: from wxwork_voice2text import load_voice2text voice2text_map = load_voice2text(dbs, log=log) if voice2text_map: log(f'[+] 语音转写缓存: {len(voice2text_map)} 条') except Exception as e: log(f'[!] 读取语音转写缓存失败 (不影响导出): {e}') if voice_asr: log('[*] 已开启无缓存语音本地识别 (较慢, 首次需下载模型)') # 导出 CSV log('[*] 步骤 2/4: 导出 CSV...') csv_path = exporter.export_messages(dbs, out_dir, date_from=date_from, date_to=date_to, personal_only=personal_only, blocked_conv_names=blocked_conv_names, voice2text_map=voice2text_map, voice_asr=voice_asr) if not csv_path: if days and days > 0: log('[-] 该时间范围内没有新消息, 无需导出') else: log('[-] CSV 导出失败') return None # 导出 Excel (跟随实际 CSV 路径, 支持按天模式) log('[*] 步骤 3/4: 生成 Excel 文件...') try: from openpyxl import Workbook from openpyxl.styles import Alignment, Font, PatternFill from openpyxl.utils import get_column_letter import csv as _csv def csv_to_xlsx(csv_p, xlsx_p, sheet, widths, wrap): with open(csv_p, encoding='utf-8-sig') as f: rows = list(_csv.reader(f)) wb = Workbook() ws = wb.active ws.title = sheet for r in rows: ws.append(r) headers = rows[0] if rows else [] fill = PatternFill('solid', fgColor='4472C4') for cell in ws[1]: cell.font = Font(bold=True, color='FFFFFF') cell.fill = fill cell.alignment = Alignment(horizontal='center', vertical='center') ws.freeze_panes = 'A2' for i, w in enumerate(widths, 1): ws.column_dimensions[get_column_letter(i)].width = w for col in wrap: for cell in ws[col][1:]: cell.alignment = Alignment(wrap_text=True, vertical='top') # 媒体文件路径列设置为超链接 (点击可直接打开图片/语音/文件) type_col = headers.index('消息类型') + 1 if '消息类型' in headers else None content_col = headers.index('内容') + 1 if '内容' in headers else None voice_text_col = (headers.index('语音转文字') + 1 if '语音转文字' in headers else None) # 内容列 + 语音转文字列 自动换行 for cidx in (content_col, voice_text_col): if cidx: for cell in ws[get_column_letter(cidx)][1:]: cell.alignment = Alignment(wrap_text=True, vertical='top') if '媒体文件路径' in headers: media_col = get_column_letter(headers.index('媒体文件路径') + 1) base_dir = os.path.dirname(xlsx_p) # 按天模式下 xlsx 位于 out_dir/按天/, 媒体文件在 out_dir/媒体文件/ if os.path.basename(base_dir) == '按天': base_dir = os.path.dirname(base_dir) for idx, cell in enumerate(ws[media_col][1:], start=2): if not cell.value: continue rel_path = str(cell.value) abs_path = os.path.abspath(os.path.join(base_dir, rel_path)) cell.hyperlink = f"file:///{abs_path.replace(os.sep, '/')}" cell.style = 'Hyperlink' # 媒体类消息的内容显示为友好提示, 不再展示原始编码串 if type_col and content_col: mt = ws.cell(idx, type_col).value if mt in ('图片', '截图', '语音', '视频', '文件'): # 语音消息若有转写文本, 内容列已是转写文本, 不再替换为 [语音] if mt == '语音' and voice_text_col: vt = ws.cell(idx, voice_text_col).value if vt and str(vt).strip(): continue tip = {'图片': '[图片]', '截图': '[截图]', '语音': '[语音]', '视频': '[视频]', '文件': '[文件]'}.get(mt, '[媒体]') ws.cell(idx, content_col).value = tip wb.save(xlsx_p) summary_path = os.path.join(out_dir, '会话汇总.csv') if not date_from else \ csv_path.replace('_聊天记录.csv', '_会话汇总.csv') csv_to_xlsx(csv_path, csv_path[:-4] + '.xlsx', '聊天记录', [16, 22, 26, 20, 18, 18, 12, 50, 50, 45, 40, 14, 12, 18, 18], []) if os.path.exists(summary_path): csv_to_xlsx(summary_path, summary_path[:-4] + '.xlsx', '会话汇总', [16, 30, 26, 12, 20, 20], []) log('[+] Excel 文件已生成') except Exception as e: log(f'[!] Excel 生成失败 (CSV 仍可用): {e}') # 导出 MySQL SQL (CRM 绑定用) if with_db: log('[*] 步骤 4/4: 生成 MySQL SQL + 导出媒体文件 (图片/语音/视频)...') try: db_out = os.path.join(out_dir, 'crm_import') db_exporter.export_to_db(dbs, db_out, date_from=date_from, date_to=date_to, personal_only=personal_only, blocked_conv_names=blocked_conv_names, voice2text_map=voice2text_map, voice_asr=voice_asr) log(f'[+] MySQL SQL 已生成 -> {db_out}') except Exception as e: log(f'[!] MySQL SQL 生成失败: {e}') # 统计 try: import csv as _csv with open(csv_path, encoding='utf-8-sig') as f: total = sum(1 for _ in _csv.reader(f)) - 1 log(f'[+] 全部完成! 共导出 {total} 条消息') except Exception: log('[+] 全部完成!') return csv_path # ================ GUI ================ class ExporterApp: def __init__(self, root): self.root = root self.msg_queue = queue.Queue() self.exporting = False self.extracting = False self.auto_thread = None self.auto_stop = threading.Event() self.config = self.load_config() root.title("企业微信聊天记录导出助手") root.geometry("860x640") root.minsize(760, 560) root.configure(bg=BG) self._build_style() self._build_ui() self._load_keys_display() # 日志队列轮询 self.root.after(100, self._poll_queue) # 启动定时线程 self._start_auto_thread() # ---------- 样式 ---------- def _build_style(self): style = ttk.Style() try: style.theme_use('vista') except tk.TclError: pass style.configure('TButton', font=(FONT, 10), padding=6) style.configure('Accent.TButton', font=(FONT, 11, 'bold'), padding=10) style.configure('TLabelframe', font=(FONT, 10, 'bold')) style.configure('TLabelframe.Label', font=(FONT, 10, 'bold')) style.configure('TLabel', font=(FONT, 10), background=BG) style.configure('TCheckbutton', font=(FONT, 10), background=BG) # ---------- UI ---------- def _build_ui(self): # 标题 header = tk.Frame(self.root, bg=PRIMARY, height=64) header.pack(fill='x') header.pack_propagate(False) tk.Label(header, text="📤 企业微信聊天记录导出助手", font=(FONT, 16, 'bold'), bg=PRIMARY, fg='white').pack(side='left', padx=20) self.status_label = tk.Label(header, text="● 就绪", font=(FONT, 10), bg=PRIMARY, fg='#cfe0ff') self.status_label.pack(side='right', padx=20) body = tk.Frame(self.root, bg=BG) body.pack(fill='both', expand=True, padx=16, pady=12) # 左侧: 密钥与设置 left = tk.Frame(body, bg=BG) left.pack(side='left', fill='y', padx=(0, 12)) # 密钥卡片 key_card = tk.Frame(left, bg=CARD, highlightthickness=1, highlightbackground='#e3e8ef') key_card.pack(fill='x', pady=(0, 12)) tk.Label(key_card, text="🔑 已识别的账号", font=(FONT, 11, 'bold'), bg=CARD).pack(anchor='w', padx=12, pady=(10, 4)) self.key_list = tk.Text(key_card, height=6, width=42, font=(FONT, 9), bg=CARD, relief='flat', state='disabled', fg='#334155') self.key_list.pack(padx=12, pady=(0, 6)) self.extract_btn = tk.Button(key_card, text="重新提取密钥 (企微需在运行)", font=(FONT, 9), bg=PRIMARY, fg='white', relief='flat', cursor='hand2', padx=8, pady=4, command=self.on_extract_keys) self.extract_btn.pack(padx=12, pady=(0, 10), fill='x') # 设置卡片 set_card = tk.Frame(left, bg=CARD, highlightthickness=1, highlightbackground='#e3e8ef') set_card.pack(fill='x') tk.Label(set_card, text="⚙ 导出设置", font=(FONT, 11, 'bold'), bg=CARD).pack(anchor='w', padx=12, pady=(10, 4)) # 企业微信数据目录 (非默认安装时可手动指定) tk.Label(set_card, text="企业微信数据目录 (WXWork):", font=(FONT, 9), bg=CARD).pack(anchor='w', padx=12) db_row = tk.Frame(set_card, bg=CARD) db_row.pack(fill='x', padx=12, pady=(2, 2)) self.db_dir_var = tk.StringVar(value=self.config.get('db_dir', DEFAULT_DB_BASE) or DEFAULT_DB_BASE) self.db_dir_entry = tk.Entry(db_row, textvariable=self.db_dir_var, font=(FONT, 9)) self.db_dir_entry.pack(side='left', fill='x', expand=True) tk.Button(db_row, text="浏览", font=(FONT, 9), bg='#eef2f7', relief='flat', cursor='hand2', command=self.on_pick_db_dir).pack(side='left', padx=(6, 0)) db_row2 = tk.Frame(set_card, bg=CARD) db_row2.pack(fill='x', padx=12, pady=(2, 8)) self.db_detect_btn = tk.Button(db_row2, text="自动检测", font=(FONT, 9), bg='#eef2f7', relief='flat', cursor='hand2', command=self.on_detect_db_dir) self.db_detect_btn.pack(side='left') self.db_dir_hint = tk.Label(db_row2, text="留空则使用默认位置, 检测不到请手动选择", font=(FONT, 8), bg=CARD, fg='#64748b') self.db_dir_hint.pack(side='left', padx=(8, 0)) self.db_dir_entry.bind('', lambda e: self.save_config()) tk.Label(set_card, text="输出目录:", font=(FONT, 9), bg=CARD).pack(anchor='w', padx=12) out_row = tk.Frame(set_card, bg=CARD) out_row.pack(fill='x', padx=12, pady=(2, 6)) self.out_var = tk.StringVar(value=self.config.get('out_dir', DEFAULT_OUT)) self.out_entry = tk.Entry(out_row, textvariable=self.out_var, font=(FONT, 9)) self.out_entry.pack(side='left', fill='x', expand=True) tk.Button(out_row, text="浏览", font=(FONT, 9), bg='#eef2f7', relief='flat', cursor='hand2', command=self.on_pick_dir).pack(side='left', padx=(6, 0)) # 导出范围 tk.Label(set_card, text="导出范围:", font=(FONT, 9), bg=CARD).pack(anchor='w', padx=12, pady=(6, 0)) self.days_var = tk.IntVar(value=self.config.get('days', 1)) self.days_combo = ttk.Combobox(set_card, state='readonly', font=(FONT, 9), values=DAYS_LABELS, width=24) self.days_combo.current(DAYS_OPTIONS.index(self.days_var.get()) if self.days_var.get() in DAYS_OPTIONS else 0) self.days_combo.pack(anchor='w', padx=12, pady=(2, 8)) self.days_combo.bind('<>', lambda e: self.on_days_change()) # MySQL SQL 导出 self.db_var = tk.BooleanVar(value=self.config.get('with_db', True)) tk.Checkbutton(set_card, text="同时生成 MySQL 导入 SQL (CRM 绑定用)", variable=self.db_var, font=(FONT, 9), bg=CARD, command=self.on_toggle_db).pack(anchor='w', padx=12, pady=(0, 4)) # 仅导出单聊 self.personal_var = tk.BooleanVar(value=self.config.get('personal_only', True)) tk.Checkbutton(set_card, text="仅导出单聊 (不含群组/应用消息)", variable=self.personal_var, font=(FONT, 9), bg=CARD, command=self.on_toggle_personal).pack(anchor='w', padx=12, pady=(0, 4)) # 语音转文字 (读取企微本地缓存) self.voice_text_var = tk.BooleanVar(value=self.config.get('voice_text', True)) tk.Checkbutton(set_card, text="语音转文字 (读取企微本地转写缓存, 写入「语音转文字」列)", variable=self.voice_text_var, font=(FONT, 9), bg=CARD, command=self.save_config).pack(anchor='w', padx=12, pady=(0, 4)) # 无缓存语音本地 AI 识别 (默认关闭, 较慢) self.voice_asr_var = tk.BooleanVar(value=self.config.get('voice_asr', False)) tk.Checkbutton(set_card, text="无缓存语音用本地 AI 识别 (较慢, 首次需下载模型 ~150MB)", variable=self.voice_asr_var, font=(FONT, 9), bg=CARD, command=self.save_config).pack(anchor='w', padx=12, pady=(0, 8)) # 按名称/ID过滤官方/系统账号 tk.Label(set_card, text="过滤的账号名称或会话ID (逗号分隔, 命中即排除整个会话):", font=(FONT, 9), bg=CARD, fg='#334155').pack(anchor='w', padx=12) self.blocked_var = tk.StringVar( value=self.config.get('blocked_names', '企业微信团队,微信团队,微信支付,腾讯客服,腾讯新闻')) name_row = tk.Frame(set_card, bg=CARD) name_row.pack(anchor='w', padx=12, pady=(2, 10), fill='x') self.blocked_entry = tk.Entry(name_row, textvariable=self.blocked_var, font=(FONT, 9)) self.blocked_entry.pack(side='left', fill='x', expand=True) self.blocked_entry.bind('', lambda e: self.save_config()) # 定时自动导出 self.auto_var = tk.BooleanVar(value=self.config.get('auto_enable', False)) tk.Checkbutton(set_card, text="每日定时自动导出", variable=self.auto_var, font=(FONT, 10), bg=CARD, command=self.on_toggle_auto ).pack(anchor='w', padx=12, pady=(4, 0)) time_row = tk.Frame(set_card, bg=CARD) time_row.pack(anchor='w', padx=12, pady=(2, 10)) tk.Label(time_row, text="时间:", font=(FONT, 9), bg=CARD).pack(side='left') hour = self.config.get('hour', 18) minute = self.config.get('minute', 30) self.hour_var = tk.StringVar(value=f"{hour:02d}") self.minute_var = tk.StringVar(value=f"{minute:02d}") self.hour_spin = tk.Spinbox(time_row, from_=0, to=23, width=3, textvariable=self.hour_var, font=(FONT, 9), format='%02.0f') self.hour_spin.pack(side='left', padx=(8, 2)) tk.Label(time_row, text=":", font=(FONT, 9), bg=CARD).pack(side='left') self.minute_spin = tk.Spinbox(time_row, from_=0, to=59, width=3, textvariable=self.minute_var, font=(FONT, 9), format='%02.0f') self.minute_spin.pack(side='left', padx=(2, 8)) tk.Label(time_row, text="(应用运行期间到点自动导出)", font=(FONT, 8), bg=CARD, fg='#64748b').pack(side='left') # 右侧: 导出操作 + 日志 right = tk.Frame(body, bg=BG) right.pack(side='left', fill='both', expand=True) act_card = tk.Frame(right, bg=CARD, highlightthickness=1, highlightbackground='#e3e8ef') act_card.pack(fill='x', pady=(0, 12)) self.export_btn = tk.Button( act_card, text=self._export_btn_text(), font=(FONT, 13, 'bold'), bg=SUCCESS, fg='white', relief='flat', cursor='hand2', padx=20, pady=12, command=self.on_export) self.export_btn.pack(fill='x', padx=14, pady=14) self.progress = ttk.Progressbar(act_card, mode='indeterminate') self.progress.pack(fill='x', padx=14, pady=(0, 14)) log_card = tk.Frame(right, bg=CARD, highlightthickness=1, highlightbackground='#e3e8ef') log_card.pack(fill='both', expand=True) tk.Label(log_card, text="📋 运行日志", font=(FONT, 11, 'bold'), bg=CARD).pack(anchor='w', padx=12, pady=(10, 4)) self.log_text = tk.Text(log_card, height=14, font=("Consolas", 9), bg='#0f172a', fg='#e2e8f0', relief='flat', state='disabled', wrap='word') self.log_text.pack(fill='both', expand=True, padx=12, pady=(0, 6)) log_btns = tk.Frame(log_card, bg=CARD) log_btns.pack(fill='x', padx=12, pady=(0, 10)) tk.Button(log_btns, text="打开输出文件夹", font=(FONT, 9), bg='#eef2f7', relief='flat', cursor='hand2', command=self.on_open_out).pack(side='left') tk.Button(log_btns, text="清空日志", font=(FONT, 9), bg='#eef2f7', relief='flat', cursor='hand2', command=self.on_clear_log).pack(side='left', padx=(8, 0)) self.log('[*] 应用已启动') # 启动时自动检测企业微信数据目录 configured = self.db_dir_var.get().strip() if configured and os.path.isdir(configured): self.db_dir_hint.configure(text=f"✓ {configured}", fg='#16a34a') self.log(f'[+] 使用企业微信数据目录: {configured}') elif configured == DEFAULT_DB_BASE: # 默认路径不存在, 尝试自动定位其他位置, 但仍保留默认路径作为占位提示 found = exporter.detect_wxwork_dir() if found: self.db_dir_var.set(found) self.save_config() self.db_dir_hint.configure(text=f"✓ 已自动检测: {found}", fg='#16a34a') self.log(f'[+] 默认位置未找到, 自动检测到企业微信数据目录: {found}') else: self.db_dir_hint.configure(text=f"✗ 默认位置未检测到: {DEFAULT_DB_BASE}", fg='#d97706') self.log(f'[-] 默认位置未检测到企业微信数据目录: {DEFAULT_DB_BASE}') self.log('[-] 若已安装在其他位置, 请在「导出设置」中手动指定或点击「自动检测」') else: # 用户自定义路径无效 self.db_dir_hint.configure(text=f"✗ 指定目录不存在: {configured}", fg='#d97706') self.log(f'[-] 指定的企业微信数据目录不存在: {configured}') self.log(f'[*] 数据目录: {self.get_db_dir()}') self.log(f'[*] 输出目录: {self.out_var.get()}') # ---------- 配置 ---------- def load_config(self): try: with open(CONFIG_FILE, encoding='utf-8') as f: return json.load(f) except Exception: return {} def save_config(self): cfg = { 'out_dir': self.out_var.get(), 'db_dir': self.db_dir_var.get(), 'auto_enable': self.auto_var.get(), 'hour': int(self.hour_var.get() or 18), 'minute': int(self.minute_var.get() or 30), 'days': self.days_var.get(), 'with_db': self.db_var.get(), 'personal_only': self.personal_var.get(), 'blocked_names': self.blocked_var.get(), 'voice_text': self.voice_text_var.get(), 'voice_asr': self.voice_asr_var.get(), } try: with open(CONFIG_FILE, 'w', encoding='utf-8') as f: json.dump(cfg, f, ensure_ascii=False, indent=2) except Exception: pass self.config = cfg # ---------- 导出范围 ---------- def _export_btn_text(self): """根据当前导出范围生成按钮文字""" days = self.days_var.get() if days <= 0: return "▶ 立即导出全部聊天记录" if days == 1: return "▶ 立即导出今天的聊天记录" return f"▶ 立即导出最近 {days} 天的聊天记录" def on_days_change(self): idx = self.days_combo.current() if 0 <= idx < len(DAYS_OPTIONS): self.days_var.set(DAYS_OPTIONS[idx]) self.export_btn.configure(text=self._export_btn_text()) self.save_config() def on_toggle_db(self): self.save_config() if self.db_var.get(): self.log('[+] 已开启: 导出时同时生成 MySQL SQL + 媒体文件 (crm_import 目录)') else: self.log('[-] 已关闭: 不再生成 MySQL SQL, 只导出 CSV/Excel') def on_toggle_personal(self): self.save_config() if self.personal_var.get(): self.log('[+] 已开启: 仅导出单聊 (群聊/应用消息/第三方应用已过滤)') else: self.log('[-] 已关闭: 导出全部会话 (含群聊和应用消息)') # ---------- 密钥 ---------- def _load_keys_display(self): keys = {} try: with open(KEYS_FILE, encoding='utf-8') as f: keys = json.load(f).get('keys', {}) except Exception: keys = {} self.key_list.configure(state='normal') self.key_list.delete('1.0', 'end') if not keys: self.key_list.insert('end', ' 暂无密钥\n 请点击下方按钮提取\n') self.status_label.configure(text="● 未提取密钥", fg='#ffd7a8') else: for uid, k in sorted(keys.items()): self.key_list.insert('end', f' {uid}\n {k[:8]}...{k[-6:]}\n') self.status_label.configure(text=f"● {len(keys)} 个账号已就绪", fg='#b8e6c9') self.key_list.configure(state='disabled') def on_extract_keys(self): if self.extracting: return if not messagebox.askyesno("提取密钥", "提取前请确认:\n\n1. 企业微信正在运行且已登录\n2. 扫描内存约需 1-2 分钟\n\n继续吗?"): return self.extracting = True self.extract_btn.configure(state='disabled', text="正在提取密钥...") threading.Thread(target=self._extract_worker, daemon=True).start() def _extract_worker(self): self.log('') self.log('=' * 50) self.log('[开始] 提取密钥') keys = extract_all_keys(self.log, self.get_db_dir()) if keys: try: old = {} if os.path.exists(KEYS_FILE): with open(KEYS_FILE, encoding='utf-8') as f: old = json.load(f) old.setdefault('keys', {}).update(keys) with open(KEYS_FILE, 'w', encoding='utf-8') as f: json.dump(old, f, ensure_ascii=False, indent=2) self.log(f'[+] 已保存 {len(keys)} 个密钥 -> {KEYS_FILE}') except Exception as e: self.log(f'[-] 密钥保存失败: {e}') self.msg_queue.put(('extract_done', None)) # ---------- 导出 ---------- def on_export(self): if self.exporting: return self.exporting = True self.export_btn.configure(state='disabled', text="⏳ 正在导出, 请稍候...") self.progress.start(12) threading.Thread(target=self._export_worker, daemon=True).start() def _export_worker(self): out_dir = self.out_var.get().strip() or DEFAULT_OUT days = self.days_var.get() try: os.makedirs(out_dir, exist_ok=True) except Exception as e: self.log(f'[-] 无法创建输出目录: {e}') self.msg_queue.put(('export_done', None)) return self.log('') self.log('=' * 50) if days <= 0: self.log(f'[开始] 导出全部聊天记录 -> {out_dir}') else: self.log(f'[开始] 导出最近 {days} 天聊天记录 -> {out_dir}') keys = {} try: with open(KEYS_FILE, encoding='utf-8') as f: keys = json.load(f).get('keys', {}) except Exception: keys = {} if not keys: self.log('[-] 没有可用密钥, 请先点击「重新提取密钥」') self.msg_queue.put(('export_done', None)) return t0 = time.time() blocked_raw = self.blocked_var.get().strip() blocked_names = tuple( kw.strip() for kw in blocked_raw.split(',') if kw.strip()) if blocked_raw else () csv_path = do_export(keys, out_dir, self.log, days=days, with_db=self.db_var.get(), personal_only=self.personal_var.get(), blocked_conv_names=blocked_names, db_dir=self.get_db_dir(), voice_text=self.voice_text_var.get(), voice_asr=self.voice_asr_var.get()) elapsed = time.time() - t0 if csv_path: self.log(f'[完成] 耗时 {elapsed:.1f} 秒') self.msg_queue.put(('export_ok', csv_path)) else: self.msg_queue.put(('export_done', None)) # ---------- 定时自动导出 ---------- def _start_auto_thread(self): self.auto_stop.clear() self.auto_thread = threading.Thread(target=self._auto_loop, daemon=True) self.auto_thread.start() def _auto_loop(self): last_trigger = None while not self.auto_stop.is_set(): try: if self.auto_var.get(): now = time.localtime() h = int(self.hour_var.get() or 0) % 24 m = int(self.minute_var.get() or 0) % 60 today_key = f"{now.tm_yday}-{h:02d}:{m:02d}" if now.tm_hour == h and now.tm_min == m and last_trigger != today_key: last_trigger = today_key self.msg_queue.put(('auto_trigger', None)) except Exception: pass time.sleep(15) def on_toggle_auto(self): self.save_config() if self.auto_var.get(): self.log(f'[+] 已开启每日定时导出: {self.hour_var.get()}:{self.minute_var.get()}') else: self.log('[-] 已关闭每日定时导出') # ---------- 辅助 ---------- def get_db_dir(self): """返回当前生效的企业微信数据目录 (留空则用默认位置)""" d = self.db_dir_var.get().strip() return d or DEFAULT_DB_BASE def on_pick_db_dir(self): d = filedialog.askdirectory(initialdir=self.get_db_dir(), title="选择企业微信数据目录 (含账号子目录的 WXWork 文件夹)") if d: self.db_dir_var.set(d) self.save_config() self.db_dir_hint.configure(text=f"✓ {d}", fg='#16a34a') self.log(f'[*] 企业微信数据目录已更改为: {d}') def on_detect_db_dir(self): self.log('[*] 正在自动检测企业微信数据目录...') found = exporter.detect_wxwork_dir([self.db_dir_var.get().strip()]) if found: self.db_dir_var.set(found) self.save_config() self.db_dir_hint.configure(text=f"✓ {found}", fg='#16a34a') self.log(f'[+] 自动检测到企业微信数据目录: {found}') else: self.db_dir_hint.configure(text="✗ 未检测到, 请手动选择", fg='#d97706') self.log('[-] 未检测到企业微信数据目录, 请手动点击「浏览」选择') def on_pick_dir(self): d = filedialog.askdirectory(initialdir=self.out_var.get()) if d: self.out_var.set(d) self.save_config() self.log(f'[*] 输出目录已更改为: {d}') def on_open_out(self): out = self.out_var.get().strip() or DEFAULT_OUT try: os.makedirs(out, exist_ok=True) os.startfile(out) except Exception: messagebox.showinfo("输出目录", f"无法打开, 目录为:\n{out}") def on_clear_log(self): self.log_text.configure(state='normal') self.log_text.delete('1.0', 'end') self.log_text.configure(state='disabled') def log(self, msg): self.msg_queue.put(('log', str(msg))) def _poll_queue(self): try: while True: kind, payload = self.msg_queue.get_nowait() if kind == 'log': self._append_log(payload) elif kind == 'export_ok': self.exporting = False self.progress.stop() self.export_btn.configure(state='normal', text=self._export_btn_text()) self.status_label.configure(text="● 导出完成", fg='#b8e6c9') if messagebox.askyesno("导出完成", f"导出成功!\n\n文件已保存到:\n{payload}\n\n是否立即打开输出文件夹?"): self.on_open_out() elif kind == 'export_done': self.exporting = False self.progress.stop() self.export_btn.configure(state='normal', text=self._export_btn_text()) self.status_label.configure(text="● 就绪", fg='#cfe0ff') elif kind == 'extract_done': self.extracting = False self.extract_btn.configure(state='normal', text="重新提取密钥 (企微需在运行)") self._load_keys_display() self.log('[*] 密钥提取流程结束') elif kind == 'auto_trigger': self._append_log('') self._append_log('=' * 50) days = self.days_var.get() if days <= 0: scope = '全部历史' elif days == 1: scope = '今天' else: scope = f'最近 {days} 天' self._append_log(f'[定时] {time.strftime("%H:%M")} 触发自动导出 (范围: {scope})') if not self.exporting: self.on_export() except queue.Empty: pass self.root.after(100, self._poll_queue) def _append_log(self, msg): self.log_text.configure(state='normal') self.log_text.insert('end', msg + '\n') self.log_text.see('end') self.log_text.configure(state='disabled') def on_close(self): self.auto_stop.set() self.save_config() self.root.destroy() def main(): root = tk.Tk() app = ExporterApp(root) root.protocol("WM_DELETE_WINDOW", app.on_close) root.mainloop() if __name__ == '__main__': main()