CookIT/tools/excel-import/cookit_import/recipe.py

"""Turns one recipe sheet of the workbook template into a Recipe Bundle recipe."""

from __future__ import annotations

import hashlib
import re
import uuid
from dataclasses import dataclass, field

from openpyxl.utils.cell import coordinate_from_string, column_index_from_string

from . import parse
from .workbook import Sheet

SKIPPED_SHEETS = {"Leeres Rezept", "Spinat-Feta-Pizza"}  # the template and its example recipe

# Stable IDs: re-running the converter yields the same Recipe IDs, so a re-import skips them.
ID_NAMESPACE = uuid.UUID("6f1d3c3e-2b7a-4c55-9a51-0c6b8f2e9d10")

# Cells the template itself fills or that this module reads explicitly.
TEMPLATE_LABELS = {"A1", "C1", "F1", "G1", "C2", "D2", "E2", "F2", "G2", "B5", "G6", "B8", "G10", "B11"}
B12_BOILERPLATE = "Löschen Sie dieses Rezeptarbeitsblatt nicht."
PLACEHOLDER_PREFIXES = ("Geben Sie", "Kopieren Sie dieses Arbeitsblatt", "Stichwort 1")
NO_COURSE = {"Gang"}
NUTRITION_LABELS = {"fett", "fette", "kohlenhydrate", "eiweiß", "ballaststoffe", "kalorien/ 100g", "kalorien gesamt"}
FIRST_INGREDIENT_ROW = 3


class Reason:
    """Review Flag reasons, shown to the cook in the app."""

    STRAY_TEXT = "Zusätzlicher Text aus der Tabelle wurde in die Notizen übernommen."
    NO_SERVINGS = "Portionen fehlen."
    UNKNOWN_UNIT = "Einheit nicht erkannt und in den Zutatennamen übernommen: {unit}"
    UNKNOWN_AMOUNT = "Menge nicht erkannt und als Hinweis übernommen: {amount}"
    MERGED_INSTRUCTIONS = "Anleitung war auf mehrere Zellen verteilt und wurde zusammengeführt."
    LONG_STEP = "Sehr langer Schritt – eventuell aufteilen."
    NAMELESS_LINES = "Zutatenzeilen ohne Zutat wurden verworfen: {values}"
    DUPLICATE_LIST = "Doppelte Zutatenliste wurde verworfen."
    ALTERNATIVE = "„ODER“-Alternative wurde als eigene Zutatengruppe übernommen."
    UNREADABLE_TIME = "Zeitangabe nicht erkannt und in die Notizen übernommen."



def is_recipe_sheet(sheet: Sheet) -> bool:
    return (
        sheet.title not in SKIPPED_SHEETS
        and parse.clean(sheet.cells.get("C2")) == "Menge"
        and parse.clean(sheet.cells.get("E2")) == "Zutat"
    )


TEMPLATE_SHEET = "Leeres Rezept"


def template_photo_hashes(sheets: list[Sheet]) -> set[str]:
    """The template's stock picture, which was copied onto every sheet made from it."""
    return {hashlib.md5(p).hexdigest() for s in sheets if s.title == TEMPLATE_SHEET for p in s.photos}


def real_photos(sheet: Sheet, placeholders: set[str]) -> list[bytes]:
    return [p for p in sheet.photos if hashlib.md5(p).hexdigest() not in placeholders]


def recipe_id(sheet_title: str) -> str:
    return str(uuid.uuid5(ID_NAMESPACE, sheet_title))


def _rc(coord: str) -> tuple[int, int]:
    col, row = coordinate_from_string(coord)
    return row, column_index_from_string(col)


@dataclass
class _Builder:
    sheet: Sheet
    notes: list[str] = field(default_factory=list)
    sources: list[str] = field(default_factory=list)
    reasons: list[str] = field(default_factory=list)
    used: set[str] = field(default_factory=set)

    def get(self, coord: str):
        self.used.add(coord)
        return self.sheet.cells.get(coord)

    def text(self, coord: str) -> str | None:
        value = parse.clean(self.get(coord))
        if value and value.startswith(PLACEHOLDER_PREFIXES):
            return None
        return value

    def flag(self, reason: str) -> None:
        if reason not in self.reasons:
            self.reasons.append(reason)

    def add_note_text(self, text: str, stray: bool) -> None:
        urls, rest = parse.extract_urls(text)
        for url in urls:
            if url not in self.sources:
                self.sources.append(url)
        if rest:
            self.notes.append(rest)
            if stray:
                self.flag(Reason.STRAY_TEXT)


