from typing import Any import httpx from app.config import get_settings class OpenFoodFactsClient: def __init__(self) -> None: settings = get_settings() self.base_url = settings.openfoodfacts_base_url.rstrip("/") def search(self, query: str, limit: int = 5) -> list[dict[str, Any]]: with httpx.Client(timeout=20.0) as client: response = client.get( f"{self.base_url}/cgi/search.pl", params={ "search_terms": query, "search_simple": 1, "action": "process", "json": 1, "page_size": limit, "lc": "ru", }, ) response.raise_for_status() products = response.json().get("products") or [] out: list[dict[str, Any]] = [] for p in products[:limit]: nutriments = p.get("nutriments") or {} out.append( { "name": p.get("product_name") or p.get("product_name_ru") or query, "brand": p.get("brands", ""), "barcode": p.get("code"), "calories_per_100g": nutriments.get("energy-kcal_100g"), "protein_g_per_100g": nutriments.get("proteins_100g"), "fat_g_per_100g": nutriments.get("fat_100g"), "carbs_g_per_100g": nutriments.get("carbohydrates_100g"), } ) return out def get_by_barcode(self, barcode: str) -> dict[str, Any] | None: with httpx.Client(timeout=20.0) as client: response = client.get(f"{self.base_url}/api/v2/product/{barcode}.json") if response.status_code == 404: return None response.raise_for_status() product = response.json().get("product") if not product: return None nutriments = product.get("nutriments") or {} return { "name": product.get("product_name") or product.get("product_name_ru"), "barcode": barcode, "calories_per_100g": nutriments.get("energy-kcal_100g"), "protein_g_per_100g": nutriments.get("proteins_100g"), "fat_g_per_100g": nutriments.get("fat_100g"), "carbs_g_per_100g": nutriments.get("carbohydrates_100g"), }