Update services/recipe_service.py

services/recipe_service.py

@@ -1,63 +1,374 @@
-def _head_token(text: str) -> str:
-    parts = (text or "").split()
-    return parts[-1] if parts else ""
-
-def _token_overlap(a: str, b: str) -> bool:
-    sa = set((a or "").split())
-    sb = set((b or "").split())
-    return bool(sa & sb)
-
-def lookup(self, ingredient: str, recipe_type: str, diet: str):
-    preferred = BAKE_CONTEXTS if recipe_type == "baked" else COOK_CONTEXTS
-    variants = ingredient_lookup_variants(ingredient)
-    query_norm = normalize_ingredient_for_lookup(ingredient)
-    query_head = _head_token(query_norm)
-
-    # 1) exact match first
-    for v in variants:
-        row = self.db.find_exact(v, preferred_contexts=preferred)
-        if row is not None:
-            return self._format_result(ingredient, row, diet, "database match", 1.0)
-
-    # 2) alias / partial match second, but only for real ingredient-length tokens
-    for v in variants:
-        if len(v) < 4:
-            continue
-        row = self.db.find_partial(v, preferred_contexts=preferred)
-        if row is not None:
-            return self._format_result(ingredient, row, diet, "partial match", 0.85)
-
-    # 3) semantic fallback only if it is actually close AND same ingredient family
-    if self.semantic and self.semantic.available:
-        best = self.semantic.most_similar(query_norm, topn=10)
-
-        for candidate_text, score in best:
-            if score < 0.82:
-                break
-
-            cand_norm = normalize_ingredient_for_lookup(candidate_text)
-            cand_head = _head_token(cand_norm)
-
-            # reject nonsense like spinach -> sparkling water
-            if query_head and cand_head and query_head != cand_head and not _token_overlap(query_norm, cand_norm):
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Dict, Iterable, List, Optional, Tuple
+
+from .classify import classify_recipe
+from .ner import extract_ingredients
+from .text_utils import (
+    as_aliases,
+    ingredient_lookup_variants,
+    normalize_ingredient_for_lookup,
+)
+
+try:
+    import pandas as pd
+except Exception:  # pragma: no cover
+    pd = None
+
+
+BAKE_CONTEXTS = {
+    "Baking & Desserts",
+    "Baking (Binder)",
+    "Baking (Leavening)",
+    "Pastries",
+    "Bagels",
+    "Puddings",
+    "Tiramisu",
+}
+
+COOK_CONTEXTS = {
+    "Main Course & Heavy Cooking",
+    "Sauces, Dips & Dressings",
+    "Soups & Savory Liquids",
+    "Cold Prep & Light Meals",
+    "Pasta",
+    "Lasagna",
+    "Roast",
+    "Stir-Fry",
+    "Appetizer",
+    "Indian",
+    "Beverages",
+    "Cheese Making",
+    "Processed",
+    "Technical & Additives",
+}
+
+
+def _clean_value(value: Any) -> str:
+    if value is None:
+        return ""
+    try:
+        if pd is not None and pd.isna(value):
+            return ""
+    except Exception:
+        pass
+    text = str(value).strip()
+    if text.lower() == "nan":
+        return ""
+    return text
+
+
+def _pick_first(row: Any, keys: Iterable[str]) -> str:
+    for key in keys:
+        if isinstance(row, dict) and key in row:
+            value = _clean_value(row.get(key))
+            if value:
+                return value
+        else:
+            try:
+                value = _clean_value(row[key])
+                if value:
+                    return value
+            except Exception:
+                pass
+    return ""
+
+
+def _row_to_dict(row: Any) -> Dict[str, Any]:
+    if row is None:
+        return {}
+    if isinstance(row, dict):
+        return dict(row)
+    if hasattr(row, "to_dict"):
+        try:
+            return row.to_dict()
+        except Exception:
+            return {}
+    return {}
+
+
+@dataclass
+class MatchResult:
+    original: str
+    normalized: str
+    compliant: bool
+    substitute: str
+    instructions: str
+    source: str
+    matched_ingredient: Optional[str]
+    confidence: float
+    notes: str = ""
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "original": self.original,
+            "normalized": self.normalized,
+            "compliant": self.compliant,
+            "substitute": self.substitute,
+            "instructions": self.instructions,
+            "source": self.source,
+            "matched_ingredient": self.matched_ingredient,
+            "confidence": self.confidence,
+            "notes": self.notes,
+        }
+
+
+class RecipeAdapterService:
+    def __init__(self, db, semantic):
+        self.db = db
+        self.semantic = semantic
+
+    def _frame(self):
+        for attr in ("df", "dataframe", "frame", "table"):
+            frame = getattr(self.db, attr, None)
+            if frame is not None:
+                return frame
+        return None
+
+    def _context_filter(self, recipe_type: str):
+        return BAKE_CONTEXTS if recipe_type == "baked" else COOK_CONTEXTS
+
+    def _friendly_col(self, diet: str) -> str:
+        if diet == "vegan":
+            return "Is_Vegan"
+        if diet == "keto":
+            return "Is_Keto_Friendly"
+        return ""
+
+    def _sub_col(self, diet: str) -> str:
+        return {
+            "vegan": "Vegan_Substitution",
+            "keto": "Keto_Substitution",
+            "both": "Vegan_Substitution",
+        }[diet]
+
+    def _instr_col(self, diet: str) -> str:
+        return {
+            "vegan": "Vegan_Instruction",
+            "keto": "Keto_Instruction",
+            "both": "Vegan_Instruction",
+        }[diet]
+
+    def _find_in_frame(self, query: str, recipe_type: str, exact: bool = True):
+        frame = self._frame()
+        if frame is None:
+            return None
+
+        preferred = self._context_filter(recipe_type)
+
+        try:
+            rows = frame
+            if "Context" in frame.columns:
+                pref_rows = frame[frame["Context"].isin(preferred)]
+                rows = pref_rows if not pref_rows.empty else frame
+        except Exception:
+            rows = frame
+
+        try:
+            if exact:
+                if "Ingredient" in rows.columns:
+                    m = rows[rows["Ingredient"].astype(str).str.lower() == query]
+                    if not m.empty:
+                        return m.iloc[0]
+            else:
+                if "Ingredient" in rows.columns:
+                    pattern = r"\b" + query.replace(" ", r"\s+") + r"\b"
+                    m = rows[rows["Ingredient"].astype(str).str.lower().str.contains(pattern, regex=True, na=False)]
+                    if not m.empty:
+                        return m.iloc[0]
+        except Exception:
+            return None
+
+        return None
+
+    def _find_alias(self, query: str, recipe_type: str):
+        frame = self._frame()
+        if frame is None or "aliases" not in frame.columns and "Aliases" not in frame.columns:
+            return None
+
+        alias_col = "aliases" if "aliases" in frame.columns else "Aliases"
+        preferred = self._context_filter(recipe_type)
+
+        try:
+            rows = frame
+            if "Context" in frame.columns:
+                pref_rows = frame[frame["Context"].isin(preferred)]
+                rows = pref_rows if not pref_rows.empty else frame
+        except Exception:
+            rows = frame
+
+        try:
+            for idx, row in rows.iterrows():
+                ingredient = _clean_value(row.get("Ingredient"))
+                aliases = as_aliases(row.get(alias_col))
+                if query == ingredient.lower() or query in aliases:
+                    return row
+        except Exception:
+            return None
+
+        return None
+
+    def _resolve_row(self, ingredient: str, diet: str, recipe_type: str, row: Any, source: str, confidence: float):
+        rowd = _row_to_dict(row)
+
+        sub = _clean_value(rowd.get(self._sub_col(diet)))
+        instr = _clean_value(rowd.get(self._instr_col(diet)))
+        ing_name = _clean_value(rowd.get("Ingredient")) or ingredient
+
+        friendly_col = self._friendly_col(diet)
+        already_ok = False
+        if friendly_col and friendly_col in rowd:
+            val = rowd.get(friendly_col)
+            already_ok = bool(val) is True
+
+        if not sub or already_ok:
+            return MatchResult(
+                original=ingredient,
+                normalized=normalize_ingredient_for_lookup(ingredient),
+                compliant=True,
+                substitute=ingredient,
+                instructions="Already compatible — no substitution needed.",
+                source=source,
+                matched_ingredient=ing_name or None,
+                confidence=1.0 if already_ok else confidence,
+                notes="",
+            ).to_dict()
+
+        return MatchResult(
+            original=ingredient,
+            normalized=normalize_ingredient_for_lookup(ingredient),
+            compliant=False,
+            substitute=sub,
+            instructions=instr,
+            source=source,
+            matched_ingredient=ing_name or None,
+            confidence=confidence,
+            notes="",
+        ).to_dict()
+
+    def _safe_semantic_match(self, query: str, recipe_type: str):
+        if self.semantic is None or not getattr(self.semantic, "available", False):
+            return None
+
+        query_norm = normalize_ingredient_for_lookup(query)
+        variants = ingredient_lookup_variants(query_norm)
+
+        # Conservative threshold so the model does not invent unrelated foods.
+        threshold = 0.82
+
+        for variant in variants:
+            try:
+                candidates = self.semantic.most_similar(variant, topn=10)
+            except Exception:
                 continue
 
