864 lines
39 KiB
Python
864 lines
39 KiB
Python
#!/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('<KeyRelease>', 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('<<ComboboxSelected>>',
|
|
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('<KeyRelease>', 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()
|