goods/ingestion/opengoods/etl/dedup.py

"""Duplicate detection and product merging.

Barcodes (GTIN) are already unique at the schema level, so duplicates here are
non-GTIN records that describe the same product (same normalized name + brand +
net content). For each duplicate group we keep the highest-quality product as
canonical and merge the rest into it: child rows (provenance, images, MSRP) are
re-pointed to the canonical product, the merged product is marked ``merged``
with ``canonical_id`` set, and a row is written to ``merge_log``.
"""

from __future__ import annotations

from typing import Any

import psycopg

from opengoods.etl.quality import update_quality


def _norm(text: str | None) -> str:
    return " ".join((text or "").lower().split())


def product_signature(name: str | None, brand: str | None, net_canonical: Any | None) -> str | None:
    """Stable signature for non-GTIN dedup, or ``None`` if too sparse to match."""
    n = _norm(name)
    if not n:
        return None
    net = "" if net_canonical is None else str(net_canonical)
    return f"{n}|{_norm(brand)}|{net}"


def choose_canonical(members: list[dict]) -> dict:
    """Pick the canonical product: best quality, then oldest, then lowest id."""
    return min(
        members,
        key=lambda m: (
            -float(m.get("quality_score") or 0.0),
            m.get("created_at"),
            str(m.get("id")),
        ),
    )


def find_duplicate_groups(conn: psycopg.Connection) -> list[list[dict]]:
    """Return groups (size >= 2) of active products sharing a signature."""
    rows = conn.execute(
        """
        SELECT p.id, p.name, b.normalized_name, p.net_content_canonical,
               p.quality_score, p.created_at
        FROM product p
        LEFT JOIN brand b ON b.id = p.brand_id
        WHERE p.status = 'active'
        """
    ).fetchall()

    groups: dict[str, list[dict]] = {}
    for r in rows:
        sig = product_signature(r[1], r[2], r[3])
        if sig is None:
            continue
        member = {
            "id": r[0],
            "name": r[1],
            "quality_score": r[4],
            "created_at": r[5],
        }
        groups.setdefault(sig, []).append(member)

    return [m for m in groups.values() if len(m) >= 2]


def merge_products(
    conn: psycopg.Connection,
    kept_id: str,
    merged_id: str,
    reason: str = "auto-dedup",
    actor: str = "ingestion",
) -> None:
    """Merge ``merged_id`` into ``kept_id`` (re-point children, mark merged)."""
    if kept_id == merged_id:
        return

    # Re-point provenance, images and MSRP to the canonical product.
    conn.execute(
        "UPDATE product_source SET product_id = %s WHERE product_id = %s",
        (kept_id, merged_id),
    )
    conn.execute(
        "UPDATE product_image SET product_id = %s WHERE product_id = %s",
        (kept_id, merged_id),
    )
    conn.execute(
        "UPDATE product_msrp SET product_id = %s WHERE product_id = %s",
        (kept_id, merged_id),
    )

    # food_detail has product_id as PK, so it can only move if the canonical
    # product does not already have one.
    kept_has_food = conn.execute(
        "SELECT 1 FROM food_detail WHERE product_id = %s", (kept_id,)
    ).fetchone()
    if not kept_has_food:
        conn.execute(
            "UPDATE food_detail SET product_id = %s WHERE product_id = %s",
            (kept_id, merged_id),
        )

    conn.execute(
        "UPDATE product SET status = 'merged', canonical_id = %s WHERE id = %s",
        (kept_id, merged_id),
    )
    conn.execute(
        """
        INSERT INTO merge_log (kept_id, merged_id, reason, actor)
        VALUES (%s, %s, %s, %s)
        """,
        (kept_id, merged_id, reason, actor),
    )

    # The canonical product gained sources, so its quality may have changed.
    update_quality(conn, kept_id)


def dedup_all(
    conn: psycopg.Connection, actor: str = "ingestion", dry_run: bool = False
) -> dict[str, int]:
    """Merge every duplicate group. Returns counts of groups and merges."""
    groups = find_duplicate_groups(conn)
    merged = 0
    for members in groups:
        canonical = choose_canonical(members)
        for m in members:
            if m["id"] == canonical["id"]:
                continue
            if not dry_run:
                merge_products(conn, canonical["id"], m["id"], actor=actor)
            merged += 1
    return {"groups": len(groups), "merged": merged}