mitai-jinkendo/backend/bls/parser.py

"""Parse official BLS 4.0 XLSX files without a hardcoded nutrient code list."""
from __future__ import annotations

import re
from io import BytesIO
from typing import Any

from openpyxl import load_workbook

# Header like: "ENERCJ Energie (Kilojoule) [kJ/100g]"
ATTR_HEADER_RE = re.compile(
    r"^([A-Z][A-Z0-9:]{1,20})\s+(.+?)(?:\s*\[([^\]]+)\])?\s*$"
)


def _cell(v: Any) -> str:
    if v is None:
        return ""
    return str(v).strip()


def parse_components_xlsx(data: bytes) -> list[dict[str, Any]]:
    wb = load_workbook(filename=BytesIO(data), read_only=True, data_only=True)
    ws = wb.active
    rows = ws.iter_rows(values_only=True)
    header = [_cell(c) for c in next(rows)]
    idx = {h.lower(): i for i, h in enumerate(header) if h}

    def col(*names: str) -> int | None:
        for n in names:
            if n.lower() in idx:
                return idx[n.lower()]
        for key, i in idx.items():
            for n in names:
                if n.lower() in key:
                    return i
        return None

    i_code = col("code", "schlüssel", "schluessel", "attr_key", "komponente")
    i_de = col("name_de", "deutsch", "bezeichnung_de", "name de")
    i_en = col("name_en", "english", "bezeichnung_en", "name en")
    i_unit = col("unit", "einheit")
    i_cat = col("category", "kategorie", "gruppe")
    if i_code is None:
        i_code = 0
    if i_de is None:
        i_de = 1 if len(header) > 1 else 0

    out = []
    sort_order = 0
    for raw in rows:
        if not raw:
            continue
        code = _cell(raw[i_code] if i_code < len(raw) else "")
        if not code or code.lower() in ("code", "schlüssel", "schluessel"):
            continue
        name_de = _cell(raw[i_de] if i_de is not None and i_de < len(raw) else "") or code
        name_en = _cell(raw[i_en] if i_en is not None and i_en < len(raw) else "") or None
        unit = _cell(raw[i_unit] if i_unit is not None and i_unit < len(raw) else "") or None
        category = _cell(raw[i_cat] if i_cat is not None and i_cat < len(raw) else "") or None
        sort_order += 1
        out.append({
            "attr_key": code,
            "name_de": name_de,
            "name_en": name_en,
            "unit": unit,
            "category": category,
            "data_type": "num_per_100g",
            "origin": "official_bls",
            "sort_order": sort_order,
        })
    wb.close()
    return out


def _parse_value_header(title: str) -> tuple[str | None, str, str | None]:
    t = title.strip()
    m = ATTR_HEADER_RE.match(t)
    if m:
        return m.group(1), m.group(2).strip(), m.group(3)
    # Fallback: first token
    parts = t.split()
    if parts and re.match(r"^[A-Z][A-Z0-9:]{1,20}$", parts[0]):
        return parts[0], " ".join(parts[1:]) or parts[0], None
    return None, t, None


def parse_foods_xlsx(data: bytes) -> dict[str, Any]:
    wb = load_workbook(filename=BytesIO(data), read_only=True, data_only=True)
    ws = wb.active
    rows = ws.iter_rows(values_only=True)
    header = [_cell(c) for c in next(rows)]
    if len(header) < 3:
        wb.close()
        raise ValueError("BLS-Datendatei: erwartet mindestens BLS-Code, Name DE, Name EN")

    triples: list[dict[str, Any]] = []
    i = 3
    while i < len(header):
        code, name_de, unit = _parse_value_header(header[i])
        origin_i = i + 1 if i + 1 < len(header) else None
        ref_i = i + 2 if i + 2 < len(header) else None
        triples.append({
            "attr_key": code or f"COL{i}",
            "name_de": name_de,
            "unit": unit,
            "value_col": i,
            "origin_col": origin_i,
            "ref_col": ref_i,
        })
        i += 3 if (origin_i is not None and ref_i is not None) else 1

    foods = []
    for raw in rows:
        if not raw:
            continue
        code = _cell(raw[0] if len(raw) else "")
        if not code:
            continue
        name_de = _cell(raw[1] if len(raw) > 1 else "") or code
        name_en = _cell(raw[2] if len(raw) > 2 else "") or None
        values = []
        for t in triples:
            vc = t["value_col"]
            raw_v = raw[vc] if vc < len(raw) else None
            is_trace = False
            num = None
            if raw_v is None or raw_v == "" or raw_v == "-":
                num = None
            elif str(raw_v).strip().upper() in ("TR", "TRACE", "SPUREN"):
                is_trace = True
            else:
                try:
                    num = float(str(raw_v).replace(",", "."))
                except (TypeError, ValueError):
                    num = None
            origin = ""
            if t["origin_col"] is not None and t["origin_col"] < len(raw):
                origin = _cell(raw[t["origin_col"]])
            ref = ""
            if t["ref_col"] is not None and t["ref_col"] < len(raw):
                ref = _cell(raw[t["ref_col"]])
            values.append({
                "attr_key": t["attr_key"],
                "value_num": num,
                "is_trace": is_trace,
                "origin_code": origin or None,
                "reference_text": ref or None,
            })
        foods.append({
            "bls_code": code,
            "name_de": name_de,
            "name_en": name_en,
            "food_group": code[0] if code else None,
            "values": values,
        })
    wb.close()
    return {"attribute_headers": triples, "foods": foods}