"""Parse FDDB lists_*.csv (name;…;produkte) into recipe + ingredients.""" from __future__ import annotations import csv import io import re from typing import Any from data_layer.food_mapping import normalize_food_name, parse_quantity_g ING_RE = re.compile( r"(?P\d+(?:[.,]\d+)?)\s*(?Pg|kg|ml|l)\b\s*(?P.+?)" r"(?=,\s*\d+(?:[.,]\d+)?\s*(?:g|kg|ml|l)\b|$)", re.IGNORECASE | re.DOTALL, ) def parse_fddb_produkte(text: str) -> list[dict[str, Any]]: raw = (text or "").strip().strip('"') if not raw: return [] out: list[dict[str, Any]] = [] for i, m in enumerate(ING_RE.finditer(raw)): qty = float(m.group("qty").replace(",", ".")) unit = m.group("unit").lower() name = re.sub(r"\s+", " ", m.group("name")).strip(" ,;") if not name: continue grams = qty if unit == "kg": grams = qty * 1000.0 elif unit == "l": grams = qty * 1000.0 elif unit == "ml": grams = qty out.append({ "source_name_raw": name, "source_name_normalized": normalize_food_name(name), "quantity_raw": f"{m.group('qty').replace(',', '.')} {unit}", "quantity_g": round(grams, 3), "sort_order": i, }) return out def parse_fddb_lists_csv(text: str) -> list[dict[str, Any]]: if text.startswith("\ufeff"): text = text[1:] reader = csv.DictReader(io.StringIO(text), delimiter=";") recipes = [] for row in reader: name = (row.get("name") or "").strip().strip('"') if not name: continue try: portions = float(str(row.get("anzahl_portionen") or "1").replace(",", ".")) except ValueError: portions = 1.0 if portions <= 0: portions = 1.0 ingredients = parse_fddb_produkte(row.get("produkte") or "") if not ingredients: leftover = (row.get("produkte") or "").strip().strip('"') if leftover: ingredients = [{ "source_name_raw": leftover, "source_name_normalized": normalize_food_name(leftover), "quantity_raw": None, "quantity_g": parse_quantity_g(leftover), "sort_order": 0, }] recipes.append({ "name_raw": name, "name_normalized": normalize_food_name(name), "portions": portions, "description": (row.get("beschreibung") or "").strip() or None, "ingredients": ingredients, }) return recipes