38 lines
1.4 KiB
Python
38 lines
1.4 KiB
Python
"""Relate saved discussions by note identity, independent of signed URLs."""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from urllib.parse import urlsplit
|
|
|
|
|
|
def note_key(source: dict) -> str:
|
|
identity = source.get("note_id")
|
|
if isinstance(identity, str) and identity:
|
|
return "note:" + identity
|
|
url = source.get("source_url") or ""
|
|
try:
|
|
parts = urlsplit(url)
|
|
except ValueError:
|
|
return "url:" + url
|
|
if parts.hostname in {"www.xiaohongshu.com", "xiaohongshu.com"}:
|
|
match = re.fullmatch(r"/(?:explore|discovery/item)/([^/]+)/?", parts.path)
|
|
if match:
|
|
return "note:" + match.group(1)
|
|
return "url:" + url
|
|
|
|
|
|
def related_context(batch: list[dict], sources: list[dict]) -> list[dict]:
|
|
by_id = {row["source_id"]: row for row in sources}
|
|
notes = {note_key(row): row for row in sources if row.get("source_type") == "note"}
|
|
own_ids = {row["source_id"] for row in batch}
|
|
related = {}
|
|
for source in batch:
|
|
note = notes.get(note_key(source))
|
|
if note and note["source_id"] not in own_ids:
|
|
related[note["source_id"]] = note
|
|
for key in ("parent_source_id", "root_source_id"):
|
|
parent = by_id.get(source.get(key))
|
|
if parent and parent["source_id"] not in own_ids:
|
|
related[parent["source_id"]] = parent
|
|
return list(related.values())
|