-            row = self.db.find_exact(cand_norm, preferred_contexts=preferred)
-            if row is None:
-                row = self.db.find_partial(cand_norm, preferred_contexts=preferred)
+            for candidate, score in candidates:
+                if score < threshold:
+                    break
+
+                candidate_norm = normalize_ingredient_for_lookup(candidate)
+                if not candidate_norm:
+                    continue
+
+                row = self._find_in_frame(candidate_norm, recipe_type, exact=True)
+                if row is None:
+                    row = self._find_alias(candidate_norm, recipe_type)
+                if row is None:
+                    row = self._find_in_frame(candidate_norm, recipe_type, exact=False)
+
+                if row is not None:
+                    return row, f"semantic match ({candidate_norm})", float(score)
+
+        return None
 
+    def adapt(self, recipe_text: str, diet: str) -> Dict[str, Any]:
+        if diet not in {"vegan", "keto", "both"}:
+            raise ValueError("diet must be vegan, keto, or both")
+
+        recipe_type = classify_recipe(recipe_text)
+        ingredients = extract_ingredients(recipe_text)
+
+        if not ingredients:
+            raise ValueError("Could not extract any ingredients from the recipe text.")
+
+        results: List[Dict[str, Any]] = []
+        substitution_count = 0
+
+        for ing in ingredients:
+            query = normalize_ingredient_for_lookup(ing)
+            row = None
+            source = ""
+            confidence = 1.0
+
+            # 1) exact match
+            row = self._find_in_frame(query, recipe_type, exact=True)
+            if row is not None:
+                source = "database match"
+                result = self._resolve_row(ing, diet, recipe_type, row, source, confidence)
+                results.append(result)
+                if not result["compliant"]:
+                    substitution_count += 1
+                continue
+
+            # 2) alias match
+            row = self._find_alias(query, recipe_type)
             if row is not None:
