diff --git a/src/wechat_decrypt_tool/routers/sns.py b/src/wechat_decrypt_tool/routers/sns.py index fcc5af19..135c69b8 100644 --- a/src/wechat_decrypt_tool/routers/sns.py +++ b/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 `` 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() diff --git a/src/wechat_decrypt_tool/sns_export_service.py b/src/wechat_decrypt_tool/sns_export_service.py index a43736f7..f912da1f 100644 --- a/src/wechat_decrypt_tool/sns_export_service.py +++ b/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,7 +1175,7 @@ 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", ""), @@ -1170,8 +1183,8 @@ def _load_sns_export_snapshot( "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, diff --git a/tests/test_sns_xml_likes.py b/tests/test_sns_xml_likes.py new file mode 100644 index 00000000..e5ff5364 --- /dev/null +++ b/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 = ('Post' + '' + 'aliceAlice' + 'bobBob' + 'deleted1' + '') + 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", 'PostaliceAlice')) + 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")