mitai-jinkendo/backend/data_layer/food_suggest.py

"""In-memory catalog suggestions: Haferflocken → Hafer Flocken, without opening a dialog."""
from __future__ import annotations

import re
import time
from typing import Any

from data_layer.food_mapping import normalize_food_name

_INDEX_TTL_SEC = 300
_index_cache: dict[str, tuple[float, dict[str, Any]]] = {}
MAX_CANDIDATES = 60
MAX_BUCKET = 40

SPLIT_RE = re.compile(r"[^a-z0-9äöüß]+")
COLLAPSE_RE = re.compile(r"[^a-z0-9äöüß]")
MIN_SCORE = 45


def collapse_key(raw: str | None) -> str:
    return COLLAPSE_RE.sub("", normalize_food_name(raw))


def name_tokens(raw: str | None) -> list[str]:
    return [t for t in SPLIT_RE.split(normalize_food_name(raw)) if len(t) >= 2]


def score_name_match(query: str, name_de: str, name_en: str | None = None) -> int:
    qn = normalize_food_name((query or "").split(",")[0])
    nn = normalize_food_name(name_de)
    if not qn or not nn:
        return 0
    qc, nc = collapse_key(qn), collapse_key(nn)
    if qn == nn:
        return 100
    if qc and qc == nc:
        return 95
    if qc and nc.startswith(qc) and len(qc) >= 4:
        return 82
    if nc and qc.startswith(nc) and len(nc) >= 4:
        return 78
    qt, nt = set(name_tokens(qn)), set(name_tokens(nn))
    if qt and qt <= nt:
        return 72
    if nt and nt <= qt:
        return 68
    if qt and nt:
        overlap = len(qt & nt) / len(qt | nt)
        if overlap >= 0.5:
            return 50 + int(overlap * 20)
    if qc and nc and len(qc) >= 4 and (qc in nc or nc in qc):
        return 55 if abs(len(qc) - len(nc)) <= 8 else 46
    en = normalize_food_name(name_en or "")
    if en and (qn == en or collapse_key(en) == qc):
        return 88
    return 0


def _public(food: dict[str, Any], score: int) -> dict[str, Any]:
    return {
        "id": str(food["id"]),
        "bls_code": food.get("bls_code"),
        "name_de": food.get("name_de"),
        "name_en": food.get("name_en"),
        "catalog_kind": food.get("catalog_kind"),
        "food_group": food.get("food_group"),
        "score": score,
    }


def invalidate_suggest_index(profile_id: str | None = None) -> None:
    if profile_id is None:
        _index_cache.clear()
        return
    _index_cache.pop(str(profile_id), None)
    _index_cache.pop("global", None)


def get_suggest_index(cur, profile_id: str | None) -> dict[str, Any]:
    key = str(profile_id or "global")
    hit = _index_cache.get(key)
    if hit and (time.monotonic() - hit[0]) < _INDEX_TTL_SEC:
        return hit[1]
    index = load_suggest_index(cur, profile_id)
    _index_cache[key] = (time.monotonic(), index)
    return index


def load_suggest_index(cur, profile_id: str | None) -> dict[str, Any]:
    cur.execute(
        """
        SELECT id, bls_code, name_de, name_en, catalog_kind, food_group
        FROM food_catalog
        WHERE is_active = true
          AND (owner_profile_id IS NULL OR owner_profile_id = %s)
        """,
        (profile_id,),
    )
    foods = [dict(r) for r in cur.fetchall()]
    by_collapse: dict[str, list] = {}
    by_token: dict[str, list] = {}
    by_prefix: dict[str, list] = {}
    by_suffix: dict[str, list] = {}
    for food in foods:
        food["_c"] = collapse_key(food.get("name_de"))
        food["_t"] = name_tokens(food.get("name_de"))
        if food["_c"]:
            by_collapse.setdefault(food["_c"], []).append(food)
            by_prefix.setdefault(food["_c"][:4], []).append(food)
            if len(food["_c"]) >= 4:
                by_suffix.setdefault(food["_c"][-4:], []).append(food)
        for tok in food["_t"]:
            by_token.setdefault(tok, []).append(food)
    return {
        "foods": foods,
        "by_collapse": by_collapse,
        "by_token": by_token,
        "by_prefix": by_prefix,
        "by_suffix": by_suffix,
    }


