diff --git a/frontend/composables/useApi.js b/frontend/composables/useApi.js index 13efb89..a5be2d2 100644 --- a/frontend/composables/useApi.js +++ b/frontend/composables/useApi.js @@ -61,6 +61,22 @@ export const useApi = () => { body: data }) } + + // 导入预览API + const importDecryptedPreview = async (data) => { + return await request('/import_decrypted/preview', { + method: 'POST', + body: data + }) + } + + // 导入已解密目录API + const importDecrypted = async (data) => { + return await request('/import_decrypted', { + method: 'POST', + body: data + }) + } // 健康检查API const healthCheck = async () => { @@ -601,10 +617,21 @@ export const useApi = () => { return `${base}/biz/proxy_image?${query.toString()}` } + const pickSystemDirectory = async (params = {}) => { + const query = new URLSearchParams() + if (params && params.title) query.set('title', params.title) + if (params && params.initial_dir) query.set('initial_dir', params.initial_dir) + const url = '/system/pick_directory' + (query.toString() ? `?${query.toString()}` : '') + return await request(url) + } + return { + pickSystemDirectory, detectWechat, detectCurrentAccount, decryptDatabase, + importDecryptedPreview, + importDecrypted, healthCheck, listChatAccounts, getChatAccountInfo, diff --git a/frontend/pages/detection-result.vue b/frontend/pages/detection-result.vue index f05d5f7..5b677fa 100644 --- a/frontend/pages/detection-result.vue +++ b/frontend/pages/detection-result.vue @@ -24,26 +24,36 @@ - -
- -
- - + +
+
+

+ 未找到想要的账号？ + +

+

+ 当前指定检测路径：{{ customPath }} + 如果自动检测漏了，您可以手动指定微信数据根目录 (通常名为 xwechat_files) 让系统重新扫描。 +

-

未找到时可填写 xwechat_files 根目录。

+
-
- +
+ @@ -51,16 +61,16 @@
-
+
-
+
- +
-

检测失败

-

{{ detectionResult.error }}

+

未找到微信数据

+

{{ detectionResult.error }}

@@ -114,89 +124,91 @@
-
-
-

微信账户详情

-
-
- -
-
-
-
-
- {{ account.account_name?.charAt(0)?.toUpperCase() || 'U' }} -
-
-
-

{{ account.account_name || '未知账户' }}

- - - - - - 当前登录 - -
-
- - - - - {{ account.database_count }} 个数据库 - - - - - - 数据目录已找到 - +
+
+

可操作的微信账户

+ 点击解密即可提取数据 +
+
+
+ +
+ + + 最近登录账户 + +
+ +
+
+
+ + + +
+
+ + +
+ +
+ + + + + {{ account.database_count }} 个库文件 + + + + + + 路径已确认 + +
+ + +
+ +
+

+ 📂 {{ account.data_dir }} +

- -
- - -
-

- 数据路径：{{ account.data_dir }} -

+ + +
+ + + +

没有在这台设备上发现微信数据

+

您可以尝试通过上方的按钮手动指定 "xwechat_files" 文件夹路径。

+
- - -
- - - -

未检测到微信账户数据

-
-
-
- - -
- - - -

暂无检测结果

