Public Access
227 lines
7.9 KiB
Python
227 lines
7.9 KiB
Python
"""Recipe Discovery — queries external APIs (Spoonacular, TheMealDB) for recipes.
|
|
|
|
Takes discovery queries from FeedbackAnalyzer and fetches normalized recipe candidates.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import time
|
|
from dataclasses import dataclass
|
|
from decimal import Decimal
|
|
from typing import List, Optional, Any
|
|
|
|
import requests
|
|
|
|
from app.config import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_SPOONACULAR_SEARCH_URL = "https://api.spoonacular.com/recipes/complexSearch"
|
|
_SPOONACULAR_INFO_URL = "https://api.spoonacular.com/recipes/{id}/information"
|
|
_THEMEALDB_SEARCH_URL = "https://www.themealdb.com/api/json/v1/1/search.php"
|
|
|
|
_RATE_LIMIT_SECS = 1.0 # polite gap between calls
|
|
_MAX_RESULTS_PER_QUERY = 5 # cap to stay within free quota
|
|
|
|
|
|
@dataclass
|
|
class ExternalRecipe:
|
|
name: str
|
|
external_source: str
|
|
external_id: str
|
|
image_url: Optional[str]
|
|
description: Optional[str]
|
|
prep_time_minutes: Optional[int]
|
|
cook_time_minutes: Optional[int]
|
|
servings: int
|
|
cuisine_tags: List[str]
|
|
dietary_tags: List[str]
|
|
protein_type: Optional[str]
|
|
calories_per_serving: Optional[int]
|
|
ingredients: List[dict] # [{"name": str, "qty": float, "unit": str}]
|
|
instructions: List[str]
|
|
source_url: Optional[str]
|
|
|
|
|
|
class RecipeDiscoveryService:
|
|
"""Fetch recipes from external sources."""
|
|
|
|
def __init__(self) -> None:
|
|
self.api_key = getattr(settings, "SPOONACULAR_API_KEY", "")
|
|
self.enabled = bool(self.api_key)
|
|
self._points_used = 0
|
|
|
|
def discover(self, queries: List[str]) -> List[ExternalRecipe]:
|
|
"""Run all discovery queries and return unique recipes."""
|
|
if not self.enabled:
|
|
logger.warning("RecipeDiscovery: SPOONACULAR_API_KEY not set — skipping")
|
|
return []
|
|
|
|
all_recipes: List[ExternalRecipe] = []
|
|
seen_ids: set[str] = set()
|
|
|
|
for query in queries:
|
|
if self._points_used >= 140: # stay under 150/day free tier
|
|
logger.warning("RecipeDiscovery: quota near limit (%d/150), stopping", self._points_used)
|
|
break
|
|
|
|
recipes = self._search_spoonacular(query)
|
|
for r in recipes:
|
|
key = f"{r.external_source}:{r.external_id}"
|
|
if key not in seen_ids:
|
|
seen_ids.add(key)
|
|
all_recipes.append(r)
|
|
|
|
time.sleep(_RATE_LIMIT_SECS)
|
|
|
|
logger.info("RecipeDiscovery: %d unique recipes from %d queries", len(all_recipes), len(queries))
|
|
return all_recipes
|
|
|
|
def _search_spoonacular(self, query: str) -> List[ExternalRecipe]:
|
|
"""Search Spoonacular and return normalized recipes."""
|
|
params = {
|
|
"apiKey": self.api_key,
|
|
"query": query,
|
|
"number": _MAX_RESULTS_PER_QUERY,
|
|
"addRecipeInformation": "true",
|
|
"fillIngredients": "true",
|
|
"instructionsRequired": "true",
|
|
}
|
|
try:
|
|
resp = requests.get(_SPOONACULAR_SEARCH_URL, params=params, timeout=30)
|
|
resp.raise_for_status()
|
|
except requests.RequestException as exc:
|
|
logger.warning("Spoonacular search failed for %r: %s", query, exc)
|
|
return []
|
|
|
|
data = resp.json()
|
|
results = data.get("results", [])
|
|
total = data.get("totalResults", 0)
|
|
# complexSearch = 1 point + 0.01 per result
|
|
self._points_used += 1 + len(results) * 0.01
|
|
|
|
logger.info("Spoonacular: %r → %d/%d results", query, len(results), total)
|
|
|
|
out = []
|
|
for item in results:
|
|
ext_id = str(item.get("id"))
|
|
if not ext_id:
|
|
continue
|
|
|
|
# Try to get full info for ingredients + instructions
|
|
full = self._fetch_recipe_info(ext_id)
|
|
if full:
|
|
normalized = self._normalize_spoonacular(item, full)
|
|
if normalized:
|
|
out.append(normalized)
|
|
time.sleep(0.5) # between info calls
|
|
else:
|
|
# Fallback: info endpoint failed, use search summary only
|
|
normalized = self._normalize_spoonacular(item, item)
|
|
if normalized:
|
|
out.append(normalized)
|
|
|
|
return out
|
|
|
|
def _fetch_recipe_info(self, recipe_id: str) -> dict | None:
|
|
"""Fetch detailed recipe info from Spoonacular."""
|
|
url = _SPOONACULAR_INFO_URL.format(id=recipe_id)
|
|
params = {
|
|
"apiKey": self.api_key,
|
|
"includeNutrition": "false",
|
|
}
|
|
try:
|
|
resp = requests.get(url, params=params, timeout=30)
|
|
resp.raise_for_status()
|
|
except requests.RequestException as exc:
|
|
logger.warning("Spoonacular info failed for %s: %s", recipe_id, exc)
|
|
return None
|
|
|
|
# info endpoint = 1 point
|
|
self._points_used += 1
|
|
return resp.json()
|
|
|
|
def _normalize_spoonacular(self, summary: dict, full: dict) -> ExternalRecipe | None:
|
|
"""Convert Spoonacular response into our ExternalRecipe dataclass."""
|
|
title = summary.get("title") or full.get("title")
|
|
if not title:
|
|
return None
|
|
|
|
# Ingredients from full info
|
|
ingredients = []
|
|
for ing in full.get("extendedIngredients", []):
|
|
qty = ing.get("amount")
|
|
unit = ing.get("unit", "")
|
|
name = ing.get("name", "")
|
|
if qty is not None and name:
|
|
ingredients.append({"name": name, "qty": float(qty), "unit": unit})
|
|
|
|
# Instructions
|
|
instructions = []
|
|
analyzed = full.get("analyzedInstructions", [])
|
|
if analyzed:
|
|
for step in analyzed[0].get("steps", []):
|
|
instructions.append(step.get("step", ""))
|
|
else:
|
|
raw = full.get("instructions", "")
|
|
if raw:
|
|
instructions = [raw] # single blob
|
|
|
|
# Cuisines + diets
|
|
cuisines = [c.lower() for c in (summary.get("cuisines") or full.get("cuisines", [])) if c]
|
|
diets = [d.lower() for d in (summary.get("diets") or full.get("diets", [])) if d]
|
|
|
|
# Protein type inference from ingredient names or summary tags
|
|
protein = _infer_protein(food=full)
|
|
|
|
# Times
|
|
prep = full.get("preparationMinutes")
|
|
cook = full.get("cookingMinutes")
|
|
if prep is None and "readyInMinutes" in full:
|
|
prep = full["readyInMinutes"] # use total as proxy
|
|
|
|
return ExternalRecipe(
|
|
name=title,
|
|
external_source="spoonacular",
|
|
external_id=str(summary.get("id") or full.get("id")),
|
|
image_url=summary.get("image") or full.get("image"),
|
|
description=full.get("summary"), # HTML summary; caller strips tags
|
|
prep_time_minutes=int(prep) if prep else None,
|
|
cook_time_minutes=int(cook) if cook else None,
|
|
servings=int(full.get("servings", 4)),
|
|
cuisine_tags=cuisines,
|
|
dietary_tags=diets,
|
|
protein_type=protein,
|
|
calories_per_serving=None, # would require nutrition endpoint
|
|
ingredients=ingredients,
|
|
instructions=instructions or ["See source for instructions."],
|
|
source_url=full.get("sourceUrl") or full.get("spoonacularSourceUrl"),
|
|
)
|
|
|
|
|
|
def _infer_protein(food: dict) -> Optional[str]:
|
|
"""Infer protein_type from recipe data."""
|
|
title = (food.get("title") or "").lower()
|
|
ings = " ".join(
|
|
i.get("name", "").lower()
|
|
for i in food.get("extendedIngredients", [])
|
|
)
|
|
|
|
proteins = {
|
|
"chicken": ["chicken"],
|
|
"beef": ["beef", "steak", "ground beef"],
|
|
"pork": ["pork", "bacon", "ham"],
|
|
"fish": ["salmon", "tilapia", "cod", "fish fillet"],
|
|
"shrimp": ["shrimp", "prawn"],
|
|
"turkey": ["turkey"],
|
|
"lamb": ["lamb"],
|
|
"vegetarian": ["tofu", "tempeh", "vegetarian"],
|
|
}
|
|
|
|
for ptype, keywords in proteins.items():
|
|
for kw in keywords:
|
|
if kw in title or kw in ings:
|
|
return ptype
|
|
|
|
return None
|