212 lines
8.0 KiB
Python
212 lines
8.0 KiB
Python
import json
|
||
import time
|
||
import sys
|
||
import os
|
||
import requests
|
||
from typing import Dict, Optional
|
||
|
||
# Ensure recipe_parser is importable
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
from recipe_parser import ParsedRecipe
|
||
|
||
_BASE_URL = "https://world.openfoodfacts.org"
|
||
_SEARCH_BASE = _BASE_URL + "/cgi/search.pl"
|
||
_PRODUCT_BASE = _BASE_URL + "/api/v2/product"
|
||
|
||
|
||
class NutritionApi:
|
||
"""Wrapper around OpenFoodFacts API for looking up ingredient nutrition."""
|
||
|
||
def __init__(self):
|
||
self.conn = None # SQLite connection set externally if using cache
|
||
self._last_request = 0.0
|
||
|
||
def set_conn(self, conn):
|
||
"""Set the SQLite connection for caching."""
|
||
self.conn = conn
|
||
|
||
def lookup(self, ingredient: str) -> Optional[Dict]:
|
||
"""Look up nutrition data for *ingredient*.
|
||
|
||
Returns a dict with keys ``calories``, ``protein``, ``fat``,
|
||
``carbs``, ``sugar``, ``sodium`` – or ``None`` if not found.
|
||
"""
|
||
# Check cache first
|
||
if self.conn:
|
||
cursor = self.conn.execute(
|
||
"SELECT data FROM ingredient_cache WHERE ingredient = ?",
|
||
(ingredient,),
|
||
)
|
||
row = cursor.fetchone()
|
||
if row:
|
||
try:
|
||
return json.loads(row[0])
|
||
except json.JSONDecodeError:
|
||
pass # Fall through to API lookup
|
||
|
||
self._rate_limit()
|
||
result = self._search_openfood(ingredient)
|
||
if result is not None:
|
||
if self.conn:
|
||
self.conn.execute(
|
||
"INSERT OR REPLACE INTO ingredient_cache (ingredient, data) VALUES (?, ?)",
|
||
(ingredient, json.dumps(result)),
|
||
)
|
||
self.conn.commit()
|
||
return result
|
||
return None
|
||
|
||
def _rate_limit(self) -> None:
|
||
"""Simple rate limiter: minimum delay between requests."""
|
||
elapsed = time.time() - self._last_request
|
||
if elapsed < 1.0: # 1 request per second minimum
|
||
time.sleep(1.0 - elapsed)
|
||
self._last_request = time.time()
|
||
|
||
def _search_openfood(self, query: str) -> Optional[Dict]:
|
||
"""Search OpenFoodFacts and return normalised nutrition dict.
|
||
|
||
Retries with increasing delays to handle rate limiting / captcha challenges.
|
||
"""
|
||
# Retry with exponential backoff for transient errors (e.g. 503, HTML captcha pages)
|
||
max_retries = 5
|
||
# Increasing delays: 2, 5, 10, 20, 30 seconds to handle rate limiting / captcha challenges
|
||
backoff_delays = [2, 5, 10, 20, 30]
|
||
for attempt in range(max_retries):
|
||
try:
|
||
if attempt > 0:
|
||
delay = backoff_delays[min(attempt - 1, len(backoff_delays) - 1)]
|
||
time.sleep(delay)
|
||
params = {"search": query.strip(), "json": "1", "page_size": "5"}
|
||
r = requests.get(_SEARCH_BASE, params=params, timeout=10,
|
||
headers={"User-Agent": "RecipeParser/1.0"})
|
||
# Handle non-JSON responses (e.g. Cloudflare captcha HTML)
|
||
if r.status_code != 200:
|
||
if attempt == max_retries - 1:
|
||
return None
|
||
continue
|
||
# Check if response is valid JSON (not HTML error page)
|
||
try:
|
||
data = r.json()
|
||
except Exception:
|
||
# Not JSON – likely a captcha/challenge page; retry with longer delay
|
||
if attempt == max_retries - 1:
|
||
print(f"OpenFoodFacts returned non-JSON for '{query}'; giving up")
|
||
return None
|
||
continue
|
||
products = data.get("products", [])
|
||
if not products:
|
||
return None
|
||
|
||
# Take the first reasonably complete product
|
||
for prod in products:
|
||
nutr = prod.get("nutriments", {})
|
||
# We need at least calories/energy; skip if totally empty
|
||
if not nutr:
|
||
continue
|
||
return {
|
||
"calories": nutr.get("energy-kcal", 0) or 0,
|
||
"protein": nutr.get("protein", 0) or 0,
|
||
"fat": nutr.get("fat", 0) or 0,
|
||
"carbs": nutr.get("carbohydrates", 0) or 0,
|
||
"sugar": nutr.get("sugars", 0) or 0,
|
||
"sodium": nutr.get("sodium", 0) or 0,
|
||
}
|
||
return None
|
||
except Exception as exc: # broad – network / JSON issues
|
||
if attempt == max_retries - 1:
|
||
print(f"OpenFoodFacts error for '{query}': {exc}")
|
||
return None
|
||
continue
|
||
return None
|
||
|
||
|
||
class RecipeParser:
|
||
"""Parses a markdown recipe string into a structured format."""
|
||
|
||
def parse(self, text: str) -> ParsedRecipe:
|
||
"""Parse a markdown recipe string.
|
||
|
||
Returns a ParsedRecipe object with 'servings' and 'ingredients' attributes.
|
||
"""
|
||
import re
|
||
lines = text.strip().split('\n')
|
||
|
||
# Default values
|
||
servings = 4
|
||
ingredients = []
|
||
|
||
# Simple parsing: look for "Makes X servings" line
|
||
for line in lines:
|
||
line_lower = line.strip().lower()
|
||
if 'makes' in line_lower and 'serving' in line_lower:
|
||
# Extract number before 'servings'
|
||
match = re.search(r'(\d+)', line)
|
||
if match:
|
||
servings = int(match.group(1))
|
||
break
|
||
|
||
# Also look for ingredient lines (lines starting with '- ')
|
||
for line in lines:
|
||
stripped = line.strip()
|
||
if stripped.startswith('- '):
|
||
ingredients.append(stripped[2:])
|
||
|
||
return ParsedRecipe(servings=servings, ingredients=ingredients)
|
||
|
||
|
||
class Recipe:
|
||
"""Represents a parsed recipe with nutrition lookups."""
|
||
|
||
def __init__(self, parsed: ParsedRecipe, api: NutritionApi, servings: int = 4):
|
||
self.parsed = parsed
|
||
self.api = api
|
||
self.servings = parsed.servings if parsed else servings
|
||
self._nutrition = {}
|
||
|
||
def _get_ingredient_nutrition(self, ingredient: str) -> Optional[Dict]:
|
||
"""Look up nutrition for a single ingredient."""
|
||
# Try exact match first, then fall back
|
||
result = self.api.lookup(ingredient)
|
||
if result is not None:
|
||
self._nutrition[ingredient] = result
|
||
return result
|
||
return None
|
||
|
||
def summary(self) -> str:
|
||
"""Return a text summary of the recipe with nutrition info."""
|
||
lines = []
|
||
lines.append(f"Recipe Summary")
|
||
lines.append(f"Servings: {self.servings}")
|
||
lines.append("")
|
||
|
||
ingredients = self.parsed.ingredients
|
||
if ingredients:
|
||
lines.append("Ingredients:")
|
||
for ing in ingredients:
|
||
# Get just the ingredient name (before any measurements/units)
|
||
ing_name = ing.split(',')[0] if ',' in ing else ing.split()[0] if ing.split() else ing
|
||
nutrition = self._nutrition.get(ing_name)
|
||
if nutrition:
|
||
lines.append(f" - {ing_name}: "
|
||
f"Cal={nutrition.get('calories', '?')}, "
|
||
f"Pro={nutrition.get('protein', '?')}g, "
|
||
f"Fat={nutrition.get('fat', '?')}g, "
|
||
f"Carbs={nutrition.get('carbs', '?')}g")
|
||
else:
|
||
lines.append(f" - {ing_name}: nutrition pending")
|
||
else:
|
||
lines.append("No ingredients found")
|
||
|
||
return '\n'.join(lines)
|
||
|
||
def to_json(self) -> str:
|
||
"""Return a JSON representation of the recipe."""
|
||
import json as _json
|
||
data = {
|
||
'servings': self.servings,
|
||
'ingredients': self.parsed.ingredients,
|
||
'nutrition': self._nutrition,
|
||
}
|
||
return _json.dumps(data, indent=2)
|