def _candidate_foods(index: dict[str, Any], query: str) -> list[dict[str, Any]]:
    qc = collapse_key(query)
    seen: set[str] = set()
    out: list[dict[str, Any]] = []

    def add(food: dict[str, Any]) -> None:
        fid = str(food["id"])
        if fid in seen:
            return
        seen.add(fid)
        out.append(food)

    if qc:
        for food in index["by_collapse"].get(qc, []):
            add(food)
        if len(qc) >= 4:
            prefix_hits = index["by_prefix"].get(qc[:4], [])
            if len(prefix_hits) > MAX_BUCKET:
                prefix_hits = [
                    food for food in prefix_hits
                    if (food.get("_c") or "").startswith(qc) or qc.startswith(food.get("_c") or "")
                ]
            for food in prefix_hits[:MAX_CANDIDATES]:
                add(food)
            suffix_hits = index["by_suffix"].get(qc[-4:], [])
            if len(suffix_hits) <= MAX_BUCKET:
                for food in suffix_hits:
                    add(food)
    for tok in name_tokens(query):
        token_hits = index["by_token"].get(tok, [])
        if len(token_hits) > MAX_BUCKET:
            token_hits = sorted(token_hits, key=lambda f: len(f.get("name_de") or ""))[:MAX_BUCKET]
        for food in token_hits:
            add(food)
        if len(out) >= MAX_CANDIDATES:
            break
    return out[:MAX_CANDIDATES]


def suggest_for_name(index: dict[str, Any], query: str, limit: int = 3) -> dict[str, Any]:
    q = (query or "").strip()
    scored: list[tuple[int, int, dict]] = []
    for food in _candidate_foods(index, q):
        score = score_name_match(q, food.get("name_de") or "", food.get("name_en"))
        if score < MIN_SCORE:
            continue
        scored.append((score, len(food.get("name_de") or ""), food))
    scored.sort(key=lambda x: (-x[0], x[1], x[2].get("name_de") or ""))
    top = [_public(food, score) for score, _nlen, food in scored[: max(limit, 3)]]
    ambiguous = False
    if len(top) >= 2 and top[0]["score"] - top[1]["score"] <= 8 and top[1]["score"] >= 60:
        ambiguous = True
    elif len(top) >= 2 and top[0]["score"] < 90:
        ambiguous = True
    return {
        "suggestions": top[:limit],
        "suggestion_count": len(top),
        "ambiguous": ambiguous,
    }


def attach_suggestions(index: dict[str, Any], rows: list[dict[str, Any]], limit: int = 3) -> list[dict[str, Any]]:
    for row in rows:
        q = row.get("source_name_raw") or row.get("source_name_normalized") or ""
        packed = suggest_for_name(index, q, limit=limit)
        row["suggestions"] = packed["suggestions"]
        row["suggestion_count"] = packed["suggestion_count"]
        row["ambiguous"] = packed["ambiguous"]
    return rows


def suggest_catalog_foods_ranked(cur, query: str, profile_id: str | None, limit: int = 8) -> list[dict]:
    q = (query or "").strip()
    if len(q) < 2:
        return []
    index = get_suggest_index(cur, profile_id)
    packed = suggest_for_name(index, q, limit=limit)
    if packed["suggestions"]:
        return packed["suggestions"]
    from data_layer.food_mapping import suggest_catalog_foods
    return suggest_catalog_foods(cur, q, profile_id, limit=limit)


def suggest_batch(cur, profile_id: str | None, names: list[str], limit: int = 3) -> dict[str, dict[str, Any]]:
    index = get_suggest_index(cur, profile_id)
    out = {}
    for raw in names[:100]:
        key = (raw or "").strip()
        if not key or key in out:
            continue
        out[key] = suggest_for_name(index, key, limit=limit)
    return out