Create-knowledge-find-decode/pipeline/dedupe.py

"""Deduplication helpers for acquisition -> decode -> ingest pipelines."""
from __future__ import annotations

from dataclasses import dataclass
from typing import Iterable

from acquisition.domain import CandidateItem


@dataclass(frozen=True)
class DedupeDecision:
    keep: bool
    key: str
    reason: str = ""


def item_dedupe_key(item: CandidateItem) -> str:
    if item.platform_item_id:
        return f"{item.platform}:id:{item.platform_item_id}"
    if item.canonical_url:
        return f"url:{item.canonical_url.strip().lower()}"
    return f"item:{item.id}"


def dedupe_candidate_items(items: Iterable[CandidateItem]) -> list[CandidateItem]:
    seen: set[str] = set()
    out: list[CandidateItem] = []
    for item in items:
        key = item_dedupe_key(item)
        if key in seen:
            continue
        seen.add(key)
        out.append(item)
    return out


def should_decode_item(
    item: CandidateItem,
    *,
    decoded_item_ids: set[str] | None = None,
) -> DedupeDecision:
    key = item_dedupe_key(item)
    decoded = decoded_item_ids or set()
    if str(item.id) in decoded or key in decoded:
        return DedupeDecision(keep=False, key=key, reason="already_decoded")
    return DedupeDecision(keep=True, key=key)