diff --git a/WeChatDecrypt.spec b/WeChatDecrypt.spec index 8f9218d..96ab499 100644 --- a/WeChatDecrypt.spec +++ b/WeChatDecrypt.spec @@ -1,7 +1,7 @@ # -*- mode: python ; coding: utf-8 -*- 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', '.'), ('key_scan_common.py', '.'), ('key_utils.py', '.'), ('decode_image.py', '.'), ('find_image_key.py', '.'), ('find_image_key_monitor.py', '.'), ('monitor.py', '.'), ('monitor_web.py', '.'), ('mcp_server.py', '.'), ('config.example.json', '.')] +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', '.'), ('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', '.')] binaries = [] hiddenimports = [] tmp_ret = collect_all('pilk') diff --git a/app_gui.py b/app_gui.py index 74b5cc1..4b00fe2 100644 --- a/app_gui.py +++ b/app_gui.py @@ -3,6 +3,9 @@ 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 @@ -12,6 +15,7 @@ if getattr(sys, "frozen", False): else: BASE_DIR = os.path.dirname(os.path.abspath(__file__)) os.chdir(BASE_DIR) +os.environ["WECHAT_DECRYPT_APP_DIR"] = BASE_DIR # ── 子任务入口（当以 --task 参数调用时直接执行对应脚本） ────────────────────── @@ -52,6 +56,9 @@ def _run_subtask(task: str): "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", } script = mapping.get(task) if not script: @@ -69,6 +76,8 @@ def _run_subtask(task: str): # 将 decrypt 命令传给 main.py if task == "decrypt": sys.argv = ["main.py", "decrypt"] + elif task == "find_image_key": + sys.argv = ["find_image_key.py"] else: sys.argv = [script] @@ -97,6 +106,260 @@ if sys.platform == "win32": 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 App(tk.Tk): def __init__(self): super().__init__() @@ -105,6 +368,13 @@ class App(tk.Tk): 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._build_ui() @@ -121,7 +391,7 @@ class App(tk.Tk): # 按钮区域 btn_frame = ttk.Frame(self) - btn_frame.pack(fill="x", padx=20, pady=(4, 8)) + btn_frame.pack(fill="x", padx=20, pady=(4, 4)) self.btn_decrypt = ttk.Button( btn_frame, text="① 解密数据库", style="Big.TButton", @@ -129,17 +399,35 @@ class App(tk.Tk): ) 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", + 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_voice = ttk.Button( - btn_frame, text="③ 转换音频", style="Big.TButton", - command=lambda: self._run_task("voice") + self.btn_sns = ttk.Button( + btn_frame, text="④ 朋友圈图片", style="Big.TButton", + command=lambda: self._run_task("decrypt_sns") ) - self.btn_voice.pack(side="left", expand=True, fill="x", padx=4) + self.btn_sns.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缩略图自动跳过）" + ) + ttk.Label(tips_frame, text=tips_text, font=("Microsoft YaHei UI", 9), + wraplength=680, justify="left").pack(anchor="w") # 进度条 self.progress = ttk.Progressbar(self, mode="indeterminate") @@ -177,56 +465,181 @@ class App(tk.Tk): 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_voice.configure(state=state) + self.btn_sns.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._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() + 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() - labels = { - "decrypt": "解密数据库", - "export": "导出消息记录", - "voice": "转换音频文件", - } - self.status_var.set(f"正在{labels[task]}...") + # ── 子进程执行 ───────────────────────────────────────────────────────── + def _run_subprocess(self, task: str) -> int: + """运行子进程，返回退出码""" + cmd = [sys.executable, "--task", task] + self.after(0, self._log, f">>> {' '.join(cmd)}\n\n") - thread = threading.Thread(target=self._exec_task, args=(task,), daemon=True) - thread.start() + env = os.environ.copy() + env["PYTHONIOENCODING"] = "utf-8" + env["WECHAT_DECRYPT_APP_DIR"] = BASE_DIR + + 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" + + 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_task(self, task: str): + """执行单一任务（解密）""" try: - cmd = [sys.executable, "--task", task] - - self._log(f">>> {' '.join(cmd)}\n\n") - - env = os.environ.copy() - env["PYTHONIOENCODING"] = "utf-8" - env["WECHAT_DECRYPT_APP_DIR"] = BASE_DIR - - 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() - rc = proc.returncode + 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})") @@ -240,6 +653,10 @@ class App(tk.Tk): 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__": diff --git a/batch_decrypt_images.py b/batch_decrypt_images.py new file mode 100644 index 0000000..23f5505 --- /dev/null +++ b/batch_decrypt_images.py @@ -0,0 +1,181 @@ +"""批量解密 .dat 图片文件 + +用法: python batch_decrypt_images.py <文件夹路径> [输出目录] + +递归扫描指定文件夹下的所有 .dat 文件并解密。 +输出目录默认为 <文件夹路径>_decoded/，保持原有子目录结构。 +""" +import os +import sys +import glob +import struct + +# Windows 控制台 UTF-8 +if sys.platform == "win32": + sys.stdout.reconfigure(encoding="utf-8", errors="replace") + +from config import load_config + +_cfg = load_config() +IMAGE_AES_KEY = _cfg.get("image_aes_key", "") +IMAGE_XOR_KEY = _cfg.get("image_xor_key", 0x88) + +# ── V2/V1 magic ────────────────────────────────────────────────────────────── +_V2_MAGIC_FULL = b'\x07\x08V2\x08\x07' +_V1_MAGIC_FULL = b'\x07\x08V1\x08\x07' + +_IMAGE_MAGICS = { + 'jpg': [0xFF, 0xD8, 0xFF], + 'png': [0x89, 0x50, 0x4E, 0x47], + 'gif': [0x47, 0x49, 0x46, 0x38], + 'webp': [0x52, 0x49, 0x46, 0x46], + 'bmp': [0x42, 0x4D], + 'tif': [0x49, 0x49, 0x2A, 0x00], +} + + +def _detect_format(header): + if header[:3] == bytes([0xFF, 0xD8, 0xFF]): + return 'jpg' + if header[:4] == bytes([0x89, 0x50, 0x4E, 0x47]): + return 'png' + if header[:3] == b'GIF': + return 'gif' + if header[:2] == b'BM': + return 'bmp' + if header[:4] == b'RIFF' and len(header) >= 12 and header[8:12] == b'WEBP': + return 'webp' + if header[:4] == bytes([0x49, 0x49, 0x2A, 0x00]): + return 'tif' + if header[:4] == b'wxgf': + return 'hevc' + return 'bin' + + +def decrypt_dat(dat_path): + """解密单个 .dat 文件，返回 (bytes, format) 或 (None, None)""" + with open(dat_path, 'rb') as f: + data = f.read() + if len(data) < 6: + return None, None + + head6 = data[:6] + + # V2 / V1 格式 (AES-ECB + XOR) + if head6 in (_V2_MAGIC_FULL, _V1_MAGIC_FULL): + if head6 == _V1_MAGIC_FULL: + aes_key = b'cfcd208495d565ef' + elif IMAGE_AES_KEY: + aes_key = IMAGE_AES_KEY.encode('ascii')[:16] if isinstance(IMAGE_AES_KEY, str) else IMAGE_AES_KEY[:16] + else: + return None, None + if len(aes_key) < 16: + return None, None + try: + from Crypto.Cipher import AES + from Crypto.Util import Padding + if len(data) < 15: + return None, None + aes_size, xor_size = struct.unpack_from(' len(data): + return None, None + cipher = AES.new(aes_key[:16], AES.MODE_ECB) + dec_aes = Padding.unpad(cipher.decrypt(data[offset:offset+aligned]), AES.block_size) + offset += aligned + raw_end = len(data) - xor_size + raw_data = data[offset:raw_end] if offset < raw_end else b'' + xor_data = data[raw_end:] + xor_key = IMAGE_XOR_KEY if isinstance(IMAGE_XOR_KEY, int) else 0x88 + dec_xor = bytes(b ^ xor_key for b in xor_data) + result = dec_aes + raw_data + dec_xor + fmt = _detect_format(result[:16]) + return result, fmt + except Exception as e: + print(f" AES 解密失败: {e}") + return None, None + + # 旧 XOR 格式 + for fmt_name, magic in _IMAGE_MAGICS.items(): + key = data[0] ^ magic[0] + match = all(i < len(data) and (data[i] ^ key) == magic[i] for i in range(len(magic))) + if match: + result = bytes(b ^ key for b in data) + fmt = _detect_format(result[:16]) + return result, fmt + + return None, None + + +def main(): + if len(sys.argv) < 2: + print("用法: python batch_decrypt_images.py <文件夹路径> [输出目录]") + print(" 递归扫描文件夹下所有 .dat 文件并解密") + sys.exit(1) + + source_dir = os.path.abspath(sys.argv[1]) + if not os.path.isdir(source_dir): + print(f"目录不存在: {source_dir}") + sys.exit(1) + + if len(sys.argv) >= 3: + output_dir = os.path.abspath(sys.argv[2]) + else: + output_dir = source_dir.rstrip(os.sep) + "_decoded" + + # 递归收集所有 .dat 文件 + dat_files = [] + for root, _dirs, files in os.walk(source_dir): + for f in files: + if f.lower().endswith('.dat'): + dat_files.append(os.path.join(root, f)) + dat_files.sort() + + print(f"源目录: {source_dir}") + print(f"输出目录: {output_dir}") + print(f"找到 {len(dat_files)} 个 .dat 文件") + print() + + total = len(dat_files) + success = 0 + skipped = 0 + failed = 0 + + for dat_path in dat_files: + # 保持相对目录结构 + rel = os.path.relpath(dat_path, source_dir) + rel_dir = os.path.dirname(rel) + out_subdir = os.path.join(output_dir, rel_dir) if rel_dir else output_dir + + fname = os.path.splitext(os.path.basename(dat_path))[0] + # 去除 _t / _h 后缀获取基础名 + base_name = fname + for suffix in ('_t', '_h'): + if base_name.endswith(suffix): + base_name = base_name[:-len(suffix)] + break + + # 检查是否已解密 + existing = glob.glob(os.path.join(out_subdir, f"{base_name}.*")) + if existing: + skipped += 1 + continue + + img_bytes, fmt = decrypt_dat(dat_path) + if not img_bytes or fmt == 'bin': + failed += 1 + continue + + os.makedirs(out_subdir, exist_ok=True) + out_path = os.path.join(out_subdir, f"{base_name}.{fmt}") + with open(out_path, 'wb') as f: + f.write(img_bytes) + success += 1 + + print(f"完成: 共 {total} 个文件, 成功 {success}, 跳过(已存在) {skipped}, 失败 {failed}") + print(f"输出: {output_dir}") + + +if __name__ == "__main__": + main() diff --git a/build.bat b/build.bat index 51f5ee6..d0b43eb 100644 --- a/build.bat +++ b/build.bat @@ -29,6 +29,8 @@ pyinstaller --noconfirm --onefile --console --name "WeChatDecrypt" ^ --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;." ^ diff --git a/config.py b/config.py index a6c9960..7398c2f 100644 --- a/config.py +++ b/config.py @@ -229,8 +229,34 @@ def load_config(): else: cfg["wechat_base_dir"] = db_dir + # 输出目录：/wechat_files// + wxid = os.path.basename(os.path.normpath(cfg["wechat_base_dir"])) + cfg["output_base_dir"] = os.path.join(base, "wechat_files", wxid) + # decoded_image_dir 默认值 if "decoded_image_dir" not in cfg: cfg["decoded_image_dir"] = os.path.join(base, "decoded_images") + # 自动检测 WeChat Files 目录（FileStorage/MsgAttach, FileStorage/Sns/Cache） + if not cfg.get("wechat_files_dir"): + wechat_files_base = os.path.join(os.path.expanduser("~"), "Documents", "WeChat Files") + if os.path.isdir(wechat_files_base): + # xwechat_files 的 wxid 可能带后缀如 _1d4c，需要模糊匹配 + wxid_prefix = wxid.rsplit("_", 1)[0] if "_" in wxid else wxid + for d in os.listdir(wechat_files_base): + if d == wxid or d == wxid_prefix or wxid.startswith(d): + candidate = os.path.join(wechat_files_base, d) + if os.path.isdir(os.path.join(candidate, "FileStorage")): + cfg["wechat_files_dir"] = candidate + break + + wf_dir = cfg.get("wechat_files_dir", "") + cfg["msgattach_dir"] = os.path.join(wf_dir, "FileStorage", "MsgAttach") if wf_dir else "" + cfg["sns_cache_dir"] = os.path.join(wf_dir, "FileStorage", "Sns", "Cache") if wf_dir else "" + + # xwechat_files 图片/缓存路径 + wb = cfg["wechat_base_dir"] + cfg["xwechat_attach_dir"] = os.path.join(wb, "msg", "attach") if wb else "" + cfg["xwechat_cache_dir"] = os.path.join(wb, "cache") if wb else "" + return cfg diff --git a/decrypt_sns.py b/decrypt_sns.py new file mode 100644 index 0000000..174ec0f --- /dev/null +++ b/decrypt_sns.py @@ -0,0 +1,272 @@ +"""解密微信朋友圈图片缓存 +来源1: WeChat Files/FileStorage/Sns/Cache//[_t|_d] +来源2: xwechat_files/cache//Sns/Img// +输出目录: /朋友圈图片// +_t 后缀为缩略图（跳过） +""" +import os +import sys +import glob +import struct + +if sys.platform == "win32": + sys.stdout.reconfigure(encoding="utf-8", errors="replace") + +from config import load_config + +_cfg = load_config() +SNS_CACHE_DIR = _cfg.get("sns_cache_dir", "") +XWECHAT_CACHE_DIR = _cfg.get("xwechat_cache_dir", "") +OUTPUT_DIR = os.path.join(_cfg["output_base_dir"], "朋友圈图片") +IMAGE_AES_KEY = _cfg.get("image_aes_key") +IMAGE_XOR_KEY = _cfg.get("image_xor_key", 0x88) + +# ── V2/V1 magic ────────────────────────────────────────────────────────────── +_V2_MAGIC_FULL = b'\x07\x08V2\x08\x07' +_V1_MAGIC_FULL = b'\x07\x08V1\x08\x07' + +_IMAGE_MAGICS = { + 'jpg': [0xFF, 0xD8, 0xFF], + 'png': [0x89, 0x50, 0x4E, 0x47], + 'gif': [0x47, 0x49, 0x46, 0x38], + 'webp': [0x52, 0x49, 0x46, 0x46], + 'bmp': [0x42, 0x4D], + 'tif': [0x49, 0x49, 0x2A, 0x00], +} + + +def _detect_format(header): + if header[:3] == bytes([0xFF, 0xD8, 0xFF]): + return 'jpg' + if header[:4] == bytes([0x89, 0x50, 0x4E, 0x47]): + return 'png' + if header[:3] == b'GIF': + return 'gif' + if header[:2] == b'BM': + return 'bmp' + if header[:4] == b'RIFF' and len(header) >= 12 and header[8:12] == b'WEBP': + return 'webp' + if header[:4] == bytes([0x49, 0x49, 0x2A, 0x00]): + return 'tif' + if header[:4] == b'wxgf': + return 'hevc' + return 'bin' + + +def decrypt_dat(dat_path): + """解密单个 .dat 文件，返回 (bytes, format) 或 (None, None)""" + with open(dat_path, 'rb') as f: + data = f.read() + if len(data) < 6: + return None, None + + head6 = data[:6] + + # V2 / V1 格式 + if head6 in (_V2_MAGIC_FULL, _V1_MAGIC_FULL): + if head6 == _V1_MAGIC_FULL: + aes_key = b'cfcd208495d565ef' + elif IMAGE_AES_KEY: + aes_key = IMAGE_AES_KEY.encode('ascii')[:16] if isinstance(IMAGE_AES_KEY, str) else IMAGE_AES_KEY[:16] + else: + return None, None + if not aes_key or len(aes_key) < 16: + return None, None + try: + from Crypto.Cipher import AES + from Crypto.Util import Padding + if len(data) < 15: + return None, None + aes_size, xor_size = struct.unpack_from(' len(data): + return None, None + cipher = AES.new(aes_key[:16], AES.MODE_ECB) + dec_aes = Padding.unpad(cipher.decrypt(data[offset:offset+aligned]), AES.block_size) + offset += aligned + raw_end = len(data) - xor_size + raw_data = data[offset:raw_end] if offset < raw_end else b'' + xor_data = data[raw_end:] + xor_key = IMAGE_XOR_KEY if isinstance(IMAGE_XOR_KEY, int) else 0x88 + dec_xor = bytes(b ^ xor_key for b in xor_data) + result = dec_aes + raw_data + dec_xor + fmt = _detect_format(result[:16]) + return result, fmt + except Exception as e: + print(f" AES 解密失败: {e}") + return None, None + + # 旧 XOR 格式 + for fmt_name, magic in _IMAGE_MAGICS.items(): + key = data[0] ^ magic[0] + match = all(i < len(data) and (data[i] ^ key) == magic[i] for i in range(len(magic))) + if match: + result = bytes(b ^ key for b in data) + fmt = _detect_format(result[:16]) + return result, fmt + + return None, None + + +def _collect_xwechat_sns_files(): + """收集 xwechat cache//Sns/Img// 下的所有文件 + 返回 {month: [(file_path, basename), ...], ...} + """ + result = {} + if not XWECHAT_CACHE_DIR or not os.path.isdir(XWECHAT_CACHE_DIR): + return result + try: + months = sorted(os.listdir(XWECHAT_CACHE_DIR)) + except OSError: + return result + for month in months: + sns_img = os.path.join(XWECHAT_CACHE_DIR, month, "Sns", "Img") + if not os.path.isdir(sns_img): + continue + files = [] + try: + hex_dirs = os.listdir(sns_img) + except OSError: + continue + for hd in hex_dirs: + hd_path = os.path.join(sns_img, hd) + if not os.path.isdir(hd_path): + continue + try: + for fname in os.listdir(hd_path): + fp = os.path.join(hd_path, fname) + if os.path.isfile(fp): + files.append((fp, fname)) + except OSError: + continue + if files: + result[month] = files + return result + + +def main(): + has_wechat = SNS_CACHE_DIR and os.path.isdir(SNS_CACHE_DIR) + has_xwechat = XWECHAT_CACHE_DIR and os.path.isdir(XWECHAT_CACHE_DIR) + + if not has_wechat and not has_xwechat: + print(f"朋友圈缓存目录不存在:") + print(f" WeChat Files: {SNS_CACHE_DIR}") + print(f" xwechat: {XWECHAT_CACHE_DIR}") + print("请确认 config.json 中的路径配置正确") + return + + print(f"输出目录: {OUTPUT_DIR}") + + total = 0 + success = 0 + skipped_thumb = 0 + skipped_exist = 0 + failed = 0 + + # ── 来源1: WeChat Files/FileStorage/Sns/Cache// ── + if has_wechat: + print(f"\n[来源1] WeChat Files: {SNS_CACHE_DIR}") + months = sorted(d for d in os.listdir(SNS_CACHE_DIR) + if os.path.isdir(os.path.join(SNS_CACHE_DIR, d))) + has_month_dirs = any(len(m) == 7 and m[4] == '-' for m in months) + + if has_month_dirs: + print(f" 时间目录: {len(months)} 个") + for month in months: + month_src = os.path.join(SNS_CACHE_DIR, month) + month_out = os.path.join(OUTPUT_DIR, month) + stats = _process_dir_stats(month_src, month_out, month) + total += stats[0]; success += stats[1]; skipped_thumb += stats[2] + skipped_exist += stats[3]; failed += stats[4] + else: + stats = _process_dir_stats(SNS_CACHE_DIR, OUTPUT_DIR, "") + total, success, skipped_thumb, skipped_exist, failed = stats + + # ── 来源2: xwechat cache//Sns/Img// ── + if has_xwechat: + print(f"\n[来源2] xwechat: {XWECHAT_CACHE_DIR}") + xw_files = _collect_xwechat_sns_files() + if not xw_files: + print(" 未找到 Sns/Img 文件") + else: + print(f" 时间目录: {len(xw_files)} 个") + for month, file_list in sorted(xw_files.items()): + month_out = os.path.join(OUTPUT_DIR, month) + stats = _process_file_list(file_list, month_out, month) + total += stats[0]; success += stats[1]; skipped_thumb += stats[2] + skipped_exist += stats[3]; failed += stats[4] + + print(f"\n完成: 共 {total} 个文件") + print(f" 成功解密: {success}") + print(f" 跳过缩略图(_t): {skipped_thumb}") + print(f" 跳过已存在: {skipped_exist}") + print(f" 解密失败: {failed}") + print(f"输出: {os.path.abspath(OUTPUT_DIR)}") + + +def _process_dir_stats(src_dir, out_dir, label): + """处理一个目录中的所有文件，返回 (total, success, skipped_thumb, skipped_exist, failed)""" + try: + all_files = sorted(os.listdir(src_dir)) + except OSError: + return (0, 0, 0, 0, 0) + + dat_files = [(os.path.join(src_dir, f), f) for f in all_files + if os.path.isfile(os.path.join(src_dir, f))] + return _process_file_list(dat_files, out_dir, label) + + +def _process_file_list(file_list, out_dir, label): + """处理文件列表 [(file_path, basename), ...], 返回 (total, success, skipped_thumb, skipped_exist, failed)""" + total = 0 + success = 0 + skipped_thumb = 0 + skipped_exist = 0 + failed = 0 + + if not file_list: + return (0, 0, 0, 0, 0) + + if label: + print(f" [{label}] {len(file_list)} 个文件") + + month_ok = 0 + for file_path, fname in file_list: + total += 1 + # 跳过缩略图 + if fname.endswith('_t'): + skipped_thumb += 1 + continue + + # 去掉 _d 后缀得到基础名 + base_name = fname + if base_name.endswith('_d'): + base_name = base_name[:-2] + + # 检查是否已存在 + existing = glob.glob(os.path.join(out_dir, f"{base_name}.*")) + if existing: + skipped_exist += 1 + continue + + img_bytes, fmt = decrypt_dat(file_path) + if not img_bytes or fmt in ('bin', 'hevc'): + failed += 1 + continue + + os.makedirs(out_dir, exist_ok=True) + out_path = os.path.join(out_dir, f"{base_name}.{fmt}") + with open(out_path, 'wb') as f: + f.write(img_bytes) + success += 1 + month_ok += 1 + + if month_ok > 0 and label: + print(f" 解密成功: {month_ok} 张") + + return (total, success, skipped_thumb, skipped_exist, failed) + + +if __name__ == "__main__": + main() diff --git a/export_messages.py b/export_messages.py index bf92e07..bf2201d 100644 --- a/export_messages.py +++ b/export_messages.py @@ -1,6 +1,8 @@ """导出微信消息记录到 CSV / HTML / JSON -目录结构: /export//messages.csv|html|json +目录结构: //messages.csv|html|json +图片导出: //image/. """ +import base64 import sqlite3 import glob import hashlib @@ -8,6 +10,7 @@ import os import json import csv import re +import struct import sys import xml.etree.ElementTree as ET from datetime import datetime @@ -23,7 +26,310 @@ from config import load_config _cfg = load_config() MSG_DB_DIR = os.path.join(_cfg["decrypted_dir"], "message") CONTACT_DB_PATH = os.path.join(_cfg["decrypted_dir"], "contact", "contact.db") -OUTPUT_DIR = os.path.join(_cfg["wechat_base_dir"], "export") +OUTPUT_DIR = _cfg["output_base_dir"] + +# 图片相关配置 +WECHAT_BASE_DIR = _cfg.get("wechat_base_dir", "") +ATTACH_DIR = os.path.join(WECHAT_BASE_DIR, "msg", "attach") if WECHAT_BASE_DIR else "" +MSGATTACH_DIR = _cfg.get("msgattach_dir", "") # WeChat Files/FileStorage/MsgAttach +IMAGE_AES_KEY = _cfg.get("image_aes_key") +IMAGE_XOR_KEY = _cfg.get("image_xor_key", 0x88) +MSG_RESOURCE_DB = os.path.join(_cfg["decrypted_dir"], "message", "message_resource.db") + +_CONTACT_FILTER = None +_filter_raw = os.environ.get("WECHAT_EXPORT_CONTACTS", "").strip() +if _filter_raw: + _CONTACT_FILTER = set(_filter_raw.split(",")) + print(f"联系人筛选: {len(_CONTACT_FILTER)} 个") + +_EXPORT_FORMATS = None +_formats_raw = os.environ.get("WECHAT_EXPORT_FORMATS", "").strip() +if _formats_raw: + _EXPORT_FORMATS = set(_formats_raw.lower().split(",")) + print(f"导出格式: {', '.join(sorted(_EXPORT_FORMATS))}") + +_EXPORT_IMAGES = os.environ.get("WECHAT_EXPORT_IMAGES", "1").strip() == "1" + + +# ─── 图片解密辅助 ─────────────────────────────────────────────────────────────── + +def _extract_md5_from_packed_info(blob): + """从 message_resource.db 的 packed_info 中提取文件 MD5""" + if not blob or not isinstance(blob, bytes): + return None + marker = b'\x12\x22\x0a\x20' + idx = blob.find(marker) + if idx >= 0 and idx + len(marker) + 32 <= len(blob): + md5_bytes = blob[idx + len(marker): idx + len(marker) + 32] + try: + md5_str = md5_bytes.decode('ascii') + int(md5_str, 16) + return md5_str + except (UnicodeDecodeError, ValueError): + pass + hex_chars = set(b'0123456789abcdef') + i = 0 + while i <= len(blob) - 32: + if blob[i] in hex_chars: + candidate = blob[i:i+32] + if all(b in hex_chars for b in candidate): + try: + return candidate.decode('ascii') + except UnicodeDecodeError: + pass + i += 32 + else: + i += 1 + return None + + +def _load_resource_md5_map(): + """加载 message_resource.db 的 (chat_username, local_id) -> file_md5 映射""" + md5_map = {} + if not os.path.exists(MSG_RESOURCE_DB): + return md5_map + try: + conn = sqlite3.connect(MSG_RESOURCE_DB) + # chat_id -> username + chat_id_map = {} + for row in conn.execute("SELECT rowid, user_name FROM ChatName2Id"): + chat_id_map[row[0]] = row[1] + for row in conn.execute( + "SELECT chat_id, message_local_id, packed_info FROM MessageResourceInfo" + ): + cid, lid, blob = row + md5 = _extract_md5_from_packed_info(blob) + if md5: + uname = chat_id_map.get(cid, "") + if uname: + md5_map[(uname, lid)] = md5 + conn.close() + print(f"图片资源映射: {len(md5_map)} 条") + except Exception as e: + print(f"读取 message_resource.db 失败: {e}") + return md5_map + + +def _find_dat_file(username_hash, file_md5): + """在 attach / MsgAttach 目录下查找 .dat 文件，优先高清版""" + search_patterns = [] + # xwechat_files 的 msg/attach 目录: //Img/*.dat + if ATTACH_DIR and os.path.isdir(ATTACH_DIR): + search_base = os.path.join(ATTACH_DIR, username_hash) + if os.path.isdir(search_base): + search_patterns.append(os.path.join(search_base, "*", "Img", f"{file_md5}*.dat")) + # WeChat Files 的 MsgAttach 目录: /Image//*.dat + if MSGATTACH_DIR and os.path.isdir(MSGATTACH_DIR): + search_base = os.path.join(MSGATTACH_DIR, username_hash) + if os.path.isdir(search_base): + search_patterns.append(os.path.join(search_base, "Image", "*", f"{file_md5}*.dat")) + + files = [] + for pat in search_patterns: + files.extend(glob.glob(pat)) + if not files: + return None + # 优先: 无后缀(原图) > _W(原图) > _h(高清) > _t/_t_W(缩略图) + # 先过滤掉缩略图 + non_thumb = [f for f in files if '_t.' not in os.path.basename(f) and '_t_' not in os.path.basename(f)] + candidates = non_thumb if non_thumb else files + selected = candidates[0] + for f in candidates: + fname = os.path.basename(f) + # 精确匹配原图（无后缀） + if fname == f"{file_md5}.dat": + return f + for f in candidates: + fname = os.path.basename(f) + if fname == f"{file_md5}_W.dat": + return f + for f in candidates: + if '_h.' in os.path.basename(f) or '_h_' in os.path.basename(f): + return f + return selected + + +def _detect_image_format(header): + """根据解密后的文件头检测图片格式""" + if header[:3] == bytes([0xFF, 0xD8, 0xFF]): + return 'jpg' + if header[:4] == bytes([0x89, 0x50, 0x4E, 0x47]): + return 'png' + if header[:3] == b'GIF': + return 'gif' + if header[:4] == b'RIFF' and len(header) >= 12 and header[8:12] == b'WEBP': + return 'webp' + if header[:4] == b'wxgf': + return 'hevc' + return 'bin' + + +# V2 格式常量 +_V2_MAGIC_FULL = b'\x07\x08V2\x08\x07' +_V1_MAGIC_FULL = b'\x07\x08V1\x08\x07' +_IMAGE_MAGICS = { + 'jpg': [0xFF, 0xD8, 0xFF], + 'png': [0x89, 0x50, 0x4E, 0x47], + 'gif': [0x47, 0x49, 0x46, 0x38], + 'webp': [0x52, 0x49, 0x46, 0x46], +} + + +def _decrypt_dat_to_bytes(dat_path): + """解密 .dat 文件，返回 (bytes, format) 或 (None, None)""" + with open(dat_path, 'rb') as f: + data = f.read() + if len(data) < 15: + return None, None + head6 = data[:6] + + # V2 / V1 格式 + if head6 in (_V2_MAGIC_FULL, _V1_MAGIC_FULL): + aes_key = None + if head6 == _V1_MAGIC_FULL: + aes_key = b'cfcd208495d565ef' + elif IMAGE_AES_KEY: + aes_key = IMAGE_AES_KEY.encode('ascii')[:16] if isinstance(IMAGE_AES_KEY, str) else IMAGE_AES_KEY[:16] + if not aes_key or len(aes_key) < 16: + return None, None + try: + from Crypto.Cipher import AES as _AES + from Crypto.Util import Padding + aes_size, xor_size = struct.unpack_from(' len(data): + return None, None + cipher = _AES.new(aes_key[:16], _AES.MODE_ECB) + dec_aes = Padding.unpad(cipher.decrypt(data[offset:offset+aligned]), _AES.block_size) + offset += aligned + raw_end = len(data) - xor_size + raw_data = data[offset:raw_end] if offset < raw_end else b'' + xor_data = data[raw_end:] + xor_key = IMAGE_XOR_KEY if isinstance(IMAGE_XOR_KEY, int) else 0x88 + dec_xor = bytes(b ^ xor_key for b in xor_data) + result = dec_aes + raw_data + dec_xor + fmt = _detect_image_format(result[:16]) + return result, fmt + except Exception: + return None, None + + # 旧 XOR 格式 + for fmt_name, magic in _IMAGE_MAGICS.items(): + key = data[0] ^ magic[0] + match = all(i < len(data) and (data[i] ^ key) == magic[i] for i in range(len(magic))) + if match: + result = bytes(b ^ key for b in data) + fmt = _detect_image_format(result[:16]) + return result, fmt + + return None, None + + +_resource_md5_map = _load_resource_md5_map() if _EXPORT_IMAGES else {} + + +def decode_chat_images(chat_username, _messages_unused, out_dir): + """直接扫描 attach 目录下该联系人的全部图片并解密 + 按月份分目录输出到 out_dir/image// + 跳过 _t 缩略图，优先 _h 高清版 + 返回 {file_md5: relative_path} 用于 HTML 嵌入 + """ + image_map = {} + username_hash = hashlib.md5(chat_username.encode()).hexdigest() + + # 收集所有来源目录: [(base_path, sub_structure), ...] + # xwechat: attach///Img/*.dat + # WeChat Files: MsgAttach//Image//*.dat + source_dirs = [] + if ATTACH_DIR: + p = os.path.join(ATTACH_DIR, username_hash) + if os.path.isdir(p): + source_dirs.append(("xwechat", p)) + if MSGATTACH_DIR: + p = os.path.join(MSGATTACH_DIR, username_hash) + if os.path.isdir(p): + source_dirs.append(("wechat", p)) + + if not source_dirs: + return image_map + + # 收集所有 dat 文件: {base_md5: (best_path, month)} + # 优先级: _h > 无后缀 > _W > 其他（跳过 _t） + file_candidates = {} # base_md5 -> (priority, dat_path, month) + + def _priority(fname): + """返回优先级数字，越小越好""" + base = fname.rsplit('.', 1)[0] + if base.endswith('_h'): + return 0 # 高清 + if '_' not in base[-3:]: + return 1 # 无后缀原图 + if base.endswith('_W'): + return 2 + return 9 # 其他 + + for src_type, base_path in source_dirs: + # xwechat: //Img/ — 直接列 base_path 得到月份 + # wechat: /Image// — 需要列 base_path/Image 得到月份 + if src_type == "xwechat": + scan_base = base_path + else: + scan_base = os.path.join(base_path, "Image") + try: + months = sorted(os.listdir(scan_base)) + except OSError: + continue + for month in months: + if src_type == "xwechat": + img_dir = os.path.join(base_path, month, "Img") + else: + img_dir = os.path.join(scan_base, month) + if not os.path.isdir(img_dir): + continue + try: + files = os.listdir(img_dir) + except OSError: + continue + for fname in files: + if not fname.endswith('.dat'): + continue + # 跳过缩略图 _t.dat 和 _t_W.dat + base_no_ext = fname.rsplit('.', 1)[0] + if '_t' in base_no_ext.split('_'): + continue + if base_no_ext.endswith('_t') or '_t_' in base_no_ext: + continue + # 提取 base md5 + base_md5 = base_no_ext.split('_')[0] + pri = _priority(fname) + existing = file_candidates.get(base_md5) + if not existing or pri < existing[0]: + file_candidates[base_md5] = (pri, os.path.join(img_dir, fname), month) + + if not file_candidates: + return image_map + + decoded_count = 0 + for base_md5, (pri, dat_path, month) in file_candidates.items(): + month_dir = os.path.join(out_dir, "image", month) + # 检查是否已解密 + existing = glob.glob(os.path.join(month_dir, f"{base_md5}.*")) + if existing: + rel = os.path.relpath(existing[0], out_dir).replace("\\", "/") + image_map[base_md5] = rel + continue + img_bytes, fmt = _decrypt_dat_to_bytes(dat_path) + if not img_bytes or fmt == 'bin': + continue + os.makedirs(month_dir, exist_ok=True) + out_path = os.path.join(month_dir, f"{base_md5}.{fmt}") + with open(out_path, 'wb') as f: + f.write(img_bytes) + image_map[base_md5] = f"image/{month}/{base_md5}.{fmt}" + decoded_count += 1 + + return image_map MSG_TYPES = { 1: "文本", @@ -127,6 +433,7 @@ body{{background:#ededed;font-family:"PingFang SC","Helvetica Neue",Arial,sans-s .bubble{{display:inline-block;padding:8px 12px;border-radius:6px;word-break:break-word;line-height:1.5;box-shadow:0 1px 2px rgba(0,0,0,.1);white-space:pre-wrap}} .received .bubble{{background:#fff;border-radius:0 6px 6px 6px}} .sent .bubble{{background:#95EC69;border-radius:6px 0 6px 6px}} +.bubble img{{max-width:100%;border-radius:4px;display:block;margin:2px 0}} .type-tag{{font-size:11px;color:#aaa;margin-top:2px}} @@ -142,7 +449,7 @@ body{{background:#ededed;font-family:"PingFang SC","Helvetica Neue",Arial,sans-s def _html_escape(s: str) -> str: return s.replace("&","&").replace("<","<").replace(">",">").replace('"','"') -def _write_html(path: str, title: str, is_group: bool, messages: list): +def _write_html(path: str, title: str, is_group: bool, messages: list, image_map: dict = None, out_dir: str = None): parts = [] last_date = None for m in messages: @@ -169,12 +476,33 @@ def _write_html(path: str, title: str, is_group: bool, messages: list): if m["type"] != 1: type_tag = f'
{m["type_name"]}
' + # 图片消息嵌入 + bubble_content = _html_escape(m["display_content"]) + if m["type"] == 3 and image_map and m["local_id"] in image_map: + rel_path = image_map[m["local_id"]] + if out_dir: + abs_img = os.path.join(out_dir, rel_path) + if os.path.exists(abs_img): + ext = os.path.splitext(abs_img)[1].lstrip('.').lower() + mime = {'jpg': 'image/jpeg', 'jpeg': 'image/jpeg', 'png': 'image/png', + 'gif': 'image/gif', 'webp': 'image/webp'}.get(ext, 'image/jpeg') + try: + with open(abs_img, 'rb') as imgf: + b64 = base64.b64encode(imgf.read()).decode('ascii') + bubble_content = f'\"图片\"' + except Exception: + bubble_content = f'\"图片\"' + else: + bubble_content = f'\"图片\"' + else: + bubble_content = f'\"图片\"' + parts.append( f'
' f'
{initial}
' f'
' f'{sender_label}' - f'
{_html_escape(m["display_content"])}
' + f'
{bubble_content}
' f'{type_tag}' f'
{m["time_str"]}
' f'
' @@ -217,6 +545,10 @@ print(f"找到 {len(db_files)} 个消息数据库") total_chats = 0 total_msgs = 0 +# ── 阶段1: 收集所有联系人的消息 ────────────────────────────────────────────── +# chat_data[chat_username] -> { dname, is_group, db_messages: [(db_name, messages)] } +chat_data: dict[str, dict] = {} + for db_path in sorted(db_files): db_name = os.path.basename(db_path) conn = sqlite3.connect(db_path) @@ -244,6 +576,8 @@ for db_path in sorted(db_files): for table_name in all_tables: h = table_name[4:] # strip "Msg_" chat_username = hash_to_username.get(h, f"unknown_{h[:8]}") + if _CONTACT_FILTER and chat_username not in _CONTACT_FILTER: + continue dname = safe_dirname(display_name(chat_username)) is_group = chat_username.endswith("@chatroom") or chat_username.endswith("@openim") @@ -287,58 +621,93 @@ for db_path in sorted(db_files): "content": content, "display_content": display_content, "is_system": is_system, - # 1-on-1: sender==chat_partner -> received(left), else sent(right) "is_received": (sender_uname == chat_username) if not is_group else True, }) - # ── 输出目录 ────────────────────────────────────────────────────────── - out_dir = os.path.join(OUTPUT_DIR, dname) - os.makedirs(out_dir, exist_ok=True) + if chat_username not in chat_data: + chat_data[chat_username] = { + "dname": dname, "is_group": is_group, "db_messages": [] + } + chat_data[chat_username]["db_messages"].append((db_name, messages)) - # ── .info 文件 ──────────────────────────────────────────────────────── - info_path = os.path.join(out_dir, ".info") - if not os.path.exists(info_path): - info = contact_map.get(chat_username, { - "username": chat_username, "alias": "", "remark": "", "nick_name": "" - }) - with open(info_path, "w", encoding="utf-8") as f: - f.write(f"username: {info['username']}\n") - f.write(f"alias: {info['alias']}\n") - f.write(f"nick_name: {info['nick_name']}\n") - f.write(f"remark: {info['remark']}\n") - f.write(f"is_group: {is_group}\n") + conn.close() - # ── CSV ─────────────────────────────────────────────────────────────── - csv_path = os.path.join(out_dir, f"{db_name}.csv") - with open(csv_path, "w", newline="", encoding="utf-8-sig") as f: - w = csv.writer(f) - w.writerow(["时间", "发送者", "消息类型", "内容", "server_id"]) +# ── 阶段2: 每个联系人解密图片一次，再写出文件 ──────────────────────────────── +total_chats = 0 +total_msgs = 0 + +for chat_username, cdata in chat_data.items(): + dname = cdata["dname"] + is_group = cdata["is_group"] + out_dir = os.path.join(OUTPUT_DIR, dname) + os.makedirs(out_dir, exist_ok=True) + + # ── .info 文件 ──────────────────────────────────────────────────────── + info_path = os.path.join(out_dir, ".info") + if not os.path.exists(info_path): + info = contact_map.get(chat_username, { + "username": chat_username, "alias": "", "remark": "", "nick_name": "" + }) + with open(info_path, "w", encoding="utf-8") as f: + f.write(f"username: {info['username']}\n") + f.write(f"alias: {info['alias']}\n") + f.write(f"nick_name: {info['nick_name']}\n") + f.write(f"remark: {info['remark']}\n") + f.write(f"is_group: {is_group}\n") + + # ── 解密图片（每个联系人只执行一次）──────────────────────────────────── + image_md5_map = {} + if _EXPORT_IMAGES: + image_md5_map = decode_chat_images(chat_username, None, out_dir) + if image_md5_map: + print(f" 图片解密: {len(image_md5_map)} 张 ({dname})") + + # ── 按 DB 写出消息文件 ──────────────────────────────────────────────── + for db_name, messages in cdata["db_messages"]: + # 建立 local_id -> rel_path 映射 + image_map = {} + if image_md5_map: for m in messages: - w.writerow([ - m["time_str"], m["sender"], m["type_name"], - m["display_content"], m["server_id"] - ]) + if m["type"] != 3: + continue + lid = m["local_id"] + file_md5 = _resource_md5_map.get((chat_username, lid)) + if file_md5 and file_md5 in image_md5_map: + image_map[lid] = image_md5_map[file_md5] - # ── JSON ────────────────────────────────────────────────────────────── - json_path = os.path.join(out_dir, f"{db_name}.json") - with open(json_path, "w", encoding="utf-8") as f: - json.dump({ - "chat_username": chat_username, - "display_name": dname, - "is_group": is_group, - "message_count": len(messages), - "messages": messages, - }, f, ensure_ascii=False, indent=2) + # ── CSV ─────────────────────────────────────────────────────────── + if not _EXPORT_FORMATS or "csv" in _EXPORT_FORMATS: + csv_path = os.path.join(out_dir, f"{db_name}.csv") + with open(csv_path, "w", newline="", encoding="utf-8-sig") as f: + w = csv.writer(f) + w.writerow(["时间", "发送者", "消息类型", "内容", "图片路径", "server_id"]) + for m in messages: + img_path = image_map.get(m["local_id"], "") if m["type"] == 3 else "" + w.writerow([ + m["time_str"], m["sender"], m["type_name"], + m["display_content"], img_path, m["server_id"] + ]) - # ── HTML ────────────────────────────────────────────────────────────── - html_path = os.path.join(out_dir, f"{db_name}.html") - _write_html(html_path, dname, is_group, messages) + # ── JSON ────────────────────────────────────────────────────────── + if not _EXPORT_FORMATS or "json" in _EXPORT_FORMATS: + json_path = os.path.join(out_dir, f"{db_name}.json") + with open(json_path, "w", encoding="utf-8") as f: + json.dump({ + "chat_username": chat_username, + "display_name": dname, + "is_group": is_group, + "message_count": len(messages), + "messages": messages, + }, f, ensure_ascii=False, indent=2) + + # ── HTML ────────────────────────────────────────────────────────── + if not _EXPORT_FORMATS or "html" in _EXPORT_FORMATS: + html_path = os.path.join(out_dir, f"{db_name}.html") + _write_html(html_path, dname, is_group, messages, image_map=image_map, out_dir=out_dir) total_chats += 1 total_msgs += len(messages) print(f" [{db_name}] {dname}: {len(messages)} 条消息") - conn.close() - print(f"\n完成: {total_chats} 个会话, 共 {total_msgs} 条消息") print(f"输出目录: {os.path.abspath(OUTPUT_DIR)}") diff --git a/export_sns.py b/export_sns.py new file mode 100644 index 0000000..c24a178 --- /dev/null +++ b/export_sns.py @@ -0,0 +1,820 @@ +"""导出微信朋友圈动态（SnsTimeLine 表） + +输出目录: //SNS/.json +媒体文件: //SNS/_. +汇总文件: //SNS/timeline.json +时间线: //SNS/timeline.html +""" +import bisect +import os +import sys +import json +import sqlite3 +import struct +import re +import xml.etree.ElementTree as ET +from datetime import datetime +from urllib.request import urlopen, Request +from urllib.error import URLError + +if sys.platform == "win32": + sys.stdout.reconfigure(encoding="utf-8", errors="replace") + +from config import load_config + +_cfg = load_config() +DECRYPTED_DIR = _cfg["decrypted_dir"] +SNS_DB_PATH = os.path.join(DECRYPTED_DIR, "sns", "sns.db") +CONTACT_DB_PATH = os.path.join(DECRYPTED_DIR, "contact", "contact.db") +OUTPUT_DIR = _cfg["output_base_dir"] + +# 图片缓存 / 解密相关配置 +IMAGE_AES_KEY = _cfg.get("image_aes_key") +IMAGE_XOR_KEY = _cfg.get("image_xor_key", 0x88) +XWECHAT_CACHE_DIR = _cfg.get("xwechat_cache_dir", "") +SNS_CACHE_DIR = _cfg.get("sns_cache_dir", "") + +# 联系人筛选（与 export_messages.py 一致） +_CONTACT_FILTER = None +_filter_raw = os.environ.get("WECHAT_EXPORT_CONTACTS", "").strip() +if _filter_raw: + _CONTACT_FILTER = set(_filter_raw.split(",")) + print(f"朋友圈联系人筛选: {len(_CONTACT_FILTER)} 个") + +# ── 媒体下载 ───────────────────────────────────────────────────────────────── + +_DOWNLOAD_TIMEOUT = 10 # 秒 + +# ── 本地缓存图片解密 & 匹配 ────────────────────────────────────────────────── + +_V2_MAGIC = b'\x07\x08V2\x08\x07' +_V1_MAGIC = b'\x07\x08V1\x08\x07' +_IMAGE_MAGICS = { + 'jpg': [0xFF, 0xD8, 0xFF], + 'png': [0x89, 0x50, 0x4E, 0x47], + 'gif': [0x47, 0x49, 0x46, 0x38], + 'webp': [0x52, 0x49, 0x46, 0x46], +} +_TIME_WINDOW = 72 * 3600 # 72 小时 + + +def _decrypt_sns_dat(dat_path): + """解密 SNS 缓存 .dat 文件，返回 bytes 或 None""" + try: + with open(dat_path, 'rb') as f: + data = f.read() + except OSError: + return None + if len(data) < 15: + return None + + head6 = data[:6] + + # V2 / V1 格式（xwechat cache） + if head6 in (_V2_MAGIC, _V1_MAGIC): + aes_key = None + if head6 == _V1_MAGIC: + aes_key = b'cfcd208495d565ef' + elif IMAGE_AES_KEY: + aes_key = IMAGE_AES_KEY.encode('ascii')[:16] if isinstance(IMAGE_AES_KEY, str) else IMAGE_AES_KEY[:16] + if not aes_key or len(aes_key) < 16: + return None + try: + from Crypto.Cipher import AES + from Crypto.Util import Padding + aes_size, xor_size = struct.unpack_from(' len(data): + return None + cipher = AES.new(aes_key[:16], AES.MODE_ECB) + dec_aes = Padding.unpad(cipher.decrypt(data[offset:offset + aligned]), AES.block_size) + offset += aligned + raw_end = len(data) - xor_size + raw_data = data[offset:raw_end] if offset < raw_end else b'' + xor_data = data[raw_end:] + xor_key = IMAGE_XOR_KEY if isinstance(IMAGE_XOR_KEY, int) else 0x88 + dec_xor = bytes(b ^ xor_key for b in xor_data) + return dec_aes + raw_data + dec_xor + except Exception: + return None + + # 旧 XOR 格式（FileStorage Sns Cache） + for magic in _IMAGE_MAGICS.values(): + key = data[0] ^ magic[0] + if all(i < len(data) and (data[i] ^ key) == magic[i] for i in range(len(magic))): + return bytes(b ^ key for b in data) + + return None + + +def _detect_format(header): + """检测解密后数据的图片格式，返回扩展名""" + if header[:3] == b'\xff\xd8\xff': + return 'jpg' + if header[:4] == b'\x89PNG': + return 'png' + if header[:3] == b'GIF': + return 'gif' + if header[:4] == b'RIFF' and len(header) >= 12 and header[8:12] == b'WEBP': + return 'webp' + return 'bin' + + +def _image_size_from_bytes(data): + """从解密后的图片数据提取 (width, height)，失败返回 (0, 0)""" + if not data or len(data) < 24: + return 0, 0 + + # PNG: IHDR 位于字节 16-24 + if data[:4] == b'\x89PNG': + w = struct.unpack('>I', data[16:20])[0] + h = struct.unpack('>I', data[20:24])[0] + return w, h + + # JPEG: 查找 SOF 标记 + if data[:2] == b'\xff\xd8': + i = 2 + while i < len(data) - 9: + if data[i] != 0xFF: + break + marker = data[i + 1] + if 0xC0 <= marker <= 0xCF and marker not in (0xC4, 0xC8, 0xCC): + h = struct.unpack('>H', data[i + 5:i + 7])[0] + w = struct.unpack('>H', data[i + 7:i + 9])[0] + return w, h + if i + 3 >= len(data): + break + seg_len = struct.unpack('>H', data[i + 2:i + 4])[0] + i += 2 + seg_len + return 0, 0 + + # WEBP VP8 + if data[:4] == b'RIFF' and len(data) >= 30 and data[8:12] == b'WEBP': + if data[12:16] == b'VP8 ': + w = struct.unpack('/YYYY-MM/Sns/Img/<2hex>/<30hex> + if XWECHAT_CACHE_DIR and os.path.isdir(XWECHAT_CACHE_DIR): + for month_dir in os.listdir(XWECHAT_CACHE_DIR): + sns_img = os.path.join(XWECHAT_CACHE_DIR, month_dir, "Sns", "Img") + if not os.path.isdir(sns_img): + continue + for sub in os.listdir(sns_img): + sub_path = os.path.join(sns_img, sub) + if not os.path.isdir(sub_path): + continue + for fname in os.listdir(sub_path): + fp = os.path.join(sub_path, fname) + if os.path.isfile(fp): + raw_paths.append(fp) + + # 2. FileStorage Sns Cache: /YYYY-MM/ + if SNS_CACHE_DIR and os.path.isdir(SNS_CACHE_DIR): + for month_dir in os.listdir(SNS_CACHE_DIR): + month_path = os.path.join(SNS_CACHE_DIR, month_dir) + if not os.path.isdir(month_path): + continue + for fname in os.listdir(month_path): + if fname.endswith('_t'): # 跳过缩略图 + continue + fp = os.path.join(month_path, fname) + if os.path.isfile(fp): + raw_paths.append(fp) + + if not raw_paths: + return [] + + print(f" 预读取 {len(raw_paths)} 个缓存文件元数据...") + + # 准备 AES key（避免在循环内重复构造） + aes_key = None + if IMAGE_AES_KEY: + aes_key = IMAGE_AES_KEY.encode('ascii')[:16] if isinstance(IMAGE_AES_KEY, str) else IMAGE_AES_KEY[:16] + + entries = [] + for path in raw_paths: + try: + fsize = os.path.getsize(path) + mtime = os.path.getmtime(path) + if fsize < 15: + continue + + with open(path, 'rb') as f: + data = f.read(min(fsize, 4096)) + + head6 = data[:6] + dec_header = None + est_dec_size = fsize + + if head6 in (_V2_MAGIC, _V1_MAGIC): + # V2/V1: 解密 AES 部分获取文件头 + k = b'cfcd208495d565ef' if head6 == _V1_MAGIC else aes_key + if not k or len(k) < 16: + continue + try: + from Crypto.Cipher import AES as _AES + aes_size, xor_size = struct.unpack_from('= hi: + lo, hi = 0, len(index) + + used_paths = set() + + for media in media_list: + mtype = media.get("type", "") + if mtype not in ("2", ""): + results.append((None, None)) + continue + + want_w = int(media.get("width") or 0) + want_h = int(media.get("height") or 0) + want_size = int(media.get("total_size") or 0) + + candidates = [] # (score, path, fmt) + + for i in range(lo, hi): + mtime_i, path_i, dec_size_i, fmt_i, w_i, h_i = index[i] + if path_i in used_paths: + continue + + # 尺寸匹配 + if want_w > 0 and want_h > 0 and w_i > 0 and h_i > 0: + if w_i != want_w or h_i != want_h: + continue + + # 大小匹配 + if want_size > 0: + if dec_size_i > want_size * 3 or dec_size_i < want_size * 0.3: + continue + + size_diff = abs(dec_size_i - want_size) if want_size > 0 else 0 + time_diff = abs(mtime_i - create_time) + candidates.append((size_diff, time_diff, path_i, fmt_i)) + + if candidates: + candidates.sort(key=lambda x: (x[0], x[1])) + best = candidates[0] + used_paths.add(best[2]) + results.append((best[2], best[3])) + else: + results.append((None, None)) + + return results + +# ContentObject type 含义（已知） +_CONTENT_TYPES = { + 1: "图文", + 2: "纯文本", + 3: "链接", + 5: "视频链接", + 7: "位置", + 15: "视频", + 28: "短视频", + 30: "音乐", + 34: "笔记", + 42: "小程序", + 54: "直播", +} + + +def _try_download_media(url, save_path): + """尝试下载媒体文件，返回 True/False + + 微信朋友圈 shmmsns.qpic.cn 图片需要携带 Referer 和 User-Agent。 + 注意: URL 返回的数据可能是加密的（enc_idx=1 的情况）， + 解密算法尚未公开，此时下载的文件无法直接查看。 + 如果下载失败返回 False，后续可替换为更复杂的下载逻辑。 + """ + if not url or not url.startswith("http"): + return False + try: + req = Request(url, headers={ + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/120.0.0.0 Safari/537.36", + "Referer": "https://weixin.qq.com/", + }) + with urlopen(req, timeout=_DOWNLOAD_TIMEOUT) as resp: + if resp.status != 200: + return False + data = resp.read() + if len(data) < 100: + return False + # 检测格式 + if data[:3] == b'\xff\xd8\xff': + ext = '.jpg' + elif data[:4] == b'\x89PNG': + ext = '.png' + elif data[:4] == b'GIF8': + ext = '.gif' + elif data[:4] == b'RIFF' and data[8:12] == b'WEBP': + ext = '.webp' + else: + ext = '.bin' + if not os.path.splitext(save_path)[1]: + save_path += ext + with open(save_path, 'wb') as f: + f.write(data) + return True + except (URLError, OSError, Exception): + return False + + +def _parse_media_list(timeline_obj): + """解析 TimelineObject 中的 mediaList，返回 media 信息列表""" + medias = [] + for media_el in timeline_obj.findall('.//media'): + media_type = media_el.findtext('type', '') + sub_type = media_el.findtext('sub_type', '') + vid_duration = media_el.findtext('videoDuration', '0') + + thumb_el = media_el.find('thumb') + url_el = media_el.find('url') + size_el = media_el.find('size') + + info = { + "type": media_type, + "sub_type": sub_type, + "video_duration": vid_duration, + } + + if thumb_el is not None: + info["thumb_url"] = thumb_el.text or "" + info["thumb_key"] = thumb_el.get("key", "") + info["thumb_token"] = thumb_el.get("token", "") + + if url_el is not None: + info["url"] = url_el.text or "" + info["url_md5"] = url_el.get("md5", "") + info["url_key"] = url_el.get("key", "") + info["url_token"] = url_el.get("token", "") + + if size_el is not None: + info["width"] = size_el.get("width", "") + info["height"] = size_el.get("height", "") + info["total_size"] = size_el.get("totalSize", "") + + medias.append(info) + return medias + + +def _parse_timeline_xml(content_xml): + """解析 SnsTimeLine 的 Content XML，返回结构化数据""" + try: + root = ET.fromstring(content_xml) + except ET.ParseError: + return None + + tl = root.find('.//TimelineObject') + if tl is None: + return None + + create_time_str = tl.findtext('createTime', '0') + try: + create_time = int(create_time_str) + except ValueError: + create_time = 0 + + content_type = tl.findtext('.//ContentObject/type', '0') + try: + content_type_int = int(content_type) + except ValueError: + content_type_int = 0 + + # 解析位置 + loc_el = tl.find('.//location') + location = None + if loc_el is not None: + lat = loc_el.get('latitude', '0') + lon = loc_el.get('longitude', '0') + if lat != '0' or lon != '0': + location = { + "latitude": lat, + "longitude": lon, + "poi_name": loc_el.get("poiName", ""), + } + + return { + "id": tl.findtext('id', ''), + "username": tl.findtext('username', ''), + "create_time": create_time, + "create_time_str": datetime.fromtimestamp(create_time).strftime("%Y-%m-%d %H:%M:%S") if create_time else "", + "content_desc": tl.findtext('contentDesc', ''), + "content_type": content_type_int, + "content_type_name": _CONTENT_TYPES.get(content_type_int, f"未知({content_type_int})"), + "nickname": root.findtext('.//LocalExtraInfo/nickname', ''), + "is_private": tl.findtext('private', '0') == '1', + "location": location, + "media": _parse_media_list(tl), + } + + +def _load_comments(conn): + """加载 SnsMessage_tmp3 评论/点赞，按 feed_id 分组""" + comments = {} + try: + rows = conn.execute( + "SELECT feed_id, create_time, type, from_username, from_nickname," + " to_username, to_nickname, content" + " FROM SnsMessage_tmp3 ORDER BY create_time" + ).fetchall() + for feed_id, ctime, ctype, from_u, from_n, to_u, to_n, content in rows: + if feed_id not in comments: + comments[feed_id] = [] + comments[feed_id].append({ + "create_time": ctime, + "create_time_str": datetime.fromtimestamp(ctime).strftime("%Y-%m-%d %H:%M:%S") if ctime else "", + "type": ctype, # 1=点赞, 2=评论 + "type_name": "点赞" if ctype == 1 else "评论" if ctype == 2 else f"未知({ctype})", + "from_username": from_u or "", + "from_nickname": from_n or "", + "to_username": to_u or "", + "to_nickname": to_n or "", + "content": content or "", + }) + except Exception as e: + print(f"读取评论数据失败: {e}") + return comments + + +def _safe_dirname(name: str) -> str: + """清理文件夹名中的非法字符""" + for ch in r'\/:*?"<>|': + name = name.replace(ch, "_") + return name.strip() or "unknown" + + +def _load_contact_map(): + """从 contact.db 加载 {username: display_name}""" + cmap = {} + if not os.path.exists(CONTACT_DB_PATH): + return cmap + try: + conn = sqlite3.connect(CONTACT_DB_PATH) + for uname, remark, nick_name in conn.execute( + "SELECT username, remark, nick_name FROM contact" + ): + dname = remark or nick_name or uname + cmap[uname] = _safe_dirname(dname) + conn.close() + except Exception as e: + print(f"读取联系人数据库失败: {e}") + return cmap + + +def _timestamp_filename(unix_ts): + """Unix 时间戳 → yyyyMMddHHmmss000 文件名（毫秒部分为 000）""" + if not unix_ts: + return "00000000000000000" + dt = datetime.fromtimestamp(unix_ts) + return dt.strftime("%Y%m%d%H%M%S") + "000" + + +def _html_escape(text): + """简单 HTML 转义""" + return (text or "").replace("&", "&").replace("<", "<").replace(">", ">").replace('"', """) + + +def _generate_timeline_html(display_name, posts, sns_dir, image_files): + """生成朋友圈时间线 HTML + + Args: + display_name: 联系人显示名 + posts: 按时间倒序排列的动态列表 + sns_dir: SNS 输出目录 + image_files: {final_name: [(rel_path, ext), ...]} 每条动态的图片文件列表 + """ + html_path = os.path.join(sns_dir, "timeline.html") + parts = [f""" + + + + +{_html_escape(display_name)} - 朋友圈 + + + +

