goods/ingestion/opengoods/etl/supplement.py

"""Apply GS1 (or other authoritative) supplements to existing products.

A supplement only fills *gaps*: a field is written only when the product does
not already have a value. Each applied supplement records field-level provenance
in ``product_source`` and refreshes the product's quality score.
"""

from __future__ import annotations

from decimal import Decimal, InvalidOperation
from typing import Any

import psycopg
from psycopg.types.json import Jsonb

from opengoods import units
from opengoods.adapters.gs1 import GS1_HOMEPAGE, GS1_LICENSE, GS1_TRUST, SOURCE_NAME
from opengoods.etl.load import _ensure_brand, _normalize_brand, ensure_source_named
from opengoods.etl.quality import update_quality


def ensure_gs1_source(conn: psycopg.Connection) -> str:
    """Upsert the GS1 source row and return its id."""
    return ensure_source_named(conn, SOURCE_NAME, GS1_HOMEPAGE, GS1_LICENSE, GS1_TRUST)


def _ensure_manufacturer(conn: psycopg.Connection, name: str | None) -> str | None:
    if not name:
        return None
    row = conn.execute(
        """
        INSERT INTO manufacturer (name, normalized_name)
        VALUES (%s, %s)
        ON CONFLICT (normalized_name) DO UPDATE SET name = manufacturer.name
        RETURNING id
        """,
        (name, _normalize_brand(name)),
    ).fetchone()
    return row[0]


def _net_content(rec: dict) -> tuple[Decimal, str, Decimal | None] | None:
    raw_value = rec.get("net_content_value")
    unit = rec.get("net_content_unit")
    if raw_value is None or not unit:
        return None
    try:
        value = Decimal(str(raw_value))
    except (InvalidOperation, ValueError):
        return None
    try:
        canonical = units.normalize(value, unit).canonical_value
    except units.UnitError:
        canonical = None
    return value, unit, canonical


def apply_supplement(conn: psycopg.Connection, rec: dict[str, Any], source_id: str) -> list[str]:
    """Fill missing fields of the GTIN-matched product from ``rec``.

    Returns the list of field names actually filled (empty if the product is
    unknown or already complete for the supplied fields).
    """
    gtin = rec.get("gtin")
    if not gtin:
        return []
    prod = conn.execute(
        """
        SELECT id, brand_id, manufacturer_id, gpc_brick_code, country_of_origin,
               net_content_value
        FROM product
        WHERE gtin = %s AND status = 'active'
        """,
        (gtin,),
    ).fetchone()
    if prod is None:
        return []

    product_id, brand_id, manufacturer_id, gpc, country, net_value = prod
    sets: list[str] = []
    params: list[Any] = []
    filled: list[str] = []

    if brand_id is None and rec.get("brand"):
        new_brand_id = _ensure_brand(conn, rec["brand"])
        if new_brand_id is not None:
            sets.append("brand_id = %s")
            params.append(new_brand_id)
            filled.append("brand")

    if manufacturer_id is None and rec.get("manufacturer"):
        new_mfr_id = _ensure_manufacturer(conn, rec["manufacturer"])
        if new_mfr_id is not None:
            sets.append("manufacturer_id = %s")
            params.append(new_mfr_id)
            filled.append("manufacturer")

    if gpc is None and rec.get("gpc_brick_code"):
        sets.append("gpc_brick_code = %s")
        params.append(rec["gpc_brick_code"])
        filled.append("gpc_brick_code")

    if country is None and rec.get("country_of_origin"):
        sets.append("country_of_origin = %s")
        params.append(rec["country_of_origin"])
        filled.append("country_of_origin")

    if net_value is None:
        net = _net_content(rec)
        if net is not None:
            value, unit, canonical = net
            sets += [
                "net_content_value = %s",
                "net_content_unit = %s",
                "net_content_canonical = %s",
            ]
            params += [value, unit, canonical]
            filled.append("net_content")

    if not filled:
        return []

    params.append(product_id)
    conn.execute(f"UPDATE product SET {', '.join(sets)} WHERE id = %s", params)
    conn.execute(
        """
        INSERT INTO product_source (product_id, source_id, url, fields, fetched_at, raw)
        VALUES (%s, %s, %s, %s, now(), %s)
        """,
        (product_id, source_id, GS1_HOMEPAGE, filled, Jsonb(rec)),
    )
    update_quality(conn, product_id)
    return filled