app/services/community/element_snapshot.py

@@ -0,0 +1,127 @@
+# app/services/community/element_snapshot.py
+# MIT License
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+# Ingredient name substrings → allergen flag
+_ALLERGEN_MAP: dict[str, str] = {
+    "milk": "dairy", "cream": "dairy", "cheese": "dairy", "butter": "dairy",
+    "yogurt": "dairy", "whey": "dairy",
+    "egg": "eggs",
+    "wheat": "gluten", "pasta": "gluten", "flour": "gluten", "bread": "gluten",
+    "barley": "gluten", "rye": "gluten",
+    "peanut": "nuts", "almond": "nuts", "cashew": "nuts", "walnut": "nuts",
+    "pecan": "nuts", "hazelnut": "nuts", "pistachio": "nuts", "macadamia": "nuts",
+    "soy": "soy", "tofu": "soy", "edamame": "soy", "miso": "soy", "tempeh": "soy",
+    "shrimp": "shellfish", "crab": "shellfish", "lobster": "shellfish",
+    "clam": "shellfish", "mussel": "shellfish", "scallop": "shellfish",
+    "fish": "fish", "salmon": "fish", "tuna": "fish", "cod": "fish",
+    "tilapia": "fish", "halibut": "fish",
+    "sesame": "sesame",
+}
+
+_MEAT_KEYWORDS = frozenset([
+    "chicken", "beef", "pork", "lamb", "turkey", "bacon", "ham", "sausage",
+    "salami", "prosciutto", "guanciale", "pancetta", "steak", "ground meat",
+    "mince", "veal", "duck", "venison", "bison", "lard",
+])
+_SEAFOOD_KEYWORDS = frozenset([
+    "fish", "shrimp", "crab", "lobster", "tuna", "salmon", "clam", "mussel",
+    "scallop", "anchovy", "sardine", "cod", "tilapia",
+])
+_ANIMAL_PRODUCT_KEYWORDS = frozenset([
+    "milk", "cream", "cheese", "butter", "egg", "honey", "yogurt", "whey",
+])
+
+
+def _detect_allergens(ingredient_names: list[str]) -> list[str]:
+    found: set[str] = set()
+    for ingredient in (n.lower() for n in ingredient_names):
+        for keyword, flag in _ALLERGEN_MAP.items():
+            if keyword in ingredient:
+                found.add(flag)
+    return sorted(found)
+
+
+def _detect_dietary_tags(ingredient_names: list[str]) -> list[str]:
+    all_text = " ".join(n.lower() for n in ingredient_names)
+    has_meat = any(k in all_text for k in _MEAT_KEYWORDS)
+    has_seafood = any(k in all_text for k in _SEAFOOD_KEYWORDS)
+    has_animal_products = any(k in all_text for k in _ANIMAL_PRODUCT_KEYWORDS)
+
+    tags: list[str] = []
+    if not has_meat and not has_seafood:
+        tags.append("vegetarian")
+    if not has_meat and not has_seafood and not has_animal_products:
+        tags.append("vegan")
+    return tags
+
+
+@dataclass(frozen=True)
+class ElementSnapshot:
+    seasoning_score: float
+    richness_score: float
+    brightness_score: float
+    depth_score: float
+    aroma_score: float
+    structure_score: float
+    texture_profile: str
+    dietary_tags: tuple
+    allergen_flags: tuple
+    flavor_molecules: tuple
+    fat_pct: float | None
+    protein_pct: float | None
+    moisture_pct: float | None
+
+
+def compute_snapshot(recipe_ids: list[int], store) -> ElementSnapshot:
+    """Compute an element snapshot from a list of recipe IDs.
+
+    Pulls SFAH scores, ingredient lists, and USDA FDC macros from the corpus.
+    Averages numeric scores across all recipes. Unions allergen flags and dietary tags.
+    Call at publish time only — snapshot is stored denormalized in community_posts.
+    """
+    _empty = ElementSnapshot(
+        seasoning_score=0.0, richness_score=0.0, brightness_score=0.0,
+        depth_score=0.0, aroma_score=0.0, structure_score=0.0,
+        texture_profile="", dietary_tags=(), allergen_flags=(),
+        flavor_molecules=(), fat_pct=None, protein_pct=None, moisture_pct=None,
+    )
+    if not recipe_ids:
+        return _empty
+
+    rows = store.get_recipes_by_ids(recipe_ids)
+    if not rows:
+        return _empty
+
+    def _avg(field: str) -> float:
+        vals = [r.get(field) or 0.0 for r in rows]
+        return sum(vals) / len(vals)
+
+    all_ingredients: list[str] = []
+    for r in rows:
+        names = r.get("ingredient_names") or []
+        if isinstance(names, list):
+            all_ingredients.extend(names)
+
+    fat_vals = [r["fat"] for r in rows if r.get("fat") is not None]
+    prot_vals = [r["protein"] for r in rows if r.get("protein") is not None]
+    moist_vals = [r["moisture"] for r in rows if r.get("moisture") is not None]
+
+    return ElementSnapshot(
+        seasoning_score=_avg("seasoning_score"),
+        richness_score=_avg("richness_score"),
+        brightness_score=_avg("brightness_score"),
+        depth_score=_avg("depth_score"),
+        aroma_score=_avg("aroma_score"),
+        structure_score=_avg("structure_score"),
+        texture_profile=rows[0].get("texture_profile") or "",
+        dietary_tags=tuple(_detect_dietary_tags(all_ingredients)),
+        allergen_flags=tuple(_detect_allergens(all_ingredients)),
+        flavor_molecules=(),  # deferred — FlavorGraph ticket
+        fat_pct=(sum(fat_vals) / len(fat_vals)) if fat_vals else None,
+        protein_pct=(sum(prot_vals) / len(prot_vals)) if prot_vals else None,
+        moisture_pct=(sum(moist_vals) / len(moist_vals)) if moist_vals else None,
+    )

