diff --git a/frontend/composables/chat/useChatSearch.js b/frontend/composables/chat/useChatSearch.js index 8a92e75..caa2915 100644 --- a/frontend/composables/chat/useChatSearch.js +++ b/frontend/composables/chat/useChatSearch.js @@ -457,6 +457,33 @@ if (scope === 'global') { } return (text.charAt(0) || '?').toString() } + +const buildTransientSearchTargetContact = ({ username, displayName = '', avatar = '', isGroup = null } = {}) => { +const u = String(username || '').trim() +if (!u) return null +const name = String(displayName || u).trim() || u +return { + id: u, + username: u, + name, + avatar: String(avatar || '').trim() || null, + avatarColor: '#4B5563', + lastMessage: '', + lastMessageTime: '', + unreadCount: 0, + isGroup: typeof isGroup === 'boolean' ? isGroup : u.endsWith('@chatroom'), + isTop: false +} +} + +const resolveSearchTargetContact = ({ username, displayName = '', avatar = '', isGroup = null } = {}) => { +const u = String(username || '').trim() +if (!u) return null +const existing = contacts.value.find((c) => String(c?.username || '').trim() === u) +if (existing) return existing +if (String(selectedContact.value?.username || '').trim() === u) return selectedContact.value +return buildTransientSearchTargetContact({ username: u, displayName, avatar, isGroup }) +} const searchContextBannerText = computed(() => { if (!searchContext.value?.active) return '' const kind = String(searchContext.value.kind || 'search') @@ -986,7 +1013,12 @@ if (!hit?.id) return const targetUsername = String(hit?.username || selectedContact.value?.username || '').trim() if (!targetUsername) return -const targetContact = contacts.value.find((c) => c?.username === targetUsername) +const targetContact = resolveSearchTargetContact({ + username: targetUsername, + displayName: String(hit?.conversationName || hit?.username || targetUsername).trim(), + avatar: String(hit?.conversationAvatar || hit?.senderAvatar || '').trim(), + isGroup: targetUsername.endsWith('@chatroom') +}) if (targetContact && selectedContact.value?.username !== targetUsername) { await selectContact(targetContact, { skipLoadMessages: true }) } @@ -1051,7 +1083,12 @@ const u = String(targetUsername || selectedContact.value?.username || '').trim() const anchor = String(anchorId || '').trim() if (!u || !anchor) return -const targetContact = contacts.value.find((c) => c?.username === u) +const targetContact = resolveSearchTargetContact({ + username: u, + displayName: String(selectedContact.value?.name || u).trim(), + avatar: String(selectedContact.value?.avatar || '').trim(), + isGroup: u.endsWith('@chatroom') +}) if (targetContact && selectedContact.value?.username !== u) { await selectContact(targetContact, { skipLoadMessages: true }) } diff --git a/frontend/pages/chat/[[username]].vue b/frontend/pages/chat/[[username]].vue index 6ec51e6..4ab20f8 100644 --- a/frontend/pages/chat/[[username]].vue +++ b/frontend/pages/chat/[[username]].vue @@ -119,6 +119,23 @@ const nextMessageLoadToken = () => { return messageLoadSequence } +const buildTransientContact = ({ username, name = '', avatar = '', isGroup = null } = {}) => { + const u = String(username || '').trim() + const displayName = String(name || u).trim() || u + return { + id: u, + username: u, + name: displayName, + avatar: String(avatar || '').trim() || null, + avatarColor: '#4B5563', + lastMessage: '', + lastMessageTime: '', + unreadCount: 0, + isGroup: typeof isGroup === 'boolean' ? isGroup : u.endsWith('@chatroom'), + isTop: false + } +} + const buildChatPath = (username) => { return username ? `/chat/${encodeURIComponent(username)}` : '/chat' } @@ -334,14 +351,28 @@ const selectContact = async (contact, options = {}) => { } const applyRouteSelection = async (options = {}) => { + const selectionReason = String(options.reason || 'route-selection').trim() || 'route-selection' + const requested = routeUsername.value || '' + if ((!contacts.value || contacts.value.length === 0) && requested) { + if (selectedContact.value?.username === requested) { + return + } + await selectContact(buildTransientContact({ username: requested }), { + syncRoute: false, + deferLoadMessages: !!options.deferLoadMessages, + reason: `${selectionReason}:transient-route-empty-list` + }) + return + } if (!contacts.value || contacts.value.length === 0) { selectedContact.value = null return } - const selectionReason = String(options.reason || 'route-selection').trim() || 'route-selection' - const requested = routeUsername.value || '' if (requested) { + if (selectedContact.value?.username === requested) { + return + } const matched = contacts.value.find((contact) => contact.username === requested) if (matched) { if (selectedContact.value?.username !== matched.username) { @@ -353,6 +384,12 @@ const applyRouteSelection = async (options = {}) => { } return } + await selectContact(buildTransientContact({ username: requested }), { + syncRoute: false, + deferLoadMessages: !!options.deferLoadMessages, + reason: `${selectionReason}:transient-route` + }) + return } await selectContact(contacts.value[0], { diff --git a/src/wechat_decrypt_tool/chat_helpers.py b/src/wechat_decrypt_tool/chat_helpers.py index 603bb81..40b0bad 100644 --- a/src/wechat_decrypt_tool/chat_helpers.py +++ b/src/wechat_decrypt_tool/chat_helpers.py @@ -14,6 +14,7 @@ from fastapi import HTTPException from .app_paths import get_output_databases_dir from .logging_config import get_logger +from .sqlite_diagnostics import collect_sqlite_diagnostics, format_sqlite_diagnostics try: import zstandard as zstd # type: ignore @@ -1755,9 +1756,10 @@ def _load_latest_message_previews(account_dir: Path, usernames: list[str]) -> di session_db_path = Path(account_dir) / "session.db" if session_db_path.exists() and remaining: - sconn = sqlite3.connect(str(session_db_path)) - sconn.row_factory = sqlite3.Row + sconn: Optional[sqlite3.Connection] = None try: + sconn = sqlite3.connect(str(session_db_path)) + sconn.row_factory = sqlite3.Row uniq = list(dict.fromkeys([u for u in remaining if u])) chunk_size = 900 for i in range(0, len(uniq), chunk_size): @@ -1786,10 +1788,24 @@ def _load_latest_message_previews(account_dir: Path, usernames: list[str]) -> di if not u: continue expected_ts_by_user[u] = int(r["last_timestamp"] or 0) + except sqlite3.DatabaseError as e: + expected_ts_by_user = {} + logger.warning( + "[sessions.preview] session timestamp lookup failed account=%s db=%s usernames=%s sample_usernames=%s error=%s diag=%s", + account_dir.name, + str(session_db_path), + len(remaining), + sorted([u for u in remaining if u])[:5], + str(e), + format_sqlite_diagnostics( + collect_sqlite_diagnostics(session_db_path, quick_check=True, table_name="SessionTable") + ), + ) except Exception: expected_ts_by_user = {} finally: - sconn.close() + if sconn is not None: + sconn.close() if _DEBUG_SESSIONS: logger.info( @@ -1800,9 +1816,16 @@ def _load_latest_message_previews(account_dir: Path, usernames: list[str]) -> di ) for db_path in db_paths: - conn = sqlite3.connect(str(db_path)) - conn.row_factory = sqlite3.Row + conn: Optional[sqlite3.Connection] = None + stage = "connect" + stage_username = "" + stage_table = "" try: + stage = "connect" + conn = sqlite3.connect(str(db_path)) + conn.row_factory = sqlite3.Row + + stage = "sqlite_master" rows = conn.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall() names = [str(r[0]) for r in rows if r and r[0]] lower_to_actual = {n.lower(): n for n in names} @@ -1818,9 +1841,12 @@ def _load_latest_message_previews(account_dir: Path, usernames: list[str]) -> di conn.text_factory = bytes for u, tn in found.items(): + stage_username = str(u) + stage_table = str(tn) quoted = _quote_ident(tn) try: try: + stage = "latest_row_with_name2id" r = conn.execute( "SELECT " "m.local_type, m.message_content, m.compress_content, m.create_time, m.sort_seq, m.local_id, " @@ -1831,6 +1857,7 @@ def _load_latest_message_previews(account_dir: Path, usernames: list[str]) -> di "LIMIT 1" ).fetchone() except Exception: + stage = "latest_row_without_name2id" r = conn.execute( "SELECT " "local_type, message_content, compress_content, create_time, sort_seq, local_id, '' AS sender_username " @@ -1838,6 +1865,20 @@ def _load_latest_message_previews(account_dir: Path, usernames: list[str]) -> di "ORDER BY sort_seq DESC, local_id DESC " "LIMIT 1" ).fetchone() + except sqlite3.DatabaseError as e: + logger.warning( + "[sessions.preview] latest row query failed account=%s db=%s username=%s table=%s stage=%s error=%s diag=%s", + account_dir.name, + str(db_path), + str(u), + str(tn), + stage, + str(e), + format_sqlite_diagnostics( + collect_sqlite_diagnostics(db_path, quick_check=True, table_name=tn) + ), + ) + continue except Exception as e: if _DEBUG_SESSIONS: logger.info( @@ -1855,6 +1896,7 @@ def _load_latest_message_previews(account_dir: Path, usernames: list[str]) -> di expected_ts = int(expected_ts_by_user.get(u) or 0) if expected_ts > 0 and create_time > 0 and create_time < expected_ts: try: + stage = "latest_row_by_create_time_with_name2id" r2 = conn.execute( "SELECT " "m.local_type, m.message_content, m.compress_content, m.create_time, m.sort_seq, m.local_id, " @@ -1866,6 +1908,7 @@ def _load_latest_message_previews(account_dir: Path, usernames: list[str]) -> di ).fetchone() except Exception: try: + stage = "latest_row_by_create_time_without_name2id" r2 = conn.execute( "SELECT " "local_type, message_content, compress_content, create_time, sort_seq, local_id, '' AS sender_username " @@ -1873,6 +1916,20 @@ def _load_latest_message_previews(account_dir: Path, usernames: list[str]) -> di "ORDER BY COALESCE(create_time, 0) DESC, COALESCE(sort_seq, 0) DESC, local_id DESC " "LIMIT 1" ).fetchone() + except sqlite3.DatabaseError as e: + logger.warning( + "[sessions.preview] latest row requery failed account=%s db=%s username=%s table=%s stage=%s error=%s diag=%s", + account_dir.name, + str(db_path), + str(u), + str(tn), + stage, + str(e), + format_sqlite_diagnostics( + collect_sqlite_diagnostics(db_path, quick_check=True, table_name=tn) + ), + ) + r2 = None except Exception: r2 = None @@ -1900,8 +1957,28 @@ def _load_latest_message_previews(account_dir: Path, usernames: list[str]) -> di prev = best.get(u) if prev is None or sort_key > prev[0]: best[u] = (sort_key, preview) + except sqlite3.DatabaseError as e: + logger.warning( + "[sessions.preview] malformed message db account=%s db=%s stage=%s username=%s table=%s remaining=%s sample_usernames=%s error=%s diag=%s", + account_dir.name, + str(db_path), + stage, + stage_username, + stage_table, + len(remaining), + sorted([u for u in remaining if u])[:5], + str(e), + format_sqlite_diagnostics( + collect_sqlite_diagnostics(db_path, quick_check=True, table_name=(stage_table or None)) + ), + ) + continue finally: - conn.close() + if conn is not None: + try: + conn.close() + except Exception: + pass previews = {u: v[1] for u, v in best.items() if v and v[1]} if _DEBUG_SESSIONS: diff --git a/src/wechat_decrypt_tool/routers/chat.py b/src/wechat_decrypt_tool/routers/chat.py index f1c423a..b6f2ec0 100644 --- a/src/wechat_decrypt_tool/routers/chat.py +++ b/src/wechat_decrypt_tool/routers/chat.py @@ -77,6 +77,7 @@ from ..session_last_message import ( get_session_last_message_status, load_session_last_messages, ) +from ..sqlite_diagnostics import collect_sqlite_diagnostics, format_sqlite_diagnostics from ..wcdb_realtime import ( WCDBRealtimeError, WCDB_REALTIME, @@ -2022,12 +2023,18 @@ def _sync_chat_realtime_messages_for_table( if backfill_limit > max_scan: backfill_limit = max_scan - msg_conn = sqlite3.connect(str(msg_db_path)) - msg_conn.row_factory = sqlite3.Row + msg_conn: Optional[sqlite3.Connection] = None + stage = "connect" try: + stage = "connect" + msg_conn = sqlite3.connect(str(msg_db_path)) + msg_conn.row_factory = sqlite3.Row + + stage = "resolve_db_storage_paths" msg_db_path_real, _res_db_path_real = _resolve_db_storage_message_paths(account_dir, msg_db_path.stem) name2id_synced = False try: + stage = "sync_name2id" name2id_result = _sync_output_name2id_from_live( msg_conn, rt_conn=rt_conn, @@ -2050,12 +2057,14 @@ def _sync_chat_realtime_messages_for_table( ) quoted_table = _quote_ident(table_name) + stage = "max_local_id" row = msg_conn.execute(f"SELECT MAX(local_id) AS mx FROM {quoted_table}").fetchone() try: max_local_id = int((row["mx"] if row is not None else 0) or 0) except Exception: max_local_id = 0 + stage = "pragma_table_info" cols = msg_conn.execute(f"PRAGMA table_info({quoted_table})").fetchall() available_cols = {str(c[1] or "") for c in cols} base_cols = [ @@ -2075,6 +2084,7 @@ def _sync_chat_realtime_messages_for_table( placeholders = ",".join(["?"] * len(insert_cols)) insert_sql = f"INSERT OR IGNORE INTO {quoted_table} ({','.join(insert_cols)}) VALUES ({placeholders})" + stage = "collect_realtime_rows" fetch_result = _collect_realtime_rows_for_session( trace_id=None, account_name=account_dir.name, @@ -2094,6 +2104,7 @@ def _sync_chat_realtime_messages_for_table( backfilled = 0 if new_rows: if not name2id_synced: + stage = "upsert_name2id_fallback" _best_effort_upsert_output_name2id_rows( msg_conn, account_name=account_dir.name, @@ -2102,6 +2113,7 @@ def _sync_chat_realtime_messages_for_table( values = [tuple(r.get(c) for c in insert_cols) for r in reversed(new_rows)] insert_t0 = time.perf_counter() + stage = "insert_new_rows" msg_conn.executemany(insert_sql, values) msg_conn.commit() insert_ms = (time.perf_counter() - insert_t0) * 1000.0 @@ -2131,6 +2143,7 @@ def _sync_chat_realtime_messages_for_table( if update_values: before_changes = msg_conn.total_changes update_t0 = time.perf_counter() + stage = "backfill_packed_info" msg_conn.executemany( f"UPDATE {quoted_table} SET packed_info_data = ? WHERE local_id = ? AND (packed_info_data IS NULL OR length(packed_info_data) = 0)", update_values, @@ -2187,8 +2200,11 @@ def _sync_chat_realtime_messages_for_table( if inserted and newest_ts: session_db_path = account_dir / "session.db" - sconn = sqlite3.connect(str(session_db_path)) + sconn: Optional[sqlite3.Connection] = None try: + stage = "open_session_db" + sconn = sqlite3.connect(str(session_db_path)) + stage = "update_session_table" sconn.execute("INSERT OR IGNORE INTO SessionTable(username) VALUES (?)", (username,)) sconn.execute( """ @@ -2217,6 +2233,7 @@ def _sync_chat_realtime_messages_for_table( ), ) + stage = "update_session_last_message" _ensure_session_last_message_table(sconn) sconn.execute( """ @@ -2239,8 +2256,25 @@ def _sync_chat_realtime_messages_for_table( ), ) sconn.commit() + except sqlite3.DatabaseError as e: + logger.warning( + "[realtime] malformed session db during sync account=%s username=%s session_db=%s stage=%s error=%s diag=%s", + account_dir.name, + username, + str(session_db_path), + stage, + str(e), + format_sqlite_diagnostics( + collect_sqlite_diagnostics(session_db_path, quick_check=True, table_name="SessionTable") + ), + ) + raise HTTPException( + status_code=500, + detail=f"Malformed session db during realtime sync: {session_db_path.name}", + ) finally: - sconn.close() + if sconn is not None: + sconn.close() return { "username": username, @@ -2250,8 +2284,23 @@ def _sync_chat_realtime_messages_for_table( "backfilled": int(backfilled), "preview": preview or "", } + except sqlite3.DatabaseError as e: + logger.warning( + "[realtime] malformed decrypted message db account=%s username=%s db=%s table=%s stage=%s error=%s diag=%s", + account_dir.name, + username, + str(msg_db_path), + table_name, + stage, + str(e), + format_sqlite_diagnostics( + collect_sqlite_diagnostics(msg_db_path, quick_check=True, table_name=table_name) + ), + ) + raise HTTPException(status_code=500, detail=f"Malformed decrypted message db: {msg_db_path.name}") finally: - msg_conn.close() + if msg_conn is not None: + msg_conn.close() @router.post("/api/chat/realtime/sync_all", summary="实时消息同步到解密库(全会话增量)") @@ -2545,20 +2594,24 @@ def sync_chat_realtime_messages_all( except HTTPException as e: errors.append(f"{uname}: {str(e.detail or '')}".strip()) logger.warning( - "[%s] sync session failed account=%s username=%s err=%s", + "[%s] sync session failed account=%s username=%s db=%s table=%s err=%s", trace_id, account_dir.name, uname, + str(msg_db_path), + str(table_name), str(e.detail or "").strip(), ) continue except Exception as e: errors.append(f"{uname}: {str(e)}".strip()) logger.exception( - "[%s] sync session crashed account=%s username=%s", + "[%s] sync session crashed account=%s username=%s db=%s table=%s", trace_id, account_dir.name, uname, + str(msg_db_path), + str(table_name), ) continue @@ -4173,6 +4226,15 @@ def list_chat_sessions( ) last_previews = load_session_last_messages(account_dir, usernames) except Exception: + logger.exception( + "[sessions.list] session_last_message preview load failed account=%s preview_mode=%s usernames=%s diag=%s", + account_dir.name, + preview_mode, + len(usernames), + format_sqlite_diagnostics( + collect_sqlite_diagnostics(account_dir / "session.db", quick_check=True, table_name="session_last_message") + ), + ) last_previews = {} def _is_generic_location_preview(value: Any) -> bool: @@ -4189,7 +4251,17 @@ def list_chat_sessions( else [u for u in usernames if u and ((u not in last_previews) or _is_generic_location_preview(last_previews.get(u)))] ) if targets: - legacy = _load_latest_message_previews(account_dir, targets) + try: + legacy = _load_latest_message_previews(account_dir, targets) + except Exception: + logger.exception( + "[sessions.list] legacy latest-message preview fallback failed account=%s preview_mode=%s targets=%s sample_targets=%s; falling back to session summaries", + account_dir.name, + preview_mode, + len(targets), + [str(u) for u in targets[:5]], + ) + legacy = {} for u, v in legacy.items(): if v: last_previews[u] = v diff --git a/src/wechat_decrypt_tool/routers/decrypt.py b/src/wechat_decrypt_tool/routers/decrypt.py index d797441..7ee6126 100644 --- a/src/wechat_decrypt_tool/routers/decrypt.py +++ b/src/wechat_decrypt_tool/routers/decrypt.py @@ -59,6 +59,12 @@ async def decrypt_databases(request: DecryptRequest): raise HTTPException(status_code=400, detail=results["message"]) logger.info(f"解密完成: 成功 {results['successful_count']}/{results['total_databases']} 个数据库") + if int(results.get("diagnostic_warning_count") or 0) > 0: + logger.warning( + "解密完成但检测到诊断告警: warning_dbs=%s total=%s", + int(results.get("diagnostic_warning_count") or 0), + int(results.get("total_databases") or 0), + ) # 成功解密后,按账号保存数据库密钥(用于前端自动回填) try: @@ -77,6 +83,7 @@ async def decrypt_databases(request: DecryptRequest): "processed_files": results["processed_files"], "failed_files": results["failed_files"], "account_results": results.get("account_results", {}), + "diagnostic_warning_count": int(results.get("diagnostic_warning_count") or 0), } except HTTPException: @@ -160,6 +167,7 @@ async def decrypt_databases_stream( processed_files: list[str] = [] failed_files: list[str] = [] account_results: dict = {} + diagnostic_warning_count = 0 overall_current = 0 for account, dbs in account_databases.items(): @@ -181,6 +189,8 @@ async def decrypt_databases_stream( account_success = 0 account_processed: list[str] = [] account_failed: list[str] = [] + account_db_diagnostics: dict[str, dict] = {} + account_diagnostic_warning_count = 0 for db_info in dbs: if await request.is_disconnected(): @@ -223,6 +233,23 @@ async def decrypt_databases_stream( ok = bool(task.result()) except Exception: ok = False + db_diagnostic = dict(getattr(decryptor, "last_result", {}) or {}) + if not db_diagnostic: + db_diagnostic = { + "db_path": str(db_path), + "db_name": str(db_name), + "output_path": str(output_path), + "success": bool(ok), + } + db_diagnostic["account"] = str(account) + account_db_diagnostics[db_name] = db_diagnostic + + if ( + (not bool(db_diagnostic.get("success", ok))) + or int(db_diagnostic.get("failed_pages") or 0) > 0 + or str(db_diagnostic.get("diagnostic_status") or "") != "ok" + ): + account_diagnostic_warning_count += 1 if ok: account_success += 1 @@ -238,18 +265,25 @@ async def decrypt_databases_stream( status = "fail" msg = "解密失败" - yield _sse( - { - "type": "progress", - "current": overall_current, - "total": total_databases, - "success_count": success_count, - "fail_count": fail_count, - "current_file": current_file, - "status": status, - "message": msg, - } - ) + payload = { + "type": "progress", + "current": overall_current, + "total": total_databases, + "success_count": success_count, + "fail_count": fail_count, + "current_file": current_file, + "status": status, + "message": msg, + } + if db_diagnostic: + payload["diagnostic_status"] = str(db_diagnostic.get("diagnostic_status") or "") + payload["page_failures"] = int(db_diagnostic.get("failed_pages") or 0) + if db_diagnostic.get("failed_page_samples"): + payload["failed_page_samples"] = db_diagnostic.get("failed_page_samples") + if db_diagnostic.get("diagnostics"): + payload["diagnostics"] = db_diagnostic.get("diagnostics") + + yield _sse(payload) if overall_current % 5 == 0: await asyncio.sleep(0) @@ -261,7 +295,10 @@ async def decrypt_databases_stream( "output_dir": str(account_output_dir), "processed_files": account_processed, "failed_files": account_failed, + "db_diagnostics": account_db_diagnostics, + "diagnostic_warning_count": int(account_diagnostic_warning_count), } + diagnostic_warning_count += int(account_diagnostic_warning_count) # Build cache table (keep behavior consistent with the POST endpoint). if os.environ.get("WECHAT_TOOL_BUILD_SESSION_LAST_MESSAGE", "1") != "0": @@ -311,6 +348,7 @@ async def decrypt_databases_stream( "processed_files": processed_files, "failed_files": failed_files, "account_results": account_results, + "diagnostic_warning_count": int(diagnostic_warning_count), } # Save db key for frontend autofill. diff --git a/src/wechat_decrypt_tool/sqlite_diagnostics.py b/src/wechat_decrypt_tool/sqlite_diagnostics.py new file mode 100644 index 0000000..2cfae67 --- /dev/null +++ b/src/wechat_decrypt_tool/sqlite_diagnostics.py @@ -0,0 +1,150 @@ +from __future__ import annotations + +import json +import sqlite3 +from pathlib import Path +from typing import Any, Mapping, Optional + +SQLITE_HEADER = b"SQLite format 3\x00" + + +def _clean_text(value: Any, *, limit: int = 240) -> str: + text = " ".join(str(value or "").split()).strip() + if len(text) > limit: + return text[: limit - 3] + "..." + return text + + +def _clean_error(exc: BaseException, *, limit: int = 240) -> str: + text = _clean_text(exc, limit=limit) + if text: + return f"{type(exc).__name__}: {text}" + return type(exc).__name__ + + +def _quote_ident(name: str) -> str: + return '"' + str(name or "").replace('"', '""') + '"' + + +def collect_sqlite_diagnostics( + path: str | Path, + *, + quick_check: bool = True, + table_name: Optional[str] = None, + table_sample_limit: int = 5, +) -> dict[str, Any]: + db_path = Path(path) + diagnostics: dict[str, Any] = { + "path": str(db_path), + "exists": bool(db_path.exists()), + } + + if not diagnostics["exists"]: + return diagnostics + + try: + diagnostics["size"] = int(db_path.stat().st_size) + except Exception as exc: + diagnostics["size_error"] = _clean_error(exc) + + try: + with db_path.open("rb") as f: + header = f.read(len(SQLITE_HEADER)) + diagnostics["header_ok"] = header == SQLITE_HEADER + diagnostics["header_hex"] = header.hex() + except Exception as exc: + diagnostics["header_error"] = _clean_error(exc) + + if not quick_check: + return diagnostics + + conn: sqlite3.Connection | None = None + try: + conn = sqlite3.connect(str(db_path)) + + try: + row = conn.execute("PRAGMA page_size").fetchone() + diagnostics["page_size"] = int((row[0] if row is not None else 0) or 0) + except Exception as exc: + diagnostics["page_size_error"] = _clean_error(exc) + + try: + row = conn.execute("PRAGMA page_count").fetchone() + diagnostics["page_count"] = int((row[0] if row is not None else 0) or 0) + except Exception as exc: + diagnostics["page_count_error"] = _clean_error(exc) + + try: + rows = conn.execute("SELECT name FROM sqlite_master WHERE type='table' ORDER BY name").fetchall() + table_names = [str(row[0]) for row in rows if row and row[0]] + diagnostics["table_count"] = len(table_names) + if table_names: + diagnostics["tables_sample"] = table_names[: max(int(table_sample_limit or 0), 1)] + except Exception as exc: + diagnostics["table_list_error"] = _clean_error(exc) + + if table_name: + diagnostics["target_table"] = str(table_name) + try: + cols = conn.execute(f"PRAGMA table_info({_quote_ident(table_name)})").fetchall() + diagnostics["target_table_exists"] = bool(cols) + if cols: + diagnostics["target_table_columns"] = [ + str(col[1]) + for col in cols[:8] + if len(col) > 1 and str(col[1] or "").strip() + ] + except Exception as exc: + diagnostics["target_table_error"] = _clean_error(exc) + + try: + rows = conn.execute("PRAGMA quick_check").fetchall() + values = [_clean_text(row[0]) for row in rows if row and row[0] is not None] + if values: + diagnostics["quick_check"] = values[0] if len(values) == 1 else values[:5] + diagnostics["quick_check_ok"] = len(values) == 1 and values[0].lower() == "ok" + if len(values) > 5: + diagnostics["quick_check_truncated"] = len(values) - 5 + else: + diagnostics["quick_check"] = "" + diagnostics["quick_check_ok"] = None + except Exception as exc: + diagnostics["quick_check_error"] = _clean_error(exc) + diagnostics["quick_check_ok"] = False + except Exception as exc: + diagnostics["connect_error"] = _clean_error(exc) + finally: + if conn is not None: + try: + conn.close() + except Exception: + pass + + return diagnostics + + +def sqlite_diagnostics_status(diagnostics: Mapping[str, Any]) -> str: + if not diagnostics: + return "not_run" + if not diagnostics.get("exists", False): + return "missing" + if diagnostics.get("header_ok") is False: + return "bad_header" + if diagnostics.get("connect_error"): + return "connect_error" + if diagnostics.get("quick_check_error"): + return "quick_check_error" + if diagnostics.get("quick_check_ok") is False: + return "quick_check_failed" + if diagnostics.get("quick_check_ok") is True: + return "ok" + return "header_only" + + +def format_sqlite_diagnostics(diagnostics: Mapping[str, Any]) -> str: + compact: dict[str, Any] = {} + for key, value in diagnostics.items(): + if value in (None, "", [], {}): + continue + compact[str(key)] = value + return json.dumps(compact, ensure_ascii=False, sort_keys=True) diff --git a/src/wechat_decrypt_tool/wechat_decrypt.py b/src/wechat_decrypt_tool/wechat_decrypt.py index 8afcb13..e714e7d 100644 --- a/src/wechat_decrypt_tool/wechat_decrypt.py +++ b/src/wechat_decrypt_tool/wechat_decrypt.py @@ -21,6 +21,7 @@ from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC from .app_paths import get_output_databases_dir +from .sqlite_diagnostics import collect_sqlite_diagnostics, sqlite_diagnostics_status # 注意:不再支持默认密钥,所有密钥必须通过参数传入 @@ -221,6 +222,7 @@ class WeChatDatabaseDecryptor: self.key_bytes = bytes.fromhex(key_hex) except ValueError: raise ValueError("密钥必须是有效的十六进制字符串") + self.last_result: dict = {} def decrypt_database(self, db_path: str, output_path: str) -> bool: """解密微信4.x版本数据库 @@ -234,6 +236,79 @@ class WeChatDatabaseDecryptor: from .logging_config import get_logger logger = get_logger(__name__) + result = { + "db_path": str(db_path), + "db_name": Path(str(db_path)).name, + "output_path": str(output_path), + "success": False, + "copied_as_sqlite": False, + "input_size": 0, + "output_size": 0, + "total_pages": 0, + "successful_pages": 0, + "failed_pages": 0, + "failed_page_samples": [], + "failure_reasons": {}, + "diagnostics": {}, + "diagnostic_status": "not_run", + "error": "", + } + self.last_result = result + + def _append_failed_page(page_num: int, reason: str, error: str = "") -> None: + result["failure_reasons"][reason] = int(result["failure_reasons"].get(reason) or 0) + 1 + if len(result["failed_page_samples"]) >= 8: + return + item = {"page": int(page_num), "reason": str(reason)} + err = " ".join(str(error or "").split()).strip() + if err: + item["error"] = err[:200] + result["failed_page_samples"].append(item) + + def _finalize(success: bool, error: str = "") -> bool: + result["success"] = bool(success) + if error: + result["error"] = " ".join(str(error).split()).strip() + + output_file = Path(str(output_path)) + if output_file.exists(): + try: + result["output_size"] = int(output_file.stat().st_size) + except Exception: + pass + + diagnostics = collect_sqlite_diagnostics(output_file, quick_check=True) + result["diagnostics"] = diagnostics + result["diagnostic_status"] = sqlite_diagnostics_status(diagnostics) + + payload = { + "db_name": result["db_name"], + "db_path": result["db_path"], + "output_path": result["output_path"], + "success": result["success"], + "copied_as_sqlite": result["copied_as_sqlite"], + "input_size": result["input_size"], + "output_size": result["output_size"], + "total_pages": result["total_pages"], + "successful_pages": result["successful_pages"], + "failed_pages": result["failed_pages"], + "failure_reasons": result["failure_reasons"], + "failed_page_samples": result["failed_page_samples"], + "diagnostic_status": result["diagnostic_status"], + "diagnostics": result["diagnostics"], + "error": result["error"], + } + log_fn = logger.info + if ( + (not result["success"]) + or int(result["failed_pages"] or 0) > 0 + or str(result["diagnostic_status"] or "") != "ok" + ): + log_fn = logger.warning + log_fn("[decrypt.diagnostic] %s", json.dumps(payload, ensure_ascii=False, sort_keys=True)) + self.last_result = result + return bool(success) + logger.info(f"开始解密数据库: {db_path}") try: @@ -241,17 +316,19 @@ class WeChatDatabaseDecryptor: encrypted_data = f.read() logger.info(f"读取文件大小: {len(encrypted_data)} bytes") + result["input_size"] = int(len(encrypted_data)) if len(encrypted_data) < 4096: logger.warning(f"文件太小,跳过解密: {db_path}") - return False + return _finalize(False, "file_too_small") # 检查是否已经是解密的数据库 if encrypted_data.startswith(SQLITE_HEADER): logger.info(f"文件已是SQLite格式,直接复制: {db_path}") with open(output_path, 'wb') as f: f.write(encrypted_data) - return True + result["copied_as_sqlite"] = True + return _finalize(True) # 提取salt (前16字节) salt = encrypted_data[:16] @@ -295,6 +372,7 @@ class WeChatDatabaseDecryptor: total_pages = len(encrypted_data) // page_size successful_pages = 0 failed_pages = 0 + result["total_pages"] = int(total_pages) # 逐页解密 for cur_page in range(total_pages): @@ -329,6 +407,7 @@ class WeChatDatabaseDecryptor: if stored_hmac != expected_hmac: logger.warning(f"页面 {page_num} HMAC验证失败") failed_pages += 1 + _append_failed_page(page_num, "hmac") continue # 提取IV和加密数据用于AES解密 @@ -354,20 +433,32 @@ class WeChatDatabaseDecryptor: except Exception as e: logger.error(f"页面 {page_num} AES解密失败: {e}") failed_pages += 1 + _append_failed_page(page_num, "aes", str(e)) continue logger.info(f"解密完成: 成功 {successful_pages} 页, 失败 {failed_pages} 页") + result["successful_pages"] = int(successful_pages) + result["failed_pages"] = int(failed_pages) # 写入解密后的文件 with open(output_path, 'wb') as f: f.write(decrypted_data) logger.info(f"解密文件大小: {len(decrypted_data)} bytes") - return True + if failed_pages > 0: + logger.warning( + "解密输出包含页失败: db=%s total_pages=%s failed_pages=%s failure_reasons=%s samples=%s", + result["db_name"], + int(total_pages), + int(failed_pages), + json.dumps(result["failure_reasons"], ensure_ascii=False, sort_keys=True), + json.dumps(result["failed_page_samples"], ensure_ascii=False), + ) + return _finalize(True) except Exception as e: logger.error(f"解密失败: {db_path}, 错误: {e}") - return False + return _finalize(False, str(e)) def decrypt_wechat_databases(db_storage_path: str = None, key: str = None) -> dict: """ @@ -493,6 +584,7 @@ def decrypt_wechat_databases(db_storage_path: str = None, key: str = None) -> di processed_files = [] failed_files = [] account_results = {} + diagnostic_warning_count = 0 for account_name, databases in account_databases.items(): logger.info(f"开始解密账号 {account_name} 的 {len(databases)} 个数据库") @@ -523,6 +615,8 @@ def decrypt_wechat_databases(db_storage_path: str = None, key: str = None) -> di account_success = 0 account_processed = [] account_failed = [] + account_db_diagnostics = {} + account_diagnostic_warning_count = 0 for db_info in databases: db_path = db_info['path'] @@ -533,7 +627,26 @@ def decrypt_wechat_databases(db_storage_path: str = None, key: str = None) -> di # 解密数据库 logger.info(f"解密 {account_name}/{db_name}") - if decryptor.decrypt_database(db_path, str(output_path)): + ok = decryptor.decrypt_database(db_path, str(output_path)) + db_diagnostic = dict(getattr(decryptor, "last_result", {}) or {}) + if not db_diagnostic: + db_diagnostic = { + "db_path": str(db_path), + "db_name": str(db_name), + "output_path": str(output_path), + "success": bool(ok), + } + db_diagnostic["account"] = str(account_name) + account_db_diagnostics[db_name] = db_diagnostic + + if ( + (not bool(db_diagnostic.get("success", ok))) + or int(db_diagnostic.get("failed_pages") or 0) > 0 + or str(db_diagnostic.get("diagnostic_status") or "") != "ok" + ): + account_diagnostic_warning_count += 1 + + if ok: account_success += 1 success_count += 1 account_processed.append(str(output_path)) @@ -551,8 +664,11 @@ def decrypt_wechat_databases(db_storage_path: str = None, key: str = None) -> di "failed": len(databases) - account_success, "output_dir": str(account_output_dir), "processed_files": account_processed, - "failed_files": account_failed + "failed_files": account_failed, + "db_diagnostics": account_db_diagnostics, + "diagnostic_warning_count": int(account_diagnostic_warning_count), } + diagnostic_warning_count += int(account_diagnostic_warning_count) # 构建“会话最后一条消息”缓存表:把耗时挪到解密阶段,后续会话列表直接查表 if os.environ.get("WECHAT_TOOL_BUILD_SESSION_LAST_MESSAGE", "1") != "0": @@ -586,6 +702,7 @@ def decrypt_wechat_databases(db_storage_path: str = None, key: str = None) -> di "failed_files": failed_files, "account_results": account_results, # 新增:按账号的详细结果 "detected_accounts": detected_accounts, + "diagnostic_warning_count": int(diagnostic_warning_count), } logger.info("=" * 60)