mitai-jinkendo/backend/bls/import_service.py

"""Apply parsed BLS 4.0 data: upsert attributes and foods, never delete official rows."""
from __future__ import annotations

from typing import Any

from psycopg2.extras import execute_values

VALUE_PAGE = 2000
FOOD_PAGE = 500


def should_persist_value(val: dict[str, Any]) -> bool:
    if val.get("is_trace"):
        return True
    return val.get("value_num") is not None


def upsert_attributes(cur, attributes: list[dict[str, Any]]) -> dict[str, int]:
    inserted = updated = 0
    for a in attributes:
        cur.execute(
            """
            INSERT INTO food_attributes
                (attr_key, name_de, name_en, unit, category, data_type, origin, sort_order, updated_at)
            VALUES (%s, %s, %s, %s, %s, %s, 'official_bls', %s, NOW())
            ON CONFLICT (attr_key) DO UPDATE SET
                name_de = EXCLUDED.name_de,
                name_en = COALESCE(EXCLUDED.name_en, food_attributes.name_en),
                unit = COALESCE(EXCLUDED.unit, food_attributes.unit),
                category = COALESCE(EXCLUDED.category, food_attributes.category),
                sort_order = EXCLUDED.sort_order,
                updated_at = NOW()
            WHERE food_attributes.origin = 'official_bls'
            RETURNING (xmax = 0) AS inserted
            """,
            (
                a["attr_key"], a["name_de"], a.get("name_en"), a.get("unit"),
                a.get("category"), a.get("data_type") or "num_per_100g",
                a.get("sort_order") or 0,
            ),
        )
        row = cur.fetchone()
        if row and row.get("inserted"):
            inserted += 1
        else:
            updated += 1
    return {"inserted": inserted, "updated": updated, "total": len(attributes)}


def upsert_foods(cur, foods: list[dict[str, Any]], bls_version: str = "4.0") -> dict[str, int]:
    cur.execute("SELECT attr_key, id FROM food_attributes")
    attr_ids = {r["attr_key"]: r["id"] for r in cur.fetchall()}
    codes = [f["bls_code"] for f in foods if f.get("bls_code")]
    existing: set[str] = set()
    if codes:
        cur.execute("SELECT bls_code FROM food_catalog WHERE bls_code = ANY(%s)", (codes,))
        existing = {r["bls_code"] for r in cur.fetchall()}
    inserted = sum(1 for c in codes if c not in existing)
    updated = len(codes) - inserted
    food_rows = [
        (
            f["bls_code"], f["name_de"], f.get("name_en"), f.get("food_group"),
            "official_bls", bls_version, "bls_4.0",
        )
        for f in foods if f.get("bls_code")
    ]
    if food_rows:
        execute_values(
            cur,
            """
            INSERT INTO food_catalog
                (bls_code, name_de, name_en, food_group, catalog_kind, bls_version, source)
            VALUES %s
            ON CONFLICT (bls_code) DO UPDATE SET
                name_de = EXCLUDED.name_de,
                name_en = EXCLUDED.name_en,
                food_group = EXCLUDED.food_group,
                bls_version = EXCLUDED.bls_version,
                catalog_kind = 'official_bls',
                source = 'bls_4.0',
                is_active = true,
                updated_at = NOW()
            WHERE food_catalog.catalog_kind = 'official_bls'
            """,
            food_rows,
            page_size=FOOD_PAGE,
        )

    food_ids: dict[str, Any] = {}
    if codes:
        cur.execute("SELECT bls_code, id FROM food_catalog WHERE bls_code = ANY(%s)", (codes,))
        food_ids = {r["bls_code"]: r["id"] for r in cur.fetchall()}

    value_rows = []
    for f in foods:
        food_id = food_ids.get(f.get("bls_code"))
        if not food_id:
            continue
        for val in f.get("values") or []:
            if not should_persist_value(val):
                continue
            aid = attr_ids.get(val["attr_key"])
            if not aid:
                continue
            value_rows.append((
                food_id,
                aid,
                None if val.get("is_trace") else val.get("value_num"),
                bool(val.get("is_trace")),
                val.get("origin_code"),
                val.get("reference_text"),
            ))

    if value_rows:
        execute_values(
            cur,
            """
            INSERT INTO food_attribute_values
                (food_id, attribute_id, value_num, is_trace, origin_code, reference_text, updated_at)
            VALUES %s
            ON CONFLICT (food_id, attribute_id) DO UPDATE SET
                value_num = EXCLUDED.value_num,
                is_trace = EXCLUDED.is_trace,
                origin_code = EXCLUDED.origin_code,
                reference_text = EXCLUDED.reference_text,
                updated_at = NOW()
            """,
            value_rows,
            template="(%s, %s, %s, %s, %s, %s, NOW())",
            page_size=VALUE_PAGE,
        )

    from data_layer.food_suggest import invalidate_suggest_index
    invalidate_suggest_index()
    return {
        "inserted": inserted,
        "updated": updated,
        "foods_inserted": inserted,
        "foods_updated": updated,
        "values_written": len(value_rows),
        "foods_total": len(foods),
    }