def build_recipe(sheet: Sheet) -> dict:
    b = _Builder(sheet)
    name = _name(b.text("C1"), sheet.title)
    course = b.text("F1")
    if course in NO_COURSE:
        course = None

    comment_b12 = b.text("B12")
    if comment_b12 and comment_b12 != B12_BOILERPLATE:
        b.add_note_text(comment_b12, stray=False)

    tags = _tags(b.text("B9"))
    prep = _minutes(b, "G3", "Vorbereitungszeit")
    cook = _minutes(b, "G7", "Kochzeit")
    servings = _servings(b)
    calories = _calories(b)
    groups = _ingredient_groups(b)
    sections = _steps(b)
    _stray_cells(b)
    _stray_comments(b)

    return {
        "id": recipe_id(sheet.title),
        "name": name,
        "course": course,
        "tags": tags,
        "servings": servings,
        "prepMinutes": prep,
        "cookMinutes": cook,
        "calories": calories,
        "ingredientGroups": groups,
        "stepSections": sections,
        "photos": [],
        "sources": b.sources,
        "notes": "\n\n".join(b.notes) or None,
        "favourite": False,
        "reviewReasons": b.reasons,
    }


def _name(title_cell: str | None, sheet_title: str) -> str:
    """Sheet tabs are cut at 31 characters, title cells are sometimes abbreviated: keep the longer one."""
    if not title_cell or title_cell in SKIPPED_SHEETS:
        return sheet_title.strip()
    return max(title_cell, sheet_title.strip(), key=len)


def _tags(text: str | None) -> list[str]:
    tags, seen = [], set()
    for tag in (text or "").replace(";", ",").split(","):
        tag = tag.strip()
        if tag and tag.casefold() not in seen:
            seen.add(tag.casefold())
            tags.append(tag)
    return tags


def _minutes(b: _Builder, coord: str, label: str) -> int | None:
    value = parse.parse_minutes(b.get(coord))
    if isinstance(value, str):
        if not value.startswith(PLACEHOLDER_PREFIXES):
            b.notes.append(f"{label}: {value}")
            b.flag(Reason.UNREADABLE_TIME)
        return None
    return value


def _servings(b: _Builder):
    raw = b.get("G11")
    if parse.clean(raw) and str(raw).startswith(PLACEHOLDER_PREFIXES):
        raw = None
    servings = parse.parse_servings(raw)
    if servings.remark:
        b.add_note_text(f"Ergebnis: {servings.remark}", stray=True)
    if servings.value is None:
        b.flag(Reason.NO_SERVINGS)
    return servings.value


def _calories(b: _Builder) -> dict | None:
    raw = b.get("B6")
    comment = b.sheet.comments.get("B6")
    b.used.add("comment:B6")
    if not parse.is_number(raw):
        text = parse.clean(raw)
        if text and not text.startswith(PLACEHOLDER_PREFIXES):
            b.add_note_text(f"Kalorien: {text}", stray=True)
        return None
    kcal = float(str(raw).replace(",", "."))
    if kcal <= 0:
        return None
    if comment and "pro portion" in comment.lower():
        basis = "perServing"
    else:
        # Guessed, but not flagged: the threshold matched every value in the workbook on review.
        basis = parse.calories_basis(kcal)
        if comment:
            b.add_note_text(f"Kalorien: {comment}", stray=True)
    return {"kcal": int(round(kcal)), "basis": basis}


# --- Ingredients -------------------------------------------------------------


def _row_cells(b: _Builder, row: int):
    return b.sheet.cells.get(f"C{row}"), b.sheet.cells.get(f"D{row}"), b.sheet.cells.get(f"E{row}")


def _ingredient_rows(b: _Builder) -> list[int]:
    rows = {
        _rc(coord)[0]
        for coord in b.sheet.cells
        if coord[0] in "CDE" and _rc(coord)[0] >= FIRST_INGREDIENT_ROW and _rc(coord)[1] in (3, 4, 5)
    }
    return sorted(rows)


def _blocks(rows: list[int]) -> list[list[int]]:
    """Runs of consecutive rows; a blank row in the sheet separates Ingredient Groups."""
    blocks: list[list[int]] = []
    for row in rows:
        if blocks and row == blocks[-1][-1] + 1:
            blocks[-1].append(row)
        else:
            blocks.append([row])
    return blocks


def _group_marker(comment: str | None) -> str | None:
    """A cell comment such as "Ab hier: Zutaten für die Sauce" names the group starting at that row."""
    if not comment or "zutaten" not in comment.lower():
        return None
    m = re.search(r"zutaten\s+für\s+(?:die|den|das)?\s*(.+)", comment, re.I | re.S)
    name = (m.group(1) if m else comment).strip().rstrip(":").strip()
    return name[:1].upper() + name[1:]


