goods/ingestion/opengoods/etl/load.py

"""Load transformed product records into the OpenGoods PostgreSQL database.

Only the ingestion side writes to the database. Every load records OFF as the
source with field-level provenance in `product_source`.
"""

from __future__ import annotations

import json
import logging
import os
from typing import Any

import psycopg
from psycopg.types.json import Jsonb

from opengoods.adapters.openfoodfacts import OFF_LICENSE, SOURCE_NAME
from opengoods.etl.quality import update_quality

OFF_HOMEPAGE = "https://world.openfoodfacts.org"

logger = logging.getLogger(__name__)


def default_dsn() -> str:
    return os.environ.get(
        "OPENGOODS_DATABASE_URL",
        "postgres://opengoods:opengoods@localhost:5432/opengoods?sslmode=disable",
    )


def _normalize_brand(name: str) -> str:
    return " ".join(name.lower().split())


def ensure_source_named(
    conn: psycopg.Connection,
    name: str,
    homepage: str,
    license: str,
    trust_weight: float,
) -> str:
    """Upsert a source row by name and return its id."""
    row = conn.execute(
        """
        INSERT INTO source (name, homepage, license, trust_weight)
        VALUES (%s, %s, %s, %s)
        ON CONFLICT (name) DO UPDATE SET homepage = EXCLUDED.homepage
        RETURNING id
        """,
        (name, homepage, license, trust_weight),
    ).fetchone()
    return row[0]


def ensure_source(conn: psycopg.Connection) -> str:
    """Upsert the Open Food Facts source row and return its id."""
    return ensure_source_named(conn, SOURCE_NAME, OFF_HOMEPAGE, OFF_LICENSE, 0.7)


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


def _category_id(conn: psycopg.Connection, path: str | None) -> tuple[str | None, str | None]:
    if not path:
        return None, None
    row = conn.execute(
        "SELECT id, gpc_brick_code FROM category WHERE path = %s::ltree", (path,)
    ).fetchone()
    return (row[0], row[1]) if row else (None, None)


def load_record(conn: psycopg.Connection, rec: dict[str, Any], source_id: str, raw: dict) -> str:
    """Upsert one transformed record; return the product id."""
    brand_id = _ensure_brand(conn, rec.get("brand"))
    category_id, gpc_brick = _category_id(conn, rec.get("category_path"))

    fields = ["name", "brand", "net_content", "category", "country_of_origin"]

    if rec.get("gtin"):
        prod = conn.execute(
            """
            INSERT INTO product (gtin, name, brand_id, category_id, gpc_brick_code,
                net_content_value, net_content_unit, net_content_canonical,
                country_of_origin, attributes)
            VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
            ON CONFLICT (gtin) WHERE gtin IS NOT NULL DO UPDATE SET
                name = EXCLUDED.name,
                brand_id = COALESCE(EXCLUDED.brand_id, product.brand_id),
                category_id = COALESCE(EXCLUDED.category_id, product.category_id),
                gpc_brick_code = COALESCE(EXCLUDED.gpc_brick_code, product.gpc_brick_code),
                net_content_value = EXCLUDED.net_content_value,
                net_content_unit = EXCLUDED.net_content_unit,
                net_content_canonical = EXCLUDED.net_content_canonical,
                country_of_origin = EXCLUDED.country_of_origin
            RETURNING id
            """,
            (
                rec["gtin"],
                rec["name"],
                brand_id,
                category_id,
                gpc_brick,
                rec.get("net_content_value"),
                rec.get("net_content_unit"),
                rec.get("net_content_canonical"),
                rec.get("country_of_origin"),
                Jsonb({}),
            ),
        ).fetchone()
    else:
        prod = conn.execute(
            """
            INSERT INTO product (name, brand_id, category_id, gpc_brick_code,
                net_content_value, net_content_unit, net_content_canonical,
                country_of_origin, attributes)
            VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)
            RETURNING id
            """,
            (
                rec["name"],
                brand_id,
                category_id,
                gpc_brick,
                rec.get("net_content_value"),
                rec.get("net_content_unit"),
                rec.get("net_content_canonical"),
                rec.get("country_of_origin"),
                Jsonb({}),
            ),
        ).fetchone()
    product_id = prod[0]

    food = rec.get("food") or {}
    conn.execute(
        """
        INSERT INTO food_detail (product_id, ingredients_text, allergens, additives,
            nutriments, nutrition_basis, serving_size, nutri_score)
        VALUES (%s,%s,%s,%s,%s,%s,%s,%s)
        ON CONFLICT (product_id) DO UPDATE SET
            ingredients_text = EXCLUDED.ingredients_text,
            allergens = EXCLUDED.allergens,
            additives = EXCLUDED.additives,
            nutriments = EXCLUDED.nutriments,
            nutrition_basis = EXCLUDED.nutrition_basis,
            serving_size = EXCLUDED.serving_size,
            nutri_score = EXCLUDED.nutri_score
        """,
        (
            product_id,
            food.get("ingredients_text"),
            food.get("allergens") or [],
            food.get("additives") or [],
            Jsonb(food.get("nutriments") or {}),
            food.get("nutrition_basis"),
            food.get("serving_size"),
            food.get("nutri_score"),
        ),
    )

    if rec.get("image_url"):
        conn.execute(
            """
            INSERT INTO product_image (product_id, url, kind, license, source_id)
            VALUES (%s,%s,'front',%s,%s)
            """,
            (product_id, rec["image_url"], "CC-BY-SA", source_id),
        )
        fields.append("image")

    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,
            f"{OFF_HOMEPAGE}/product/{rec.get('gtin') or ''}",
            fields,
            Jsonb(_jsonable(raw)),
        ),
    )

    # Recompute the data-quality score now that all facts + provenance exist.
    update_quality(conn, product_id)
    return product_id


def load_record_safe(
    conn: psycopg.Connection, rec: dict[str, Any], source_id: str, raw: dict
) -> bool:
    """Load one record inside a savepoint.

    On success the record's writes stay in the surrounding transaction. On any
    error, only this record's writes are rolled back (to the savepoint) and the
    batch continues, so a single malformed source record cannot abort a large
    import. Returns True if loaded, False if skipped due to an error.
    """
    try:
        with conn.transaction():
            load_record(conn, rec, source_id, raw)
        return True
    except Exception as exc:  # noqa: BLE001 - per-record isolation is intentional
        logger.warning("skipping record gtin=%s: %s", rec.get("gtin"), exc)
        return False


def _jsonable(raw: dict) -> dict:
    """Drop values that are not JSON-serializable from a raw record."""
    try:
        json.dumps(raw)
        return raw
    except (TypeError, ValueError):
        return {k: v for k, v in raw.items() if _is_jsonable(v)}


def _is_jsonable(v: object) -> bool:
    try:
        json.dumps(v)
        return True
    except (TypeError, ValueError):
        return False