Spaces:
Running
Running
| """ | |
| USDA FoodData Central API client. | |
| Uses DEMO_KEY by default (30 req/hr, no registration needed). | |
| Falls back gracefully when rate-limited. | |
| """ | |
| import os | |
| import requests | |
| from typing import Optional | |
| from src.cache import get_cached, set_cached | |
| USDA_BASE_URL = "https://api.nal.usda.gov/fdc/v1" | |
| USDA_API_KEY = os.environ.get("USDA_API_KEY", "DEMO_KEY") | |
| KEY_NUTRIENTS = { | |
| 1008: "Calories (kcal)", 1003: "Protein (g)", 1004: "Total Fat (g)", | |
| 1005: "Carbohydrates (g)", 1079: "Fiber (g)", 2000: "Sugars (g)", | |
| 1093: "Sodium (mg)", 1087: "Calcium (mg)", 1089: "Iron (mg)", | |
| 1092: "Potassium (mg)", 1106: "Vitamin A (mcg)", 1162: "Vitamin C (mg)", | |
| 1114: "Vitamin D (mcg)", 1090: "Magnesium (mg)", 1095: "Zinc (mg)", | |
| 1253: "Cholesterol (mg)", 1258: "Saturated Fat (g)", | |
| } | |
| def search_food(query: str, max_results: int = 1) -> Optional[dict]: | |
| """Search USDA for a food item. Returns None on any failure.""" | |
| cache_key = f"usda:{query}" | |
| cached = get_cached("nutrition_cache", cache_key) | |
| if cached is not None: | |
| return cached | |
| try: | |
| resp = requests.get( | |
| f"{USDA_BASE_URL}/foods/search", | |
| params={ | |
| "api_key": USDA_API_KEY, "query": query, | |
| "pageSize": max_results, "dataType": "Foundation,SR Legacy", | |
| }, | |
| timeout=10, | |
| ) | |
| if resp.status_code == 429: | |
| print(f"USDA rate limit hit for '{query}'") | |
| return None | |
| resp.raise_for_status() | |
| data = resp.json() | |
| results = [] | |
| for food in data.get("foods", []): | |
| nutrients = {} | |
| for n in food.get("foodNutrients", []): | |
| nid = n.get("nutrientId") | |
| if nid in KEY_NUTRIENTS: | |
| nutrients[KEY_NUTRIENTS[nid]] = round(n.get("value", 0), 2) | |
| results.append({ | |
| "description": food.get("description", ""), | |
| "nutrients": nutrients, | |
| }) | |
| result = {"query": query, "foods": results} | |
| set_cached("nutrition_cache", cache_key, result, ttl_days=30) | |
| return result | |
| except Exception as e: | |
| print(f"USDA error for '{query}': {e}") | |
| return None | |
| def lookup_ingredients(ingredient_list: list[str]) -> dict: | |
| """ | |
| Look up nutritional data for ingredients. | |
| Returns dict mapping ingredient -> nutrition data (or None if lookup failed). | |
| """ | |
| results = {} | |
| api_failures = 0 | |
| for ingredient in ingredient_list: | |
| data = search_food(ingredient.strip()) | |
| if data and data.get("foods"): | |
| best = data["foods"][0] | |
| results[ingredient] = { | |
| "matched_food": best["description"], | |
| "nutrients": best["nutrients"], | |
| "source": "USDA FoodData Central", | |
| } | |
| else: | |
| api_failures += 1 | |
| results[ingredient] = None # signal fallback needed | |
| return results, api_failures | |