- - 返回首页开始检测 -
@@ -204,34 +216,79 @@ - - + \ No newline at end of file diff --git a/frontend/pages/import.vue b/frontend/pages/import.vue new file mode 100644 index 0000000..18eb5fe --- /dev/null +++ b/frontend/pages/import.vue @@ -0,0 +1,276 @@ + + + + + diff --git a/frontend/pages/index.vue b/frontend/pages/index.vue index 95995d7..c788bf0 100644 --- a/frontend/pages/index.vue +++ b/frontend/pages/index.vue @@ -41,6 +41,14 @@ 直接解密 + + + + + + 数据导入 + @@ -70,6 +78,8 @@ import { onMounted } from 'vue' import { useApi } from '~/composables/useApi' import { DESKTOP_SETTING_DEFAULT_TO_CHAT_KEY, readLocalBoolSetting } from '~/lib/desktop-settings' +const { listChatAccounts } = useApi() + onMounted(async () => { if (!process.client || typeof window === 'undefined') return @@ -77,8 +87,7 @@ onMounted(async () => { if (!enabled) return try { - const api = useApi() - const resp = await api.listChatAccounts() + const resp = await listChatAccounts() const accounts = resp?.accounts || [] if (accounts.length) { await navigateTo('/chat', { replace: true }) diff --git a/src/wechat_decrypt_tool/api.py b/src/wechat_decrypt_tool/api.py index bc2df4d..34a25c0 100644 --- a/src/wechat_decrypt_tool/api.py +++ b/src/wechat_decrypt_tool/api.py @@ -25,6 +25,7 @@ from .routers.chat_contacts import router as _chat_contacts_router from .routers.chat_export import router as _chat_export_router from .routers.chat_media import router as _chat_media_router from .routers.decrypt import router as _decrypt_router +from .routers.import_decrypted import router as _import_decrypted_router from .routers.health import router as _health_router from .routers.admin import router as _admin_router from .routers.keys import router as _keys_router @@ -37,6 +38,7 @@ from .request_logging import log_server_errors_middleware from .sns_stage_timing import add_sns_stage_timing_headers from .wcdb_realtime import WCDB_REALTIME, shutdown as _wcdb_shutdown from .routers.biz import router as _biz_router +from .routers.system import router as _system_router app = FastAPI( title="微信数据库解密工具", @@ -87,6 +89,7 @@ async def _log_server_errors(request: Request, call_next): app.include_router(_health_router) app.include_router(_admin_router) app.include_router(_wechat_detection_router) +app.include_router(_import_decrypted_router) app.include_router(_decrypt_router) app.include_router(_keys_router) app.include_router(_media_router) @@ -98,6 +101,7 @@ app.include_router(_sns_router) app.include_router(_sns_export_router) app.include_router(_wrapped_router) app.include_router(_biz_router) +app.include_router(_system_router) class _SPAStaticFiles(StaticFiles): diff --git a/src/wechat_decrypt_tool/routers/import_decrypted.py b/src/wechat_decrypt_tool/routers/import_decrypted.py new file mode 100644 index 0000000..fec4884 --- /dev/null +++ b/src/wechat_decrypt_tool/routers/import_decrypted.py @@ -0,0 +1,271 @@ +from __future__ import annotations + +import os +import shutil +import json +import asyncio +from pathlib import Path +from typing import Optional +from fastapi import APIRouter, HTTPException, Query +from fastapi.responses import StreamingResponse +from pydantic import BaseModel, Field + +from ..app_paths import get_output_databases_dir +from ..logging_config import get_logger +from ..path_fix import PathFixRoute +from ..session_last_message import build_session_last_message_table +from ..media_helpers import _wxgf_to_image_bytes + +logger = get_logger(__name__) + +router = APIRouter(route_class=PathFixRoute) + +class ImportRequest(BaseModel): + import_path: str = Field(..., description="已解密的数据库和资源所在目录的绝对路径") + +def _is_valid_sqlite(path: Path) -> bool: + SQLITE_HEADER = b"SQLite format 3\x00" + try: + if not path.exists() or not path.is_file(): + return False + with path.open("rb") as f: + return f.read(len(SQLITE_HEADER)) == SQLITE_HEADER + except Exception: + return False + +def _validate_import_structure(import_path: Path) -> dict: + """ + 验证导入目录结构： + - databases/ (必须包含 contact.db, session.db) + - resource/ (可选) + - account.json (必须包含 username, nick) + """ + db_dir = import_path / "databases" + account_json_path = import_path / "account.json" + + if not db_dir.exists() or not db_dir.is_dir(): + raise HTTPException(status_code=400, detail="未找到 databases 目录") + + if not account_json_path.exists(): + raise HTTPException(status_code=400, detail="未找到 account.json 文件") + + # 验证关键数据库 + required_dbs = ["contact.db", "session.db"] + for db_name in required_dbs: + if not _is_valid_sqlite(db_dir / db_name): + raise HTTPException(status_code=400, detail=f"databases 目录中未找到有效的 {db_name}") + + # 解析 account.json + try: + account_info = json.loads(account_json_path.read_text(encoding="utf-8")) + except Exception as e: + raise HTTPException(status_code=400, detail=f"解析 account.json 失败: {e}") + + username = account_info.get("username") + nick = account_info.get("nick") + + if not username or not nick: + raise HTTPException(status_code=400, detail="account.json 中缺少 username 或 nick") + + return { + "username": username, + "nick": nick, + "avatar_url": account_info.get("avatar_url", ""), + "has_resource": (import_path / "resource").exists() + } + +@router.post("/api/import_decrypted/preview", summary="预览待导入的账号信息") +async def preview_import(request: ImportRequest): + import_path = Path(request.import_path.strip()) + if not import_path.exists() or not import_path.is_dir(): + raise HTTPException(status_code=400, detail="导入路径不存在或不是目录") + + return _validate_import_structure(import_path) + +@router.get("/api/import_decrypted", summary="执行导入已解密的数据库和资源目录 (SSE)") +async def import_decrypted_directory( + import_path: str = Query(..., description="已解密的数据库和资源所在目录的绝对路径") +): + import_path_obj = Path(import_path.strip()) + + def _sse(data: dict): + return f"data: {json.dumps(data, ensure_ascii=False)}\n\n" + + async def generate_progress(): + try: + if not import_path_obj.exists() or not import_path_obj.is_dir(): + yield _sse({"type": "error", "message": "导入路径不存在或不是目录"}) + return + + yield _sse({"type": "progress", "percent": 5, "message": "正在验证目录结构..."}) + # 1. 验证并获取账号信息 + try: + info = await asyncio.to_thread(_validate_import_structure, import_path_obj) + except HTTPException as e: + yield _sse({"type": "error", "message": e.detail}) + return + except Exception as e: + yield _sse({"type": "error", "message": f"验证失败: {e}"}) + return + + account_name = info["username"] + yield _sse({"type": "progress", "percent": 10, "message": f"验证成功: {account_name}"}) + + # 2. 准备输出目录 + output_base = get_output_databases_dir() + account_output_dir = output_base / account_name + await asyncio.to_thread(account_output_dir.mkdir, parents=True, exist_ok=True) + + yield _sse({"type": "progress", "percent": 15, "message": "正在准备目标目录..."}) + + # 3. 导入 databases 目录下的 .db 文件 + db_src_dir = import_path_obj / "databases" + db_files = [f for f in db_src_dir.iterdir() if f.is_file() and f.suffix == ".db"] + imported_files = [] + + for i, item in enumerate(db_files): + target = account_output_dir / item.name + def _do_import_db(src, dst): + if dst.exists(): + dst.unlink() + try: + os.link(src, dst) + except Exception: + shutil.copy2(src, dst) + + try: + await asyncio.to_thread(_do_import_db, item, target) + imported_files.append(item.name) + except Exception as e: + logger.error(f"导入数据库失败: {item.name}, error: {e}") + + percent = 15 + int((i + 1) / (len(db_files) or 1) * 15) + yield _sse({"type": "progress", "percent": percent, "message": f"正在导入数据库: {item.name}"}) + + # 4. 导入 resource 目录 + resource_src = import_path_obj / "resource" + if resource_src.exists() and resource_src.is_dir(): + yield _sse({"type": "progress", "percent": 30, "message": "正在导入资源文件 (这可能需要一些时间)..."}) + resource_dst = account_output_dir / "resource" + + def _do_import_resource(src, dst): + if dst.exists(): + if dst.is_symlink() or dst.is_file(): + dst.unlink() + else: + shutil.rmtree(dst) + try: + os.symlink(src, dst, target_is_directory=True) + except Exception: + shutil.copytree(src, dst, dirs_exist_ok=True) + + try: + await asyncio.to_thread(_do_import_resource, resource_src, resource_dst) + except Exception as e: + logger.error(f"导入 resource 目录失败: {e}") + + # 5. 转换 .wxgf 资源 (新增加的流程) + yield _sse({"type": "progress", "percent": 50, "message": "正在搜索并转换 .wxgf 图片..."}) + + if resource_dst.exists(): + # 搜索 wxgf 文件 + def _find_wxgf(root_dir): + found = [] + for root, _, files in os.walk(root_dir): + for f in files: + if f.lower().endswith(".wxgf"): + found.append(Path(root) / f) + return found + + wxgf_files = await asyncio.to_thread(_find_wxgf, resource_dst) + + if wxgf_files: + total_wxgf = len(wxgf_files) + converted_count = 0 + for i, wxgf_path in enumerate(wxgf_files): + def _convert_one(p): + jpg_p = p.with_suffix(".wxgf.jpg") + if not jpg_p.exists(): + data = p.read_bytes() + if data.startswith(b"wxgf"): + converted = _wxgf_to_image_bytes(data) + if converted: + jpg_p.write_bytes(converted) + return True + else: + return True # 已经存在视为成功 + return False + + try: + success = await asyncio.to_thread(_convert_one, wxgf_path) + if success: + converted_count += 1 + except Exception as e: + logger.error(f"转换 wxgf 失败: {wxgf_path}, {e}") + + if i % max(1, total_wxgf // 20) == 0 or i == total_wxgf - 1: + progress_val = 50 + int((i + 1) / total_wxgf * 30) + yield _sse({"type": "progress", "percent": progress_val, "message": f"转换 wxgf 图片: {i+1}/{total_wxgf}"}) + + logger.info(f"账号 {account_name} 转换完成: {converted_count}/{total_wxgf} 个 .wxgf 文件") + + # 6. 复制 account.json + yield _sse({"type": "progress", "percent": 85, "message": "正在更新账号配置..."}) + try: + await asyncio.to_thread(shutil.copy2, import_path_obj / "account.json", account_output_dir / "account.json") + except Exception: + pass + + # 7. 保存来源信息 + def _save_source_info(dst, path, info): + (dst / "_source.json").write_text( + json.dumps( + { + "db_storage_path": str(path), + "import_mode": "manual_import", + "imported_at": __import__('datetime').datetime.now().isoformat(), + "original_info": info + }, + ensure_ascii=False, + indent=2, + ), + encoding="utf-8", + ) + + try: + await asyncio.to_thread(_save_source_info, account_output_dir, import_path_obj, info) + except Exception: + pass + + # 8. 构建缓存 + yield _sse({"type": "progress", "percent": 90, "message": "正在构建会话缓存 (这可能需要较长时间)..."}) + try: + await asyncio.to_thread( + build_session_last_message_table, + account_output_dir, + rebuild=True, + include_hidden=True, + include_official=True, + ) + except Exception as e: + logger.error(f"构建会话缓存失败: {e}") + + yield _sse({ + "type": "complete", + "status": "success", + "account": account_name, + "nick": info["nick"], + "message": f"成功导入账号 {info['nick']} ({account_name})" + }) + + except Exception as e: + logger.error(f"导入过程中发生异常: {e}", exc_info=True) + yield _sse({"type": "error", "message": f"导入失败: {str(e)}"}) + + headers = { + "Content-Type": "text/event-stream", + "Cache-Control": "no-cache", + "Connection": "keep-alive", + "X-Accel-Buffering": "no" + } + return StreamingResponse(generate_progress(), headers=headers) diff --git a/src/wechat_decrypt_tool/routers/system.py b/src/wechat_decrypt_tool/routers/system.py new file mode 100644 index 0000000..eabd9bd --- /dev/null +++ b/src/wechat_decrypt_tool/routers/system.py @@ -0,0 +1,35 @@ +from fastapi import APIRouter +from pydantic import BaseModel +import asyncio +from concurrent.futures import ThreadPoolExecutor + +router = APIRouter() + + +def _open_folder_dialog(title: str, initial_dir: str) -> str: + # 延迟导入并放在独立线程运行，避免阻塞 FastAPI 主线程或发生 GUI 线程冲突 + import tkinter as tk + from tkinter import filedialog + + root = tk.Tk() + root.withdraw() # 隐藏主窗口 + root.attributes('-topmost', True) # 确保弹窗在最前 + + folder_path = filedialog.askdirectory( + parent=root, + title=title, + initialdir=initial_dir + ) + + root.destroy() + return folder_path + + +@router.get("/api/system/pick_directory", summary="唤起本地原生目录选择器") +async def pick_directory(title: str = "请选择目录", initial_dir: str = ""): + loop = asyncio.get_running_loop() + with ThreadPoolExecutor() as pool: + # 在子线程中执行 GUI 操作 + folder_path = await loop.run_in_executor(pool, _open_folder_dialog, title, initial_dir) + + return {"path": folder_path} \ No newline at end of file diff --git a/src/wechat_decrypt_tool/routers/wechat_detection.py b/src/wechat_decrypt_tool/routers/wechat_detection.py index b2fe1c5..f64fc37 100644 --- a/src/wechat_decrypt_tool/routers/wechat_detection.py +++ b/src/wechat_decrypt_tool/routers/wechat_detection.py @@ -21,7 +21,21 @@ async def detect_wechat_detailed(data_root_path: Optional[str] = None): # 检测当前登录账号 current_account_info = detect_current_logged_in_account(data_root_path) + + # 【新增特性】目录匹配校验：处理目录名 wxid_xxxx_yyyy 与真实 wxid_xxxx 的适配 + if current_account_info and current_account_info.get("current_account"): + base_wxid = current_account_info["current_account"] + current_account_info["matched_folder"] = base_wxid # 默认兜底 + + # 遍历寻找以该 wxid 开头的用户文件夹（支持后缀匹配） + for acc in info.get("accounts", []): + acc_name = acc["account_name"] + if acc_name == base_wxid or acc_name.startswith(f"{base_wxid}_"): + current_account_info["matched_folder"] = acc_name + break + info['current_account'] = current_account_info + # logger.info(current_account_info) # 添加一些统计信息 stats = { diff --git a/src/wechat_decrypt_tool/wechat_detection.py b/src/wechat_decrypt_tool/wechat_detection.py index ced8056..d00dc1d 100644 --- a/src/wechat_decrypt_tool/wechat_detection.py +++ b/src/wechat_decrypt_tool/wechat_detection.py @@ -14,7 +14,6 @@ from ctypes import wintypes from datetime import datetime - def get_wx_db(msg_dir: str = None, db_types: Union[List[str], str] = None, wxids: Union[List[str], str] = None) -> List[dict]: @@ -49,7 +48,7 @@ def get_wx_db(msg_dir: str = None, wxid_dirs[os.path.basename(sub_dir)] = os.path.join(msg_dir, sub_dir) else: wxid_dirs[os.path.basename(msg_dir)] = msg_dir - + for wxid, wxid_dir in wxid_dirs.items(): if wxids and wxid not in wxids: # 如果指定wxid,则过滤掉其他wxid continue @@ -70,6 +69,7 @@ def get_wx_db(msg_dir: str = None, result.append({"wxid": wxid, "db_type": db_type, "db_path": db_path, "wxid_dir": wxid_dir}) return result + # Windows API 常量和结构 PROCESS_QUERY_INFORMATION = 0x0400 PROCESS_VM_READ = 0x0010 @@ -87,6 +87,7 @@ CreateToolhelp32Snapshot = kernel32.CreateToolhelp32Snapshot Process32FirstW = kernel32.Process32FirstW Process32NextW = kernel32.Process32NextW + class PROCESSENTRY32W(ctypes.Structure): _fields_ = [ ('dwSize', wintypes.DWORD), @@ -105,6 +106,98 @@ class PROCESSENTRY32W(ctypes.Structure): # 删除了WeChatDecryptor类，解密功能已移至独立的wechat_decrypt.py脚本 +def parse_global_config(base_path: str) -> dict: + """ + 解析 all_users/config/global_config 获取最近登录用户信息 + 基于 AES-128-CFB 解密，并解析 MMKV 的 Varint 格式 + """ + try: + import os + config_path = os.path.join(base_path, 'all_users', 'config', 'global_config') + if not os.path.exists(config_path): + return None + + with open(config_path, 'rb') as f: + full_data = f.read() + + if len(full_data) <= 4: + return None + + encrypted_data = full_data[4:] + + # 核心修复 1：强制截断取前 16 字节，等同于 Rust 中的 b"xwechat_crypt_ke" + key = b'xwechat_crypt_key'[:16] + iv = b'\0' * 16 + + # 尝试主流的两种密码库 + try: + from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes + from cryptography.hazmat.backends import default_backend + cipher = Cipher(algorithms.AES(key), modes.CFB(iv), backend=default_backend()) + decryptor = cipher.decryptor() + decrypted = decryptor.update(encrypted_data) + decryptor.finalize() + except ImportError: + from Crypto.Cipher import AES + # PyCryptodome 中 CFB 模式默认 segment_size 是 8，需要指定为 128 + cipher = AES.new(key, AES.MODE_CFB, iv=iv, segment_size=128) + decrypted = cipher.decrypt(encrypted_data) + + # MMKV Varint 长度解码 + def decode_varint(data, offset): + result = 0 + shift = 0 + while offset < len(data): + byte = data[offset] + offset += 1 + result |= (byte & 0x7f) << shift + if not (byte & 0x80): + break + shift += 7 + return result, offset + + def extract_mmkv_string(data: bytes, key_str: str) -> str: + key_bytes = key_str.encode('utf-8') + idx = data.find(key_bytes) + if idx == -1: return None + + offset = idx + len(key_bytes) + try: + value_len, offset = decode_varint(data, offset) + if value_len <= 0 or offset >= len(data): + return None + + str_len, offset = decode_varint(data, offset) + + if str_len > 0 and offset + str_len <= len(data): + return data[offset:offset + str_len].decode('utf-8', errors='ignore') + except Exception: + pass + return None + + + wxid = extract_mmkv_string(decrypted, 'mmkv_key_user_name') + nickname = extract_mmkv_string(decrypted, 'mmkv_key_nick_name') + avatar_url = extract_mmkv_string(decrypted, 'mmkv_key_head_img_url') + + # 核心修复 2：参考 Rust 逻辑，头像链接往往以 "/0" 结尾（微信头像的尺寸标识） + if not avatar_url and b'http' in decrypted: + http_idx = decrypted.find(b'http') + slash_zero_idx = decrypted.find(b'/0', http_idx) + if slash_zero_idx != -1: + # 包含 "/0" 这两个字符本身，所以是 +2 + avatar_url = decrypted[http_idx:slash_zero_idx + 2].decode('utf-8', errors='ignore') + + if wxid or nickname: + return { + "wxid": wxid, + "nickname": nickname, + "avatar": avatar_url + } + return None + except Exception as e: + print(f"[DEBUG] 解析 global_config 失败: {e}") + return None + def find_wechat_databases() -> List[str]: """在新的xwechat_files目录中查找微信数据库文件 @@ -119,13 +212,13 @@ def find_wechat_databases() -> List[str]: # 检查新的微信4.0+目录结构 wechat_dirs = [ documents_dir / "xwechat_files", # 新版微信4.0+ - documents_dir / "WeChat Files" # 旧版微信 + documents_dir / "WeChat Files" # 旧版微信 ] - + for wechat_dir in wechat_dirs: if not wechat_dir.exists(): continue - + # 查找用户目录（wxid_*模式） for user_dir in wechat_dir.iterdir(): if not user_dir.is_dir(): @@ -149,7 +242,7 @@ def find_wechat_databases() -> List[str]: for db_file in multi_dir.glob("*.db"): if db_file.is_file(): db_files.append(str(db_file)) - + return db_files @@ -158,7 +251,7 @@ def get_process_exe_path(process_id): h_process = OpenProcess(PROCESS_QUERY_INFORMATION | PROCESS_VM_READ, False, process_id) if not h_process: return None - + exe_path = ctypes.create_unicode_buffer(MAX_PATH) if GetModuleFileNameExW(h_process, None, exe_path, MAX_PATH) > 0: CloseHandle(h_process) @@ -167,35 +260,37 @@ def get_process_exe_path(process_id): CloseHandle(h_process) return None + def get_process_list(): """获取系统进程列表""" h_process_snap = CreateToolhelp32Snapshot(TH32CS_SNAPPROCESS, 0) if h_process_snap == ctypes.wintypes.HANDLE(-1).value: return [] - + pe32 = PROCESSENTRY32W() pe32.dwSize = ctypes.sizeof(PROCESSENTRY32W) process_list = [] - + if not Process32FirstW(h_process_snap, ctypes.byref(pe32)): CloseHandle(h_process_snap) return [] - + while True: process_list.append((pe32.th32ProcessID, pe32.szExeFile)) if not Process32NextW(h_process_snap, ctypes.byref(pe32)): break - + CloseHandle(h_process_snap) return process_list + def auto_detect_wechat_data_dirs(): """ 自动检测微信数据目录 - 多策略组合检测 :return: 检测到的微信数据目录列表 """ detected_dirs = [] - + # 策略1：注册表检测已移除 # 策略2和策略3：注册表相关检测已移除 @@ -211,24 +306,24 @@ def auto_detect_wechat_data_dirs(): for drive in drives: if not os.path.exists(drive): continue - + try: # 扫描驱动器根目录和常见目录 scan_paths = [ drive + os.sep, os.path.join(drive + os.sep, "Users"), ] - + for scan_path in scan_paths: if not os.path.exists(scan_path): continue - + try: for item in os.listdir(scan_path): item_path = os.path.join(scan_path, item) if not os.path.isdir(item_path): continue - + # 检查是否匹配微信目录模式 for pattern in common_wechat_patterns: if pattern.lower() in item.lower(): @@ -242,7 +337,7 @@ def auto_detect_wechat_data_dirs(): continue except (PermissionError, OSError): continue - + # 策略2：进程内存分析（简化版） try: process_list = get_process_list() @@ -263,7 +358,7 @@ def auto_detect_wechat_data_dirs(): current = parent else: break - + for parent_dir in parent_dirs: for pattern in common_wechat_patterns: potential_dir = os.path.join(parent_dir, pattern) @@ -275,7 +370,7 @@ def auto_detect_wechat_data_dirs(): pass except: pass - + return detected_dirs @@ -300,6 +395,7 @@ def has_wxid_directories(directory): except: return False + def get_wx_dir_by_reg(wxid="all"): """ 通过多种方法获取微信目录 - 改进的自动检测 @@ -327,6 +423,7 @@ def get_wx_dir_by_reg(wxid="all"): return wx_dir if os.path.exists(wx_dir) else None + def detect_wechat_accounts_from_backup(backup_base_path: str = None) -> List[Dict[str, Any]]: """ 从指定的备份路径检测微信账号 @@ -393,8 +490,8 @@ def detect_wechat_accounts_from_backup(backup_base_path: str = None) -> List[Dic for data_item in os.listdir(backup_base_path): data_item_path = os.path.join(backup_base_path, data_item) if (os.path.isdir(data_item_path) and - data_item.startswith(f"{account_name}_") and - data_item != "Backup"): + data_item.startswith(f"{account_name}_") and + data_item != "Backup"): data_dir = data_item_path break except (PermissionError, OSError): @@ -525,9 +622,9 @@ def detect_wechat_accounts_from_login(login_base_path: str = None) -> List[Dict[ for data_item in os.listdir(base_path): data_item_path = os.path.join(base_path, data_item) if ( - os.path.isdir(data_item_path) - and data_item.startswith(f"{account_name}_") - and data_item not in ["Backup", "all_users"] + os.path.isdir(data_item_path) + and data_item.startswith(f"{account_name}_") + and data_item not in ["Backup", "all_users"] ): data_dir = data_item_path break @@ -551,6 +648,7 @@ def detect_wechat_accounts_from_login(login_base_path: str = None) -> List[Dict[ return accounts + def collect_account_databases(data_dir: str, account_name: str) -> List[Dict[str, Any]]: """ 收集指定账号数据目录下的所有数据库文件 @@ -801,40 +899,37 @@ def detect_wechat_installation(data_root_path: str | None = None) -> Dict[str, A def detect_current_logged_in_account(base_path: str = None) -> Dict[str, Any]: """ - 通过key_info.db文件时间检测当前登录的微信账号 - - Args: - base_path: 微信数据根目录，如果为None则自动检测 - - Returns: - 当前登录账号信息 + 通过 global_config 解析 或 key_info.db 时间检测当前登录的微信账号 """ - current_account = None - latest_time = None - - # 添加调试信息 - print(f"[DEBUG] 开始检测当前登录账号，提供的base_path: {base_path}") - - # 如果没有指定路径，尝试自动检测 + # print(f"[DEBUG] 开始检测当前登录账号，提供的base_path: {base_path}") + if base_path is None: detected_dirs = auto_detect_wechat_data_dirs() - print(f"[DEBUG] 自动检测到的目录: {detected_dirs}") if not detected_dirs: - return { - "current_account": None, - "latest_time": None, - "message": "未检测到微信数据目录" - } + return {"current_account": None, "message": "未检测到微信数据目录"} base_path = detected_dirs[0] - - print(f"[DEBUG] 使用的base_path: {base_path}") - - # 查找登录信息目录 - 尝试多个可能的路径 + + # 1. 新特性：优先尝试从 global_config 解析完整用户信息 + parsed_config = parse_global_config(base_path) + if parsed_config and parsed_config.get('wxid'): + print(f"[DEBUG] 从 global_config 成功解析出账号: {parsed_config['wxid']}") + return { + "current_account": parsed_config["wxid"], # 不带校验位的 wxid + "nickname": parsed_config.get("nickname"), + "avatar": parsed_config.get("avatar"), + "latest_time": None, + "message": f"通过 global_config 检测到最近登录账号: {parsed_config['wxid']}" + } + + # 2. 降级回退机制：原先基于 key_info.db 的时间探测逻辑 + latest_time = None + current_account = None + possible_login_paths = [ - os.path.join(base_path, "all_users", "login"), # 标准路径 - os.path.join(base_path, "login"), # 备选路径1 + os.path.join(base_path, "all_users", "login"), + os.path.join(base_path, "login"), ] - + # 也尝试在子目录中查找 try: for item in os.listdir(base_path): @@ -842,11 +937,11 @@ def detect_current_logged_in_account(base_path: str = None) -> Dict[str, Any]: if os.path.isdir(item_path): possible_login_paths.extend([ os.path.join(item_path, "all_users", "login"), # 子目录中的标准路径 - os.path.join(item_path, "login"), # 子目录中的备选路径 + os.path.join(item_path, "login"), # 子目录中的备选路径 ]) except (PermissionError, OSError): pass - + login_dir = None for path in possible_login_paths: print(f"[DEBUG] 检查路径: {path}") @@ -854,49 +949,49 @@ def detect_current_logged_in_account(base_path: str = None) -> Dict[str, Any]: login_dir = path print(f"[DEBUG] 找到登录目录: {login_dir}") break - + if not login_dir: return { "current_account": None, "latest_time": None, "message": f"未找到登录信息目录，尝试的路径: {possible_login_paths}" } - + try: # 遍历登录目录下的所有账号文件夹 items = os.listdir(login_dir) print(f"[DEBUG] 登录目录内容: {items}") - + for item in items: item_path = os.path.join(login_dir, item) print(f"[DEBUG] 检查项目: {item}, 路径: {item_path}, 是否为目录: {os.path.isdir(item_path)}") - + if not os.path.isdir(item_path): continue - + # 检查key_info.db文件 key_info_path = os.path.join(item_path, "key_info.db") print(f"[DEBUG] 检查key_info.db文件: {key_info_path}, 是否存在: {os.path.exists(key_info_path)}") - + if not os.path.exists(key_info_path): continue - + # 获取文件修改时间 try: file_time = os.path.getmtime(key_info_path) file_datetime = datetime.fromtimestamp(file_time) print(f"[DEBUG] 找到key_info.db文件: {key_info_path}, 修改时间: {file_datetime}") - + # 更新最新登录的账号 if latest_time is None or file_time > latest_time: latest_time = file_time current_account = item print(f"[DEBUG] 更新最新登录账号: {current_account}, 时间: {file_datetime}") - + except OSError as e: print(f"[DEBUG] 无法获取文件时间: {key_info_path}, 错误: {e}") continue - + except (PermissionError, OSError) as e: print(f"[DEBUG] 无法访问登录目录: {login_dir}, 错误: {e}") return { @@ -904,7 +999,7 @@ def detect_current_logged_in_account(base_path: str = None) -> Dict[str, Any]: "latest_time": None, "message": f"无法访问登录目录: {e}" } - + if current_account: print(f"[DEBUG] 最终结果: 当前登录账号 {current_account}, 时间 {latest_time}") return {