From 56f0c05e2eeef35e5fc66d4d0a0ba74ab1113961 Mon Sep 17 00:00:00 2001 From: STDquantum <405720329@qq.com> Date: Wed, 11 Nov 2026 18:29:29 +0800 Subject: [PATCH] fix: resolve Moments comment reply names --- src/wechat_decrypt_tool/routers/sns.py | 32 +++++++++++++++++-- src/wechat_decrypt_tool/sns_export_service.py | 10 ++++-- tests/test_sns_xml_likes.py | 27 ++++++++++++++++ 3 files changed, 64 insertions(+), 5 deletions(-) diff --git a/src/wechat_decrypt_tool/routers/sns.py b/src/wechat_decrypt_tool/routers/sns.py index 135c69b8..66f6220f 100644 --- a/src/wechat_decrypt_tool/routers/sns.py +++ b/src/wechat_decrypt_tool/routers/sns.py @@ -853,6 +853,32 @@ def _resolve_sns_like_names(names: list[str], contact_db_path: Path, nicknames: return [nicknames.get(name) or name for name in names] +def _resolve_sns_comment_names(comments: list[dict[str, Any]], contact_db_path: Path, contacts: Optional[dict[str, dict[str, Any]]] = None) -> list[dict[str, Any]]: + """Resolve reply targets from contacts or other comments on the same post.""" + entries = [c for c in comments if isinstance(c, dict)] + names = [str(c.get("refUsername") or c.get("refUserName") or "").strip() for c in entries] + if contacts is None: + try: + contacts = _load_sns_like_contacts([name for name in names if name], contact_db_path) + except sqlite3.Error: + logger.warning("[sns] unable to resolve comment contact names", exc_info=True) + contacts = {} + participants = {str(c.get("username") or "").strip(): str(c.get("nickname") or "").strip() + for c in entries if c.get("username") and c.get("nickname")} + resolved = [] + for comment in comments: + if not isinstance(comment, dict): + resolved.append(comment) + continue + item = dict(comment) + username = str(item.get("refUsername") or item.get("refUserName") or "").strip() + if username: + fallback = str(item.get("refNickname") or participants.get(username) or username).strip() + item["refNickname"] = _pick_display_name(contacts.get(username), fallback) + resolved.append(item) + return resolved + + def _parse_timeline_xml(xml_text: str, fallback_username: str) -> dict[str, Any]: out: dict[str, Any] = { "username": fallback_username, @@ -2491,7 +2517,7 @@ def _list_from_decrypted_sqlite() -> dict[str, Any]: "sourceName": str(parsed2.get("sourceName") or ""), "media": parsed2.get("media") or [], "likes": _resolve_sns_like_names(parsed2.get("likes") or [], contact_db_path, parsed2.get("likeNicknames")), - "comments": parsed2.get("comments") or [], + "comments": _resolve_sns_comment_names(parsed2.get("comments") or [], contact_db_path), "type": post_type2, "title": parsed2.get("title", ""), "contentUrl": parsed2.get("contentUrl", ""), @@ -2700,7 +2726,7 @@ def _q(v: str) -> str: "sourceName": str(parsed3.get("sourceName") or ""), "media": parsed3.get("media") or [], "likes": parsed3.get("likes") or [], - "comments": parsed3.get("comments") or [], + "comments": _resolve_sns_comment_names(parsed3.get("comments") or [], contact_db_path), "type": post_type3, "title": parsed3.get("title", ""), "contentUrl": parsed3.get("contentUrl", ""), @@ -3015,7 +3041,7 @@ def _clean_name(v: Any) -> str: "sourceName": str(source_name or ""), "media": media, "likes": likes, - "comments": comments, + "comments": _resolve_sns_comment_names(comments, contact_db_path), "type": post_type, "title": title, "contentUrl": content_url, diff --git a/src/wechat_decrypt_tool/sns_export_service.py b/src/wechat_decrypt_tool/sns_export_service.py index f912da1f..ec56a49b 100644 --- a/src/wechat_decrypt_tool/sns_export_service.py +++ b/src/wechat_decrypt_tool/sns_export_service.py @@ -54,6 +54,7 @@ _image_size_from_bytes, _load_sns_like_contacts, _resolve_sns_like_names, + _resolve_sns_comment_names, _parse_timeline_xml, _resolve_sns_cached_video_path, _to_unsigned_i64_str, @@ -1122,8 +1123,10 @@ def _load_sns_export_snapshot( 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 []] + comment_usernames = [str(c.get("refUsername") or c.get("refUserName") or "").strip() + for _, parsed in parsed_rows for c in parsed.get("comments") or [] if isinstance(c, dict)] try: - like_contacts = _load_sns_like_contacts(like_usernames, contact_db_path) + like_contacts = _load_sns_like_contacts(like_usernames + comment_usernames, contact_db_path) except sqlite3.Error: logger.warning("[sns] unable to resolve like contact names", exc_info=True) like_contacts = {} @@ -1162,9 +1165,12 @@ def _load_sns_export_snapshot( official_usernames.add(official_username) resolved_likes = _resolve_sns_like_names(parsed.get("likes") or [], contact_db_path, parsed.get("likeNicknames"), like_contacts) + resolved_comments = _resolve_sns_comment_names(parsed.get("comments") or [], contact_db_path, 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") + if resolved_comments: + fingerprint_data += b"\0sns-comment-names\0" + json.dumps(resolved_comments, ensure_ascii=False).encode("utf-8") post = { "id": post_id, "tid": tid, @@ -1176,7 +1182,7 @@ def _load_sns_export_snapshot( "sourceName": str(parsed.get("sourceName") or ""), "media": parsed.get("media") or [], "likes": resolved_likes, - "comments": parsed.get("comments") or [], + "comments": resolved_comments, "type": post_type, "title": parsed.get("title", ""), "contentUrl": parsed.get("contentUrl", ""), diff --git a/tests/test_sns_xml_likes.py b/tests/test_sns_xml_likes.py index e5ff5364..29fe1f93 100644 --- a/tests/test_sns_xml_likes.py +++ b/tests/test_sns_xml_likes.py @@ -8,6 +8,33 @@ from wechat_decrypt_tool import sns_export_service +def test_comment_reply_names_and_incremental_fingerprint(tmp_path): + account = tmp_path / "account" + account.mkdir() + xml = ('Post' + 'aliceAliceHi' + 'authorReplyalice' + 'authorOtherunknown' + '') + 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", xml)) + 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", "", "", "")) + with mock.patch.object(sns, "_resolve_account_dir", return_value=account): + timeline = sns.list_sns_timeline(account="account", source="decrypted", limit=20, offset=0) + snapshot = sns_export_service._load_sns_export_snapshot(account)[0]["posts"][0] + assert timeline["timeline"][0]["comments"][1]["refNickname"] == "备注" + assert snapshot["comments"][1]["refNickname"] == "备注" + assert snapshot["comments"][2]["refNickname"] == "unknown" + with sqlite3.connect(account / "contact.db") as conn: + conn.execute("DELETE FROM contact") + updated = sns_export_service._load_sns_export_snapshot(account)[0]["posts"][0] + assert updated["comments"][1]["refNickname"] == "Alice" + assert updated["_contentFingerprint"] != snapshot["_contentFingerprint"] + + def test_xml_likes_reach_timeline_and_export_snapshot(tmp_path): account = tmp_path / "account" account.mkdir()