From f3c32d4ca24b1e6af6e317fbee50bb5e4a805d63 Mon Sep 17 00:00:00 2001 From: phoenixray2000 Date: Mon, 18 May 2026 01:53:20 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=94=AF=E6=8C=81=E5=8F=8C=E6=A8=A1?= =?UTF-8?q?=E5=BC=8F=20exe=20=E5=85=A5=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- EXE_USAGE.md | 25 ++++++- README.md | 15 +++- WeChatDecrypt.spec | 19 +++-- build.bat | 3 +- config.py | 13 +++- mcp_server.py | 29 +++++--- tests/test_exe_launcher.py | 147 +++++++++++++++++++++++++++++++++++++ wechat_decrypt_launcher.py | 139 +++++++++++++++++++++++++++++++++++ 8 files changed, 363 insertions(+), 27 deletions(-) create mode 100644 tests/test_exe_launcher.py create mode 100644 wechat_decrypt_launcher.py diff --git a/EXE_USAGE.md b/EXE_USAGE.md index adf8907..a0ffbbf 100644 --- a/EXE_USAGE.md +++ b/EXE_USAGE.md @@ -1,4 +1,4 @@ -# WeChatDecrypt 使用说明 (Web UI 版) +# WeChatDecrypt 使用说明 (Web UI + CLI 双模式) ## 快速开始 @@ -7,6 +7,25 @@ 3. 浏览器**自动打开** `http://localhost:5678` (没自动开就手动复制粘贴) 4. 右上角点 **🛠️ 工具** 展开工具箱, 按 tab 切到你要的板块 +## CLI 用法 + +同一个 `WeChatDecrypt.exe` 带参数运行时会作为命令行入口: + +```powershell +.\WeChatDecrypt.exe --help +.\WeChatDecrypt.exe status +.\WeChatDecrypt.exe decrypt +.\WeChatDecrypt.exe export --from-plan-csv export_plan.csv +.\WeChatDecrypt.exe export-all --write-plan-csv export_plan.csv +``` + +无参数或 `web` 子命令仍启动 Web UI: + +```powershell +.\WeChatDecrypt.exe +.\WeChatDecrypt.exe web +``` + ## 工具箱 3 个 tab ### 📱 个人微信 @@ -101,6 +120,6 @@ python app_gui.py # → 弹 tkinter 窗口 - Windows-only (不跨平台) - 没法远程访问 -单 exe **默认入口是 Web UI**。要打 tkinter 版的 exe, 改 -`WeChatDecrypt.spec` 里 `Analysis(['monitor_web.py'])` 为 +单 exe **默认入口是 Web UI + CLI 分发器**。要打 tkinter 版的 exe, 改 +`WeChatDecrypt.spec` 里 `Analysis(['wechat_decrypt_launcher.py'])` 为 `Analysis(['app_gui.py'])` 再 `build.bat`。 diff --git a/README.md b/README.md index 88d6c42..fb2f2e7 100644 --- a/README.md +++ b/README.md @@ -517,8 +517,19 @@ build.bat # → dist\WeChatDecrypt.exe (~20MB) ``` **单 exe 默认入口是 Web UI** (双击 → 自动开浏览器), 因为 Web UI 体验更好。 -要打包成 tkinter 入口的话改 `WeChatDecrypt.spec` 里的 `Analysis(['monitor_web.py'])` -为 `Analysis(['app_gui.py'])` 再 `build.bat`。 +同一个 exe 也支持 CLI 子命令: + +```powershell +.\WeChatDecrypt.exe --help +.\WeChatDecrypt.exe status +.\WeChatDecrypt.exe decrypt +.\WeChatDecrypt.exe export --from-plan-csv export_plan.csv +.\WeChatDecrypt.exe export-all --write-plan-csv export_plan.csv +``` + +要打包成 tkinter 入口的话改 `WeChatDecrypt.spec` 里的 +`Analysis(['wechat_decrypt_launcher.py'])` 为 `Analysis(['app_gui.py'])` +再 `build.bat`。 > 语音转 MP3 需要系统安装 [FFmpeg](https://ffmpeg.org/download.html) 并加入 PATH。 diff --git a/WeChatDecrypt.spec b/WeChatDecrypt.spec index 5bddc07..0ceb6a5 100644 --- a/WeChatDecrypt.spec +++ b/WeChatDecrypt.spec @@ -1,17 +1,18 @@ # -*- mode: python ; coding: utf-8 -*- -# PyInstaller spec — 打包 monitor_web.py 为单 exe。 +# PyInstaller spec — 打包 wechat_decrypt_launcher.py 为双模式单 exe。 # # 启动后: -# 1. 起 HTTP 服务 + 监听线程 (如 keys 已就绪) -# 2. 自动开浏览器到 http://localhost:5678 -# 3. 用户在 Web UI 工具箱里操作 (解密 / 导出 / 朋友圈 / 企微 / 语音) +# 1. 无参数/ web: 起 HTTP 服务 + 自动开浏览器到 http://localhost:5678 +# 2. 带 CLI 参数: 分发到 main.py / export_all_chats.py / 其他工具脚本 +# 3. 兼容 Web UI 内部 sys.executable + script.py 的子进程调用 # -# (历史: 这个 spec 之前打包 tkinter app_gui.py, 现已删除 — 完全切到 Web UI) +# (历史: 这个 spec 之前打包 tkinter app_gui.py, 现默认是 Web UI + CLI) from PyInstaller.utils.hooks import collect_all # 所有子进程 (工具按钮触发) 需要的脚本, 打进 exe 同目录 datas = [ + ('monitor_web.py', '.'), ('main.py', '.'), ('config.py', '.'), ('config.example.json', '.'), @@ -44,7 +45,13 @@ datas = [ binaries = [] hiddenimports = [ # 显式列, 避免 PyInstaller 漏 detect 导致打包后 import 报错 + 'argparse', 'csv', 'glob', 'hashlib', 'hmac', 'http.server', 'json', + 'platform', 'queue', 'socketserver', 'sqlite3', '_sqlite3', + 'subprocess', 'tempfile', 'threading', 'urllib.parse', 'uuid', 'wave', + 'xml.etree.ElementTree', + 'mcp', 'mcp.server', 'mcp.server.fastmcp', 'Crypto', 'Crypto.Cipher', 'Crypto.Cipher.AES', + 'Crypto.Util', 'Crypto.Util.Padding', 'zstandard', 'pilk', ] @@ -55,7 +62,7 @@ datas += tmp_ret[0]; binaries += tmp_ret[1]; hiddenimports += tmp_ret[2] a = Analysis( - ['monitor_web.py'], # ← 入口从 app_gui.py 改成 monitor_web.py + ['wechat_decrypt_launcher.py'], pathex=[], binaries=binaries, datas=datas, diff --git a/build.bat b/build.bat index 8462b44..75ba079 100644 --- a/build.bat +++ b/build.bat @@ -1,7 +1,7 @@ @echo off chcp 65001 >nul echo ======================================== -echo WeChatDecrypt 打包脚本 (Web UI 版) +echo WeChatDecrypt 打包脚本 (Web UI + CLI 双模式) echo ======================================== echo. @@ -32,6 +32,7 @@ for %%F in (dist\WeChatDecrypt.exe) do echo 大小: %%~zF bytes echo. echo 使用: 双击 WeChatDecrypt.exe echo ^→ 自动启动浏览器到 http://localhost:5678 (Web UI) +echo CLI: WeChatDecrypt.exe --help echo ======================================== echo. pause diff --git a/config.py b/config.py index bbd050e..697451c 100644 --- a/config.py +++ b/config.py @@ -10,18 +10,23 @@ import sys CONFIG_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "config.json") -# 打包后 __file__ 指向临时目录，优先使用环境变量或5 cwd +# 打包后 __file__ 指向临时目录，优先使用环境变量指定的 exe 所在目录。 def _app_base_dir(): d = os.environ.get("WECHAT_DECRYPT_APP_DIR") if d and os.path.isdir(d): return d return os.path.dirname(os.path.abspath(__file__)) + + def _config_file_path(): - base = _app_base_dir() - p = os.path.join(base, "config.json") + if os.environ.get("WECHAT_DECRYPT_APP_DIR"): + return os.path.join(_app_base_dir(), "config.json") + p = os.path.join(_app_base_dir(), "config.json") if os.path.exists(p): return p return CONFIG_FILE + + _SYSTEM = platform.system().lower() if _SYSTEM == "linux": @@ -278,7 +283,7 @@ def load_config(): # "all_keys.json"(项目根相对),也能写 "~/Documents/wechat_decrypted" / # "$HOME/wechat" / "%USERPROFILE%\\wechat"(跨用户便携)。 # 空字串 / null 不再触发 TypeError(用 cfg.get 而非 in)。 - base = os.path.dirname(os.path.abspath(__file__)) + base = _app_base_dir() if cfg.get("db_dir"): cfg["db_dir"] = os.path.expanduser(os.path.expandvars(cfg["db_dir"])) for key in ("keys_file", "decrypted_dir", "decoded_image_dir"): diff --git a/mcp_server.py b/mcp_server.py index 6ddff11..8da488b 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -16,7 +16,8 @@ import xml.etree.ElementTree as ET from Crypto.Cipher import AES from mcp.server.fastmcp import FastMCP import zstandard as zstd -from decode_image import ImageResolver +from config import _config_file_path, _DEFAULT +from decode_image import ImageResolver from key_utils import get_key_info, key_path_variants, strip_key_metadata # ============ 加密常量 ============ @@ -29,14 +30,17 @@ WAL_HEADER_SZ = 32 WAL_FRAME_HEADER_SZ = 24 # ============ 配置加载 ============ -SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) -CONFIG_FILE = os.path.join(SCRIPT_DIR, "config.json") - -with open(CONFIG_FILE, encoding="utf-8") as f: - _cfg = json.load(f) -for _key in ("keys_file", "decrypted_dir"): - if _key in _cfg and not os.path.isabs(_cfg[_key]): - _cfg[_key] = os.path.join(SCRIPT_DIR, _cfg[_key]) +SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) +CONFIG_FILE = _config_file_path() + +try: + with open(CONFIG_FILE, encoding="utf-8") as f: + _cfg = json.load(f) +except FileNotFoundError: + _cfg = dict(_DEFAULT) +for _key in ("keys_file", "decrypted_dir"): + if _key in _cfg and not os.path.isabs(_cfg[_key]): + _cfg[_key] = os.path.join(os.path.dirname(CONFIG_FILE), _cfg[_key]) DB_DIR = _cfg["db_dir"] KEYS_FILE = _cfg["keys_file"] @@ -55,8 +59,11 @@ if not DECODED_IMAGE_DIR: elif not os.path.isabs(DECODED_IMAGE_DIR): DECODED_IMAGE_DIR = os.path.join(SCRIPT_DIR, DECODED_IMAGE_DIR) -with open(KEYS_FILE, encoding="utf-8") as f: - ALL_KEYS = strip_key_metadata(json.load(f)) +try: + with open(KEYS_FILE, encoding="utf-8") as f: + ALL_KEYS = strip_key_metadata(json.load(f)) +except FileNotFoundError: + ALL_KEYS = {} # ============ 解密函数 ============ diff --git a/tests/test_exe_launcher.py b/tests/test_exe_launcher.py new file mode 100644 index 0000000..ffce716 --- /dev/null +++ b/tests/test_exe_launcher.py @@ -0,0 +1,147 @@ +import io +import os +import subprocess +import sys +import tempfile +import unittest +from unittest.mock import patch + +import wechat_decrypt_launcher +import config + + +class LauncherDispatchTests(unittest.TestCase): + def _dispatch(self, argv): + calls = [] + stdout = io.StringIO() + stderr = io.StringIO() + + def runner(script, script_args): + calls.append((script, script_args)) + return 0 + + code = wechat_decrypt_launcher.dispatch( + argv, + script_runner=runner, + stdout=stdout, + stderr=stderr, + ) + return code, calls, stdout.getvalue(), stderr.getvalue() + + def test_no_args_starts_web_ui(self): + code, calls, _, _ = self._dispatch([]) + + self.assertEqual(code, 0) + self.assertEqual(calls, [("monitor_web.py", [])]) + + def test_web_command_starts_web_ui(self): + code, calls, _, _ = self._dispatch(["web"]) + + self.assertEqual(code, 0) + self.assertEqual(calls, [("monitor_web.py", [])]) + + def test_decrypt_command_routes_to_main(self): + code, calls, _, _ = self._dispatch(["decrypt"]) + + self.assertEqual(code, 0) + self.assertEqual(calls, [("main.py", ["decrypt"])]) + + def test_export_all_command_routes_to_export_all_chats(self): + code, calls, _, _ = self._dispatch([ + "export-all", + "--write-plan-csv", + "export_plan.csv", + ]) + + self.assertEqual(code, 0) + self.assertEqual( + calls, + [( + "export_all_chats.py", + ["--write-plan-csv", "export_plan.csv"], + )], + ) + + def test_script_style_main_py_call_routes_to_main(self): + code, calls, _, _ = self._dispatch(["main.py", "decrypt"]) + + self.assertEqual(code, 0) + self.assertEqual(calls, [("main.py", ["decrypt"])]) + + def test_unknown_command_returns_error_without_running_script(self): + code, calls, stdout, stderr = self._dispatch(["unknown"]) + + self.assertEqual(code, 2) + self.assertEqual(calls, []) + self.assertIn("未知命令", stderr) + self.assertIn("WeChatDecrypt.exe", stdout) + + def test_prepare_runtime_overrides_stale_meipass_app_dir(self): + with tempfile.TemporaryDirectory() as exe_dir, \ + tempfile.TemporaryDirectory() as stale_dir, \ + patch.object(wechat_decrypt_launcher.sys, "frozen", True, create=True), \ + patch.object( + wechat_decrypt_launcher.sys, + "executable", + os.path.join(exe_dir, "WeChatDecrypt.exe"), + ), \ + patch.dict(os.environ, {"WECHAT_DECRYPT_APP_DIR": stale_dir}): + old_cwd = os.getcwd() + try: + wechat_decrypt_launcher._prepare_runtime() + self.assertEqual(os.environ["WECHAT_DECRYPT_APP_DIR"], exe_dir) + self.assertEqual(os.getcwd(), exe_dir) + finally: + os.chdir(old_cwd) + + def test_configure_stdio_uses_utf8(self): + stream = io.TextIOWrapper(io.BytesIO(), encoding="cp936") + + with patch.object(wechat_decrypt_launcher.sys, "stdout", stream), \ + patch.object(wechat_decrypt_launcher.sys, "stderr", stream): + wechat_decrypt_launcher._configure_stdio() + + self.assertEqual(stream.encoding.lower().replace("_", "-"), "utf-8") + + def test_pyinstaller_spec_includes_sqlite_hidden_imports(self): + with open("WeChatDecrypt.spec", encoding="utf-8") as f: + spec = f.read() + + self.assertIn("'sqlite3'", spec) + self.assertIn("'_sqlite3'", spec) + self.assertIn("'wave'", spec) + self.assertIn("'mcp.server.fastmcp'", spec) + self.assertIn("'Crypto.Util'", spec) + self.assertIn("'Crypto.Util.Padding'", spec) + + def test_config_file_path_uses_app_dir_even_when_config_missing(self): + with tempfile.TemporaryDirectory() as app_dir, \ + patch.dict(os.environ, {"WECHAT_DECRYPT_APP_DIR": app_dir}): + expected = os.path.join(app_dir, "config.json") + + self.assertEqual(config._config_file_path(), expected) + + def test_mcp_server_import_allows_missing_keys_file(self): + with tempfile.TemporaryDirectory() as app_dir: + env = dict(os.environ) + env["WECHAT_DECRYPT_APP_DIR"] = app_dir + result = subprocess.run( + [ + sys.executable, + "-X", + "utf8", + "-c", + "import mcp_server; print(mcp_server.ALL_KEYS)", + ], + cwd=os.getcwd(), + env=env, + capture_output=True, + text=True, + ) + + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn("{}", result.stdout) + + +if __name__ == "__main__": + unittest.main() diff --git a/wechat_decrypt_launcher.py b/wechat_decrypt_launcher.py new file mode 100644 index 0000000..8c892c6 --- /dev/null +++ b/wechat_decrypt_launcher.py @@ -0,0 +1,139 @@ +#!/usr/bin/env python3 +"""WeChatDecrypt.exe 双模式入口。 + +无参数时启动 Web UI；带参数时分发到现有 CLI 脚本。该入口同时兼容 +PyInstaller onefile 里 Web UI 子进程使用的 ``sys.executable script.py`` +调用形态。 +""" + +import os +import runpy +import sys + + +MAIN_COMMANDS = {"decrypt", "export", "all", "status", "-s", "decode-images"} +HELP_COMMANDS = {"help", "-h", "--help"} +SCRIPT_ENTRYPOINTS = { + "main.py", + "monitor_web.py", + "export_all_chats.py", + "find_image_key.py", + "decrypt_sns.py", + "export_sns.py", + "find_wxwork_keys.py", + "decrypt_wxwork_db.py", + "export_wxwork_messages.py", + "voice_to_mp3.py", + "batch_decrypt_images.py", + "decrypt_db.py", +} + + +def _app_base_dir(): + if getattr(sys, "frozen", False): + return os.path.dirname(sys.executable) + return os.path.dirname(os.path.abspath(__file__)) + + +def _script_dir(): + return getattr(sys, "_MEIPASS", _app_base_dir()) + + +def _prepare_runtime(): + base_dir = _app_base_dir() + os.environ["WECHAT_DECRYPT_APP_DIR"] = base_dir + os.chdir(base_dir) + + +def _configure_stdio(): + for stream in (sys.stdout, sys.stderr): + if hasattr(stream, "reconfigure"): + stream.reconfigure(encoding="utf-8", errors="replace") + + +def print_usage(stream=None): + stream = stream or sys.stdout + stream.write( + "WeChatDecrypt.exe 用法:\n" + "\n" + " WeChatDecrypt.exe 启动 Web UI\n" + " WeChatDecrypt.exe web 启动 Web UI\n" + " WeChatDecrypt.exe status 查看状态\n" + " WeChatDecrypt.exe decrypt 提取密钥并解密数据库\n" + " WeChatDecrypt.exe export [args...] 解密并批量导出聊天\n" + " WeChatDecrypt.exe all 密钥 -> 解密 -> 导出\n" + " WeChatDecrypt.exe decode-images [args...] 批量解密 .dat 图片\n" + " WeChatDecrypt.exe export-all [args...] 直接调用 export_all_chats.py\n" + "\n" + "兼容脚本式调用:\n" + " WeChatDecrypt.exe main.py decrypt\n" + " WeChatDecrypt.exe export_all_chats.py --write-plan-csv export_plan.csv\n" + ) + + +def _resolve_script(argv): + if not argv: + return "monitor_web.py", [] + + cmd = argv[0] + if cmd == "web": + return "monitor_web.py", argv[1:] + if cmd in MAIN_COMMANDS: + return "main.py", argv + if cmd == "export-all": + return "export_all_chats.py", argv[1:] + if cmd in SCRIPT_ENTRYPOINTS: + return cmd, argv[1:] + return None, None + + +def run_script(script, script_args): + _prepare_runtime() + script_path = os.path.join(_script_dir(), script) + if not os.path.exists(script_path): + script_path = os.path.join(_app_base_dir(), script) + if not os.path.exists(script_path): + raise FileNotFoundError(f"脚本不存在: {script_path}") + + old_argv = sys.argv[:] + try: + sys.argv = [script] + list(script_args) + try: + runpy.run_path(script_path, run_name="__main__") + except SystemExit as exc: + if exc.code is None: + return 0 + if isinstance(exc.code, int): + return exc.code + print(exc.code, file=sys.stderr) + return 1 + return 0 + finally: + sys.argv = old_argv + + +def dispatch(argv=None, script_runner=None, stdout=None, stderr=None): + argv = list(sys.argv[1:] if argv is None else argv) + stdout = stdout or sys.stdout + stderr = stderr or sys.stderr + runner = script_runner or run_script + + if argv and argv[0] in HELP_COMMANDS: + print_usage(stdout) + return 0 + + script, script_args = _resolve_script(argv) + if script is None: + print_usage(stdout) + print(f"未知命令: {argv[0]}", file=stderr) + return 2 + return runner(script, script_args) + + +def main(): + _configure_stdio() + sys.exit(dispatch()) + + +if __name__ == "__main__": + main()