Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 29 additions & 3 deletions src/wechat_decrypt_tool/routers/sns.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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", ""),
Expand Down Expand Up @@ -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", ""),
Expand Down Expand Up @@ -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,
Expand Down
10 changes: 8 additions & 2 deletions src/wechat_decrypt_tool/sns_export_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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 = {}
Expand Down Expand Up @@ -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,
Expand All @@ -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", ""),
Expand Down
27 changes: 27 additions & 0 deletions tests/test_sns_xml_likes.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 = ('<SnsDataItem><TimelineObject><contentDesc>Post</contentDesc></TimelineObject>'
'<comment><username>alice</username><nickname>Alice</nickname><content>Hi</content></comment>'
'<comment><username>author</username><content>Reply</content><refUserName>alice</refUserName></comment>'
'<comment><username>author</username><content>Other</content><refUserName>unknown</refUserName></comment>'
'</SnsDataItem>')
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()
Expand Down
Loading