pyr0ball
890216a1f0
Cloud mode: attach shared read-only corpus DB (RECIPE_DB_PATH env var)
as "corpus" schema so per-user SQLite DBs can access 3.19M recipes.
All corpus table references now use self._cp prefix ("corpus." in cloud,
"" in local). FTS5 pseudo-column kept unqualified per SQLite spec.
compose.cloud.yml: bind-mount /Library/Assets/kiwi/kiwi.db read-only.
Also fix batch of audit issues:
- #101: OCR approval used source="receipt_ocr" for inventory_items — use "receipt"
- #89/#100: Shopping confirm-purchase used source="shopping_list" — use "manual"
- #103: Frontend inventory filter sent ?status= but API expects ?item_status=
- #104: InventoryItemUpdate schema missing purchase_date field; store.py allowed set also missing it
- #105: Guest cookie Secure flag tied to CLOUD_MODE instead of X-Forwarded-Proto; broke HTTP direct-port access
127 lines
4.5 KiB
Python
127 lines
4.5 KiB
Python
"""
|
|
SubstitutionEngine — deterministic ingredient swap candidates with compensation hints.
|
|
|
|
Powered by:
|
|
- substitution_pairs table (derived from lishuyang/recipepairs)
|
|
- ingredient_profiles functional metadata (USDA FDC)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from dataclasses import dataclass, field
|
|
from typing import TYPE_CHECKING
|
|
|
|
if TYPE_CHECKING:
|
|
from app.db.store import Store
|
|
|
|
# Compensation threshold — if |delta| exceeds this, surface a hint
|
|
_FAT_THRESHOLD = 5.0 # grams per 100g
|
|
_GLUTAMATE_THRESHOLD = 1.0 # mg per 100g
|
|
_MOISTURE_THRESHOLD = 15.0 # grams per 100g
|
|
|
|
_RICHNESS_COMPENSATORS = ["olive oil", "coconut oil", "butter", "shortening", "full-fat coconut milk"]
|
|
_DEPTH_COMPENSATORS = ["nutritional yeast", "soy sauce", "miso", "mushroom powder",
|
|
"better than bouillon not-beef", "smoked paprika"]
|
|
_MOISTURE_BINDERS = ["cornstarch", "flour", "arrowroot", "breadcrumbs"]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CompensationHint:
|
|
ingredient: str
|
|
reason: str
|
|
element: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SubstitutionSwap:
|
|
original_name: str
|
|
substitute_name: str
|
|
constraint_label: str
|
|
fat_delta: float
|
|
moisture_delta: float
|
|
glutamate_delta: float
|
|
protein_delta: float
|
|
occurrence_count: int
|
|
compensation_hints: list[dict] = field(default_factory=list)
|
|
explanation: str = ""
|
|
|
|
|
|
class SubstitutionEngine:
|
|
def __init__(self, store: "Store") -> None:
|
|
self._store = store
|
|
|
|
def find_substitutes(
|
|
self,
|
|
ingredient_name: str,
|
|
constraint: str,
|
|
) -> list[SubstitutionSwap]:
|
|
c = self._store._cp
|
|
rows = self._store._fetch_all(f"""
|
|
SELECT substitute_name, constraint_label,
|
|
fat_delta, moisture_delta, glutamate_delta, protein_delta,
|
|
occurrence_count, compensation_hints
|
|
FROM {c}substitution_pairs
|
|
WHERE original_name = ? AND constraint_label = ?
|
|
ORDER BY occurrence_count DESC
|
|
""", (ingredient_name.lower(), constraint))
|
|
|
|
return [self._row_to_swap(ingredient_name, row) for row in rows]
|
|
|
|
def _row_to_swap(self, original: str, row: dict) -> SubstitutionSwap:
|
|
hints = self._build_hints(row)
|
|
explanation = self._build_explanation(original, row, hints)
|
|
return SubstitutionSwap(
|
|
original_name=original,
|
|
substitute_name=row["substitute_name"],
|
|
constraint_label=row["constraint_label"],
|
|
fat_delta=row.get("fat_delta") or 0.0,
|
|
moisture_delta=row.get("moisture_delta") or 0.0,
|
|
glutamate_delta=row.get("glutamate_delta") or 0.0,
|
|
protein_delta=row.get("protein_delta") or 0.0,
|
|
occurrence_count=row.get("occurrence_count") or 1,
|
|
compensation_hints=[{"ingredient": h.ingredient, "reason": h.reason, "element": h.element} for h in hints],
|
|
explanation=explanation,
|
|
)
|
|
|
|
def _build_hints(self, row: dict) -> list[CompensationHint]:
|
|
hints = []
|
|
fat_delta = row.get("fat_delta") or 0.0
|
|
glutamate_delta = row.get("glutamate_delta") or 0.0
|
|
moisture_delta = row.get("moisture_delta") or 0.0
|
|
|
|
if fat_delta < -_FAT_THRESHOLD:
|
|
missing = abs(fat_delta)
|
|
sugg = _RICHNESS_COMPENSATORS[0]
|
|
hints.append(CompensationHint(
|
|
ingredient=sugg,
|
|
reason=f"substitute has ~{missing:.0f}g/100g less fat — add {sugg} to restore Richness",
|
|
element="Richness",
|
|
))
|
|
|
|
if glutamate_delta < -_GLUTAMATE_THRESHOLD:
|
|
sugg = _DEPTH_COMPENSATORS[0]
|
|
hints.append(CompensationHint(
|
|
ingredient=sugg,
|
|
reason=f"substitute is lower in umami — add {sugg} to restore Depth",
|
|
element="Depth",
|
|
))
|
|
|
|
if moisture_delta > _MOISTURE_THRESHOLD:
|
|
sugg = _MOISTURE_BINDERS[0]
|
|
hints.append(CompensationHint(
|
|
ingredient=sugg,
|
|
reason=f"substitute adds ~{moisture_delta:.0f}g/100g more moisture — add {sugg} to maintain Structure",
|
|
element="Structure",
|
|
))
|
|
|
|
return hints
|
|
|
|
def _build_explanation(
|
|
self, original: str, row: dict, hints: list[CompensationHint]
|
|
) -> str:
|
|
sub = row["substitute_name"]
|
|
count = row.get("occurrence_count") or 1
|
|
base = f"Replace {original} with {sub} (seen in {count} recipes)."
|
|
if hints:
|
|
base += " To compensate: " + "; ".join(h.reason for h in hints) + "."
|
|
return base
|