From e826b1a565c563d2f67159b92c1eae717a8324d0 Mon Sep 17 00:00:00 2001 From: ylytdeng Date: Sun, 17 May 2026 19:22:47 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20=E5=88=A0=E9=99=A4=20tkinter=20app?= =?UTF-8?q?=5Fgui.py,=20=E5=AE=8C=E5=85=A8=E5=88=87=E5=88=B0=20Web=20UI=20?= =?UTF-8?q?=E4=BD=9C=E4=B8=BA=E5=94=AF=E4=B8=80=E5=85=A5=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 决策背景 用户反馈 "windows GUI 太难看了, 还是完全用 webUI 吧, 用户可以选择"。 tkinter app_gui.py (PR #107 引入, 929 行) 的问题: - 中文字体下渲染糊 ("WeChat Decrypt 工具箱" 标题模糊) - 90 年代 Windows 控件风格, 不暗色不现代 - 只能跑 Windows, 不跨平台 - 没法远程访问 - 跟 Web UI 维护两套, 重复 Web UI (monitor_web.py) 已经完全对齐功能: - 8 个工具按钮 (3 tab 分组: 个人微信 / 企微 / 工具) - 终止按钮 + 实时日志推送 - 导出筛选模态框 (会话搜索/复选/格式选择, close #112) - 跟实时消息监听共享 SSE 通道 - Lucide SVG icon 统一风格 (无 emoji) - 暗色主题 + design tokens + 玻璃质感顶栏 ## 改动 ### 删除 - \`app_gui.py\` (929 行 tkinter GUI) ### monitor_web.py 拆 \`main()\` → \`_start_monitor_if_ready()\` + 精简的 \`main()\`: 之前: keys 不存在 → \`sys.exit(1)\` 直接挂 现在: keys 不存在 → 跳过监听线程启动, 仅起 Web UI 服务。 用户从工具箱点 "① 提取密钥 + 解密数据库" 跑完后重启进程, 监听自动激活。 这样 exe 用户第一次双击时不会报错挂掉, 而是看到 Web UI 工具箱可以 直接用。 新启动流程: \`_start_monitor_if_ready\` 检查 keys 文件 / session.db 密钥 / session.db 路径都 OK 才启 monitor_thread, 任一不满足都给友好 提示但不退出。 ### WeChatDecrypt.spec - 入口 \`app_gui.py\` → \`monitor_web.py\` - datas 清单补全 (加 export_all_chats / chat_export_helpers / batch_decrypt_images / transcribe_chat 等之前漏的) - hiddenimports 显式列 Crypto / zstandard / pilk (避免 PyInstaller 漏 detect) ### build.bat - 删 30 行重复的 --add-data 清单 (跟 .spec 漂移风险) - 改成 \`pyinstaller --noconfirm WeChatDecrypt.spec\` - 单一 source of truth 是 .spec - 完成提示从 "GUI 启动" 改成 "双击 → 浏览器开 Web UI" ### EXE_USAGE.md 完全重写为 Web UI 视角: - 快速开始: 双击 exe → 浏览器自动打开 - 工具箱 3 tab 各自能力详解 - 导出筛选模态框使用说明 - 任务终止说明 - 输出目录布局 - 远程访问说明 - 结尾解释为什么去掉了 tkinter ### README.md - 三平台 quick-start 里 "Windows GUI / EXE" → "Windows Web UI / EXE" - 文件清单: \`app_gui.py\` 那行 → \`monitor_web.py\` (新身份: Web UI 总入口) - 技术细节里 "GUI 工具箱" 章节重写: 强调 Web UI 优势 (筛选模态框 / 终止按钮 / 跨平台 / 远程访问 / 跟监听共存) + 末尾 1 句历史说明 解释 tkinter 已删除 ## 实测 - python monitor_web.py: keys 在 → 正常启动监听 + Web UI - python monitor_web.py: 假装 keys 不在 → 跳过监听, Web UI 仍能开, 用户从工具箱点 "① 提取密钥" 后重启即激活 - 测试 185/185 通过 ## 没改 (留 follow-up) - 实际打包 .spec 验证 (需要 Win 跑 pyinstaller, 没在 CI 跑) - monitor_web 启动后**自动**检测 keys 文件 mtime 变化重启监听, 不需要 用户手动重启进程 (现在的设计是"重启进程才激活") --- EXE_USAGE.md | 184 ++++----- README.md | 46 ++- WeChatDecrypt.spec | 54 ++- app_gui.py | 939 --------------------------------------------- build.bat | 40 +- monitor_web.py | 57 ++- 6 files changed, 213 insertions(+), 1107 deletions(-) delete mode 100644 app_gui.py diff --git a/EXE_USAGE.md b/EXE_USAGE.md index 649dc1a..020bd44 100644 --- a/EXE_USAGE.md +++ b/EXE_USAGE.md @@ -1,125 +1,101 @@ -# WeChat Decrypt 工具箱 使用说明 +# WeChatDecrypt 使用说明 (Web UI 版) ## 快速开始 -1. **启动微信**并登录账号；如果要解密企业微信，也请先启动企业微信 -2. 双击 `WeChatDecrypt.exe` 打开工具箱 -3. 根据需要点击按钮： - - **① 微信解密** → 从微信进程提取密钥并解密数据库到 `decrypted/` 目录 - - **② 图片密钥** → 从微信进程提取新版图片 AES 密钥 - - **③ 导出数据** → 将聊天记录导出为 CSV / HTML / JSON 到 `wechat_files//<联系人>/` 目录 - - **④ 朋友圈图片** → 解密朋友圈缓存图片 - - **⑤ 企业微信解密** → 从企业微信进程提取密钥并解密数据库到 `wxwork_decrypted/` 目录 - - **⑥ 企业微信导出** → 选择某个人或群，导出 CSV / HTML / JSON 到 `wxwork_export/` 目录 +1. **启动微信** (个人微信 / 企业微信, 哪个想解密就启动哪个) +2. **双击 `WeChatDecrypt.exe`** +3. 浏览器**自动打开** `http://localhost:5678` (没自动开就手动复制粘贴) +4. 右上角点 **🛠️ 工具** 展开工具箱, 按 tab 切到你要的板块 + +## 工具箱 3 个 tab + +### 📱 个人微信 + +| 步骤 | 操作 | +|---|---| +| Step 1 — 解密 | ① 提取密钥 + 解密数据库 / ② 提取图片密钥 | +| Step 2 — 导出/解码 | ③ 导出聊天 (弹模态框选会话+格式) / ④ 批量解密图片 / ⑤ 朋友圈解密+导出 | + +**前置**: 微信 PC 版正在运行且已登录 + +### 🏢 企业微信 + +| 步骤 | 操作 | +|---|---| +| Step 1 — 解密 | ① 提取密钥 + 解密数据库 | +| Step 2 — 导出 | ② 导出聊天 (弹模态框选会话+CSV/HTML/JSON) | + +**前置**: 企业微信 PC 版正在运行且已登录 (独立于个人微信) + +### 🔧 工具 + +跟微信/企微进程无关, 只读已解密产物: +- 语音转 MP3 (需 [ffmpeg](https://ffmpeg.org/) 在 PATH) + +## 实时消息监听 + +- 工具箱下方就是消息流, 按时间降序排列 (最新在顶) +- SSE 推送, 毫秒级延迟 +- 图片自动解密预览, 表情/链接/转账等富媒体内联渲染 +- 右上角 **⚙️** 配置消息通知规则 (按群名/发送人匹配, 桌面通知 + 声音) + +## 导出筛选 (重要) + +点 ③ 导出聊天 / ② 企微导出 后, **会弹模态框**: + +- 🔍 搜索框按会话名/wxid 过滤 +- 复选框选要导的会话 (3142 个个人微信会话 / 14 个企微会话按时间降序) +- 一键 [全选] / [清空] / [选最近 30 天活跃] +- 选格式 (CSV / HTML / JSON, 企微支持; 个人微信目前只输出 JSON) +- 点 [确认导出 →] 才真正跑 + +**不会再"一点就跑全量"**。 + +## 任务终止 + +任何任务跑起来后, 触发的按钮会变成红色 **🛑 终止**。再点一下立刻 SIGTERM/kill 子进程。 ## 前置要求 - Windows 10 / 11 -- 微信 PC 版已登录（解密微信时需要微信进程运行） -- 企业微信 PC 版已登录（解密企业微信时需要企业微信进程运行） -- [FFmpeg](https://ffmpeg.org/download.html) 已安装并加入 PATH（转换音频需要） +- 微信 / 企业微信 PC 版已登录 (跑解密前需要进程在运行) +- [FFmpeg](https://ffmpeg.org/download.html) 已安装并加入 PATH (仅"语音转 MP3"需要) -### 检查 FFmpeg +## 输出目录 -打开命令提示符，输入： -``` -ffmpeg -version -``` -如果提示"不是内部或外部命令"，需要先安装 FFmpeg。 - -## 输出目录说明 - -运行后在 exe 所在目录下生成以下文件夹： +在 exe 所在目录下生成: ``` WeChatDecrypt.exe -config.json ← 首次运行自动生成的配置文件 -decrypted/ ← ① 解密后的数据库文件 -wxwork_decrypted/ ← ⑤ 解密后的企业微信数据库文件 -wxwork_export/ ← ⑥ 导出的企业微信聊天记录 - 群名_R_123/ - .info - messages.csv +config.json ← 首次运行自动生成 +decrypted/ ← 个人微信解密后的 SQLite 数据库 +wxwork_decrypted/ ← 企业微信解密后的 SQLite 数据库 +wxwork_keys.json ← 企微 keys (含明文 raw key, 已 chmod 0600) +all_keys.json ← 个人微信 keys (同上) +wechat_files// ← 导出的聊天记录 (按 wxid + 联系人分子目录) + 张三/ + messages.csv ← (用 export_messages.py 时) messages.html messages.json -wechat_files/ ← ③ 导出的聊天记录 (按 wxid + 联系人组织) - / - 张三/ - .info ← 联系人信息（username/alias/remark/nick_name） - messages.csv ← CSV 格式（Excel 可直接打开） - messages.html ← HTML 格式（浏览器打开，微信气泡样式） - messages.json ← JSON 格式（程序处理用） - image/ ← 该联系人聊天涉及的图片 - 朋友圈图片/ ← ④ 解密后的朋友圈缓存图片 - 李四/ - ... + 朋友圈图片/ ← 朋友圈缓存图片解密后 + data/ ← 语音转 MP3 输出 (有的话) +exported_chats/ ← 用 ③ 导出全部聊天 (JSON) 时的输出 (export_all_chats.py) +wxwork_export/ ← 企微聊天导出 ``` -## 导出格式说明 +## 远程访问 / 多设备 -### CSV -- 编码：UTF-8 with BOM，Excel 双击即可正确显示中文 -- 字段：时间、发送者、消息类型、内容、server_id +monitor_web 默认 bind `0.0.0.0`, 同局域网其他设备可以访问 +`http://<你的本机 IP>:5678`。如需要只允许本机访问, 修改源码 `PORT` 那一行附近的 bind 地址改成 `127.0.0.1`。 -### HTML -- 浏览器打开，模拟微信聊天界面 -- 左侧气泡为接收消息，右侧为发送消息 -- 按日期自动分组 +## 历史: 为什么没有 tkinter GUI 了 -### JSON -- 完整结构化数据，包含所有元信息 -- 适合程序二次处理或 AI 训练 +旧版本 (commit b986413 ~ a01d326) 提供过 `app_gui.py` tkinter 桌面 GUI, 但: -## 配置文件 +- tkinter 在中文字体下渲染糊 +- 90 年代 Windows 控件风格, 难看 +- 只能跑 Windows, 不跨平台 +- 没法远程访问 +- 维护两套 UI (tkinter + Web) 代码重复 -首次运行会自动检测微信数据目录并生成 `config.json`： - -```json -{ - "db_dir": "D:\\xwechat_files\\wxid_xxx\\db_storage", - "keys_file": "all_keys.json", - "decrypted_dir": "decrypted", - "wechat_process": "Weixin.exe", - "wxwork_db_dir": "C:\\Users\\<用户>\\Documents\\WXWork\\\\Data", - "wxwork_keys_file": "wxwork_keys.json", - "wxwork_decrypted_dir": "wxwork_decrypted", - "wxwork_export_dir": "wxwork_export" -} -``` - -如果自动检测失败，请手动修改 `db_dir` 为你的微信数据目录。 -路径可在：微信设置 → 文件管理 中找到。 - -## 常见问题 - -**Q: 点击"解密数据库"提示未检测到微信进程** -A: 请确保微信 PC 版已启动并登录，然后重试。 - -**Q: 解密失败 / 密钥提取失败** -A: 检查 `config.json` 中的 `db_dir` 是否与当前登录的微信账号匹配。切换账号后需要删除 `all_keys.json` 重新提取。 - -**Q: 企业微信解密失败 / 找不到企业微信数据目录** -A: 确认企业微信 PC 版已启动并登录。若自动检测失败，请在 `config.json` 中设置 `wxwork_db_dir`，路径通常类似 `C:\Users\<用户>\Documents\WXWork\\Data`。切换企业微信账号后删除 `wxwork_keys.json` 重新提取。 - -**Q: 企业微信导出为空 / 找不到会话** -A: 先执行"⑤ 企业微信解密"，确认 `wxwork_decrypted/message.db` 和 `wxwork_decrypted/session.db` 存在，然后再执行"⑥ 企业微信导出"。 - -**Q: 转换音频没有输出** -A: 确认已安装 FFmpeg 并加入系统 PATH。确认已先执行"① 解密数据库"。 - -**Q: 导出消息为空** -A: 确认已先执行"① 解密数据库"，且 `decrypted/message/` 下有 `.db` 文件。 - -**Q: 目录名是 wxid_xxx 而不是昵称** -A: 该联系人不在通讯录中（contact.db 无记录），会使用原始 username。 - -## 自行打包 - -安装依赖后双击 `build.bat` 即可重新打包： - -``` -pip install pyinstaller pycryptodome zstandard pilk -build.bat -``` - -输出文件：`dist\WeChatDecrypt.exe` +commit (这次) 起完全切到 Web UI: 浏览器渲染清晰、跨平台、远程可访问、跟实时监听共享一套 SSE 通道, 同时没有功能损失 (8 个工具按钮 + 模态框筛选 + 终止 + 状态都在 Web 上)。 diff --git a/README.md b/README.md index 5ec2f5c..c0f2319 100644 --- a/README.md +++ b/README.md @@ -62,17 +62,21 @@ python export_all_chats.py
-Windows — GUI / EXE (推荐非技术用户) +Windows — Web UI / EXE (推荐非技术用户) ```bash -# 直接跑 GUI (开发模式) -python app_gui.py +# 直接跑 Web UI (开发模式) +python monitor_web.py # → 浏览器自动开 http://localhost:5678 # 或打包成单 exe 分发给别人 build.bat # 输出 dist/WeChatDecrypt.exe + # 双击 exe → 自动开浏览器 → Web UI 工具箱 ``` -按钮覆盖: 解密 / 导出 / 朋友圈 / 企业微信 / 语音转 MP3。详见 [EXE_USAGE.md](./EXE_USAGE.md)。 +工具箱 3 个 tab (📱 个人微信 / 🏢 企业微信 / 🔧 工具), +覆盖解密 / 导出 / 朋友圈 / 语音转 MP3 全部场景, 导出带模态框筛选 (不会动不动跑全量)。 + +实时消息监听跟工具箱在同一页面。详见 [EXE_USAGE.md](./EXE_USAGE.md)。
@@ -324,7 +328,7 @@ make help # 列出所有命令 | 文件 | 说明 | |---|---| | `main.py` | **CLI 总入口** — 子命令 `decrypt` / `export` / `all` / `status` / `decode-images` / `help` | -| `app_gui.py` | **Windows GUI** — tkinter 界面整合所有功能 (PR #107) | +| `monitor_web.py` | **Web UI 总入口** — 实时监听 + 8 个工具按钮 + 导出筛选模态框 (`python monitor_web.py` 即用, PyInstaller 打包成单 exe 给非技术用户) | | `setup.sh` | 一键安装依赖 (macOS / Linux / Windows Git Bash) | | `setup.py` | 交互式配置向导 (`python setup.py --check` 仅检查环境) | | `cleanup.py` | 磁盘清理工具 (`status` 查看用量 / `--dry-run` 预览) | @@ -381,7 +385,7 @@ make help # 列出所有命令 | 文件 | 说明 | |---|---| -| `monitor_web.py` | Web UI (浏览器实时收消息, SSE 推送, http://localhost:5678) | +| `monitor_web.py` | (见 ① 入口) — 同一个文件既是 Web UI 总入口也是实时消息监听 | | `monitor.py` | 命令行实时监听 | | `mcp_server.py` | **MCP Server** — Claude AI 查询微信数据 (含 `get_chat_history` / `decode_voice` / `decode_refer` 等 20+ 工具) | | `decode_transfer.py` | CLI: 查单条转账消息 (mcp_server `decode_transfer` 工具的命令行包装) | @@ -446,35 +450,43 @@ WCDB (微信的 SQLCipher 封装) 会在进程内存中缓存派生后的 raw ke `export_sns.py` 解析 SnsTimeLine 的 XML 时已加 **XXE 防护**(拒绝 `` / `` + 200KB 大小上限),避免恶意朋友圈 XML 通过 entity expansion 或外部实体引用执行 SSRF / 读取本地文件。`mcp_server.py` 解析其他类型 appmsg XML 同样有这层保护。 -### GUI 工具箱 & 单 exe 打包 +### Web UI 工具箱 & 单 exe 打包 -提供 tkinter 图形界面 (`app_gui.py`)，集成核心功能： +`monitor_web.py` 既是实时消息监听 (浏览器 Web UI), 又是工具箱总入口。 +右上角 🛠️ 工具 按钮展开 3 个 tab: -1. **解密数据库** — 调用 `main.py decrypt` -2. **导出消息** — 调用 `export_messages.py`，输出 CSV / HTML / JSON -3. **转换音频** — 调用 `voice_to_mp3.py`，SILK_V3 → MP3 -4. **企业微信解密** — 调用 `find_wxwork_keys.py` + `decrypt_wxwork_db.py` -5. **企业微信导出** — 调用 `export_wxwork_messages.py`，按个人/群导出 CSV / HTML / JSON +- **📱 个人微信**: Step 1 解密 / 图片密钥 → Step 2 导出聊天 / 批量解图片 / 朋友圈 +- **🏢 企业微信**: Step 1 解密 → Step 2 导出聊天 (CSV/HTML/JSON) +- **🔧 工具**: 语音转 MP3 等 + +特点: +- **导出筛选模态框**: 不会一点就跑全量, 弹框选会话 (含搜索/全选/选最近 30 天) +- **任务终止**: 按钮变红 🛑, 一点立刻 SIGTERM 子进程 +- **跟实时监听共存**: 同一页面下方就是消息流, 互不影响 +- **跨平台 + 远程可访问**: 浏览器渲染清晰, 默认 bind 0.0.0.0 同局域网可用 #### 直接运行 ```bash -python app_gui.py +python monitor_web.py # → 浏览器自动开 http://localhost:5678 ``` #### 打包为单 exe ```bash pip install pyinstaller -build.bat +build.bat # → dist\WeChatDecrypt.exe ``` -输出 `dist\WeChatDecrypt.exe`（约 18MB），双击即可使用，无需安装 Python。 +输出约 20MB 的单 exe, 双击即用, 无需安装 Python。 -> 转换音频需要系统安装 [FFmpeg](https://ffmpeg.org/download.html) 并加入 PATH。 +> 语音转 MP3 需要系统安装 [FFmpeg](https://ffmpeg.org/download.html) 并加入 PATH。 详细说明见 [EXE_USAGE.md](EXE_USAGE.md)。 +> **历史**: 旧版本提供过 tkinter `app_gui.py` 桌面 GUI, 在 commit 273fe65 后完全移除。 +> 原因: 渲染糊 / Windows-only / 维护两套 UI。Web UI 功能完全对齐且更好。 + ### WAL 处理 微信使用 SQLite WAL 模式，WAL 文件是**预分配固定大小** (4MB)。检测变化时： diff --git a/WeChatDecrypt.spec b/WeChatDecrypt.spec index 30fa867..5bddc07 100644 --- a/WeChatDecrypt.spec +++ b/WeChatDecrypt.spec @@ -1,15 +1,61 @@ # -*- mode: python ; coding: utf-8 -*- +# PyInstaller spec — 打包 monitor_web.py 为单 exe。 +# +# 启动后: +# 1. 起 HTTP 服务 + 监听线程 (如 keys 已就绪) +# 2. 自动开浏览器到 http://localhost:5678 +# 3. 用户在 Web UI 工具箱里操作 (解密 / 导出 / 朋友圈 / 企微 / 语音) +# +# (历史: 这个 spec 之前打包 tkinter app_gui.py, 现已删除 — 完全切到 Web UI) + from PyInstaller.utils.hooks import collect_all -datas = [('main.py', '.'), ('config.py', '.'), ('decrypt_db.py', '.'), ('export_messages.py', '.'), ('voice_to_mp3.py', '.'), ('find_all_keys.py', '.'), ('find_all_keys_windows.py', '.'), ('find_all_keys_linux.py', '.'), ('find_wxwork_keys.py', '.'), ('decrypt_wxwork_db.py', '.'), ('export_wxwork_messages.py', '.'), ('wxwork_crypto.py', '.'), ('key_scan_common.py', '.'), ('key_utils.py', '.'), ('decode_image.py', '.'), ('find_image_key.py', '.'), ('find_image_key_monitor.py', '.'), ('decrypt_sns.py', '.'), ('export_sns.py', '.'), ('monitor.py', '.'), ('monitor_web.py', '.'), ('mcp_server.py', '.'), ('config.example.json', '.')] +# 所有子进程 (工具按钮触发) 需要的脚本, 打进 exe 同目录 +datas = [ + ('main.py', '.'), + ('config.py', '.'), + ('config.example.json', '.'), + ('decrypt_db.py', '.'), + ('decode_image.py', '.'), + ('decrypt_sns.py', '.'), + ('export_sns.py', '.'), + ('export_messages.py', '.'), + ('export_all_chats.py', '.'), + ('export_chat.py', '.'), + ('chat_export_helpers.py', '.'), + ('voice_to_mp3.py', '.'), + ('transcribe_chat.py', '.'), + ('find_all_keys.py', '.'), + ('find_all_keys_windows.py', '.'), + ('find_all_keys_linux.py', '.'), + ('find_image_key.py', '.'), + ('find_image_key_monitor.py', '.'), + ('find_wxwork_keys.py', '.'), + ('decrypt_wxwork_db.py', '.'), + ('export_wxwork_messages.py', '.'), + ('wxwork_crypto.py', '.'), + ('key_scan_common.py', '.'), + ('key_utils.py', '.'), + ('batch_decrypt_images.py', '.'), + ('monitor.py', '.'), + ('mcp_server.py', '.'), +] + binaries = [] -hiddenimports = [] +hiddenimports = [ + # 显式列, 避免 PyInstaller 漏 detect 导致打包后 import 报错 + 'Crypto', 'Crypto.Cipher', 'Crypto.Cipher.AES', + 'zstandard', + 'pilk', +] + +# pilk 是 SILK 解码扩展, 需要把它的资源全收 tmp_ret = collect_all('pilk') datas += tmp_ret[0]; binaries += tmp_ret[1]; hiddenimports += tmp_ret[2] a = Analysis( - ['app_gui.py'], + ['monitor_web.py'], # ← 入口从 app_gui.py 改成 monitor_web.py pathex=[], binaries=binaries, datas=datas, @@ -36,7 +82,7 @@ exe = EXE( upx=True, upx_exclude=[], runtime_tmpdir=None, - console=True, + console=True, # 保留 console 显示后端日志 + 错误 disable_windowed_traceback=False, argv_emulation=False, target_arch=None, diff --git a/app_gui.py b/app_gui.py deleted file mode 100644 index 0bdc39a..0000000 --- a/app_gui.py +++ /dev/null @@ -1,939 +0,0 @@ -"""WeChat Decrypt GUI — 一键解密 / 导出消息 / 转换音频""" -import os -import sys -import subprocess -import threading -import sqlite3 -import hashlib -import glob as globmod -import tkinter as tk -from tkinter import ttk, scrolledtext - -# 确保工作目录为脚本所在目录（打包后也适用） -if getattr(sys, "frozen", False): - BASE_DIR = os.path.dirname(sys.executable) -else: - BASE_DIR = os.path.dirname(os.path.abspath(__file__)) -os.chdir(BASE_DIR) -os.environ["WECHAT_DECRYPT_APP_DIR"] = BASE_DIR - - -# ── 子任务入口（当以 --task 参数调用时直接执行对应脚本） ────────────────────── - -# 显式导入：让 PyInstaller 收集子脚本需要的所有依赖 -# （这些脚本通过 exec 动态加载，PyInstaller 无法自动检测） -import importlib.util # noqa: F401 - used for dynamic loading -if False: # noqa: never executed, only for PyInstaller dependency detection - import sqlite3, hashlib, csv, json, re, glob, tempfile # noqa: F401 - import xml.etree.ElementTree # noqa: F401 - import functools, platform, ctypes, ctypes.wintypes # noqa: F401 - import zstandard # noqa: F401 - import pilk # noqa: F401 - import Crypto, Crypto.Cipher, Crypto.Cipher.AES, Crypto.Util.Padding # noqa: F401 - import wxwork_crypto # noqa: F401 - import export_wxwork_messages # noqa: F401 - - -def _run_subtask(task: str): - """在子进程中被调用，直接执行对应脚本逻辑""" - # 强制 stdout/stderr 为 UTF-8 - if sys.platform == "win32": - for s in (sys.stdout, sys.stderr): - if hasattr(s, "reconfigure"): - s.reconfigure(encoding="utf-8", errors="replace") - - # onefile: _MEIPASS 临时目录; onedir: _internal/; 开发: BASE_DIR - if getattr(sys, "frozen", False): - script_dir = getattr(sys, "_MEIPASS", os.path.join(os.path.dirname(sys.executable), "_internal")) - else: - script_dir = BASE_DIR - - # 让 import 能找到脚本同目录的模块 - if script_dir not in sys.path: - sys.path.insert(0, script_dir) - if BASE_DIR not in sys.path: - sys.path.insert(0, BASE_DIR) - - mapping = { - "decrypt": "main.py", - "export": "export_messages.py", - "voice": "voice_to_mp3.py", - "find_image_key": "find_image_key.py", - "decrypt_sns": "decrypt_sns.py", - "export_sns": "export_sns.py", - "find_wxwork_keys": "find_wxwork_keys.py", - "decrypt_wxwork": "decrypt_wxwork_db.py", - "export_wxwork": "export_wxwork_messages.py", - } - script = mapping.get(task) - if not script: - print(f"未知任务: {task}", flush=True) - sys.exit(1) - - script_path = os.path.join(script_dir, script) - if not os.path.exists(script_path): - # 开发模式回退到 BASE_DIR - script_path = os.path.join(BASE_DIR, script) - if not os.path.exists(script_path): - print(f"脚本不存在: {script_path}", flush=True) - sys.exit(1) - - # 将 decrypt 命令传给 main.py - if task == "decrypt": - sys.argv = ["main.py", "decrypt"] - elif task == "find_image_key": - sys.argv = ["find_image_key.py"] - elif task == "find_wxwork_keys": - sys.argv = ["find_wxwork_keys.py"] - elif task == "decrypt_wxwork": - sys.argv = ["decrypt_wxwork_db.py"] - elif task == "export_wxwork": - sys.argv = ["export_wxwork_messages.py"] - else: - sys.argv = [script] - - # 设置环境变量，让 config.py 等脚本知道真正的应用目录 - os.environ["WECHAT_DECRYPT_APP_DIR"] = BASE_DIR - os.chdir(BASE_DIR) - - # 加载并执行脚本 - spec = importlib.util.spec_from_file_location("__main__", script_path) - mod = importlib.util.module_from_spec(spec) - mod.__name__ = "__main__" - spec.loader.exec_module(mod) - - -# ── 检查是否为子任务模式 ────────────────────────────────────────────────────── -if len(sys.argv) >= 3 and sys.argv[1] == "--task": - _run_subtask(sys.argv[2]) - sys.exit(0) - -# ── GUI 模式：隐藏控制台窗口 ──────────────────────────────────────────────── -if sys.platform == "win32": - try: - import ctypes - ctypes.windll.user32.ShowWindow(ctypes.windll.kernel32.GetConsoleWindow(), 0) - except Exception: - pass - - -# ── 联系人发现 ──────────────────────────────────────────────────────────────── - -def _load_contact_map(decrypted_dir): - """从 contact.db 加载联系人映射 {username: {remark, nick_name, ...}}""" - contact_map = {} - db_path = os.path.join(decrypted_dir, "contact", "contact.db") - if not os.path.exists(db_path): - return contact_map - try: - conn = sqlite3.connect(db_path) - for uname, alias, remark, nick_name in conn.execute( - "SELECT username, alias, remark, nick_name FROM contact" - ): - contact_map[uname] = { - "remark": remark or "", - "nick_name": nick_name or "", - } - conn.close() - except Exception: - pass - return contact_map - - -def _display_name(username, contact_map): - info = contact_map.get(username, {}) - return info.get("remark") or info.get("nick_name") or username - - -def _discover_contacts(): - """扫描所有联系人/会话，返回 (contacts, has_voice) - contacts: [(username, display_name), ...] - has_voice: 是否存在语音数据 - """ - from config import load_config - cfg = load_config() - decrypted_dir = cfg["decrypted_dir"] - - if not os.path.isdir(decrypted_dir): - raise FileNotFoundError(f"解密目录不存在: {decrypted_dir}\n请先运行「解密数据库」") - - contact_map = _load_contact_map(decrypted_dir) - usernames = set() - has_voice = False - - # 从消息数据库扫描 - msg_dir = os.path.join(decrypted_dir, "message") - if os.path.isdir(msg_dir): - db_files = [ - f for f in globmod.glob(os.path.join(msg_dir, "message_*.db")) - if not f.endswith(("_fts.db", "_resource.db")) - ] - print(f"找到 {len(db_files)} 个消息数据库", flush=True) - for db_path in db_files: - try: - conn = sqlite3.connect(db_path) - hash_to_uname = {} - for row in conn.execute("SELECT rowid, user_name FROM Name2Id"): - uname = row[1] - if uname: - h = hashlib.md5(uname.encode()).hexdigest() - hash_to_uname[h] = uname - for (tbl,) in conn.execute( - "SELECT name FROM sqlite_master WHERE type='table' AND name LIKE 'Msg_%'" - ): - h = tbl[4:] - uname = hash_to_uname.get(h) - if uname: - usernames.add(uname) - conn.close() - except Exception as e: - print(f" 读取 {os.path.basename(db_path)} 失败: {e}", flush=True) - continue - else: - print(f"消息目录不存在: {msg_dir}", flush=True) - - # 从语音数据库扫描 - voice_db = os.path.join(msg_dir, "media_0.db") - if os.path.exists(voice_db): - try: - conn = sqlite3.connect(voice_db) - name_map = {} - for rowid, uname in conn.execute("SELECT rowid, user_name FROM Name2Id"): - name_map[rowid] = uname - for (cid,) in conn.execute("SELECT DISTINCT chat_name_id FROM VoiceInfo"): - uname = name_map.get(cid) - if uname: - usernames.add(uname) - has_voice = True - conn.close() - except Exception as e: - print(f" 读取语音数据库失败: {e}", flush=True) - - print(f"共发现 {len(usernames)} 个会话", flush=True) - result = [(u, _display_name(u, contact_map)) for u in usernames] - result.sort(key=lambda x: x[1].lower()) - return result, has_voice - - -# ── 导出选项对话框 ────────────────────────────────────────────────────────── - -class ExportOptionsDialog(tk.Toplevel): - def __init__(self, parent, contacts, has_voice=False): - """contacts: [(username, display_name), ...] - has_voice: 是否检测到语音数据 - """ - super().__init__(parent) - self.title("导出选项") - self.geometry("460x600") - self.transient(parent) - self.grab_set() - self.result = None - self.configure(bg="#f0f0f0") - self._contacts = contacts - self._vars = {} # username -> BooleanVar - - # ── 导出格式 ── - fmt_frame = ttk.LabelFrame(self, text="导出格式", padding=6) - fmt_frame.pack(fill="x", padx=12, pady=(10, 4)) - - self._fmt_csv = tk.BooleanVar(value=True) - self._fmt_html = tk.BooleanVar(value=False) - self._fmt_json = tk.BooleanVar(value=False) - - ttk.Checkbutton(fmt_frame, text="CSV（默认）", variable=self._fmt_csv).pack(side="left", padx=10) - ttk.Checkbutton(fmt_frame, text="HTML", variable=self._fmt_html).pack(side="left", padx=10) - ttk.Checkbutton(fmt_frame, text="JSON", variable=self._fmt_json).pack(side="left", padx=10) - - # ── 其他选项 ── - opt_frame = ttk.LabelFrame(self, text="其他选项", padding=6) - opt_frame.pack(fill="x", padx=12, pady=(0, 4)) - - self._image_var = tk.BooleanVar(value=True) - ttk.Checkbutton(opt_frame, text="导出并解密图片", - variable=self._image_var).pack(anchor="w", padx=8) - - self._sns_var = tk.BooleanVar(value=False) - ttk.Checkbutton(opt_frame, text="导出朋友圈动态（文案/评论）", - variable=self._sns_var).pack(anchor="w", padx=8) - - self._sns_media_var = tk.BooleanVar(value=False) - ttk.Checkbutton(opt_frame, text=" ↳ 尝试下载朋友圈媒体（可能较慢）", - variable=self._sns_media_var).pack(anchor="w", padx=24) - - self._voice_var = tk.BooleanVar(value=False) - if has_voice: - ttk.Checkbutton(opt_frame, text="同时转换语音为 MP3", - variable=self._voice_var).pack(anchor="w", padx=8) - - # ── 联系人选择 ── - top = ttk.Frame(self) - top.pack(fill="x", padx=12, pady=(4, 4)) - - ttk.Label(top, text=f"共 {len(contacts)} 个会话", - font=("Microsoft YaHei UI", 10)).pack(side="left") - - self._all_selected = True - self._toggle_btn = ttk.Button(top, text="取消全选", command=self._toggle_all) - self._toggle_btn.pack(side="right") - - # 搜索框 - search_frame = ttk.Frame(self) - search_frame.pack(fill="x", padx=12, pady=(0, 4)) - self._search_var = tk.StringVar() - self._search_var.trace_add("write", lambda *_: self._filter_list()) - ttk.Entry(search_frame, textvariable=self._search_var, - font=("Microsoft YaHei UI", 10)).pack(fill="x") - - # 滚动区域 - container = ttk.Frame(self) - container.pack(fill="both", expand=True, padx=12, pady=4) - - self._canvas = tk.Canvas(container, bg="#ffffff", highlightthickness=0) - scrollbar = ttk.Scrollbar(container, orient="vertical", command=self._canvas.yview) - self._inner = ttk.Frame(self._canvas) - - self._inner.bind("", - lambda e: self._canvas.configure(scrollregion=self._canvas.bbox("all"))) - self._canvas.create_window((0, 0), window=self._inner, anchor="nw") - self._canvas.configure(yscrollcommand=scrollbar.set) - - self._canvas.pack(side="left", fill="both", expand=True) - scrollbar.pack(side="right", fill="y") - - # 绑定鼠标滚轮 - self._canvas.bind("", lambda e: self._bind_mousewheel()) - self._canvas.bind("", lambda e: self._unbind_mousewheel()) - - # 创建 Checkbutton 列表 - self._cb_widgets = [] - for username, dname in contacts: - var = tk.BooleanVar(value=True) - self._vars[username] = var - label = f"{dname} ({username})" if dname != username else username - cb = ttk.Checkbutton(self._inner, text=label, variable=var) - cb.pack(anchor="w", padx=6, pady=1) - self._cb_widgets.append((username, dname, cb)) - - # 底部按钮 - bottom = ttk.Frame(self) - bottom.pack(fill="x", padx=12, pady=(4, 10)) - - ttk.Button(bottom, text="确定", command=self._on_ok).pack(side="right", padx=4) - ttk.Button(bottom, text="取消", command=self._on_cancel).pack(side="right", padx=4) - - def _bind_mousewheel(self): - self._canvas.bind_all("", - lambda e: self._canvas.yview_scroll(-1 * (e.delta // 120), "units")) - - def _unbind_mousewheel(self): - self._canvas.unbind_all("") - - def _toggle_all(self): - self._all_selected = not self._all_selected - for var in self._vars.values(): - var.set(self._all_selected) - self._toggle_btn.configure(text="取消全选" if self._all_selected else "全选") - - def _filter_list(self): - keyword = self._search_var.get().strip().lower() - for username, dname, cb in self._cb_widgets: - if not keyword or keyword in dname.lower() or keyword in username.lower(): - cb.pack(anchor="w", padx=6, pady=1) - else: - cb.pack_forget() - - def _on_ok(self): - formats = [] - if self._fmt_csv.get(): - formats.append("csv") - if self._fmt_html.get(): - formats.append("html") - if self._fmt_json.get(): - formats.append("json") - - if not formats and not self._voice_var.get() and not self._sns_var.get(): - from tkinter import messagebox - messagebox.showwarning("提示", "请至少选择一种导出格式、朋友圈导出或语音转换", parent=self) - return - - self.result = { - "contacts": [u for u, var in self._vars.items() if var.get()], - "formats": formats, - "include_voice": self._voice_var.get(), - "include_images": self._image_var.get(), - "include_sns": self._sns_var.get(), - "include_sns_media": self._sns_media_var.get(), - } - self.destroy() - - def _on_cancel(self): - self.result = None - self.destroy() - - -class WxworkExportOptionsDialog(tk.Toplevel): - def __init__(self, parent, conversations): - """conversations: [{conversation_id, display_name, kind, message_count, last_time}, ...]""" - super().__init__(parent) - self.title("企业微信导出选项") - self.geometry("560x620") - self.transient(parent) - self.grab_set() - self.result = None - self.configure(bg="#f0f0f0") - self._conversations = conversations - self._vars = {} - - fmt_frame = ttk.LabelFrame(self, text="导出格式", padding=6) - fmt_frame.pack(fill="x", padx=12, pady=(10, 4)) - - self._fmt_csv = tk.BooleanVar(value=True) - self._fmt_html = tk.BooleanVar(value=False) - self._fmt_json = tk.BooleanVar(value=False) - - ttk.Checkbutton(fmt_frame, text="CSV（默认）", variable=self._fmt_csv).pack(side="left", padx=10) - ttk.Checkbutton(fmt_frame, text="HTML", variable=self._fmt_html).pack(side="left", padx=10) - ttk.Checkbutton(fmt_frame, text="JSON", variable=self._fmt_json).pack(side="left", padx=10) - - top = ttk.Frame(self) - top.pack(fill="x", padx=12, pady=(4, 4)) - ttk.Label(top, text=f"共 {len(conversations)} 个企业微信会话", - font=("Microsoft YaHei UI", 10)).pack(side="left") - - self._all_selected = True - self._toggle_btn = ttk.Button(top, text="取消全选", command=self._toggle_all) - self._toggle_btn.pack(side="right") - - search_frame = ttk.Frame(self) - search_frame.pack(fill="x", padx=12, pady=(0, 4)) - self._search_var = tk.StringVar() - self._search_var.trace_add("write", lambda *_: self._filter_list()) - ttk.Entry(search_frame, textvariable=self._search_var, - font=("Microsoft YaHei UI", 10)).pack(fill="x") - - container = ttk.Frame(self) - container.pack(fill="both", expand=True, padx=12, pady=4) - - self._canvas = tk.Canvas(container, bg="#ffffff", highlightthickness=0) - scrollbar = ttk.Scrollbar(container, orient="vertical", command=self._canvas.yview) - self._inner = ttk.Frame(self._canvas) - self._inner.bind("", - lambda e: self._canvas.configure(scrollregion=self._canvas.bbox("all"))) - self._canvas.create_window((0, 0), window=self._inner, anchor="nw") - self._canvas.configure(yscrollcommand=scrollbar.set) - self._canvas.pack(side="left", fill="both", expand=True) - scrollbar.pack(side="right", fill="y") - - self._canvas.bind("", lambda e: self._bind_mousewheel()) - self._canvas.bind("", lambda e: self._unbind_mousewheel()) - - self._cb_widgets = [] - for conv in conversations: - cid = conv["conversation_id"] - var = tk.BooleanVar(value=True) - self._vars[cid] = var - last_time = self._format_time(conv.get("last_time")) - suffix = f" · {last_time}" if last_time else "" - label = ( - f"[{conv.get('kind', '会话')}] {conv.get('display_name') or cid}" - f" · {conv.get('message_count', 0)} 条{suffix}" - ) - cb = ttk.Checkbutton(self._inner, text=label, variable=var) - cb.pack(anchor="w", padx=6, pady=1) - self._cb_widgets.append((cid, label.lower(), cb)) - - bottom = ttk.Frame(self) - bottom.pack(fill="x", padx=12, pady=(4, 10)) - ttk.Button(bottom, text="确定", command=self._on_ok).pack(side="right", padx=4) - ttk.Button(bottom, text="取消", command=self._on_cancel).pack(side="right", padx=4) - - def _format_time(self, value): - if not value: - return "" - try: - from datetime import datetime - return datetime.fromtimestamp(int(value)).strftime("%Y-%m-%d") - except Exception: - return "" - - def _bind_mousewheel(self): - self._canvas.bind_all("", - lambda e: self._canvas.yview_scroll(-1 * (e.delta // 120), "units")) - - def _unbind_mousewheel(self): - self._canvas.unbind_all("") - - def _toggle_all(self): - self._all_selected = not self._all_selected - for var in self._vars.values(): - var.set(self._all_selected) - self._toggle_btn.configure(text="取消全选" if self._all_selected else "全选") - - def _filter_list(self): - keyword = self._search_var.get().strip().lower() - for _cid, label, cb in self._cb_widgets: - if not keyword or keyword in label: - cb.pack(anchor="w", padx=6, pady=1) - else: - cb.pack_forget() - - def _on_ok(self): - formats = [] - if self._fmt_csv.get(): - formats.append("csv") - if self._fmt_html.get(): - formats.append("html") - if self._fmt_json.get(): - formats.append("json") - if not formats: - from tkinter import messagebox - messagebox.showwarning("提示", "请至少选择一种导出格式", parent=self) - return - self.result = { - "conversations": [cid for cid, var in self._vars.items() if var.get()], - "formats": formats, - } - self.destroy() - - def _on_cancel(self): - self.result = None - self.destroy() - - -class App(tk.Tk): - def __init__(self): - super().__init__() - self.title("WeChat Decrypt 工具箱") - self.geometry("820x600") - self.resizable(True, True) - self.configure(bg="#f0f0f0") - self._running = False - self._auto_export = False - self._selected_contacts = None - self._export_formats = None - self._include_voice = False - self._include_images = True - self._include_sns = False - self._include_sns_media = False - self._selected_wxwork_conversations = None - self._wxwork_export_formats = None - - self._build_ui() - - # ── UI 构建 ──────────────────────────────────────────────────────────── - def _build_ui(self): - style = ttk.Style(self) - style.theme_use("clam") - style.configure("Big.TButton", font=("Microsoft YaHei UI", 11), padding=(16, 10)) - style.configure("TLabel", font=("Microsoft YaHei UI", 10), background="#f0f0f0") - - # 标题 - title = ttk.Label(self, text="WeChat Decrypt 工具箱", font=("Microsoft YaHei UI", 16, "bold")) - title.pack(pady=(14, 6)) - - # 按钮区域 - btn_frame = ttk.Frame(self) - btn_frame.pack(fill="x", padx=20, pady=(4, 4)) - - self.btn_decrypt = ttk.Button( - btn_frame, text="① 微信解密", style="Big.TButton", - command=lambda: self._run_task("decrypt") - ) - self.btn_decrypt.pack(side="left", expand=True, fill="x", padx=4) - - self.btn_imgkey = ttk.Button( - btn_frame, text="② 图片密钥", style="Big.TButton", - command=lambda: self._run_task("find_image_key") - ) - self.btn_imgkey.pack(side="left", expand=True, fill="x", padx=4) - - self.btn_export = ttk.Button( - btn_frame, text="③ 导出数据", style="Big.TButton", - command=lambda: self._run_task("export") - ) - self.btn_export.pack(side="left", expand=True, fill="x", padx=4) - - self.btn_sns = ttk.Button( - btn_frame, text="④ 朋友圈图片", style="Big.TButton", - command=lambda: self._run_task("decrypt_sns") - ) - self.btn_sns.pack(side="left", expand=True, fill="x", padx=4) - - wxwork_frame = ttk.Frame(self) - wxwork_frame.pack(fill="x", padx=20, pady=(0, 6)) - - self.btn_wxwork = ttk.Button( - wxwork_frame, text="⑤ 企业微信解密", style="Big.TButton", - command=lambda: self._run_task("wxwork_decrypt") - ) - self.btn_wxwork.pack(side="left", expand=True, fill="x", padx=4) - - self.btn_wxwork_export = ttk.Button( - wxwork_frame, text="⑥ 企业微信导出", style="Big.TButton", - command=lambda: self._run_task("wxwork_export") - ) - self.btn_wxwork_export.pack(side="left", expand=True, fill="x", padx=4) - - # 提示信息 - tips_frame = ttk.LabelFrame(self, text="使用提示", padding=6) - tips_frame.pack(fill="x", padx=20, pady=(0, 4)) - tips_text = ( - "• 微信解密：需要微信正在运行中，会自动提取密钥并解密\n" - "• 查找图片密钥：先在微信中打开 2-3 张图片查看，然后立即运行\n" - "• 导出数据：选择联系人和格式，可同时导出消息/图片/语音\n" - "• 朋友圈图片：解密朋友圈缓存图片（_t缩略图自动跳过）\n" - "• 企业微信解密：需要企业微信正在运行中，输出到 wxwork_decrypted/\n" - "• 企业微信导出：选择某个人或群，输出 CSV / HTML / JSON 到 wxwork_export/" - ) - ttk.Label(tips_frame, text=tips_text, font=("Microsoft YaHei UI", 9), - wraplength=760, justify="left").pack(anchor="w") - - # 进度条 - self.progress = ttk.Progressbar(self, mode="indeterminate") - self.progress.pack(fill="x", padx=20, pady=(0, 4)) - - # 日志区域 - log_label = ttk.Label(self, text="运行日志：") - log_label.pack(anchor="w", padx=20) - - self.log = scrolledtext.ScrolledText( - self, wrap="word", height=18, - font=("Consolas", 10), bg="#1e1e1e", fg="#d4d4d4", - insertbackground="#fff", state="disabled" - ) - self.log.pack(fill="both", expand=True, padx=20, pady=(2, 10)) - - # 底部状态 - self.status_var = tk.StringVar(value="就绪") - status = ttk.Label(self, textvariable=self.status_var, font=("Microsoft YaHei UI", 9)) - status.pack(anchor="w", padx=20, pady=(0, 8)) - - # ── 日志写入 ─────────────────────────────────────────────────────────── - def _log(self, text: str): - self.log.configure(state="normal") - self.log.insert("end", text) - self.log.see("end") - self.log.configure(state="disabled") - - def _clear_log(self): - self.log.configure(state="normal") - self.log.delete("1.0", "end") - self.log.configure(state="disabled") - - # ── 按钮状态 ─────────────────────────────────────────────────────────── - def _set_buttons(self, enabled: bool): - state = "normal" if enabled else "disabled" - self.btn_decrypt.configure(state=state) - self.btn_imgkey.configure(state=state) - self.btn_export.configure(state=state) - self.btn_sns.configure(state=state) - self.btn_wxwork.configure(state=state) - self.btn_wxwork_export.configure(state=state) - - # ── 任务调度 ─────────────────────────────────────────────────────────── - def _run_task(self, task: str): - if self._running: - return - self._running = True - self._selected_contacts = None - self._export_formats = None - self._include_voice = False - self._include_sns = False - self._include_sns_media = False - self._selected_wxwork_conversations = None - self._wxwork_export_formats = None - self._clear_log() - self._set_buttons(False) - - if task == "export": - self.progress.start(15) - self.status_var.set("正在扫描联系人...") - threading.Thread( - target=self._discover_and_select, daemon=True - ).start() - elif task == "find_image_key": - self.progress.start(15) - self.status_var.set("正在扫描微信进程内存...") - threading.Thread(target=self._exec_task, args=(task,), daemon=True).start() - elif task == "decrypt_sns": - self.progress.start(15) - self.status_var.set("正在解密朋友圈图片...") - threading.Thread(target=self._exec_task, args=(task,), daemon=True).start() - elif task == "wxwork_decrypt": - self.progress.start(15) - self.status_var.set("正在解密企业微信数据库...") - threading.Thread(target=self._exec_wxwork_decrypt, daemon=True).start() - elif task == "wxwork_export": - self.progress.start(15) - self.status_var.set("正在扫描企业微信会话...") - threading.Thread(target=self._discover_wxwork_and_select, daemon=True).start() - else: - self.progress.start(15) - labels = {"decrypt": "解密数据库"} - self.status_var.set(f"正在{labels.get(task, task)}...") - threading.Thread(target=self._exec_task, args=(task,), daemon=True).start() - - def _discover_and_select(self): - """后台扫描联系人，然后在主线程弹出选择对话框""" - try: - contacts, has_voice = _discover_contacts() - except Exception as e: - self.after(0, self._log, f"扫描联系人失败: {e}\n") - self.after(0, self._on_task_done) - return - - if not contacts: - self.after(0, self._log, "未找到任何联系人/会话\n") - self.after(0, self._on_task_done) - return - - self.after(0, self._show_contact_dialog, contacts, has_voice) - - def _show_contact_dialog(self, contacts, has_voice): - self.progress.stop() - self.status_var.set(f"请选择导出选项 ({len(contacts)} 个会话)") - - dlg = ExportOptionsDialog(self, contacts, has_voice=has_voice) - self.wait_window(dlg) - - if dlg.result is None: - self._on_task_done() - return - - if not dlg.result["contacts"]: - self._log("未选择任何联系人\n") - self._on_task_done() - return - - self._selected_contacts = dlg.result["contacts"] - self._export_formats = dlg.result["formats"] - self._include_voice = dlg.result["include_voice"] - self._include_images = dlg.result["include_images"] - self._include_sns = dlg.result["include_sns"] - self._include_sns_media = dlg.result["include_sns_media"] - - self._clear_log() - self.progress.start(15) - n_sel = len(dlg.result["contacts"]) - parts = [] - if self._export_formats: - parts.append(f"导出 {'/'.join(f.upper() for f in self._export_formats)}") - if self._include_sns: - parts.append("朋友圈") - if self._include_voice: - parts.append("转换语音") - action = " + ".join(parts) or "处理" - self.status_var.set(f"正在{action}...（{n_sel}/{len(contacts)} 个联系人）") - threading.Thread(target=self._exec_combined, daemon=True).start() - - def _discover_wxwork_and_select(self): - """后台扫描企业微信会话，然后在主线程弹出选择对话框""" - try: - from export_wxwork_messages import discover_conversations - conversations = discover_conversations() - except Exception as e: - self.after(0, self._log, f"扫描企业微信会话失败: {e}\n") - self.after(0, self._on_task_done) - return - - if not conversations: - self.after(0, self._log, "未找到任何企业微信会话，请先运行「企业微信解密」\n") - self.after(0, self._on_task_done) - return - - self.after(0, self._show_wxwork_dialog, conversations) - - def _show_wxwork_dialog(self, conversations): - self.progress.stop() - self.status_var.set(f"请选择企业微信导出选项 ({len(conversations)} 个会话)") - - dlg = WxworkExportOptionsDialog(self, conversations) - self.wait_window(dlg) - - if dlg.result is None: - self._on_task_done() - return - - if not dlg.result["conversations"]: - self._log("未选择任何企业微信会话\n") - self._on_task_done() - return - - self._selected_wxwork_conversations = dlg.result["conversations"] - self._wxwork_export_formats = dlg.result["formats"] - - self._clear_log() - self.progress.start(15) - n_sel = len(dlg.result["conversations"]) - self.status_var.set( - f"正在导出企业微信 {'/'.join(f.upper() for f in self._wxwork_export_formats)}..." - f"（{n_sel}/{len(conversations)} 个会话）" - ) - threading.Thread(target=self._exec_wxwork_export, daemon=True).start() - - # ── 子进程执行 ───────────────────────────────────────────────────────── - def _run_subprocess(self, task: str) -> int: - """运行子进程，返回退出码 - - 打包成 exe (`sys.frozen=True`) 时 sys.executable 就是 exe 本身, - `--task` 是它的子命令,直接走 [exe, --task, ...] 即可。 - 开发模式 (python app_gui.py) 时 sys.executable 是 python 解释器, - 需要把当前脚本路径 __file__ 加上,否则 `python --task` 会被解释器 - 误判为 `python --task` (报 unknown option)。 - """ - if getattr(sys, "frozen", False): - cmd = [sys.executable, "--task", task] - else: - cmd = [sys.executable, os.path.abspath(__file__), "--task", task] - self.after(0, self._log, f">>> {' '.join(cmd)}\n\n") - - env = os.environ.copy() - env["PYTHONIOENCODING"] = "utf-8" - env["WECHAT_DECRYPT_APP_DIR"] = BASE_DIR - env["WECHAT_DECRYPT_GUI"] = "1" - env["WECHAT_DECRYPT_NONINTERACTIVE"] = "1" - - if self._selected_contacts: - env["WECHAT_EXPORT_CONTACTS"] = ",".join(self._selected_contacts) - if self._export_formats: - env["WECHAT_EXPORT_FORMATS"] = ",".join(self._export_formats) - env["WECHAT_EXPORT_IMAGES"] = "1" if getattr(self, '_include_images', True) else "0" - if getattr(self, '_include_sns_media', False): - env["WECHAT_SNS_DOWNLOAD_MEDIA"] = "1" - if self._selected_wxwork_conversations: - env["WXWORK_EXPORT_CONVERSATIONS"] = ",".join(self._selected_wxwork_conversations) - if self._wxwork_export_formats: - env["WXWORK_EXPORT_FORMATS"] = ",".join(self._wxwork_export_formats) - - proc = subprocess.Popen( - cmd, - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, - cwd=BASE_DIR, - env=env, - creationflags=subprocess.CREATE_NO_WINDOW if sys.platform == "win32" else 0, - ) - - for raw in proc.stdout: - line = raw.decode("utf-8", errors="replace") - self.after(0, self._log, line) - - proc.wait() - return proc.returncode - - def _exec_combined(self): - """执行导出（消息 + 可选语音）""" - try: - if self._export_formats: - rc = self._run_subprocess("export") - if rc != 0: - self.after(0, self._log, f"\n❌ 导出失败 (返回码 {rc})\n") - self.after(0, self.status_var.set, f"失败 (返回码 {rc})") - return - - if getattr(self, '_include_sns', False): - if self._export_formats: - self.after(0, self._log, "\n\n━━━ 开始导出朋友圈 ━━━\n\n") - rc = self._run_subprocess("export_sns") - if rc != 0: - self.after(0, self._log, f"\n❌ 朋友圈导出失败 (返回码 {rc})\n") - self.after(0, self.status_var.set, f"朋友圈导出失败 (返回码 {rc})") - return - - if self._include_voice: - if self._export_formats or getattr(self, '_include_sns', False): - self.after(0, self._log, "\n\n━━━ 开始转换语音 ━━━\n\n") - rc = self._run_subprocess("voice") - if rc != 0: - self.after(0, self._log, f"\n❌ 语音转换失败 (返回码 {rc})\n") - self.after(0, self.status_var.set, f"语音转换失败 (返回码 {rc})") - return - - self.after(0, self._log, "\n✅ 全部完成！\n") - self.after(0, self.status_var.set, "完成") - except Exception as e: - self.after(0, self._log, f"\n❌ 异常: {e}\n") - self.after(0, self.status_var.set, "异常") - finally: - self._selected_contacts = None - self._export_formats = None - self._include_voice = False - self._include_images = True - self._include_sns = False - self._include_sns_media = False - self.after(0, self._on_task_done) - - def _exec_wxwork_decrypt(self): - """执行企业微信 key 提取 + 数据库解密。""" - try: - self.after(0, self._log, "━━━ 开始提取企业微信密钥 ━━━\n\n") - rc = self._run_subprocess("find_wxwork_keys") - if rc != 0: - self.after(0, self._log, f"\n❌ 企业微信密钥提取失败 (返回码 {rc})\n") - self.after(0, self.status_var.set, f"企业微信密钥提取失败 (返回码 {rc})") - return - - self.after(0, self._log, "\n\n━━━ 开始解密企业微信数据库 ━━━\n\n") - rc = self._run_subprocess("decrypt_wxwork") - if rc != 0: - self.after(0, self._log, f"\n❌ 企业微信数据库解密失败 (返回码 {rc})\n") - self.after(0, self.status_var.set, f"企业微信解密失败 (返回码 {rc})") - return - - self.after(0, self._log, "\n✅ 企业微信解密完成！输出目录: wxwork_decrypted\n") - self.after(0, self.status_var.set, "企业微信解密完成") - except Exception as e: - self.after(0, self._log, f"\n❌ 异常: {e}\n") - self.after(0, self.status_var.set, "异常") - finally: - self.after(0, self._on_task_done) - - def _exec_wxwork_export(self): - """执行企业微信消息导出。""" - try: - rc = self._run_subprocess("export_wxwork") - if rc != 0: - self.after(0, self._log, f"\n❌ 企业微信导出失败 (返回码 {rc})\n") - self.after(0, self.status_var.set, f"企业微信导出失败 (返回码 {rc})") - return - self.after(0, self._log, "\n✅ 企业微信导出完成！输出目录: wxwork_export\n") - self.after(0, self.status_var.set, "企业微信导出完成") - except Exception as e: - self.after(0, self._log, f"\n❌ 异常: {e}\n") - self.after(0, self.status_var.set, "异常") - finally: - self._selected_wxwork_conversations = None - self._wxwork_export_formats = None - self.after(0, self._on_task_done) - - def _exec_task(self, task: str): - """执行单一任务（解密）""" - try: - rc = self._run_subprocess(task) - if rc == 0: - self.after(0, self._log, "\n✅ 完成！\n") - self.after(0, self.status_var.set, "完成") - if task == "decrypt": - self._auto_export = True - else: - self.after(0, self._log, f"\n❌ 进程退出，返回码: {rc}\n") - self.after(0, self.status_var.set, f"失败 (返回码 {rc})") - except Exception as e: - self.after(0, self._log, f"\n❌ 异常: {e}\n") - self.after(0, self.status_var.set, "异常") - finally: - self.after(0, self._on_task_done) - - def _on_task_done(self): - self._running = False - self.progress.stop() - self._set_buttons(True) - if self._auto_export: - self._auto_export = False - self._log("\n解密完成，自动进入导出流程...\n\n") - self.after(500, lambda: self._run_task("export")) - - -if __name__ == "__main__": - app = App() - app.mainloop() diff --git a/build.bat b/build.bat index 25a3c07..8462b44 100644 --- a/build.bat +++ b/build.bat @@ -1,46 +1,21 @@ @echo off chcp 65001 >nul echo ======================================== -echo WeChatDecrypt 打包脚本 +echo WeChatDecrypt 打包脚本 (Web UI 版) echo ======================================== echo. -:: 检查 pyinstaller +REM 检查 pyinstaller where pyinstaller >nul 2>&1 if errorlevel 1 ( - echo [!] 未找到 pyinstaller，正在安装... + echo [!] 未找到 pyinstaller, 正在安装... pip install pyinstaller ) -echo [*] 开始打包... +echo [*] 调用 PyInstaller (走 WeChatDecrypt.spec, 不重复维护文件清单)... echo. -pyinstaller --noconfirm --onefile --console --name "WeChatDecrypt" ^ - --add-data "main.py;." ^ - --add-data "config.py;." ^ - --add-data "decrypt_db.py;." ^ - --add-data "export_messages.py;." ^ - --add-data "voice_to_mp3.py;." ^ - --add-data "find_all_keys.py;." ^ - --add-data "find_all_keys_windows.py;." ^ - --add-data "find_all_keys_linux.py;." ^ - --add-data "find_wxwork_keys.py;." ^ - --add-data "decrypt_wxwork_db.py;." ^ - --add-data "export_wxwork_messages.py;." ^ - --add-data "wxwork_crypto.py;." ^ - --add-data "key_scan_common.py;." ^ - --add-data "key_utils.py;." ^ - --add-data "decode_image.py;." ^ - --add-data "find_image_key.py;." ^ - --add-data "find_image_key_monitor.py;." ^ - --add-data "decrypt_sns.py;." ^ - --add-data "export_sns.py;." ^ - --add-data "monitor.py;." ^ - --add-data "monitor_web.py;." ^ - --add-data "mcp_server.py;." ^ - --add-data "config.example.json;." ^ - --collect-all pilk ^ - app_gui.py +pyinstaller --noconfirm WeChatDecrypt.spec if errorlevel 1 ( echo. @@ -51,9 +26,12 @@ if errorlevel 1 ( echo. echo ======================================== -echo 打包完成！ +echo 打包完成! echo 输出: dist\WeChatDecrypt.exe for %%F in (dist\WeChatDecrypt.exe) do echo 大小: %%~zF bytes +echo. +echo 使用: 双击 WeChatDecrypt.exe +echo ^→ 自动启动浏览器到 http://localhost:5678 (Web UI) echo ======================================== echo. pause diff --git a/monitor_web.py b/monitor_web.py index 3322c2f..8b3679b 100644 --- a/monitor_web.py +++ b/monitor_web.py @@ -2973,20 +2973,43 @@ class ThreadedServer(ThreadingMixIn, HTTPServer): allow_reuse_address = True -def main(): - print("=" * 60, flush=True) - print(" 微信实时监听 (WAL增量 + SSE推送)", flush=True) - print("=" * 60, flush=True) +def _start_monitor_if_ready(): + """如果 keys 已存在且能解出 session.db, 启动监听线程; 否则跳过。 - with open(KEYS_FILE, encoding="utf-8") as f: - keys = strip_key_metadata(json.load(f)) + 用户在 Web UI 工具箱点 "① 提取密钥 + 解密数据库" 后, keys 文件就会 + 存在。届时刷新页面即可激活监听 (重启 monitor_web 或者下次访问 UI + 时再尝试启动)。 + + 返回 True = 监听已启动, False = 未启动 (UI 仍可用作工具箱)。 + """ + if not os.path.exists(KEYS_FILE): + print(f"[!] 未找到 keys 文件: {KEYS_FILE}", flush=True) + print(" Web UI 仍可用作工具箱 (从右上角 🛠️ 工具 启动 '① 提取密钥')", + flush=True) + print(" 解密完成后重启本进程, 监听会自动启动\n", flush=True) + return False + + try: + with open(KEYS_FILE, encoding="utf-8") as f: + keys = strip_key_metadata(json.load(f)) + except Exception as e: + print(f"[!] 读取 keys 文件失败: {e}", flush=True) + print(" Web UI 仍可用作工具箱\n", flush=True) + return False session_key_info = get_key_info(keys, os.path.join("session", "session.db")) if not session_key_info: - print("[ERROR] 找不到 session.db 的密钥", flush=True) - sys.exit(1) + print("[!] keys 文件里没有 session.db 密钥", flush=True) + print(" 可能 keys 是部分提取的, 工具箱 → '① 提取密钥' 重跑一次\n", + flush=True) + return False + enc_key = bytes.fromhex(session_key_info["enc_key"]) session_db = os.path.join(DB_DIR, "session", "session.db") + if not os.path.exists(session_db): + print(f"[!] session.db 不存在: {session_db}", flush=True) + print(" 检查 config.json 的 db_dir 是否对应当前微信账号\n", flush=True) + return False print("加载联系人...", flush=True) contact_names = load_contact_names() @@ -3014,7 +3037,7 @@ def main(): db_cache = MonitorDBCache(keys, MONITOR_CACHE_DIR) - # 后台预热所有 message DB（图片/emoji 解密必需） + # 后台预热所有 message DB def _warmup(): try: t0 = time.perf_counter() @@ -3032,16 +3055,26 @@ def main(): print(f"[warmup] {k} 失败: {e}", flush=True) except Exception as e: print(f"[warmup] 异常: {e}", flush=True) - # 构建 emoji 映射（独立解密，不走 cache） _build_emoji_lookup(keys) print(f"[warmup] 全部完成 {(time.perf_counter()-t0)*1000:.0f}ms", flush=True) threading.Thread(target=_warmup, daemon=True).start() - t = threading.Thread(target=monitor_thread, args=(enc_key, session_db, contact_names, db_cache, username_db_map), daemon=True) + t = threading.Thread(target=monitor_thread, + args=(enc_key, session_db, contact_names, db_cache, username_db_map), + daemon=True) t.start() + return True + + +def main(): + print("=" * 60, flush=True) + print(" WeChat Decrypt — Web UI + 实时监听", flush=True) + print("=" * 60, flush=True) + + _start_monitor_if_ready() server = ThreadedServer(('0.0.0.0', PORT), Handler) - print(f"\n=> http://localhost:{PORT}", flush=True) + print(f"=> http://localhost:{PORT}", flush=True) print("Ctrl+C 停止\n", flush=True) try: