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