pyr0ball
c166e5216a
Pantry tracker app with: - FastAPI backend + Vue 3 SPA frontend - SQLite via circuitforge-core (migrations 001-005) - Inventory CRUD, barcode scan, receipt OCR pipeline - Expiry prediction (deterministic + LLM fallback) - CF-core tier system integration - Cloud session support (menagerie)
234 lines
7.4 KiB
Python
234 lines
7.4 KiB
Python
"""
|
|
OpenFoodFacts API integration service.
|
|
|
|
This module provides functionality to look up product information
|
|
from the OpenFoodFacts database using barcodes (UPC/EAN).
|
|
"""
|
|
|
|
import httpx
|
|
from typing import Optional, Dict, Any
|
|
from app.core.config import settings
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class OpenFoodFactsService:
|
|
"""
|
|
Service for interacting with the OpenFoodFacts API.
|
|
|
|
OpenFoodFacts is a free, open database of food products with
|
|
ingredients, allergens, and nutrition facts.
|
|
"""
|
|
|
|
BASE_URL = "https://world.openfoodfacts.org/api/v2"
|
|
USER_AGENT = "Kiwi/0.1.0 (https://circuitforge.tech)"
|
|
|
|
async def lookup_product(self, barcode: str) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Look up a product by barcode in the OpenFoodFacts database.
|
|
|
|
Args:
|
|
barcode: UPC/EAN barcode (8-13 digits)
|
|
|
|
Returns:
|
|
Dictionary with product information, or None if not found
|
|
|
|
Example response:
|
|
{
|
|
"name": "Organic Milk",
|
|
"brand": "Horizon",
|
|
"categories": ["Dairy", "Milk"],
|
|
"image_url": "https://...",
|
|
"nutrition_data": {...},
|
|
"raw_data": {...} # Full API response
|
|
}
|
|
"""
|
|
try:
|
|
async with httpx.AsyncClient() as client:
|
|
url = f"{self.BASE_URL}/product/{barcode}.json"
|
|
|
|
response = await client.get(
|
|
url,
|
|
headers={"User-Agent": self.USER_AGENT},
|
|
timeout=10.0,
|
|
)
|
|
|
|
if response.status_code == 404:
|
|
logger.info(f"Product not found in OpenFoodFacts: {barcode}")
|
|
return None
|
|
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
|
|
if data.get("status") != 1:
|
|
logger.info(f"Product not found in OpenFoodFacts: {barcode}")
|
|
return None
|
|
|
|
return self._parse_product_data(data, barcode)
|
|
|
|
except httpx.HTTPError as e:
|
|
logger.error(f"HTTP error looking up barcode {barcode}: {e}")
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"Error looking up barcode {barcode}: {e}")
|
|
return None
|
|
|
|
def _parse_product_data(self, data: Dict[str, Any], barcode: str) -> Dict[str, Any]:
|
|
"""
|
|
Parse OpenFoodFacts API response into our product format.
|
|
|
|
Args:
|
|
data: Raw API response
|
|
barcode: Original barcode
|
|
|
|
Returns:
|
|
Parsed product dictionary
|
|
"""
|
|
product = data.get("product", {})
|
|
|
|
# Extract basic info
|
|
name = (
|
|
product.get("product_name")
|
|
or product.get("product_name_en")
|
|
or f"Unknown Product ({barcode})"
|
|
)
|
|
|
|
brand = product.get("brands", "").split(",")[0].strip() if product.get("brands") else None
|
|
|
|
# Categories (comma-separated string to list)
|
|
categories_str = product.get("categories", "")
|
|
categories = [c.strip() for c in categories_str.split(",") if c.strip()]
|
|
category = categories[0] if categories else None
|
|
|
|
# Description
|
|
description = product.get("generic_name") or product.get("generic_name_en")
|
|
|
|
# Image
|
|
image_url = product.get("image_url") or product.get("image_front_url")
|
|
|
|
# Nutrition data
|
|
nutrition_data = self._extract_nutrition_data(product)
|
|
|
|
# Allergens and dietary info
|
|
allergens = product.get("allergens_tags", [])
|
|
labels = product.get("labels_tags", [])
|
|
|
|
return {
|
|
"name": name,
|
|
"brand": brand,
|
|
"category": category,
|
|
"categories": categories,
|
|
"description": description,
|
|
"image_url": image_url,
|
|
"nutrition_data": nutrition_data,
|
|
"allergens": allergens,
|
|
"labels": labels,
|
|
"raw_data": product, # Store full response for debugging
|
|
}
|
|
|
|
def _extract_nutrition_data(self, product: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""
|
|
Extract nutrition facts from product data.
|
|
|
|
Args:
|
|
product: Product data from OpenFoodFacts
|
|
|
|
Returns:
|
|
Dictionary of nutrition facts
|
|
"""
|
|
nutriments = product.get("nutriments", {})
|
|
|
|
# Extract common nutrients (per 100g)
|
|
nutrition = {}
|
|
|
|
# Energy
|
|
if "energy-kcal_100g" in nutriments:
|
|
nutrition["calories"] = nutriments["energy-kcal_100g"]
|
|
elif "energy_100g" in nutriments:
|
|
# Convert kJ to kcal (1 kcal = 4.184 kJ)
|
|
nutrition["calories"] = round(nutriments["energy_100g"] / 4.184, 1)
|
|
|
|
# Macronutrients
|
|
if "fat_100g" in nutriments:
|
|
nutrition["fat_g"] = nutriments["fat_100g"]
|
|
if "saturated-fat_100g" in nutriments:
|
|
nutrition["saturated_fat_g"] = nutriments["saturated-fat_100g"]
|
|
if "carbohydrates_100g" in nutriments:
|
|
nutrition["carbohydrates_g"] = nutriments["carbohydrates_100g"]
|
|
if "sugars_100g" in nutriments:
|
|
nutrition["sugars_g"] = nutriments["sugars_100g"]
|
|
if "fiber_100g" in nutriments:
|
|
nutrition["fiber_g"] = nutriments["fiber_100g"]
|
|
if "proteins_100g" in nutriments:
|
|
nutrition["protein_g"] = nutriments["proteins_100g"]
|
|
|
|
# Minerals
|
|
if "salt_100g" in nutriments:
|
|
nutrition["salt_g"] = nutriments["salt_100g"]
|
|
elif "sodium_100g" in nutriments:
|
|
# Convert sodium to salt (1g sodium = 2.5g salt)
|
|
nutrition["salt_g"] = round(nutriments["sodium_100g"] * 2.5, 2)
|
|
|
|
# Serving size
|
|
if "serving_size" in product:
|
|
nutrition["serving_size"] = product["serving_size"]
|
|
|
|
return nutrition
|
|
|
|
async def search_products(
|
|
self,
|
|
query: str,
|
|
page: int = 1,
|
|
page_size: int = 20
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Search for products by name in OpenFoodFacts.
|
|
|
|
Args:
|
|
query: Search query
|
|
page: Page number (1-indexed)
|
|
page_size: Number of results per page
|
|
|
|
Returns:
|
|
Dictionary with search results and metadata
|
|
"""
|
|
try:
|
|
async with httpx.AsyncClient() as client:
|
|
url = f"{self.BASE_URL}/search"
|
|
|
|
response = await client.get(
|
|
url,
|
|
params={
|
|
"search_terms": query,
|
|
"page": page,
|
|
"page_size": page_size,
|
|
"json": 1,
|
|
},
|
|
headers={"User-Agent": self.USER_AGENT},
|
|
timeout=10.0,
|
|
)
|
|
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
|
|
products = [
|
|
self._parse_product_data({"product": p}, p.get("code", ""))
|
|
for p in data.get("products", [])
|
|
]
|
|
|
|
return {
|
|
"products": products,
|
|
"count": data.get("count", 0),
|
|
"page": data.get("page", page),
|
|
"page_size": data.get("page_size", page_size),
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error searching OpenFoodFacts: {e}")
|
|
return {
|
|
"products": [],
|
|
"count": 0,
|
|
"page": page,
|
|
"page_size": page_size,
|
|
}
|