#!/usr/bin/env python3 """Versioned local BLV enrichment; never overwrites YAZIO nutrient rows.""" from __future__ import annotations import argparse import hashlib import json import math import re import sqlite3 import unicodedata from datetime import datetime, timezone from pathlib import Path from openpyxl import load_workbook from dashboard_v5.nutrition_contract import NUTRIENT_CONTRACTS from dashboard_v5.sprint6f_b_schema import apply_pipeline_schema SOURCE_NAME = "Schweizer Nährwertdatenbank BLV" SOURCE_URL = "https://naehrwertdaten.ch/de/downloads/" SOURCE_HIERARCHY = ( "YAZIO existing values", "Schweizer Nährwertdatenbank BLV", "USDA FoodData Central fallback", "exact manufacturer or label data", "Open Food Facts identity/barcode/ingredients only", ) LICENSE_NOTE = "Source terms apply; private local use only; dashboard does not redistribute the source workbook." VERSION_RE = re.compile(r"\bV\s*([0-9]+(?:\.[0-9]+)*)\b") HEADER_MAP = { "Energie, Kilokalorien (kcal)": "energy.energy", "Fett, total (g)": "nutrient.fat", "Fettsäuren, gesättigt (g)": "nutrient.saturated", "Fettsäuren, einfach ungesättigt (g)": "nutrient.monounsaturated", "Fettsäuren, mehrfach ungesättigt (g)": "nutrient.polyunsaturated", "Kohlenhydrate, verfügbar (g)": "nutrient.carb", "Zucker (g)": "nutrient.sugar", "Nahrungsfasern (g)": "nutrient.dietaryfiber", "Protein (g)": "nutrient.protein", "Vitamin A-Aktivität, RAE (µg)": "vitamin.a", "Vitamin B1 (Thiamin) (mg)": "vitamin.b1", "Vitamin B2 (Riboflavin) (mg)": "vitamin.b2", "Vitamin B6 (Pyridoxin) (mg)": "vitamin.b6", "Vitamin B12 (Cobalamin) (µg)": "vitamin.b12", "Niacin (mg)": "vitamin.b3", "Folat (µg)": "vitamin.b11", "Pantothensäure (mg)": "vitamin.b5", "Vitamin C (Ascorbinsäure) (mg)": "vitamin.c", "Vitamin D (Calciferol) (µg)": "vitamin.d", "Vitamin E (α-Tocopherol) (mg)": "vitamin.e", "Kalium (K) (mg)": "mineral.potassium", "Natrium (Na) (mg)": "nutrient.sodium", "Calcium (Ca) (mg)": "mineral.calcium", "Magnesium (Mg) (mg)": "mineral.magnesium", "Eisen (Fe) (mg)": "mineral.iron", "Jod (I) (µg)": "mineral.iodine", "Zink (Zn) (mg)": "mineral.zinc", "Selen (Se) (µg)": "mineral.selenium", } def normalize(value: object) -> str: text = unicodedata.normalize("NFKD", str(value or "")).encode("ascii", "ignore").decode().casefold() return " ".join(re.sub(r"[^a-z0-9]+", " ", text).split()) def simple_yazio_name(value: str) -> bool: return bool(value) and not re.search(r"\d|[%/()+]", value) and 1 <= len(normalize(value).split()) <= 5 def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for block in iter(lambda: handle.read(1024 * 1024), b""): digest.update(block) return digest.hexdigest() def workbook_rows(path: Path, expected_version: str) -> tuple[str, list[dict[str, object]]]: workbook = load_workbook(path, read_only=True, data_only=True) sheet = workbook["Generische Lebensmittel"] title = str(sheet.cell(1, 1).value or "") match = VERSION_RE.search(title) workbook_version = match.group(1) if match else "" if workbook_version != expected_version: raise RuntimeError("source version mismatch: workbook and website version differ") headers = [str(value or "") for value in next(sheet.iter_rows(min_row=3, max_row=3, values_only=True))] indexes = {name: headers.index(name) for name in HEADER_MAP if name in headers} rows: list[dict[str, object]] = [] for values in sheet.iter_rows(min_row=4, values_only=True): entry_id, name = values[0], values[3] if not entry_id or not name: continue nutrients = {} for header, nutrient_key in HEADER_MAP.items(): if header not in indexes: continue raw = values[indexes[header]] if type(raw) not in {int, float} or not math.isfinite(float(raw)): continue nutrients[nutrient_key] = float(raw) density = values[6] rows.append({ "id": str(entry_id), "name": str(name), "synonyms": str(values[4] or ""), "density": float(density) if type(density) in {int, float} else None, "basis": str(values[7] or ""), "nutrients": nutrients, }) return workbook_version, rows def enrich(database: Path, source: Path, expected_site_version: str, report_path: Path) -> dict[str, object]: source_hash = sha256(source) version, source_rows = workbook_rows(source, expected_site_version) candidates: dict[str, list[tuple[dict[str, object], str]]] = {} for row in source_rows: candidates.setdefault(normalize(row["name"]), []).append((row, "exact_name")) for synonym in str(row["synonyms"]).split(","): if normalize(synonym): candidates.setdefault(normalize(synonym), []).append((row, "unique_synonym")) connection = sqlite3.connect(database) connection.row_factory = sqlite3.Row apply_pipeline_schema(connection) now = datetime.now(timezone.utc).isoformat(timespec="seconds") connection.execute( "INSERT OR IGNORE INTO food_enrichment_sources(source_name,source_version,source_hash,source_url,license_note,retrieved_at) VALUES(?,?,?,?,?,?)", (SOURCE_NAME, version, source_hash, SOURCE_URL, LICENSE_NOTE, now), ) names = connection.execute("SELECT name,count(*) AS uses FROM nutrition_items GROUP BY name ORDER BY name").fetchall() counters = {"yazio_names": len(names), "auto_applied_names": 0, "ambiguous_names": 0, "unmatched_names": 0, "nutrient_rows": 0} review: list[dict[str, object]] = [] for item in names: yazio_name = str(item["name"] or "") key = normalize(yazio_name) matches = candidates.get(key, []) if simple_yazio_name(yazio_name) else [] unique = {(str(row["id"]), method): row for row, method in matches} source_ids = {entry_id for entry_id, _method in unique} if len(source_ids) != 1: bucket = "ambiguous_names" if matches else "unmatched_names" counters[bucket] += 1 review.append({"yazio_name": yazio_name, "status": "needs_user_input" if matches else "not_matched", "reason": "multiple_source_entries" if matches else "no_unique_exact_match", "uses": int(item["uses"])}) continue selected_id = next(iter(source_ids)) options = [(row, method) for (entry_id, method), row in unique.items() if entry_id == selected_id] row, method = sorted(options, key=lambda pair: pair[1])[0] inserted = 0 for nutrient_key, value in row["nutrients"].items(): contract = NUTRIENT_CONTRACTS.get(str(nutrient_key)) if contract is None: continue cursor = connection.execute( """INSERT OR IGNORE INTO food_nutrient_enrichment (normalized_yazio_name,yazio_name,source_entry_id,source_entry_name,source_name,source_version, portion_basis,density,preparation_state,mapping_method,confidence,is_estimated,nutrient_key, value_per_100g,unit,review_status,created_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", (key, yazio_name, row["id"], row["name"], SOURCE_NAME, version, row["basis"], row["density"], "as_documented_in_source_name", method, "exact", 1, nutrient_key, value, contract.unit, "auto_applied", now), ) inserted += max(0, cursor.rowcount) if inserted: counters["auto_applied_names"] += 1 counters["nutrient_rows"] += inserted connection.commit() connection.close() report = { "status": "completed", "source": SOURCE_NAME, "source_version": version, "source_sha256": source_hash, "source_url": SOURCE_URL, "retrieved_at": now, "license_note": LICENSE_NOTE, "source_hierarchy": SOURCE_HIERARCHY, "counters": counters, "review_items": review, "rules": ["YAZIO values are never overwritten", "only unique exact generic matches auto-apply", "all added values remain labelled estimates"], } report_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) report_path.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n") report_path.chmod(0o600) return report def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--db", type=Path, required=True) parser.add_argument("--source", type=Path, required=True) parser.add_argument("--expected-site-version", required=True) parser.add_argument("--report", type=Path, required=True) args = parser.parse_args() report = enrich(args.db.resolve(), args.source.resolve(), args.expected_site_version, args.report.resolve()) print(json.dumps(report["counters"], sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())