-                return self._format_result(ingredient, row, diet, f"semantic match ({candidate_text})", float(score))
-
-    # 4) do NOT guess
-    return {
-        "original": ingredient,
-        "normalized": query_norm,
-        "compliant": False,
-        "substitute": ingredient,
-        "instructions": "No reliable match found — verify manually.",
-        "source": "unmatched",
-        "matched_ingredient": None,
-        "confidence": 0.0,
-        "notes": "No exact, alias, or safe semantic match found.",
-    }
+                source = "alias match"
+                result = self._resolve_row(ing, diet, recipe_type, row, source, 0.9)
+                results.append(result)
+                if not result["compliant"]:
+                    substitution_count += 1
+                continue
+
+            # 3) partial match, but only as a last non-semantic direct match
+            row = self._find_in_frame(query, recipe_type, exact=False)
+            if row is not None and len(query) >= 4:
+                source = "partial match"
+                result = self._resolve_row(ing, diet, recipe_type, row, source, 0.8)
+                results.append(result)
+                if not result["compliant"]:
+                    substitution_count += 1
+                continue
+
+            # 4) semantic match with conservative threshold
+            semantic_hit = self._safe_semantic_match(ing, recipe_type)
+            if semantic_hit is not None:
+                row, source, confidence = semantic_hit
+                result = self._resolve_row(ing, diet, recipe_type, row, source, confidence)
+                results.append(result)
+                if not result["compliant"]:
+                    substitution_count += 1
+                continue
+
+            # 5) do not guess
+            results.append(
+                MatchResult(
+                    original=ing,
+                    normalized=query,
+                    compliant=False,
+                    substitute=ing,
+                    instructions="No reliable match found — verify manually.",
+                    source="unmatched",
+                    matched_ingredient=None,
+                    confidence=0.0,
+                    notes="No exact, alias, or safe semantic match found.",
+                ).to_dict()
+            )
+
+        return {
+            "diet": diet,
+            "recipe_type": recipe_type,
+            "ingredients": results,
+            "ingredients_found": len(results),
+            "substitution_count": substitution_count,
+            "model_metadata": {
+                "ner_model": "ingredient-parser-nlp (comma-separated parsing)",
+                "qa_model": "keyword fallback only",
+                "semantic_model": getattr(self.semantic, "model_name", "unknown"),
+                "semantic_available": bool(getattr(self.semantic, "available", False)),
+                "semantic_mode": getattr(self.semantic, "_kind", "disabled"),
+                "dataset_path": str(getattr(self.db, "path", "")),
+            },
+            "disclaimer": "This is an assistive recipe tool, not nutritional or allergen medical advice. Verify substitutions before cooking.",
+        }