Create-knowledge-find-decode/acquisition/unique_key.py

"""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