{_html_escape(display_name)} 的朋友圈

+
共 {len(posts)} 条动态
+"""] + + for post in posts: + final_name = post.get("_final_name", "") + time_str = _html_escape(post.get("create_time_str", "")) + type_name = _html_escape(post.get("content_type_name", "")) + text = _html_escape(post.get("content_desc", "")) + is_private = post.get("is_private", False) + + parts.append('
') + parts.append(f'
{time_str}{type_name}') + if is_private: + parts.append('仅自己可见') + parts.append('
') + + if text: + parts.append(f'
{text}
') + + # 图片 + imgs = image_files.get(final_name, []) + if imgs: + parts.append('
') + for rel_path, _ in imgs: + parts.append(f'') + parts.append('
') + + # 位置 + loc = post.get("location") + if loc and loc.get("poi_name"): + parts.append(f'
📍 {_html_escape(loc["poi_name"])}
') + + # 评论 + comments = post.get("comments", []) + if comments: + parts.append('
') + for c in comments: + if c.get("type") == 1: + parts.append(f'
❤️ {_html_escape(c["from_nickname"])}
') + else: + to_part = "" + if c.get("to_nickname"): + to_part = f' 回复 {_html_escape(c["to_nickname"])}' + parts.append(f'
{_html_escape(c["from_nickname"])}{to_part}: {_html_escape(c.get("content", ""))}
') + parts.append('
') + + parts.append('
') + + parts.append('') + + with open(html_path, 'w', encoding='utf-8') as f: + f.write('\n'.join(parts)) + return html_path + + +def export_sns_timeline(): + """导出朋友圈动态主函数""" + if not os.path.exists(SNS_DB_PATH): + print(f"朋友圈数据库不存在: {SNS_DB_PATH}") + print("请先运行「解密数据库」") + return + + # 加载联系人 + contact_map = _load_contact_map() + print(f"联系人: {len(contact_map)} 条") + + conn = sqlite3.connect(SNS_DB_PATH) + + # 加载评论 + print("加载评论数据...") + comments_map = _load_comments(conn) + print(f"评论/点赞: {sum(len(v) for v in comments_map.values())} 条") + + # 读取所有动态 + print("读取朋友圈动态...") + rows = conn.execute( + "SELECT tid, user_name, content FROM SnsTimeLine WHERE content IS NOT NULL" + ).fetchall() + conn.close() + + print(f"共 {len(rows)} 条动态") + if not rows: + return + + # 是否尝试下载媒体 + try_download = os.environ.get("WECHAT_SNS_DOWNLOAD_MEDIA", "0").strip() == "1" + + # ── 构建缓存索引 ───────────────────────────────────────────────────── + print("扫描 SNS 图片缓存...") + cache_index = _build_sns_cache_index() + index_mtimes = [e[0] for e in cache_index] + print(f"缓存索引: {len(cache_index)} 个有效图片文件") + + # ── 按 user_name 分组 ───────────────────────────────────────────────── + user_posts: dict[str, list] = {} # user_name -> [post, ...] + user_nicknames: dict[str, str] = {} # user_name -> nickname (从 XML 提取) + skipped = 0 + + for tid, user_name, content_xml in rows: + if not content_xml: + continue + if _CONTACT_FILTER and user_name not in _CONTACT_FILTER: + skipped += 1 + continue + + post = _parse_timeline_xml(content_xml) + if not post: + continue + + post["tid"] = tid + post["db_user_name"] = user_name or "" + post["comments"] = comments_map.get(tid, []) + + key = user_name or "unknown" + if key not in user_posts: + user_posts[key] = [] + user_posts[key].append(post) + + # 记录 nickname（取第一个非空的） + nick = post.get("nickname", "") + if nick and key not in user_nicknames: + user_nicknames[key] = nick + + if skipped: + print(f"筛选跳过: {skipped} 条") + + # ── 按联系人输出 ────────────────────────────────────────────────────── + total_posts = 0 + cache_match_ok = 0 + cache_match_fail = 0 + media_download_ok = 0 + media_download_fail = 0 + + for user_name, posts in user_posts.items(): + dname = contact_map.get(user_name) or _safe_dirname( + user_nicknames.get(user_name) or user_name + ) + sns_dir = os.path.join(OUTPUT_DIR, dname, "SNS") + os.makedirs(sns_dir, exist_ok=True) + + # 用 set 处理同一秒多条动态的文件名冲突 + used_names = set() + # image_files: {final_name: [(rel_path, ext), ...]} 用于 HTML 生成 + image_files: dict[str, list] = {} + + posts.sort(key=lambda p: p.get("create_time", 0)) + + for post in posts: + ts_name = _timestamp_filename(post.get("create_time")) + # 处理冲突: 递增末尾毫秒 + final_name = ts_name + counter = 1 + while final_name in used_names: + final_name = ts_name[:-3] + f"{counter:03d}" + counter += 1 + used_names.add(final_name) + post["_final_name"] = final_name + + # ── 缓存图片匹配 ───────────────────────────────────────── + media_list = post.get("media", []) + if media_list and cache_index: + matches = _match_cache_images( + post.get("create_time", 0), media_list, + cache_index, index_mtimes, + ) + for i, (matched_path, fmt) in enumerate(matches): + if matched_path is not None: + dec_bytes = _decrypt_sns_dat(matched_path) + if dec_bytes: + ext = _detect_format(dec_bytes[:16]) + img_name = f"{final_name}_{i}.{ext}" + img_path = os.path.join(sns_dir, img_name) + with open(img_path, 'wb') as f: + f.write(dec_bytes) + if final_name not in image_files: + image_files[final_name] = [] + image_files[final_name].append((img_name, ext)) + cache_match_ok += 1 + continue + cache_match_fail += 1 + else: + cache_match_fail += 1 + + # ── 网络下载（仅对缓存未匹配的媒体尝试） ───────────────── + if try_download and media_list: + existing_imgs = image_files.get(final_name, []) + existing_indices = {int(p.rsplit('_', 1)[1].split('.')[0]) for p, _ in existing_imgs} if existing_imgs else set() + for i, media in enumerate(media_list): + if i in existing_indices: + continue + media_url = media.get("url", "") or media.get("thumb_url", "") + if not media_url: + continue + save_name = os.path.join(sns_dir, f"{final_name}_{i}") + if _try_download_media(media_url, save_name): + # 下载成功后更新 image_files + for cand_ext in ('.jpg', '.png', '.gif', '.webp', '.bin'): + if os.path.exists(save_name + cand_ext): + if final_name not in image_files: + image_files[final_name] = [] + image_files[final_name].append((f"{final_name}_{i}{cand_ext}", cand_ext[1:])) + break + media_download_ok += 1 + else: + media_download_fail += 1 + + # 保存 JSON（去掉内部字段） + post_out = {k: v for k, v in post.items() if not k.startswith("_")} + post_file = os.path.join(sns_dir, f"{final_name}.json") + with open(post_file, 'w', encoding='utf-8') as f: + json.dump(post_out, f, ensure_ascii=False, indent=2) + + total_posts += 1 + + # 每个联系人的汇总 JSON + posts.sort(key=lambda p: p.get("create_time", 0), reverse=True) + summary_posts = [{k: v for k, v in p.items() if not k.startswith("_")} for p in posts] + summary_path = os.path.join(sns_dir, "timeline.json") + with open(summary_path, 'w', encoding='utf-8') as f: + json.dump({ + "user_name": user_name, + "display_name": dname, + "export_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), + "total_posts": len(posts), + "posts": summary_posts, + }, f, ensure_ascii=False, indent=2) + + # 生成 HTML 时间线 + _generate_timeline_html(dname, posts, sns_dir, image_files) + + print(f" {dname}: {len(posts)} 条动态") + + print(f"\n完成: {len(user_posts)} 个联系人, 共 {total_posts} 条动态") + if cache_index: + print(f"缓存匹配: 成功 {cache_match_ok}, 失败 {cache_match_fail}") + if try_download: + print(f"媒体下载: 成功 {media_download_ok}, 失败 {media_download_fail}") + print(f"输出目录: {os.path.abspath(OUTPUT_DIR)}") + + +if __name__ == "__main__": + export_sns_timeline() diff --git a/find_image_key.py b/find_image_key.py index afb9ad1..b034660 100644 --- a/find_image_key.py +++ b/find_image_key.py @@ -333,9 +333,15 @@ def verify_and_decrypt(attach_dir, aes_key_str, xor_key): def main(): - config_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'config.json') - with open(config_path, encoding="utf-8") as f: - config = json.load(f) + from config import _config_file_path, load_config + config_path = _config_file_path() + # 加载完整配置用于逻辑，读取原始 JSON 用于保存 + config = load_config() + try: + with open(config_path, encoding="utf-8") as f: + config_raw = json.load(f) + except (FileNotFoundError, json.JSONDecodeError): + config_raw = {} db_dir = config['db_dir'] base_dir = os.path.dirname(db_dir) @@ -392,8 +398,11 @@ def main(): config['image_aes_key'] = aes_key if xor_key is not None: config['image_xor_key'] = xor_key + config_raw['image_aes_key'] = aes_key + if xor_key is not None: + config_raw['image_xor_key'] = xor_key with open(config_path, 'w', encoding="utf-8") as f: - json.dump(config, f, indent=2, ensure_ascii=False) + json.dump(config_raw, f, indent=2, ensure_ascii=False) print(f"Saved to {config_path}", flush=True) print("\n=== Verify decrypt ===", flush=True) diff --git a/find_image_key_monitor.py b/find_image_key_monitor.py index 437a47f..fb68cee 100644 --- a/find_image_key_monitor.py +++ b/find_image_key_monitor.py @@ -226,9 +226,14 @@ def verify_and_decrypt(attach_dir, aes_key_str, xor_key): def main(): - config_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'config.json') - with open(config_path, encoding="utf-8") as f: - config = json.load(f) + from config import _config_file_path, load_config + config_path = _config_file_path() + config = load_config() + try: + with open(config_path, encoding="utf-8") as f: + config_raw = json.load(f) + except (FileNotFoundError, json.JSONDecodeError): + config_raw = {} db_dir = config['db_dir'] base_dir = os.path.dirname(db_dir) @@ -292,8 +297,11 @@ def main(): config['image_aes_key'] = aes_key if xor_key is not None: config['image_xor_key'] = xor_key + config_raw['image_aes_key'] = aes_key + if xor_key is not None: + config_raw['image_xor_key'] = xor_key with open(config_path, 'w', encoding="utf-8") as f: - json.dump(config, f, indent=2, ensure_ascii=False) + json.dump(config_raw, f, indent=2, ensure_ascii=False) print(f"Saved to {config_path}", flush=True) verify_and_decrypt(attach_dir, aes_key, xor_key) diff --git a/voice_to_mp3.py b/voice_to_mp3.py index e8d7ebe..b113806 100644 --- a/voice_to_mp3.py +++ b/voice_to_mp3.py @@ -15,7 +15,13 @@ from config import load_config _cfg = load_config() DB_PATH = os.path.join(_cfg["decrypted_dir"], "message", "media_0.db") CONTACT_DB_PATH = os.path.join(_cfg["decrypted_dir"], "contact", "contact.db") -OUTPUT_DIR = os.path.join(_cfg["wechat_base_dir"], "voice") +OUTPUT_DIR = _cfg["output_base_dir"] + +_CONTACT_FILTER = None +_filter_raw = os.environ.get("WECHAT_EXPORT_CONTACTS", "").strip() +if _filter_raw: + _CONTACT_FILTER = set(_filter_raw.split(",")) + print(f"联系人筛选: {len(_CONTACT_FILTER)} 个") def silk_to_mp3(voice_data, output_path): """将微信 SILK 语音数据转换为 MP3""" @@ -92,15 +98,17 @@ success = 0 fail = 0 for chat_name_id, create_time, local_id, voice_data in rows: user_name = name_map.get(chat_name_id, f"unknown_{chat_name_id}") + if _CONTACT_FILTER and user_name not in _CONTACT_FILTER: + continue dname = safe_dirname(display_name(user_name)) dt = datetime.fromtimestamp(create_time) filename = dt.strftime("%Y%m%d_%H%M%S") + f"_{local_id}.mp3" - user_dir = os.path.join(OUTPUT_DIR, dname) + user_dir = os.path.join(OUTPUT_DIR, dname, "voice") os.makedirs(user_dir, exist_ok=True) - # 写入 .info 文件（只写一次） - info_path = os.path.join(user_dir, ".info") + # 写入 .info 文件（只写一次，写到联系人根目录） + info_path = os.path.join(OUTPUT_DIR, dname, ".info") if not os.path.exists(info_path): info = contact_map.get(user_name, {"username": user_name, "alias": "", "remark": "", "nick_name": ""}) with open(info_path, "w", encoding="utf-8") as f: