From 8679f090ce9639367566aeaba61199b4e94608ed Mon Sep 17 00:00:00 2001 From: 2977094657 <2977094657@qq.com> Date: Fri, 19 Dec 2025 16:17:46 +0800 Subject: [PATCH] =?UTF-8?q?refactor(api):=20=E9=87=8D=E6=9E=84=E4=B8=BA?= =?UTF-8?q?=E6=A8=A1=E5=9D=97=E5=8C=96=E8=B7=AF=E7=94=B1=E6=9E=B6=E6=9E=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增routers目录，按功能划分路由模块 - health.py: 健康检查端点 - chat.py: 聊天会话与消息查询 - media.py: 媒体资源解密 - decrypt.py: 数据库解密 - wechat_detection.py: 微信安装检测 - chat_media.py: 聊天媒体资源访问 --- src/wechat_decrypt_tool/routers/__init__.py | 0 src/wechat_decrypt_tool/routers/chat.py | 717 +++++++++++++++++ src/wechat_decrypt_tool/routers/chat_media.py | 728 ++++++++++++++++++ src/wechat_decrypt_tool/routers/decrypt.py | 66 ++ src/wechat_decrypt_tool/routers/health.py | 22 + src/wechat_decrypt_tool/routers/media.py | 458 +++++++++++ .../routers/wechat_detection.py | 73 ++ 7 files changed, 2064 insertions(+) create mode 100644 src/wechat_decrypt_tool/routers/__init__.py create mode 100644 src/wechat_decrypt_tool/routers/chat.py create mode 100644 src/wechat_decrypt_tool/routers/chat_media.py create mode 100644 src/wechat_decrypt_tool/routers/decrypt.py create mode 100644 src/wechat_decrypt_tool/routers/health.py create mode 100644 src/wechat_decrypt_tool/routers/media.py create mode 100644 src/wechat_decrypt_tool/routers/wechat_detection.py diff --git a/src/wechat_decrypt_tool/routers/__init__.py b/src/wechat_decrypt_tool/routers/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/wechat_decrypt_tool/routers/chat.py b/src/wechat_decrypt_tool/routers/chat.py new file mode 100644 index 0000000..6f0dd36 --- /dev/null +++ b/src/wechat_decrypt_tool/routers/chat.py @@ -0,0 +1,717 @@ +import os +import sqlite3 +from pathlib import Path +from typing import Any, Optional +from urllib.parse import quote + +from fastapi import APIRouter, HTTPException, Request +from ..logging_config import get_logger +from ..chat_helpers import ( + _build_avatar_url, + _decode_message_content, + _decode_sqlite_text, + _extract_sender_from_group_xml, + _extract_xml_attr, + _extract_xml_tag_or_attr, + _extract_xml_tag_text, + _format_session_time, + _infer_last_message_brief, + _infer_message_brief_by_local_type, + _infer_transfer_status_text, + _iter_message_db_paths, + _list_decrypted_accounts, + _load_contact_rows, + _load_latest_message_previews, + _lookup_resource_md5, + _parse_app_message, + _parse_pat_message, + _pick_avatar_url, + _pick_display_name, + _query_head_image_usernames, + _quote_ident, + _resolve_account_dir, + _resolve_msg_table_name, + _resource_lookup_chat_id, + _should_keep_session, + _split_group_sender_prefix, +) +from ..media_helpers import _try_find_decrypted_resource +from ..path_fix import PathFixRoute + +logger = get_logger(__name__) + +_DEBUG_SESSIONS = os.environ.get("WECHAT_TOOL_DEBUG_SESSIONS", "0") == "1" + +router = APIRouter(route_class=PathFixRoute) + + +@router.get("/api/chat/accounts", summary="列出已解密账号") +async def list_chat_accounts(): + """列出 output/databases 下可用于聊天预览的账号目录""" + accounts = _list_decrypted_accounts() + if not accounts: + return { + "status": "error", + "accounts": [], + "default_account": None, + "message": "No decrypted databases found. Please decrypt first.", + } + + return { + "status": "success", + "accounts": accounts, + "default_account": accounts[0], + } + + +@router.get("/api/chat/sessions", summary="获取会话列表（聊天左侧列表）") +async def list_chat_sessions( + request: Request, + account: Optional[str] = None, + limit: int = 400, + include_hidden: bool = False, + include_official: bool = False, +): + """从 session.db + contact.db 读取会话列表，用于前端聊天界面动态渲染联系人""" + if limit <= 0: + raise HTTPException(status_code=400, detail="Invalid limit.") + if limit > 2000: + limit = 2000 + + account_dir = _resolve_account_dir(account) + session_db_path = account_dir / "session.db" + contact_db_path = account_dir / "contact.db" + head_image_db_path = account_dir / "head_image.db" + base_url = str(request.base_url).rstrip("/") + + sconn = sqlite3.connect(str(session_db_path)) + sconn.row_factory = sqlite3.Row + try: + rows = sconn.execute( + """ + SELECT + username, + unread_count, + is_hidden, + summary, + draft, + last_timestamp, + sort_timestamp, + last_msg_type, + last_msg_sub_type + FROM SessionTable + ORDER BY sort_timestamp DESC + LIMIT ? + """, + (int(limit),), + ).fetchall() + finally: + sconn.close() + + filtered: list[sqlite3.Row] = [] + usernames: list[str] = [] + for r in rows: + username = r["username"] or "" + if not username: + continue + if not include_hidden and int(r["is_hidden"] or 0) == 1: + continue + if not _should_keep_session(username, include_official=include_official): + continue + filtered.append(r) + usernames.append(username) + + contact_rows = _load_contact_rows(contact_db_path, usernames) + local_avatar_usernames = _query_head_image_usernames(head_image_db_path, usernames) + latest_previews = _load_latest_message_previews(account_dir, usernames) + if _DEBUG_SESSIONS: + logger.info( + f"[sessions.preview] endpoint account={account_dir.name} sessions={len(usernames)} previews={len(latest_previews)}" + ) + + sessions: list[dict[str, Any]] = [] + for r in filtered: + username = r["username"] + c_row = contact_rows.get(username) + + display_name = _pick_display_name(c_row, username) + avatar_url = _pick_avatar_url(c_row) + if not avatar_url and username in local_avatar_usernames: + avatar_url = base_url + _build_avatar_url(account_dir.name, username) + + if str(latest_previews.get(username) or "").strip(): + last_message = str(latest_previews.get(username) or "").strip() + else: + last_message = _infer_last_message_brief(r["last_msg_type"], r["last_msg_sub_type"]) + + last_time = _format_session_time(r["sort_timestamp"] or r["last_timestamp"]) + + sessions.append( + { + "id": username, + "username": username, + "name": display_name, + "avatar": avatar_url, + "lastMessage": last_message, + "lastMessageTime": last_time, + "unreadCount": int(r["unread_count"] or 0), + "isGroup": bool(username.endswith("@chatroom")), + } + ) + + return { + "status": "success", + "account": account_dir.name, + "total": len(sessions), + "sessions": sessions, + } + + +@router.get("/api/chat/messages", summary="获取会话消息列表") +async def list_chat_messages( + request: Request, + username: str, + account: Optional[str] = None, + limit: int = 50, + offset: int = 0, + order: str = "asc", +): + if not username: + raise HTTPException(status_code=400, detail="Missing username.") + if limit <= 0: + raise HTTPException(status_code=400, detail="Invalid limit.") + if limit > 500: + limit = 500 + if offset < 0: + offset = 0 + + account_dir = _resolve_account_dir(account) + db_paths = _iter_message_db_paths(account_dir) + contact_db_path = account_dir / "contact.db" + head_image_db_path = account_dir / "head_image.db" + message_resource_db_path = account_dir / "message_resource.db" + base_url = str(request.base_url).rstrip("/") + if not db_paths: + return { + "status": "error", + "account": account_dir.name, + "username": username, + "total": 0, + "messages": [], + "message": "No message databases found for this account.", + } + + resource_conn: Optional[sqlite3.Connection] = None + resource_chat_id: Optional[int] = None + try: + if message_resource_db_path.exists(): + resource_conn = sqlite3.connect(str(message_resource_db_path)) + resource_conn.row_factory = sqlite3.Row + resource_chat_id = _resource_lookup_chat_id(resource_conn, username) + except Exception: + if resource_conn is not None: + try: + resource_conn.close() + except Exception: + pass + resource_conn = None + resource_chat_id = None + + want_asc = str(order or "").lower() != "desc" + take = int(limit) + int(offset) + take_probe = take + 1 + merged: list[dict[str, Any]] = [] + sender_usernames: list[str] = [] + pat_usernames: set[str] = set() + is_group = bool(username.endswith("@chatroom")) + has_more_any = False + + for db_path in db_paths: + conn = sqlite3.connect(str(db_path)) + conn.row_factory = sqlite3.Row + try: + table_name = _resolve_msg_table_name(conn, username) + if not table_name: + continue + + my_wxid = account_dir.name + my_rowid = None + try: + r = conn.execute( + "SELECT rowid FROM Name2Id WHERE user_name = ? LIMIT 1", + (my_wxid,), + ).fetchone() + if r is not None: + my_rowid = int(r[0]) + except Exception: + my_rowid = None + + quoted_table = _quote_ident(table_name) + sql_with_join = ( + "SELECT " + "m.local_id, m.server_id, m.local_type, m.sort_seq, m.real_sender_id, m.create_time, " + "m.message_content, m.compress_content, n.user_name AS sender_username " + f"FROM {quoted_table} m " + "LEFT JOIN Name2Id n ON m.real_sender_id = n.rowid " + "ORDER BY m.create_time DESC, m.sort_seq DESC, m.local_id DESC " + "LIMIT ?" + ) + sql_no_join = ( + "SELECT " + "m.local_id, m.server_id, m.local_type, m.sort_seq, m.real_sender_id, m.create_time, " + "m.message_content, m.compress_content, '' AS sender_username " + f"FROM {quoted_table} m " + "ORDER BY m.create_time DESC, m.sort_seq DESC, m.local_id DESC " + "LIMIT ?" + ) + + # Force sqlite3 to return TEXT as raw bytes for this query, so we can zstd-decompress + # compress_content reliably. + conn.text_factory = bytes + + try: + rows = conn.execute(sql_with_join, (take_probe,)).fetchall() + except Exception: + rows = conn.execute(sql_no_join, (take_probe,)).fetchall() + if len(rows) > take: + has_more_any = True + rows = rows[:take] + + for r in rows: + local_id = int(r["local_id"] or 0) + create_time = int(r["create_time"] or 0) + sort_seq = int(r["sort_seq"] or 0) if r["sort_seq"] is not None else 0 + local_type = int(r["local_type"] or 0) + sender_username = _decode_sqlite_text(r["sender_username"]).strip() + + is_sent = False + if my_rowid is not None: + try: + is_sent = int(r["real_sender_id"] or 0) == int(my_rowid) + except Exception: + is_sent = False + + raw_text = _decode_message_content(r["compress_content"], r["message_content"]) + raw_text = raw_text.strip() + + sender_prefix = "" + if is_group and not raw_text.startswith("<") and not raw_text.startswith('"<'): + sender_prefix, raw_text = _split_group_sender_prefix(raw_text) + + if is_group and sender_prefix: + sender_username = sender_prefix + + if is_group and (raw_text.startswith("<") or raw_text.startswith('"<')): + xml_sender = _extract_sender_from_group_xml(raw_text) + if xml_sender: + sender_username = xml_sender + + if is_sent: + sender_username = account_dir.name + elif (not is_group) and (not sender_username): + sender_username = username + + if sender_username: + sender_usernames.append(sender_username) + + render_type = "text" + content_text = raw_text + title = "" + url = "" + image_md5 = "" + emoji_md5 = "" + emoji_url = "" + thumb_url = "" + image_url = "" + video_md5 = "" + video_thumb_md5 = "" + video_url = "" + video_thumb_url = "" + voice_length = "" + quote_title = "" + quote_content = "" + amount = "" + cover_url = "" + file_size = "" + pay_sub_type = "" + transfer_status = "" + file_md5 = "" + transfer_id = "" + + if local_type == 10000: + render_type = "system" + if "revokemsg" in raw_text: + content_text = "撤回了一条消息" + else: + import re + + content_text = re.sub(r"]*>", "", raw_text) + content_text = re.sub(r"\s+", " ", content_text).strip() or "[系统消息]" + elif local_type == 49: + parsed = _parse_app_message(raw_text) + render_type = str(parsed.get("renderType") or "text") + content_text = str(parsed.get("content") or "") + title = str(parsed.get("title") or "") + url = str(parsed.get("url") or "") + quote_title = str(parsed.get("quoteTitle") or "") + quote_content = str(parsed.get("quoteContent") or "") + amount = str(parsed.get("amount") or "") + cover_url = str(parsed.get("coverUrl") or "") + thumb_url = str(parsed.get("thumbUrl") or "") + file_size = str(parsed.get("size") or "") + pay_sub_type = str(parsed.get("paySubType") or "") + file_md5 = str(parsed.get("fileMd5") or "") + transfer_id = str(parsed.get("transferId") or "") + + if render_type == "transfer": + # 直接从原始 XML 提取 transferid（可能在 wcpayinfo 内） + if not transfer_id: + transfer_id = _extract_xml_tag_or_attr(raw_text, "transferid") or "" + transfer_status = _infer_transfer_status_text( + is_sent=is_sent, + paysubtype=pay_sub_type, + receivestatus=str(parsed.get("receiveStatus") or ""), + sendertitle=str(parsed.get("senderTitle") or ""), + receivertitle=str(parsed.get("receiverTitle") or ""), + senderdes=str(parsed.get("senderDes") or ""), + receiverdes=str(parsed.get("receiverDes") or ""), + ) + if not content_text: + content_text = transfer_status or "转账" + elif local_type == 266287972401: + render_type = "system" + template = _extract_xml_tag_text(raw_text, "template") + if template: + import re + + pat_usernames.update({m.group(1) for m in re.finditer(r"\$\{([^}]+)\}", template) if m.group(1)}) + content_text = "[拍一拍]" + else: + content_text = "[拍一拍]" + elif local_type == 244813135921: + render_type = "quote" + parsed = _parse_app_message(raw_text) + content_text = str(parsed.get("content") or "[引用消息]") + quote_title = str(parsed.get("quoteTitle") or "") + quote_content = str(parsed.get("quoteContent") or "") + elif local_type == 3: + render_type = "image" + image_md5 = _extract_xml_attr(raw_text, "md5") + # Extract CDN URL and validate it looks like a proper URL + _cdn_url = ( + _extract_xml_attr(raw_text, "cdnthumburl") + or _extract_xml_attr(raw_text, "cdnmidimgurl") + or _extract_xml_attr(raw_text, "cdnbigimgurl") + ) + image_url = _cdn_url if _cdn_url.startswith(("http://", "https://")) else "" + if (not image_md5) and resource_conn is not None: + image_md5 = _lookup_resource_md5( + resource_conn, + resource_chat_id, + message_local_type=local_type, + server_id=int(r["server_id"] or 0), + local_id=local_id, + create_time=create_time, + ) + content_text = "[图片]" + elif local_type == 34: + render_type = "voice" + duration = _extract_xml_attr(raw_text, "voicelength") + voice_length = duration + content_text = f"[语音 {duration}秒]" if duration else "[语音]" + elif local_type == 43 or local_type == 62: + render_type = "video" + video_md5 = _extract_xml_attr(raw_text, "md5") + video_thumb_md5 = _extract_xml_attr(raw_text, "cdnthumbmd5") + video_thumb_url = _extract_xml_attr(raw_text, "cdnthumburl") + video_url = _extract_xml_attr(raw_text, "cdnvideourl") + if (not video_thumb_md5) and resource_conn is not None: + video_thumb_md5 = _lookup_resource_md5( + resource_conn, + resource_chat_id, + message_local_type=local_type, + server_id=int(r["server_id"] or 0), + local_id=local_id, + create_time=create_time, + ) + content_text = "[视频]" + elif local_type == 47: + render_type = "emoji" + emoji_md5 = _extract_xml_attr(raw_text, "md5") + if not emoji_md5: + emoji_md5 = _extract_xml_tag_text(raw_text, "md5") + emoji_url = _extract_xml_attr(raw_text, "cdnurl") + if not emoji_url: + emoji_url = _extract_xml_tag_text(raw_text, "cdn_url") + if (not emoji_md5) and resource_conn is not None: + emoji_md5 = _lookup_resource_md5( + resource_conn, + resource_chat_id, + message_local_type=local_type, + server_id=int(r["server_id"] or 0), + local_id=local_id, + create_time=create_time, + ) + content_text = "[表情]" + elif local_type != 1: + if not content_text: + content_text = _infer_message_brief_by_local_type(local_type) + else: + if content_text.startswith("<") or content_text.startswith('"<'): + if " tuple[int, int, int]: + sseq = int(m.get("sortSeq") or 0) + cts = int(m.get("createTime") or 0) + lid = int(m.get("localId") or 0) + return (cts, sseq, lid) + + merged.sort(key=sort_key, reverse=True) + has_more_global = bool(has_more_any or (len(merged) > (int(offset) + int(limit)))) + page = merged[int(offset) : int(offset) + int(limit)] + if want_asc: + page = list(reversed(page)) + + return { + "status": "success", + "account": account_dir.name, + "username": username, + "total": int(offset) + len(page) + (1 if has_more_global else 0), + "hasMore": bool(has_more_global), + "messages": page, + } diff --git a/src/wechat_decrypt_tool/routers/chat_media.py b/src/wechat_decrypt_tool/routers/chat_media.py new file mode 100644 index 0000000..3cde08e --- /dev/null +++ b/src/wechat_decrypt_tool/routers/chat_media.py @@ -0,0 +1,728 @@ +import asyncio +import ipaddress +import mimetypes +import os +import sqlite3 +import subprocess +from pathlib import Path +from typing import Optional +from urllib.parse import urlparse + +import requests +from fastapi import APIRouter, HTTPException +from fastapi.responses import FileResponse, Response +from pydantic import BaseModel, Field + +from ..logging_config import get_logger +from ..media_helpers import ( + _convert_silk_to_wav, + _detect_image_extension, + _detect_image_media_type, + _ensure_decrypted_resource_for_md5, + _fallback_search_media_by_md5, + _get_decrypted_resource_path, + _get_resource_dir, + _guess_media_type_by_path, + _iter_emoji_source_candidates, + _read_and_maybe_decrypt_media, + _resolve_account_db_storage_dir, + _resolve_account_dir, + _resolve_account_wxid_dir, + _resolve_media_path_for_kind, + _resolve_media_path_from_hardlink, + _try_find_decrypted_resource, + _try_strip_media_prefix, +) +from ..path_fix import PathFixRoute + +logger = get_logger(__name__) + +router = APIRouter(route_class=PathFixRoute) + + +@router.get("/api/chat/avatar", summary="获取联系人头像") +async def get_chat_avatar(username: str, account: Optional[str] = None): + if not username: + raise HTTPException(status_code=400, detail="Missing username.") + account_dir = _resolve_account_dir(account) + head_image_db_path = account_dir / "head_image.db" + if not head_image_db_path.exists(): + raise HTTPException(status_code=404, detail="head_image.db not found.") + + conn = sqlite3.connect(str(head_image_db_path)) + try: + row = conn.execute( + "SELECT image_buffer FROM head_image WHERE username = ? ORDER BY update_time DESC LIMIT 1", + (username,), + ).fetchone() + finally: + conn.close() + + if not row or row[0] is None: + raise HTTPException(status_code=404, detail="Avatar not found.") + + data = bytes(row[0]) if isinstance(row[0], (memoryview, bytearray)) else row[0] + if not isinstance(data, (bytes, bytearray)): + data = bytes(data) + media_type = _detect_image_media_type(data) + return Response(content=data, media_type=media_type) + + +class EmojiDownloadRequest(BaseModel): + account: Optional[str] = Field(None, description="账号目录名（可选，默认使用第一个）") + md5: str = Field(..., description="表情 MD5") + emoji_url: str = Field(..., description="表情 CDN URL") + force: bool = Field(False, description="是否强制重新下载并覆盖") + + +def _is_valid_md5(s: str) -> bool: + import re + + v = str(s or "").strip().lower() + return bool(re.fullmatch(r"[0-9a-f]{32}", v)) + + +def _is_safe_http_url(url: str) -> bool: + u = str(url or "").strip() + if not u: + return False + try: + p = urlparse(u) + except Exception: + return False + if p.scheme not in ("http", "https"): + return False + host = (p.hostname or "").strip() + if not host: + return False + if host in {"localhost"}: + return False + try: + ip = ipaddress.ip_address(host) + if ip.is_private or ip.is_loopback or ip.is_link_local: + return False + except Exception: + pass + return True + + +def _detect_media_type_and_ext(data: bytes) -> tuple[bytes, str, str]: + payload = data + media_type = "application/octet-stream" + ext = "dat" + + try: + payload2, mt2 = _try_strip_media_prefix(payload) + if mt2 != "application/octet-stream": + payload = payload2 + media_type = mt2 + except Exception: + pass + + if media_type == "application/octet-stream": + mt0 = _detect_image_media_type(payload[:32]) + if mt0 != "application/octet-stream": + media_type = mt0 + + if media_type == "application/octet-stream": + try: + if len(payload) >= 8 and payload[4:8] == b"ftyp": + media_type = "video/mp4" + except Exception: + pass + + if media_type.startswith("image/"): + ext = _detect_image_extension(payload) + elif media_type == "video/mp4": + ext = "mp4" + else: + ext = "dat" + + return payload, media_type, ext + + +@router.post("/api/chat/media/emoji/download", summary="下载表情消息资源到本地 resource") +async def download_chat_emoji(req: EmojiDownloadRequest): + md5 = str(req.md5 or "").strip().lower() + emoji_url = str(req.emoji_url or "").strip() + + if not _is_valid_md5(md5): + raise HTTPException(status_code=400, detail="Invalid md5.") + if not _is_safe_http_url(emoji_url): + raise HTTPException(status_code=400, detail="Invalid emoji_url (only public http/https allowed).") + + account_dir = _resolve_account_dir(req.account) + + existing = _try_find_decrypted_resource(account_dir, md5) + if existing and existing.exists() and (not req.force): + return { + "status": "success", + "account": account_dir.name, + "md5": md5, + "saved": True, + "already_exists": True, + "path": str(existing), + "resource_dir": str(existing.parent), + } + + def _download_bytes() -> bytes: + headers = { + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120 Safari/537.36", + "Accept": "*/*", + } + r = requests.get(emoji_url, headers=headers, timeout=20, stream=True) + try: + r.raise_for_status() + max_bytes = 30 * 1024 * 1024 + chunks: list[bytes] = [] + total = 0 + for ch in r.iter_content(chunk_size=64 * 1024): + if not ch: + continue + chunks.append(ch) + total += len(ch) + if total > max_bytes: + raise HTTPException(status_code=400, detail="Emoji download too large (>30MB).") + return b"".join(chunks) + finally: + try: + r.close() + except Exception: + pass + + try: + data = await asyncio.to_thread(_download_bytes) + except HTTPException: + raise + except Exception as e: + logger.warning(f"emoji_download failed: md5={md5} url={emoji_url} err={e}") + raise HTTPException(status_code=500, detail=f"Emoji download failed: {e}") + + if not data: + raise HTTPException(status_code=500, detail="Emoji download returned empty body.") + + payload, media_type, ext = _detect_media_type_and_ext(data) + out_path = _get_decrypted_resource_path(account_dir, md5, ext) + out_path.parent.mkdir(parents=True, exist_ok=True) + + if out_path.exists() and (not req.force): + return { + "status": "success", + "account": account_dir.name, + "md5": md5, + "saved": True, + "already_exists": True, + "path": str(out_path), + "resource_dir": str(out_path.parent), + "media_type": media_type, + "bytes": len(payload), + } + + try: + out_path.write_bytes(payload) + except Exception as e: + raise HTTPException(status_code=500, detail=f"Failed to write emoji file: {e}") + + logger.info(f"emoji_download: md5={md5} url={emoji_url} -> {out_path} bytes={len(payload)} mt={media_type}") + return { + "status": "success", + "account": account_dir.name, + "md5": md5, + "saved": True, + "already_exists": False, + "path": str(out_path), + "resource_dir": str(out_path.parent), + "media_type": media_type, + "bytes": len(payload), + } + + +@router.get("/api/chat/media/image", summary="获取图片消息资源") +async def get_chat_image(md5: str, account: Optional[str] = None, username: Optional[str] = None): + if not md5: + raise HTTPException(status_code=400, detail="Missing md5.") + account_dir = _resolve_account_dir(account) + + # 优先从解密资源目录读取（更快） + decrypted_path = _try_find_decrypted_resource(account_dir, md5.lower()) + if decrypted_path: + data = decrypted_path.read_bytes() + media_type = _detect_image_media_type(data[:32]) + if media_type == "application/octet-stream": + guessed = mimetypes.guess_type(str(decrypted_path))[0] + if guessed: + media_type = guessed + return Response(content=data, media_type=media_type) + + # 回退到原始逻辑：从微信数据目录实时解密 + wxid_dir = _resolve_account_wxid_dir(account_dir) + hardlink_db_path = account_dir / "hardlink.db" + extra_roots: list[Path] = [] + db_storage_dir = _resolve_account_db_storage_dir(account_dir) + if db_storage_dir: + extra_roots.append(db_storage_dir) + + roots: list[Path] = [] + if wxid_dir: + roots.append(wxid_dir) + roots.append(wxid_dir / "msg" / "attach") + roots.append(wxid_dir / "msg" / "file") + roots.append(wxid_dir / "msg" / "video") + roots.append(wxid_dir / "cache") + if db_storage_dir: + roots.append(db_storage_dir) + if not roots: + raise HTTPException( + status_code=404, + detail="wxid_dir/db_storage_path not found. Please decrypt with db_storage_path to enable media lookup.", + ) + p = _resolve_media_path_from_hardlink( + hardlink_db_path, + roots[0], + md5=str(md5), + kind="image", + username=username, + extra_roots=roots[1:], + ) + if (not p) and wxid_dir: + hit = _fallback_search_media_by_md5(str(wxid_dir), str(md5)) + if hit: + p = Path(hit) + if not p: + raise HTTPException(status_code=404, detail="Image not found.") + + logger.info(f"chat_image: md5={md5} resolved_source={p}") + + data, media_type = _read_and_maybe_decrypt_media(p, account_dir=account_dir, weixin_root=wxid_dir) + if media_type.startswith("image/"): + try: + out_md5 = str(md5).lower() + ext = _detect_image_extension(data) + out_path = _get_decrypted_resource_path(account_dir, out_md5, ext) + out_path.parent.mkdir(parents=True, exist_ok=True) + if not out_path.exists(): + out_path.write_bytes(data) + except Exception: + pass + logger.info(f"chat_image: md5={md5} media_type={media_type} bytes={len(data)}") + return Response(content=data, media_type=media_type) + + +@router.get("/api/chat/media/emoji", summary="获取表情消息资源") +async def get_chat_emoji(md5: str, account: Optional[str] = None, username: Optional[str] = None): + if not md5: + raise HTTPException(status_code=400, detail="Missing md5.") + account_dir = _resolve_account_dir(account) + + # 优先从解密资源目录读取（更快） + decrypted_path = _try_find_decrypted_resource(account_dir, md5.lower()) + if decrypted_path: + data = decrypted_path.read_bytes() + media_type = _detect_image_media_type(data[:32]) + if media_type == "application/octet-stream": + guessed = mimetypes.guess_type(str(decrypted_path))[0] + if guessed: + media_type = guessed + return Response(content=data, media_type=media_type) + + wxid_dir = _resolve_account_wxid_dir(account_dir) + p = _resolve_media_path_for_kind(account_dir, kind="emoji", md5=str(md5), username=username) + if not p: + raise HTTPException(status_code=404, detail="Emoji not found.") + + data, media_type = _read_and_maybe_decrypt_media(p, account_dir=account_dir, weixin_root=wxid_dir) + if media_type.startswith("image/"): + try: + out_md5 = str(md5).lower() + ext = _detect_image_extension(data) + out_path = _get_decrypted_resource_path(account_dir, out_md5, ext) + out_path.parent.mkdir(parents=True, exist_ok=True) + if not out_path.exists(): + out_path.write_bytes(data) + except Exception: + pass + return Response(content=data, media_type=media_type) + + +@router.get("/api/chat/media/video_thumb", summary="获取视频缩略图资源") +async def get_chat_video_thumb(md5: str, account: Optional[str] = None, username: Optional[str] = None): + if not md5: + raise HTTPException(status_code=400, detail="Missing md5.") + account_dir = _resolve_account_dir(account) + + # 优先从解密资源目录读取（更快） + decrypted_path = _try_find_decrypted_resource(account_dir, md5.lower()) + if decrypted_path: + data = decrypted_path.read_bytes() + media_type = _detect_image_media_type(data[:32]) + return Response(content=data, media_type=media_type) + + # 回退到原始逻辑 + wxid_dir = _resolve_account_wxid_dir(account_dir) + hardlink_db_path = account_dir / "hardlink.db" + extra_roots: list[Path] = [] + db_storage_dir = _resolve_account_db_storage_dir(account_dir) + if db_storage_dir: + extra_roots.append(db_storage_dir) + + roots: list[Path] = [] + if wxid_dir: + roots.append(wxid_dir) + if db_storage_dir: + roots.append(db_storage_dir) + if not roots: + raise HTTPException( + status_code=404, + detail="wxid_dir/db_storage_path not found. Please decrypt with db_storage_path to enable media lookup.", + ) + p = _resolve_media_path_from_hardlink( + hardlink_db_path, + roots[0], + md5=str(md5), + kind="video_thumb", + username=username, + extra_roots=roots[1:], + ) + if (not p) and wxid_dir: + hit = _fallback_search_media_by_md5(str(wxid_dir), str(md5)) + if hit: + p = Path(hit) + if not p: + raise HTTPException(status_code=404, detail="Video thumbnail not found.") + + data, media_type = _read_and_maybe_decrypt_media(p, account_dir=account_dir, weixin_root=wxid_dir) + return Response(content=data, media_type=media_type) + + +@router.get("/api/chat/media/video", summary="获取视频资源") +async def get_chat_video(md5: str, account: Optional[str] = None, username: Optional[str] = None): + if not md5: + raise HTTPException(status_code=400, detail="Missing md5.") + account_dir = _resolve_account_dir(account) + wxid_dir = _resolve_account_wxid_dir(account_dir) + hardlink_db_path = account_dir / "hardlink.db" + extra_roots: list[Path] = [] + db_storage_dir = _resolve_account_db_storage_dir(account_dir) + if db_storage_dir: + extra_roots.append(db_storage_dir) + + roots: list[Path] = [] + if wxid_dir: + roots.append(wxid_dir) + if db_storage_dir: + roots.append(db_storage_dir) + if not roots: + raise HTTPException( + status_code=404, + detail="wxid_dir/db_storage_path not found. Please decrypt with db_storage_path to enable media lookup.", + ) + p = _resolve_media_path_from_hardlink( + hardlink_db_path, + roots[0], + md5=str(md5), + kind="video", + username=username, + extra_roots=roots[1:], + ) + if (not p) and wxid_dir: + hit = _fallback_search_media_by_md5(str(wxid_dir), str(md5)) + if hit: + p = Path(hit) + if not p: + raise HTTPException(status_code=404, detail="Video not found.") + media_type = _guess_media_type_by_path(p, fallback="video/mp4") + return FileResponse(str(p), media_type=media_type) + + +@router.get("/api/chat/media/voice", summary="获取语音消息资源") +async def get_chat_voice(server_id: int, account: Optional[str] = None): + if not server_id: + raise HTTPException(status_code=400, detail="Missing server_id.") + account_dir = _resolve_account_dir(account) + media_db_path = account_dir / "media_0.db" + if not media_db_path.exists(): + raise HTTPException(status_code=404, detail="media_0.db not found.") + + conn = sqlite3.connect(str(media_db_path)) + conn.row_factory = sqlite3.Row + try: + row = conn.execute( + "SELECT voice_data FROM VoiceInfo WHERE svr_id = ? ORDER BY create_time DESC LIMIT 1", + (int(server_id),), + ).fetchone() + except Exception: + row = None + finally: + conn.close() + + if not row or row[0] is None: + raise HTTPException(status_code=404, detail="Voice not found.") + + data = bytes(row[0]) if isinstance(row[0], (memoryview, bytearray)) else row[0] + if not isinstance(data, (bytes, bytearray)): + data = bytes(data) + + # Try to convert SILK to WAV for browser playback + wav_data = _convert_silk_to_wav(data) + if wav_data != data: + return Response( + content=wav_data, + media_type="audio/wav", + ) + + # Fallback to raw SILK if conversion fails + return Response( + content=data, + media_type="audio/silk", + headers={"Content-Disposition": f"attachment; filename=voice_{int(server_id)}.silk"}, + ) + + +@router.post("/api/chat/media/open_folder", summary="在资源管理器中打开媒体文件所在位置") +async def open_chat_media_folder( + kind: str, + md5: Optional[str] = None, + server_id: Optional[int] = None, + account: Optional[str] = None, + username: Optional[str] = None, +): + account_dir = _resolve_account_dir(account) + + kind_key = str(kind or "").strip().lower() + if kind_key not in {"image", "emoji", "video", "video_thumb", "file", "voice"}: + raise HTTPException(status_code=400, detail="Unsupported kind.") + + p: Optional[Path] = None + if kind_key == "voice": + if not server_id: + raise HTTPException(status_code=400, detail="Missing server_id.") + + media_db_path = account_dir / "media_0.db" + if not media_db_path.exists(): + raise HTTPException(status_code=404, detail="media_0.db not found.") + + conn = sqlite3.connect(str(media_db_path)) + conn.row_factory = sqlite3.Row + try: + row = conn.execute( + "SELECT voice_data FROM VoiceInfo WHERE svr_id = ? ORDER BY create_time DESC LIMIT 1", + (int(server_id),), + ).fetchone() + except Exception: + row = None + finally: + conn.close() + + if not row or row[0] is None: + raise HTTPException(status_code=404, detail="Voice not found.") + + data = bytes(row[0]) if isinstance(row[0], (memoryview, bytearray)) else row[0] + if not isinstance(data, (bytes, bytearray)): + data = bytes(data) + + export_dir = account_dir / "_exports" + export_dir.mkdir(parents=True, exist_ok=True) + p = export_dir / f"voice_{int(server_id)}.silk" + try: + p.write_bytes(data) + except Exception as e: + raise HTTPException(status_code=500, detail=f"Failed to export voice: {e}") + else: + if not md5: + raise HTTPException(status_code=400, detail="Missing md5.") + p = _resolve_media_path_for_kind(account_dir, kind=kind_key, md5=str(md5), username=username) + + resolved_before_materialize = p + materialized_ok = False + opened_kind = "resolved" + + if p and kind_key in {"image", "emoji", "video_thumb"}: + wxid_dir = _resolve_account_wxid_dir(account_dir) + source_path = p + if kind_key == "emoji": + candidates: list[Path] = [] + try: + md5s = str(md5 or "").lower().strip() + except Exception: + md5s = str(md5) + + try: + if p is not None and p.exists() and p.is_file(): + if (not str(p.suffix or "")) and md5s and str(p.name).lower() == md5s: + candidates.extend(_iter_emoji_source_candidates(p.parent, md5s)) + if p not in candidates: + candidates.append(p) + else: + candidates.append(p) + candidates.extend(_iter_emoji_source_candidates(p.parent, md5s)) + else: + candidates = _iter_emoji_source_candidates(p, md5s) + except Exception: + candidates = _iter_emoji_source_candidates(p, str(md5)) + + # de-dup while keeping order + seen: set[str] = set() + uniq: list[Path] = [] + for c in candidates: + try: + k = str(c.resolve()) + except Exception: + k = str(c) + if k in seen: + continue + seen.add(k) + uniq.append(c) + candidates = uniq + + try: + preferred: list[Path] = [] + if md5s: + for c in candidates: + try: + if md5s in str(c.name).lower(): + preferred.append(c) + except Exception: + continue + if preferred: + rest = [c for c in candidates if c not in preferred] + candidates = preferred + rest + except Exception: + pass + if not candidates and p is not None: + candidates = [p] + for cand in candidates: + source_path = cand + materialized = _ensure_decrypted_resource_for_md5( + account_dir, + md5=str(md5), + source_path=source_path, + weixin_root=wxid_dir, + ) + if materialized: + p = materialized + materialized_ok = True + opened_kind = "decrypted" + break + + if not materialized_ok: + try: + sz = -1 + head_hex = "" + try: + if source_path and source_path.exists() and source_path.is_file(): + sz = int(source_path.stat().st_size) + with open(source_path, "rb") as f: + head_hex = f.read(32).hex() + except Exception: + pass + logger.info( + f"open_folder: emoji materialize failed: resolved={str(resolved_before_materialize)} source={str(source_path)} size={sz} head32={head_hex}" + ) + except Exception: + pass + + try: + resource_dir = _get_resource_dir(account_dir) + sub_dir = str(md5).lower()[:2] if len(str(md5)) >= 2 else "00" + fallback_dir = resource_dir / sub_dir + fallback_dir.mkdir(parents=True, exist_ok=True) + p = fallback_dir + opened_kind = "resource_dir" + except Exception: + try: + resource_dir = _get_resource_dir(account_dir) + sub_dir = str(md5).lower()[:2] if len(str(md5)) >= 2 else "00" + fallback_dir = resource_dir / sub_dir + fallback_dir.mkdir(parents=True, exist_ok=True) + p = fallback_dir + opened_kind = "resource_dir" + except Exception: + pass + else: + materialized = _ensure_decrypted_resource_for_md5( + account_dir, + md5=str(md5), + source_path=source_path, + weixin_root=wxid_dir, + ) + if materialized: + p = materialized + materialized_ok = True + opened_kind = "decrypted" + + if kind_key == "emoji" and md5: + try: + existing2 = _try_find_decrypted_resource(account_dir, str(md5).lower()) + if existing2: + p = existing2 + opened_kind = "decrypted" + except Exception: + pass + + if not p: + if kind_key == "emoji": + wxid_dir = _resolve_account_wxid_dir(account_dir) + resource_dir = _get_resource_dir(account_dir) + candidates: list[Path] = [] + if md5: + sub_dir = str(md5).lower()[:2] if len(str(md5)) >= 2 else "00" + c1 = resource_dir / sub_dir + if c1.exists() and c1.is_dir(): + candidates.append(c1) + if resource_dir.exists() and resource_dir.is_dir(): + candidates.append(resource_dir) + if wxid_dir: + for c in [ + wxid_dir / "msg" / "emoji", + wxid_dir / "msg" / "emoticon", + wxid_dir / "emoji", + wxid_dir, + ]: + try: + if c.exists() and c.is_dir(): + candidates.append(c) + except Exception: + continue + candidates.append(account_dir) + p = candidates[0] + else: + raise HTTPException(status_code=404, detail="File not found.") + + try: + target = str(p.resolve()) + except Exception: + target = str(p) + + logger.info(f"open_folder: kind={kind_key} md5={md5} server_id={server_id} -> {target}") + + if os.name != "nt": + raise HTTPException(status_code=400, detail="open_folder is only supported on Windows.") + + try: + tp = Path(target) + if tp.exists() and tp.is_dir(): + subprocess.Popen(["explorer.exe", str(tp)]) + elif tp.exists(): + subprocess.Popen(["explorer.exe", "/select,", str(tp)]) + else: + subprocess.Popen(["explorer.exe", str(tp.parent)]) + except Exception as e: + raise HTTPException(status_code=500, detail=f"Failed to open explorer: {e}") + + file_found = False + try: + tp2 = Path(target) + if kind_key == "emoji": + file_found = bool(tp2.exists()) + else: + if tp2.exists() and tp2.is_file(): + file_found = True + except Exception: + pass + + resp = {"status": "success", "path": target} + if kind_key == "emoji": + resp["file_found"] = bool(file_found) + resp["materialized"] = bool(materialized_ok) if "materialized_ok" in locals() else bool(file_found) + resp["opened"] = str(opened_kind) if "opened_kind" in locals() else "unknown" + return resp diff --git a/src/wechat_decrypt_tool/routers/decrypt.py b/src/wechat_decrypt_tool/routers/decrypt.py new file mode 100644 index 0000000..376f30f --- /dev/null +++ b/src/wechat_decrypt_tool/routers/decrypt.py @@ -0,0 +1,66 @@ +from fastapi import APIRouter, HTTPException +from pydantic import BaseModel, Field + +from ..logging_config import get_logger +from ..path_fix import PathFixRoute +from ..wechat_decrypt import decrypt_wechat_databases + +logger = get_logger(__name__) + +router = APIRouter(route_class=PathFixRoute) + + +class DecryptRequest(BaseModel): + """解密请求模型""" + + key: str = Field(..., description="解密密钥，64位十六进制字符串") + db_storage_path: str = Field(..., description="数据库存储路径，必须是绝对路径") + + +@router.post("/api/decrypt", summary="解密微信数据库") +async def decrypt_databases(request: DecryptRequest): + """使用提供的密钥解密指定账户的微信数据库 + + 参数: + - key: 解密密钥（必选）- 64位十六进制字符串 + - db_storage_path: 数据库存储路径（必选），如 D:\\wechatMSG\\xwechat_files\\{微信id}\\db_storage + + 注意： + - 一个密钥只能解密对应账户的数据库 + - 必须提供具体的db_storage_path，不支持自动检测多账户 + - 支持自动处理Windows路径中的反斜杠转义问题 + """ + logger.info(f"开始解密请求: db_storage_path={request.db_storage_path}") + try: + # 验证密钥格式 + if not request.key or len(request.key) != 64: + logger.warning(f"密钥格式无效: 长度={len(request.key) if request.key else 0}") + raise HTTPException(status_code=400, detail="密钥格式无效，必须是64位十六进制字符串") + + # 使用新的解密API + results = decrypt_wechat_databases( + db_storage_path=request.db_storage_path, + key=request.key, + ) + + if results["status"] == "error": + logger.error(f"解密失败: {results['message']}") + raise HTTPException(status_code=400, detail=results["message"]) + + logger.info(f"解密完成: 成功 {results['successful_count']}/{results['total_databases']} 个数据库") + + return { + "status": "completed" if results["status"] == "success" else "failed", + "total_databases": results["total_databases"], + "success_count": results["successful_count"], + "failure_count": results["failed_count"], + "output_directory": results["output_directory"], + "message": results["message"], + "processed_files": results["processed_files"], + "failed_files": results["failed_files"], + "account_results": results.get("account_results", {}), + } + + except Exception as e: + logger.error(f"解密API异常: {str(e)}") + raise HTTPException(status_code=500, detail=str(e)) diff --git a/src/wechat_decrypt_tool/routers/health.py b/src/wechat_decrypt_tool/routers/health.py new file mode 100644 index 0000000..5e8caf9 --- /dev/null +++ b/src/wechat_decrypt_tool/routers/health.py @@ -0,0 +1,22 @@ +from fastapi import APIRouter + +from ..logging_config import get_logger +from ..path_fix import PathFixRoute + +logger = get_logger(__name__) + +router = APIRouter(route_class=PathFixRoute) + + +@router.get("/", summary="根端点") +async def root(): + """根端点""" + logger.info("访问根端点") + return {"message": "微信数据库解密工具 API"} + + +@router.get("/api/health", summary="健康检查端点") +async def health_check(): + """健康检查端点""" + logger.debug("健康检查请求") + return {"status": "healthy", "service": "微信解密工具"} diff --git a/src/wechat_decrypt_tool/routers/media.py b/src/wechat_decrypt_tool/routers/media.py new file mode 100644 index 0000000..40dc7e7 --- /dev/null +++ b/src/wechat_decrypt_tool/routers/media.py @@ -0,0 +1,458 @@ +import asyncio +import json +from typing import Optional + +from fastapi import APIRouter, HTTPException +from fastapi.responses import Response, StreamingResponse +from pydantic import BaseModel, Field + +from ..logging_config import get_logger +from ..media_helpers import ( + _collect_all_dat_files, + _decrypt_and_save_resource, + _detect_image_media_type, + _extract_wechat_aes_key_from_process, + _find_wechat_xor_key, + _get_resource_dir, + _get_wechat_template_most_common_last2, + _get_wechat_v2_ciphertext, + _load_media_keys, + _resolve_account_dir, + _resolve_account_wxid_dir, + _save_media_keys, + _try_find_decrypted_resource, +) +from ..path_fix import PathFixRoute + +logger = get_logger(__name__) + +router = APIRouter(route_class=PathFixRoute) + + +class MediaKeysRequest(BaseModel): + """媒体密钥请求模型""" + + account: Optional[str] = Field(None, description="账号目录名（可选，默认使用第一个）") + force_extract: bool = Field(False, description="是否强制从微信进程重新提取密钥") + + +class MediaDecryptRequest(BaseModel): + """媒体解密请求模型""" + + account: Optional[str] = Field(None, description="账号目录名（可选，默认使用第一个）") + xor_key: Optional[str] = Field(None, description="XOR密钥（十六进制，如 0xA5 或 A5）") + aes_key: Optional[str] = Field(None, description="AES密钥（16字符ASCII字符串）") + + +@router.get("/api/media/keys", summary="获取图片解密密钥") +async def get_media_keys(account: Optional[str] = None, force_extract: bool = False): + """获取图片解密密钥（XOR和AES） + + 如果已缓存密钥且不强制提取，直接返回缓存的密钥。 + 否则尝试从微信进程中提取密钥。 + + 注意：提取AES密钥需要微信进程正在运行。 + """ + account_dir = _resolve_account_dir(account) + wxid_dir = _resolve_account_wxid_dir(account_dir) + + # 尝试加载已缓存的密钥 + cached_keys = _load_media_keys(account_dir) + if cached_keys and not force_extract: + xor_key = cached_keys.get("xor") + aes_key = cached_keys.get("aes") + if xor_key is not None and aes_key: + return { + "status": "success", + "source": "cache", + "xor_key": f"0x{int(xor_key):02X}", + "aes_key": str(aes_key)[:16] if aes_key else "", + "message": "已从缓存加载密钥", + } + + if not wxid_dir: + return { + "status": "error", + "message": "未找到微信数据目录，请确保已正确配置 db_storage_path", + } + + # 尝试提取XOR密钥 + xor_key = _find_wechat_xor_key(str(wxid_dir)) + if xor_key is None: + return { + "status": "error", + "message": "无法提取XOR密钥，请确保微信数据目录中存在 _t.dat 模板文件", + } + + # 尝试提取AES密钥（需要微信进程运行） + aes_key16: Optional[bytes] = None + aes_message = "" + + most_common = _get_wechat_template_most_common_last2(str(wxid_dir)) + if most_common: + ct = _get_wechat_v2_ciphertext(wxid_dir, most_common) + if ct: + aes_key16 = _extract_wechat_aes_key_from_process(ct) + if aes_key16: + aes_message = "已从微信进程提取AES密钥" + # 保存密钥到缓存 + _save_media_keys(account_dir, xor_key, aes_key16) + else: + aes_message = "无法从微信进程提取AES密钥（微信是否正在运行？）" + else: + aes_message = "未找到V2加密模板文件" + else: + aes_message = "未找到足够的模板文件用于提取AES密钥" + + return { + "status": "success", + "source": "extracted", + "xor_key": f"0x{xor_key:02X}", + "aes_key": aes_key16.decode("ascii", errors="ignore") if aes_key16 else "", + "message": f"XOR密钥提取成功。{aes_message}", + } + + +@router.post("/api/media/keys", summary="保存图片解密密钥") +async def save_media_keys_api(request: MediaKeysRequest, xor_key: str, aes_key: str): + """手动保存图片解密密钥 + + 参数: + - xor_key: XOR密钥（十六进制格式，如 0xA5 或 A5） + - aes_key: AES密钥（16字符ASCII字符串） + """ + account_dir = _resolve_account_dir(request.account) + + # 解析XOR密钥 + try: + xor_hex = xor_key.strip().lower().replace("0x", "") + xor_int = int(xor_hex, 16) + except Exception: + raise HTTPException(status_code=400, detail="XOR密钥格式无效，请使用十六进制格式如 0xA5") + + # 验证AES密钥 + aes_str = aes_key.strip() + if len(aes_str) < 16: + raise HTTPException(status_code=400, detail="AES密钥长度不足，需要至少16个字符") + + # 保存密钥 + _save_media_keys(account_dir, xor_int, aes_str[:16].encode("ascii", errors="ignore")) + + return { + "status": "success", + "message": "密钥已保存", + "xor_key": f"0x{xor_int:02X}", + "aes_key": aes_str[:16], + } + + +@router.post("/api/media/decrypt_all", summary="批量解密所有图片资源") +async def decrypt_all_media(request: MediaDecryptRequest): + """批量解密所有图片资源到 output/databases/{账号}/resource 目录 + + 解密后的图片按MD5哈希命名，存储在 resource/{md5前2位}/{md5}.{ext} 路径下。 + 这样可以快速通过MD5定位资源文件。 + + 参数: + - account: 账号目录名（可选） + - xor_key: XOR密钥（可选，不提供则从缓存读取） + - aes_key: AES密钥（可选，不提供则从缓存读取） + """ + account_dir = _resolve_account_dir(request.account) + wxid_dir = _resolve_account_wxid_dir(account_dir) + + if not wxid_dir: + raise HTTPException( + status_code=400, + detail="未找到微信数据目录，请确保已正确配置 db_storage_path", + ) + + # 获取密钥 + xor_key_int: Optional[int] = None + aes_key16: Optional[bytes] = None + + if request.xor_key: + try: + xor_hex = request.xor_key.strip().lower().replace("0x", "") + xor_key_int = int(xor_hex, 16) + except Exception: + raise HTTPException(status_code=400, detail="XOR密钥格式无效") + + if request.aes_key: + aes_str = request.aes_key.strip() + if len(aes_str) >= 16: + aes_key16 = aes_str[:16].encode("ascii", errors="ignore") + + # 如果未提供密钥，尝试从缓存加载 + if xor_key_int is None or aes_key16 is None: + cached = _load_media_keys(account_dir) + if xor_key_int is None: + xor_key_int = cached.get("xor") + if aes_key16 is None: + aes_str = str(cached.get("aes") or "").strip() + if len(aes_str) >= 16: + aes_key16 = aes_str[:16].encode("ascii", errors="ignore") + + # 如果仍然没有XOR密钥，尝试自动提取 + if xor_key_int is None: + xor_key_int = _find_wechat_xor_key(str(wxid_dir)) + + if xor_key_int is None: + raise HTTPException( + status_code=400, + detail="未找到XOR密钥，请先调用 /api/media/keys 获取密钥或手动提供", + ) + + # 收集所有.dat文件 + logger.info(f"开始扫描 {wxid_dir} 中的.dat文件...") + dat_files = _collect_all_dat_files(wxid_dir) + total_files = len(dat_files) + logger.info(f"共发现 {total_files} 个.dat文件") + + if total_files == 0: + return { + "status": "success", + "message": "未发现需要解密的.dat文件", + "total": 0, + "success_count": 0, + "skip_count": 0, + "fail_count": 0, + "output_dir": str(_get_resource_dir(account_dir)), + } + + # 开始解密 + success_count = 0 + skip_count = 0 + fail_count = 0 + failed_files: list[dict] = [] + + resource_dir = _get_resource_dir(account_dir) + resource_dir.mkdir(parents=True, exist_ok=True) + + for dat_path, md5 in dat_files: + # 检查是否已解密 + existing = _try_find_decrypted_resource(account_dir, md5) + if existing: + skip_count += 1 + continue + + # 解密并保存 + success, msg = _decrypt_and_save_resource( + dat_path, md5, account_dir, xor_key_int, aes_key16 + ) + + if success: + success_count += 1 + else: + fail_count += 1 + if len(failed_files) < 100: # 只记录前100个失败 + failed_files.append( + { + "file": str(dat_path), + "md5": md5, + "error": msg, + } + ) + + logger.info(f"解密完成: 成功={success_count}, 跳过={skip_count}, 失败={fail_count}") + + return { + "status": "success", + "message": f"解密完成: 成功 {success_count}, 跳过 {skip_count}, 失败 {fail_count}", + "total": total_files, + "success_count": success_count, + "skip_count": skip_count, + "fail_count": fail_count, + "output_dir": str(resource_dir), + "failed_files": failed_files[:20] if failed_files else [], + } + + +@router.get("/api/media/resource/{md5}", summary="获取已解密的资源文件") +async def get_decrypted_resource(md5: str, account: Optional[str] = None): + """直接从解密资源目录获取图片 + + 如果资源已解密，直接返回解密后的文件。 + 这比实时解密更快，适合频繁访问的场景。 + """ + if not md5 or len(md5) != 32: + raise HTTPException(status_code=400, detail="无效的MD5") + + account_dir = _resolve_account_dir(account) + p = _try_find_decrypted_resource(account_dir, md5.lower()) + + if not p: + raise HTTPException(status_code=404, detail="资源未找到，请先执行批量解密") + + data = p.read_bytes() + media_type = _detect_image_media_type(data[:32]) + return Response(content=data, media_type=media_type) + + +@router.get("/api/media/decrypt_all_stream", summary="批量解密所有图片资源（SSE实时进度）") +async def decrypt_all_media_stream( + account: Optional[str] = None, + xor_key: Optional[str] = None, + aes_key: Optional[str] = None, +): + """批量解密所有图片资源，通过SSE实时推送进度 + + 返回格式为Server-Sent Events，每条消息包含: + - type: progress/complete/error + - current: 当前处理数量 + - total: 总文件数 + - success_count: 成功数 + - skip_count: 跳过数（已解密） + - fail_count: 失败数 + - current_file: 当前处理的文件名 + - status: 当前文件状态（success/skip/fail） + - message: 状态消息 + + 跳过原因：文件已经解密过 + 失败原因： + - 文件为空 + - V4-V2版本需要AES密钥但未提供 + - 未知加密版本 + - 解密结果为空 + - 解密后非有效图片格式 + """ + + async def generate_progress(): + try: + account_dir = _resolve_account_dir(account) + wxid_dir = _resolve_account_wxid_dir(account_dir) + + if not wxid_dir: + yield f"data: {json.dumps({'type': 'error', 'message': '未找到微信数据目录'})}\n\n" + return + + # 获取密钥 + xor_key_int: Optional[int] = None + aes_key16: Optional[bytes] = None + + if xor_key: + try: + xor_hex = xor_key.strip().lower().replace("0x", "") + xor_key_int = int(xor_hex, 16) + except Exception: + yield f"data: {json.dumps({'type': 'error', 'message': 'XOR密钥格式无效'})}\n\n" + return + + if aes_key: + aes_str = aes_key.strip() + if len(aes_str) >= 16: + aes_key16 = aes_str[:16].encode("ascii", errors="ignore") + + # 如果未提供密钥，尝试从缓存加载 + if xor_key_int is None or aes_key16 is None: + cached = _load_media_keys(account_dir) + if xor_key_int is None: + xor_key_int = cached.get("xor") + if aes_key16 is None: + aes_str = str(cached.get("aes") or "").strip() + if len(aes_str) >= 16: + aes_key16 = aes_str[:16].encode("ascii", errors="ignore") + + # 如果仍然没有XOR密钥，尝试自动提取 + if xor_key_int is None: + xor_key_int = _find_wechat_xor_key(str(wxid_dir)) + + if xor_key_int is None: + yield f"data: {json.dumps({'type': 'error', 'message': '未找到XOR密钥，请先获取密钥'})}\n\n" + return + + # 收集所有.dat文件 + logger.info(f"[SSE] 开始扫描 {wxid_dir} 中的.dat文件...") + yield f"data: {json.dumps({'type': 'scanning', 'message': '正在扫描图片文件...'})}\n\n" + await asyncio.sleep(0) + + dat_files = _collect_all_dat_files(wxid_dir) + total_files = len(dat_files) + logger.info(f"[SSE] 共发现 {total_files} 个.dat文件（仅图片）") + + if total_files == 0: + yield f"data: {json.dumps({'type': 'complete', 'message': '未发现需要解密的图片文件', 'total': 0, 'success_count': 0, 'skip_count': 0, 'fail_count': 0})}\n\n" + return + + # 发送总数信息 + yield f"data: {json.dumps({'type': 'start', 'total': total_files, 'message': f'开始解密 {total_files} 个图片文件'})}\n\n" + await asyncio.sleep(0) + + # 开始解密 + success_count = 0 + skip_count = 0 + fail_count = 0 + failed_files: list[dict] = [] + + resource_dir = _get_resource_dir(account_dir) + resource_dir.mkdir(parents=True, exist_ok=True) + + for i, (dat_path, md5) in enumerate(dat_files): + current = i + 1 + file_name = dat_path.name + + # 检查是否已解密 + existing = _try_find_decrypted_resource(account_dir, md5) + if existing: + skip_count += 1 + # 每100个跳过的文件发送一次进度，减少消息量 + if skip_count % 100 == 0 or current == total_files: + yield ( + f"data: {json.dumps({'type': 'progress', 'current': current, 'total': total_files, 'success_count': success_count, 'skip_count': skip_count, 'fail_count': fail_count, 'current_file': file_name, 'status': 'skip', 'message': '已存在'})}\n\n" + ) + await asyncio.sleep(0) + continue + + # 解密并保存 + success, msg = _decrypt_and_save_resource( + dat_path, md5, account_dir, xor_key_int, aes_key16 + ) + + if success: + success_count += 1 + status = "success" + status_msg = "解密成功" + logger.debug(f"[SSE] 解密成功: {file_name}") + else: + fail_count += 1 + status = "fail" + status_msg = msg + logger.warning(f"[SSE] 解密失败: {file_name} - {msg}") + if len(failed_files) < 100: + failed_files.append( + { + "file": file_name, + "md5": md5, + "error": msg, + } + ) + + # 每处理一个文件发送进度（成功或失败都发送） + yield ( + f"data: {json.dumps({'type': 'progress', 'current': current, 'total': total_files, 'success_count': success_count, 'skip_count': skip_count, 'fail_count': fail_count, 'current_file': file_name, 'status': status, 'message': status_msg})}\n\n" + ) + + # 每处理10个文件让出一次控制权，避免阻塞 + if current % 10 == 0: + await asyncio.sleep(0) + + logger.info(f"[SSE] 解密完成: 成功={success_count}, 跳过={skip_count}, 失败={fail_count}") + + # 发送完成消息 + yield ( + f"data: {json.dumps({'type': 'complete', 'total': total_files, 'success_count': success_count, 'skip_count': skip_count, 'fail_count': fail_count, 'output_dir': str(resource_dir), 'failed_files': failed_files[:20], 'message': f'解密完成: 成功 {success_count}, 跳过 {skip_count}, 失败 {fail_count}'})}\n\n" + ) + + except Exception as e: + logger.error(f"[SSE] 解密过程出错: {e}") + yield f"data: {json.dumps({'type': 'error', 'message': str(e)})}\n\n" + + return StreamingResponse( + generate_progress(), + media_type="text/event-stream", + headers={ + "Cache-Control": "no-cache", + "Connection": "keep-alive", + "X-Accel-Buffering": "no", + }, + ) diff --git a/src/wechat_decrypt_tool/routers/wechat_detection.py b/src/wechat_decrypt_tool/routers/wechat_detection.py new file mode 100644 index 0000000..ae0077d --- /dev/null +++ b/src/wechat_decrypt_tool/routers/wechat_detection.py @@ -0,0 +1,73 @@ +from typing import Optional + +from fastapi import APIRouter + +from ..logging_config import get_logger +from ..path_fix import PathFixRoute + +logger = get_logger(__name__) + +router = APIRouter(route_class=PathFixRoute) + + +@router.get("/api/wechat-detection", summary="详细检测微信安装信息") +async def detect_wechat_detailed(data_root_path: Optional[str] = None): + """详细检测微信安装信息，包括版本、路径、消息目录等。""" + logger.info("开始执行微信检测") + try: + from ..wechat_detection import detect_wechat_installation, detect_current_logged_in_account + + info = detect_wechat_installation(data_root_path=data_root_path) + + # 检测当前登录账号 + current_account_info = detect_current_logged_in_account(data_root_path) + info['current_account'] = current_account_info + + # 添加一些统计信息 + stats = { + 'total_databases': len(info['databases']), + 'total_user_accounts': len(info['user_accounts']), + 'total_message_dirs': len(info['message_dirs']), + 'has_wechat_installed': info['wechat_install_path'] is not None, + 'detection_time': __import__('datetime').datetime.now().isoformat(), + } + + logger.info(f"微信检测完成: 检测到 {stats['total_user_accounts']} 个账户, {stats['total_databases']} 个数据库") + + return { + 'status': 'success', + 'data': info, + 'statistics': stats, + } + except Exception as e: + logger.error(f"微信检测失败: {str(e)}") + return { + 'status': 'error', + 'error': str(e), + 'data': None, + 'statistics': None, + } + + +@router.get("/api/current-account", summary="检测当前登录账号") +async def detect_current_account(data_root_path: Optional[str] = None): + """检测当前登录的微信账号""" + logger.info("开始检测当前登录账号") + try: + from ..wechat_detection import detect_current_logged_in_account + + result = detect_current_logged_in_account(data_root_path) + + logger.info(f"当前账号检测完成: {result.get('message', '无结果')}") + + return { + 'status': 'success', + 'data': result, + } + except Exception as e: + logger.error(f"当前账号检测失败: {str(e)}") + return { + 'status': 'error', + 'error': str(e), + 'data': None, + }