修复朋友圈点赞人显示和导出

src/wechat_decrypt_tool/routers/sns.py

@@ -831,6 +831,28 @@ def _clean(v: str) -> str:
     return ""
 
 
+def _load_sns_like_contacts(names: list[str], contact_db_path: Path) -> dict[str, dict[str, Any]]:
+    contacts: dict[str, dict[str, Any]] = {}
+    unique_names = list(dict.fromkeys(names))
+    if contact_db_path.exists():
+        for start in range(0, len(unique_names), 500):
+            contacts.update(_load_contact_rows(contact_db_path, unique_names[start:start + 500]))
+    return contacts
+
+
+def _resolve_sns_like_names(names: list[str], contact_db_path: Path, nicknames: Optional[dict[str, str]] = None, contacts: Optional[dict[str, dict[str, Any]]] = None) -> list[str]:
+    nicknames = nicknames or {}
+    if not names:
+        return []
+    try:
+        if contacts is None:
+            contacts = _load_sns_like_contacts(names, contact_db_path)
+        return [_pick_display_name(contacts.get(name), nicknames.get(name) or name) for name in names]
+    except sqlite3.Error:
+        logger.warning("[sns] unable to resolve like contact names", exc_info=True)
+        return [nicknames.get(name) or name for name in names]
+
+
 def _parse_timeline_xml(xml_text: str, fallback_username: str) -> dict[str, Any]:
     out: dict[str, Any] = {
         "username": fallback_username,
@@ -840,6 +862,7 @@ def _parse_timeline_xml(xml_text: str, fallback_username: str) -> dict[str, Any]
         "sourceName": "",
         "media": [],
         "likes": [],
+        "likeNicknames": {},
         "comments": [],
         "type": 1,  # 默认类型
         "title": "",
@@ -1168,6 +1191,20 @@ def _parse_comment_images(comment_node: ET.Element) -> list[dict[str, Any]]:
             v = str(u.text).strip()
             if v:
                 likes.append(v)
+        nicknames: dict[str, str] = {}
+        for entry in root.findall(".//LocalExtraInfo/like_user_list/user_comment"):
+            if _safe_int(entry.findtext("b_deleted")) or _safe_int(entry.findtext("type")) == 2:
+                continue
+            if entry.findtext("content") or entry.find(".//emojiinfo") is not None or entry.find(".//imageinfo") is not None:
+                continue
+            username = (entry.findtext("username") or "").strip()
+            nickname = (entry.findtext("nickname") or "").strip()
+            name = username or nickname
+            if name and name not in likes:
+                likes.append(name)
+            if username and nickname:
+                nicknames[username] = nickname
+        out["likeNicknames"] = nicknames
     except Exception:
         likes = []
     out["likes"] = likes
@@ -2453,7 +2490,7 @@ def _list_from_decrypted_sqlite() -> dict[str, Any]:
                     "location": str(parsed2.get("location") or ""),
                     "sourceName": str(parsed2.get("sourceName") or ""),
                     "media": parsed2.get("media") or [],
-                    "likes": parsed2.get("likes") or [],
+                    "likes": _resolve_sns_like_names(parsed2.get("likes") or [], contact_db_path, parsed2.get("likeNicknames")),
                     "comments": parsed2.get("comments") or [],
                     "type": post_type2,
                     "title": parsed2.get("title", ""),
@@ -2856,7 +2893,7 @@ def _clean_name(v: Any) -> str:
             content_desc = str(r.get("contentDesc") or "")
             media = r.get("media") if isinstance(r.get("media"), list) else []
             likes = r.get("likes") if isinstance(r.get("likes"), list) else []
-            likes = [_clean_name(x) for x in likes if _clean_name(x)]
+            likes = _resolve_sns_like_names([_clean_name(x) for x in likes if _clean_name(x)], contact_db_path)
             comments = r.get("comments") if isinstance(r.get("comments"), list) else []
 
             # WeFlow: live photo / SNS video decryption key comes from `<enc key="...">` in raw XML.
@@ -2890,6 +2927,8 @@ def _clean_name(v: Any) -> str:
                 xml = content_by_tid.get(int(tid_s))
                 if xml:
                     parsed = _parse_timeline_xml(xml, uname)
+                    if parsed.get("likes"):
+                        likes = _resolve_sns_like_names(parsed["likes"], contact_db_path, parsed.get("likeNicknames"))
                     if parsed.get("location"):
                         location = str(parsed.get("location") or "")
                     sn0 = str(parsed.get("sourceName") or "").strip()

src/wechat_decrypt_tool/sns_export_service.py

@@ -52,6 +52,8 @@
     _extract_sns_video_key,
     _get_biz_to_official_index,
     _image_size_from_bytes,
+    _load_sns_like_contacts,
+    _resolve_sns_like_names,
     _parse_timeline_xml,
     _resolve_sns_cached_video_path,
     _to_unsigned_i64_str,
@@ -1118,12 +1120,19 @@ def _load_sns_export_snapshot(
     official_usernames: set[str] = set()
     grouped: dict[str, dict[str, Any]] = {}
 
-    for row in rows:
+    parsed_rows = [(row, _parse_timeline_xml(str(row["content"] or ""), str(row["user_name"] or "").strip())) for row in rows]
+    like_usernames = [name for _, parsed in parsed_rows for name in parsed.get("likes") or []]
+    try:
+        like_contacts = _load_sns_like_contacts(like_usernames, contact_db_path)
+    except sqlite3.Error:
+        logger.warning("[sns] unable to resolve like contact names", exc_info=True)
+        like_contacts = {}
+
+    for row, parsed in parsed_rows:
         uname = str(row["user_name"] or "").strip()
         if not uname:
             continue
         content_xml = str(row["content"] or "")
-        parsed = _parse_timeline_xml(content_xml, uname)
         post_type = int(parsed.get("type", 1) or 1)
         tid = row["tid"]
         post_id = _to_unsigned_i64_str(tid) if tid is not None else (str(parsed.get("createTime") or "") or uname)
@@ -1152,6 +1161,10 @@ def _load_sns_export_snapshot(
             if official_username:
                 official_usernames.add(official_username)
 
+        resolved_likes = _resolve_sns_like_names(parsed.get("likes") or [], contact_db_path, parsed.get("likeNicknames"), like_contacts)
+        fingerprint_data = content_xml.encode("utf-8", errors="replace")
+        if resolved_likes:
+            fingerprint_data += b"\0sns-like-names\0" + json.dumps(resolved_likes, ensure_ascii=False).encode("utf-8")
         post = {
             "id": post_id,
             "tid": tid,
@@ -1162,16 +1175,16 @@ def _load_sns_export_snapshot(
             "location": str(parsed.get("location") or ""),
             "sourceName": str(parsed.get("sourceName") or ""),
             "media": parsed.get("media") or [],
-            "likes": parsed.get("likes") or [],
+            "likes": resolved_likes,
             "comments": parsed.get("comments") or [],
             "type": post_type,
             "title": parsed.get("title", ""),
             "contentUrl": parsed.get("contentUrl", ""),
             "finderFeed": parsed.get("finderFeed", {}),
             "finderLive": parsed.get("finderLive", {}),
             "official": official,
-            # 原始 XML 指纹会同时覆盖正文、评论、点赞和媒体变化。
-            "_contentFingerprint": hashlib.sha256(content_xml.encode("utf-8", errors="replace")).hexdigest(),
+            # Include display names so incremental exports notice remark changes.
+            "_contentFingerprint": hashlib.sha256(fingerprint_data).hexdigest(),
         }
         item = grouped.setdefault(
             uname,

tests/test_sns_xml_likes.py

@@ -0,0 +1,72 @@
+import sqlite3
+import json
+import zipfile
+from unittest import mock
+
+import pytest
+from wechat_decrypt_tool.routers import sns
+from wechat_decrypt_tool import sns_export_service
+
+
+def test_xml_likes_reach_timeline_and_export_snapshot(tmp_path):
+    account = tmp_path / "account"
+    account.mkdir()
+    with sqlite3.connect(account / "sns.db") as conn:
+        conn.execute("CREATE TABLE SnsTimeLine (tid INTEGER, user_name TEXT, content TEXT)")
+        conn.execute("CREATE TABLE SnsMessage_tmp3 (local_id INTEGER, feed_id INTEGER, type INTEGER, from_username TEXT, from_nickname TEXT, del_status INTEGER)")
+        xml = ('<SnsDataItem><TimelineObject><contentDesc>Post</contentDesc></TimelineObject>'
+               '<LocalExtraInfo><like_user_list>'
+               '<user_comment><username>alice</username><nickname>Alice</nickname></user_comment>'
+               '<user_comment><username>bob</username><nickname>Bob</nickname></user_comment>'
+               '<user_comment><username>deleted</username><b_deleted>1</b_deleted></user_comment>'
+               '</like_user_list></LocalExtraInfo></SnsDataItem>')
+        conn.execute("INSERT INTO SnsTimeLine VALUES (?, ?, ?)", (42, "author", xml))
+        conn.executemany("INSERT INTO SnsMessage_tmp3 VALUES (?, ?, ?, ?, ?, ?)", [
+            (1, 42, 1, "notification_only", "Not in XML", 0),
+        ])
+
+    with sqlite3.connect(account / "contact.db") as conn:
+        conn.execute("CREATE TABLE contact (username TEXT, remark TEXT, nick_name TEXT, alias TEXT, big_head_url TEXT, small_head_url TEXT)")
+        conn.execute("INSERT INTO contact VALUES (?, ?, ?, ?, ?, ?)", ("alice", "Alice 备注", "Alice 昵称", "", "", ""))
+
+    with mock.patch.object(sns, "_resolve_account_dir", return_value=account):
+        result = sns.list_sns_timeline(account="account", source="decrypted", limit=20, offset=0)
+    snapshot = sns_export_service._load_sns_export_snapshot(account)
+    assert result["timeline"][0]["likes"] == ["Alice 备注", "Bob"]
+    assert snapshot[0]["posts"][0]["likes"] == ["Alice 备注", "Bob"]
+
+    old_fingerprint = snapshot[0]["posts"][0]["_contentFingerprint"]
+    with sqlite3.connect(account / "contact.db") as conn:
+        conn.execute("UPDATE contact SET remark = '新备注' WHERE username = 'alice'")
+    updated = sns_export_service._load_sns_export_snapshot(account)
+    assert updated[0]["posts"][0]["likes"] == ["新备注", "Bob"]
+    assert updated[0]["posts"][0]["_contentFingerprint"] != old_fingerprint
+
+
+@pytest.mark.parametrize("export_format", ["html", "json", "txt", "excel"])
+def test_all_export_formats_include_xml_likes(tmp_path, export_format):
+    account = tmp_path / "account"
+    account.mkdir()
+    with sqlite3.connect(account / "sns.db") as conn:
+        conn.execute("CREATE TABLE SnsTimeLine (tid INTEGER, user_name TEXT, content TEXT)")
+        conn.execute("INSERT INTO SnsTimeLine VALUES (?, ?, ?)", (42, "author", '<SnsDataItem><TimelineObject><contentDesc>Post</contentDesc></TimelineObject><LocalExtraInfo><like_user_list><user_comment><username>alice</username><nickname>Alice</nickname></user_comment></like_user_list></LocalExtraInfo></SnsDataItem>'))
+    job = sns_export_service.ExportJob(export_id=f"likes-{export_format}", account="account", options={"scope": "all", "format": export_format, "useCache": True, "outputDir": str(tmp_path), "fileName": f"{export_format}.zip"})
+    assets = {"cssPath": "assets/a.css", "cssPayload": "", "jsPath": "assets/a.js", "jsPayload": ""}
+    with (mock.patch.object(sns_export_service, "sync_sns_realtime_timeline_latest", return_value={"status": "noop"}),
+          mock.patch.object(sns_export_service, "export_css", return_value=""),
+          mock.patch.object(sns_export_service, "prepare_html_zip_assets", return_value=assets),
+          mock.patch.object(sns_export_service, "protect_html_document_with_external_assets", side_effect=lambda doc, _: doc),
+          mock.patch.object(sns_export_service, "write_active_html_zip_integrity"),
+          mock.patch.object(sns_export_service, "write_zip_integrity_sidecars")):
+        output = sns_export_service.SnsExportManager()._run_job(job, account)
+    with zipfile.ZipFile(output) as archive:
+        extension = {"html": "html", "json": "json", "txt": "txt", "excel": "xlsx"}[export_format]
+        payload = archive.read(f"sns_author.{extension}")
+        if export_format == "json":
+            assert json.loads(payload)["posts"][0]["likes"] == ["Alice"]
+        elif export_format == "excel":
+            import io
+            with zipfile.ZipFile(io.BytesIO(payload)) as workbook:
+                assert any(b"Alice" in workbook.read(name) for name in workbook.namelist() if name.endswith(".xml"))
+        else:
+            assert "Alice" in payload.decode("utf-8")