diff --git a/README.md b/README.md index 1c1ef25..e9497be 100644 --- a/README.md +++ b/README.md @@ -182,6 +182,8 @@ MCP 仅暴露读取数据与获取媒体资源 URL/参数的能力；系统设 - `wechat.biz`: 公众号/服务号与微信支付记录 - `wechat.analytics`: 年度总结与聚合分析读取；年度总结只读取应用内已生成的缓存，未生成时请先在应用内打开年度总结 +会话列表、联系人和头像相关接口均采用 best-effort 读取策略。即使 `contact.db` 中某些头像字段损坏或无法按 UTF-8 解码，也会继续返回昵称、会话摘要和其他可用内容，头像则自动降级为空或占位，不会阻塞整页数据加载。 + 媒体和视频不会直接塞进 MCP JSON 响应；相关工具返回可访问 URL 或资源参数。 配套 skill 可通过 HTTP 加载，访问时需要带 MCP token： diff --git a/desktop/package-lock.json b/desktop/package-lock.json index 65a4a43..34e433e 100644 --- a/desktop/package-lock.json +++ b/desktop/package-lock.json @@ -1,12 +1,12 @@ { "name": "wechat-data-analysis-desktop", - "version": "1.9.1", + "version": "1.9.2", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "wechat-data-analysis-desktop", - "version": "1.9.1", + "version": "1.9.2", "dependencies": { "electron-updater": "^6.7.3" }, diff --git a/desktop/package.json b/desktop/package.json index 5d8838e..eda2ab9 100644 --- a/desktop/package.json +++ b/desktop/package.json @@ -1,7 +1,7 @@ { "name": "wechat-data-analysis-desktop", "private": true, - "version": "1.9.1", + "version": "1.9.2", "main": "src/main.cjs", "scripts": { "dev": "node scripts/dev.cjs", diff --git a/pyproject.toml b/pyproject.toml index 7608dd8..aa8c50b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "wechat-decrypt-tool" -version = "1.9.1" +version = "1.9.2" description = "Modern WeChat database decryption tool with React frontend" readme = "README.md" requires-python = ">=3.11" diff --git a/src/wechat_decrypt_tool/__init__.py b/src/wechat_decrypt_tool/__init__.py index 3f1e130..7866317 100644 --- a/src/wechat_decrypt_tool/__init__.py +++ b/src/wechat_decrypt_tool/__init__.py @@ -1,5 +1,5 @@ """微信数据库解密工具 """ -__version__ = "1.9.1" +__version__ = "1.9.2" __author__ = "WeChat Decrypt Tool" diff --git a/src/wechat_decrypt_tool/chat_helpers.py b/src/wechat_decrypt_tool/chat_helpers.py index b02ae11..12de700 100644 --- a/src/wechat_decrypt_tool/chat_helpers.py +++ b/src/wechat_decrypt_tool/chat_helpers.py @@ -1992,45 +1992,93 @@ def _load_latest_message_previews(account_dir: Path, usernames: list[str]) -> di return previews -def _pick_display_name(contact_row: Optional[sqlite3.Row], fallback_username: str) -> str: +def _row_get_value(row: Any, key: str, default: Any = None) -> Any: + if row is None: + return default + try: + return row[key] + except Exception: + pass + if isinstance(row, dict): + return row.get(key, default) + return default + + +def _normalize_contact_text(value: Any) -> str: + return _decode_sqlite_text(value).strip() + + +def _normalize_avatar_url(value: Any) -> str: + if value is None: + return "" + if isinstance(value, memoryview): + value = value.tobytes() + if isinstance(value, (bytes, bytearray)): + raw = bytes(value) + if not raw: + return "" + # Avatar URLs should be ASCII/UTF-8 HTTP(S) URLs. If invalid bytes were + # stored in the TEXT column, ignore that avatar instead of failing the + # surrounding chat/contact response. + try: + text = raw.decode("utf-8") + except UnicodeDecodeError: + return "" + else: + text = str(value or "") + + text = text.strip() + if text.lower().startswith(("http://", "https://")): + return text + return "" + + +def _contact_row_to_dict(row: Any) -> dict[str, Any]: + username = _normalize_contact_text(_row_get_value(row, "username", "")) + return { + "username": username, + "remark": _normalize_contact_text(_row_get_value(row, "remark", "")), + "nick_name": _normalize_contact_text(_row_get_value(row, "nick_name", "")), + "alias": _normalize_contact_text(_row_get_value(row, "alias", "")), + "big_head_url": _normalize_avatar_url(_row_get_value(row, "big_head_url", "")), + "small_head_url": _normalize_avatar_url(_row_get_value(row, "small_head_url", "")), + } + + +def _pick_display_name(contact_row: Optional[Any], fallback_username: str) -> str: if contact_row is None: return fallback_username for key in ("remark", "nick_name", "alias"): - try: - v = contact_row[key] - except Exception: - v = None - if isinstance(v, str) and v.strip(): - return v.strip() + v = _normalize_contact_text(_row_get_value(contact_row, key, "")) + if v: + return v return fallback_username -def _pick_avatar_url(contact_row: Optional[sqlite3.Row]) -> Optional[str]: +def _pick_avatar_url(contact_row: Optional[Any]) -> Optional[str]: if contact_row is None: return None for key in ("big_head_url", "small_head_url"): - try: - v = contact_row[key] - except Exception: - v = None - if isinstance(v, str) and v.strip(): - return v.strip() + v = _normalize_avatar_url(_row_get_value(contact_row, key, "")) + if v: + return v return None -def _load_contact_rows(contact_db_path: Path, usernames: list[str]) -> dict[str, sqlite3.Row]: +def _load_contact_rows(contact_db_path: Path, usernames: list[str]) -> dict[str, dict[str, Any]]: uniq = list(dict.fromkeys([u for u in usernames if u])) if not uniq: return {} - result: dict[str, sqlite3.Row] = {} + result: dict[str, dict[str, Any]] = {} conn = sqlite3.connect(str(contact_db_path)) conn.row_factory = sqlite3.Row + conn.text_factory = bytes try: def query_table(table: str, targets: list[str]) -> None: if not targets: @@ -2043,7 +2091,10 @@ def _load_contact_rows(contact_db_path: Path, usernames: list[str]) -> dict[str, """ rows = conn.execute(sql, targets).fetchall() for r in rows: - result[r["username"]] = r + item = _contact_row_to_dict(r) + username = str(item.get("username") or "").strip() + if username: + result[username] = item query_table("contact", uniq) missing = [u for u in uniq if u not in result] diff --git a/tests/test_chat_sessions_realtime_sender_preview.py b/tests/test_chat_sessions_realtime_sender_preview.py index dd92bad..c19ce12 100644 --- a/tests/test_chat_sessions_realtime_sender_preview.py +++ b/tests/test_chat_sessions_realtime_sender_preview.py @@ -1,5 +1,6 @@ import sys import threading +import sqlite3 import unittest from pathlib import Path from tempfile import TemporaryDirectory @@ -97,7 +98,101 @@ class TestChatSessionsRealtimeSenderPreview(unittest.TestCase): self.assertEqual(len(sessions), 1) self.assertEqual(sessions[0].get("lastMessage"), "群名片B: https://example.com/x") + def test_sessions_ignore_invalid_utf8_avatar_url(self): + with TemporaryDirectory() as td: + account_dir = Path(td) / "acc" + account_dir.mkdir(parents=True, exist_ok=True) + + session_conn = sqlite3.connect(str(account_dir / "session.db")) + try: + session_conn.execute( + """ + CREATE TABLE SessionTable ( + username TEXT, + unread_count INTEGER, + is_hidden INTEGER, + summary TEXT, + draft TEXT, + last_timestamp INTEGER, + sort_timestamp INTEGER, + last_msg_locald_id INTEGER, + last_msg_type INTEGER, + last_msg_sub_type INTEGER, + last_msg_sender TEXT, + last_sender_display_name TEXT + ) + """ + ) + session_conn.execute( + "INSERT INTO SessionTable VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + ("wxid_bad_avatar", 0, 0, "hello", "", 100, 100, 1, 1, 0, "", ""), + ) + session_conn.commit() + finally: + session_conn.close() + + contact_conn = sqlite3.connect(str(account_dir / "contact.db")) + try: + contact_conn.execute( + """ + CREATE TABLE contact ( + username TEXT, + remark TEXT, + nick_name TEXT, + alias TEXT, + flag INTEGER, + big_head_url TEXT, + small_head_url TEXT + ) + """ + ) + contact_conn.execute( + """ + CREATE TABLE stranger ( + username TEXT, + remark TEXT, + nick_name TEXT, + alias TEXT, + flag INTEGER, + big_head_url TEXT, + small_head_url TEXT + ) + """ + ) + contact_conn.execute( + """ + INSERT INTO contact + (username, remark, nick_name, alias, flag, big_head_url, small_head_url) + VALUES (?, ?, ?, ?, ?, CAST(x'fffe687474703a2f2f6578616d706c652e746573742f612e706e67' AS TEXT), ?) + """, + ("wxid_bad_avatar", "", "坏头像好友", "", 0, ""), + ) + contact_conn.commit() + finally: + contact_conn.close() + + with ( + patch.object(chat_router, "_resolve_account_dir", return_value=account_dir), + patch.object(chat_router.WCDB_REALTIME, "get_status", return_value={}), + patch.object(chat_router, "load_session_last_messages", return_value={}), + patch.object(chat_router, "_load_latest_message_previews", return_value={}), + ): + resp = chat_router.list_chat_sessions( + _DummyRequest(), + account="acc", + limit=50, + include_hidden=True, + include_official=True, + preview="session", + ) + + self.assertEqual(resp.get("status"), "success") + sessions = resp.get("sessions") or [] + self.assertEqual(len(sessions), 1) + self.assertEqual(sessions[0].get("name"), "坏头像好友") + self.assertEqual(sessions[0].get("lastMessage"), "hello") + self.assertIn("/api/chat/avatar", sessions[0].get("avatar") or "") + if __name__ == "__main__": unittest.main() - diff --git a/uv.lock b/uv.lock index 04eb8fd..355d727 100644 --- a/uv.lock +++ b/uv.lock @@ -872,7 +872,7 @@ wheels = [ [[package]] name = "wechat-decrypt-tool" -version = "1.9.1" +version = "1.9.2" source = { editable = "." } dependencies = [ { name = "aiofiles" },