def _ingredient_groups(b: _Builder) -> list[dict]:
    groups: list[dict] = []
    seen_names: set[str] = set()
    pending_name: str | None = None
    dropped: list[str] = []

    for block in _blocks(_ingredient_rows(b)):
        for row in block:
            b.used.update({f"C{row}", f"D{row}", f"E{row}"})

        block_rows = [(row, *_row_cells(b, row)) for row in block]

        # Nutrition tables pasted into the ingredient columns belong in the Notes.
        if any((parse.clean(d) or "").lower() in NUTRITION_LABELS for _, _, d, _ in block_rows):
            lines = [" ".join(str(v).strip() for v in (c, d, e) if parse.clean(v)) for _, c, d, e in block_rows]
            b.add_note_text("\n".join(lines), stray=True)
            continue

        named = [parse.clean(e).casefold() for _, _, _, e in block_rows if parse.clean(e)]
        if len(named) >= 3 and sum(n in seen_names for n in named) / len(named) >= 0.8:
            b.flag(Reason.DUPLICATE_LIST)
            continue

        current: dict | None = None
        for row, c, d, e in block_rows:
            if any((parse.clean(v) or "").upper() == "ODER" for v in (c, e)) and not parse.clean(d):
                pending_name = "Alternative"
                b.flag(Reason.ALTERNATIVE)
                current = None
                continue
            if not parse.clean(e):
                if all(v is None or parse.is_number(v) for v in (c, d)):
                    continue  # leftover numbers from rescaling a recipe
                if parse.clean(c) or parse.clean(d):
                    dropped.append(" ".join(str(v).strip() for v in (c, d) if parse.clean(v)))
                continue

            marker = None
            for col in "CDE":
                marker = marker or _group_marker(b.sheet.comments.get(f"{col}{row}"))
            if marker:
                pending_name = marker
                current = None

            if current is None:
                current = {"name": pending_name, "lines": []}
                pending_name = None
                groups.append(current)
            current["lines"].append(_line(b, row, c, d, e))
            seen_names.add(parse.clean(e).casefold())

    if dropped:
        b.flag(Reason.NAMELESS_LINES.format(values=", ".join(dropped)))
    return [g for g in groups if g["lines"]]


def _line(b: _Builder, row: int, c, d, e) -> dict:
    name = parse.clean(e)
    remarks: list[str] = []
    amount_text = parse.clean(c)
    if amount_text and amount_text.startswith("(") and name.endswith(")"):
        # "(2 | Teelöffel | Oregano)": the whole line is in parentheses, i.e. optional.
        c, name = amount_text[1:], name[:-1].strip()
        remarks.append("optional")
    amount = parse.parse_amount(c)
    unit = parse.parse_unit(d)
    unit_text = parse.clean(d)

    if amount is None:
        b.flag(Reason.UNKNOWN_AMOUNT.format(amount=parse.clean(c)))
        remarks.append(parse.clean(c))
        amount = (None, None)

    if unit_text and unit is None:
        combined = parse.split_amount_unit(unit_text)
        if combined and amount == (None, None):
            amount, unit = (combined[0], None), combined[1]
        else:
            b.flag(Reason.UNKNOWN_UNIT.format(unit=unit_text))
            name = f"{unit_text} {name}"

    for col in "CDE":
        comment = b.sheet.comments.get(f"{col}{row}")
        b.used.add(f"comment:{col}{row}")
        if comment and not _group_marker(comment):
            remarks.append(comment)

    return {
        "amount": amount[0],
        "amountMax": amount[1],
        "unit": unit,
        "name": name,
        "remark": "; ".join(remarks) or None,
    }


# --- Steps -------------------------------------------------------------------


def _steps(b: _Builder) -> list[dict]:
    instruction_cells = sorted((c for c in b.sheet.cells if c.startswith("F") and c not in ("F1", "F2")), key=_rc)
    texts = []
    for coord in instruction_cells:
        text = b.text(coord)
        if text:
            texts.append(text)
    b.used.update(instruction_cells)
    if len(texts) > 1:
        b.flag(Reason.MERGED_INSTRUCTIONS)

    sections = parse.split_steps("\n\n".join(texts))
    if any(len(step) > parse.LONG_STEP_CHARS for s in sections for step in s.steps):
        b.flag(Reason.LONG_STEP)
    return [{"title": s.title, "steps": s.steps} for s in sections]


# --- Everything else -----------------------------------------------------------


def _stray_cells(b: _Builder) -> None:
    rows: dict[int, list[tuple[int, str]]] = {}
    for coord, value in b.sheet.cells.items():
        if coord in b.used or coord in TEMPLATE_LABELS:
            continue
        text = parse.clean(value)
        if not text or text == B12_BOILERPLATE or text.startswith(PLACEHOLDER_PREFIXES):
            continue
        row, col = _rc(coord)
        rows.setdefault(row, []).append((col, text))
    for row in sorted(rows):
        line = " ".join(text for _, text in sorted(rows[row]))
        b.add_note_text(line, stray=True)


def _stray_comments(b: _Builder) -> None:
    for coord, comment in b.sheet.comments.items():
        if f"comment:{coord}" in b.used or _group_marker(comment):
            continue
        b.add_note_text(comment, stray=True)