"""Stable business identity helpers for acquired candidate items.""" from __future__ import annotations from typing import Any from urllib.parse import parse_qs, urlparse def _first(params: dict[str, list[str]], key: str) -> str: values = params.get(key) or [] return values[0].strip() if values and values[0] else "" def parse_weixin_unique_key(url: str | None) -> str | None: """Build the Weixin article key from URL params when all parts exist.""" if not url: return None params = parse_qs(urlparse(url).query, keep_blank_values=False) biz = _first(params, "__biz") mid = _first(params, "mid") idx = _first(params, "idx") sn = _first(params, "sn") if not all([biz, mid, idx, sn]): return None return f"wx:{biz}:{mid}:{idx}:{sn}" def _raw_nested(raw_payload: dict[str, Any] | None, path: tuple[str, ...]) -> str: current: Any = raw_payload or {} for key in path: if not isinstance(current, dict): return "" current = current.get(key) return str(current).strip() if current else "" def build_unique_key( platform: str, *, platform_item_id: str | None = None, url: str | None = None, raw_payload: dict[str, Any] | None = None, ) -> str | None: """Return the formal candidate item unique_key, or None if unknown.""" item_id = (platform_item_id or "").strip() if platform == "xiaohongshu": item_id = item_id or _raw_nested(raw_payload, ("candidate", "source_id")) item_id = item_id or _raw_nested(raw_payload, ("detail", "data", "data", "channel_content_id")) return f"xhs:{item_id}" if item_id else None if platform == "douyin": item_id = item_id or _raw_nested(raw_payload, ("candidate", "source_id")) item_id = item_id or _raw_nested(raw_payload, ("candidate", "raw", "id")) item_id = item_id or _raw_nested(raw_payload, ("detail", "data", "data", "channel_content_id")) return f"dy:{item_id}" if item_id else None if platform == "weixin": return parse_weixin_unique_key(url) return None