fix(chat): tolerate invalid avatar URLs

Author: 2977094657

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：

desktop/package-lock.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",

desktop/package.json

@@ -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"

pyproject.toml

@@ -1,5 +1,5 @@
 """微信数据库解密工具
 """
 
-__version__ = "1.9.1"
+__version__ = "1.9.2"
 __author__ = "WeChat Decrypt Tool"

src/wechat_decrypt_tool/__init__.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 query_table(table: str, targets: list[str]) -> None:
             """
             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]

src/wechat_decrypt_tool/chat_helpers.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 @@ def test_realtime_sessions_group_url_summary_keeps_scheme(self):
         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()
-