tests/services/community/test_element_snapshot.py

@@ -0,0 +1,77 @@
+# tests/services/community/test_element_snapshot.py
+import pytest
+from unittest.mock import MagicMock
+from app.services.community.element_snapshot import compute_snapshot, ElementSnapshot
+
+
+def make_mock_store(recipe_rows: list[dict]) -> MagicMock:
+    store = MagicMock()
+    store.get_recipes_by_ids.return_value = recipe_rows
+    return store
+
+
+RECIPE_ROW = {
+    "id": 1,
+    "name": "Spaghetti Carbonara",
+    "ingredient_names": ["pasta", "eggs", "guanciale", "pecorino"],
+    "keywords": ["italian", "quick", "dinner"],
+    "category": "dinner",
+    "fat": 22.0,
+    "protein": 18.0,
+    "moisture": 45.0,
+    "seasoning_score": 0.7,
+    "richness_score": 0.8,
+    "brightness_score": 0.2,
+    "depth_score": 0.6,
+    "aroma_score": 0.5,
+    "structure_score": 0.9,
+    "texture_profile": "creamy",
+}
+
+
+def test_compute_snapshot_basic():
+    store = make_mock_store([RECIPE_ROW])
+    snap = compute_snapshot(recipe_ids=[1], store=store)
+    assert isinstance(snap, ElementSnapshot)
+    assert 0.0 <= snap.seasoning_score <= 1.0
+    assert snap.texture_profile == "creamy"
+
+
+def test_compute_snapshot_averages_multiple_recipes():
+    row2 = {**RECIPE_ROW, "id": 2, "seasoning_score": 0.3, "richness_score": 0.2}
+    store = make_mock_store([RECIPE_ROW, row2])
+    snap = compute_snapshot(recipe_ids=[1, 2], store=store)
+    # seasoning: average of 0.7 and 0.3 = 0.5
+    assert abs(snap.seasoning_score - 0.5) < 0.01
+
+
+def test_compute_snapshot_allergen_flags_detected():
+    row = {**RECIPE_ROW, "ingredient_names": ["pasta", "eggs", "milk", "shrimp", "peanuts"]}
+    store = make_mock_store([row])
+    snap = compute_snapshot(recipe_ids=[1], store=store)
+    assert "gluten" in snap.allergen_flags   # pasta
+    assert "dairy" in snap.allergen_flags    # milk
+    assert "shellfish" in snap.allergen_flags  # shrimp
+    assert "nuts" in snap.allergen_flags     # peanuts
+
+
+def test_compute_snapshot_dietary_tags_vegetarian():
+    row = {**RECIPE_ROW, "ingredient_names": ["pasta", "eggs", "tomato", "basil"]}
+    store = make_mock_store([row])
+    snap = compute_snapshot(recipe_ids=[1], store=store)
+    assert "vegetarian" in snap.dietary_tags
+
+
+def test_compute_snapshot_no_recipes_returns_defaults():
+    store = make_mock_store([])
+    snap = compute_snapshot(recipe_ids=[], store=store)
+    assert snap.seasoning_score == 0.0
+    assert snap.dietary_tags == ()
+    assert snap.allergen_flags == ()
+
+
+def test_element_snapshot_immutable():
+    store = make_mock_store([RECIPE_ROW])
+    snap = compute_snapshot(recipe_ids=[1], store=store)
+    with pytest.raises((AttributeError, TypeError)):
+        snap.seasoning_score = 0.0  # type: ignore