182 lines
6.5 KiB
Python
182 lines
6.5 KiB
Python
from uuid import uuid4
|
|
|
|
import httpx
|
|
from fastapi import APIRouter, HTTPException, Query, status
|
|
from pydantic import ValidationError
|
|
from sqlmodel import select
|
|
|
|
from app import llm
|
|
from app.access import product_barcode_accessible
|
|
from app.deps import CurrentUserDep, SessionDep
|
|
from app.models import FoodLog, Product
|
|
from app.schemas import (
|
|
NutritionEstimateRequest,
|
|
NutritionEstimateResponse,
|
|
ProductCreate,
|
|
ProductRead,
|
|
ProductSearchResult,
|
|
)
|
|
|
|
router = APIRouter(prefix="/api/products", tags=["products"])
|
|
|
|
OFF_API_URL = "https://world.openfoodfacts.org/api/v2/product/{barcode}.json"
|
|
|
|
ESTIMATE_SYSTEM_PROMPT = """Du bist ein Ernährungsexperte. Schätze die Nährwerte des vom Nutzer beschriebenen Lebensmittels oder Gerichts.
|
|
|
|
Antworte AUSSCHLIESSLICH mit einem JSON-Objekt in genau diesem Format:
|
|
{"name": "kurzer Produktname", "calories": 250.0, "carbs": 30.0, "protein": 12.0, "fat": 9.0, "sugar": 3.0, "fiber": 2.0, "saturated_fat": 3.5, "salt": 1.2, "portion_g": 350.0}
|
|
|
|
Regeln:
|
|
- Alle Nährwerte beziehen sich auf 100 g (calories in kcal, alles andere in Gramm).
|
|
- portion_g ist die geschätzte übliche Portionsgröße der Beschreibung in Gramm.
|
|
- name ist ein kurzer, gut lesbarer deutscher Produktname ohne Mengenangabe.
|
|
- Nutze realistische Durchschnittswerte. Keine Erklärungen, kein Text außerhalb des JSON."""
|
|
|
|
|
|
def _extract_macros(nutriments: dict) -> dict:
|
|
calories = nutriments.get("energy-kcal_100g")
|
|
if calories is None:
|
|
energy_kj = nutriments.get("energy_100g")
|
|
calories = energy_kj / 4.184 if energy_kj is not None else 0.0
|
|
|
|
return {
|
|
"calories": calories,
|
|
"carbs": nutriments.get("carbohydrates_100g", 0.0) or 0.0,
|
|
"protein": nutriments.get("proteins_100g", 0.0) or 0.0,
|
|
"fat": nutriments.get("fat_100g", 0.0) or 0.0,
|
|
"sugar": nutriments.get("sugars_100g", 0.0) or 0.0,
|
|
"fiber": nutriments.get("fiber_100g", 0.0) or 0.0,
|
|
"saturated_fat": nutriments.get("saturated-fat_100g", 0.0) or 0.0,
|
|
"salt": nutriments.get("salt_100g", 0.0) or 0.0,
|
|
}
|
|
|
|
|
|
async def _fetch_from_open_food_facts(barcode: str) -> Product | None:
|
|
try:
|
|
async with httpx.AsyncClient(timeout=10.0) as client:
|
|
response = await client.get(OFF_API_URL.format(barcode=barcode))
|
|
except httpx.RequestError:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail="Open Food Facts ist gerade nicht erreichbar",
|
|
)
|
|
|
|
if response.status_code != 200:
|
|
return None
|
|
|
|
data = response.json()
|
|
if data.get("status") != 1:
|
|
return None
|
|
|
|
product_data = data.get("product", {})
|
|
macros = _extract_macros(product_data.get("nutriments", {}))
|
|
name = product_data.get("product_name") or product_data.get("generic_name") or "Unbekanntes Produkt"
|
|
|
|
return Product(barcode=barcode, name=name, **macros)
|
|
|
|
|
|
@router.post("", response_model=ProductRead, status_code=status.HTTP_201_CREATED)
|
|
def create_product(product_in: ProductCreate, current_user: CurrentUserDep, session: SessionDep):
|
|
barcode = product_in.barcode or f"manual-{uuid4().hex[:12]}"
|
|
|
|
existing = session.exec(select(Product).where(Product.barcode == barcode)).first()
|
|
if existing:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Barcode existiert bereits")
|
|
|
|
product = Product(**{**product_in.model_dump(), "barcode": barcode})
|
|
session.add(product)
|
|
session.commit()
|
|
session.refresh(product)
|
|
return product
|
|
|
|
|
|
@router.post("/estimate", response_model=NutritionEstimateResponse)
|
|
def estimate_nutrition(estimate_in: NutritionEstimateRequest, current_user: CurrentUserDep):
|
|
if not llm.is_configured():
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail="KI-Schätzung ist auf diesem Server nicht konfiguriert",
|
|
)
|
|
|
|
try:
|
|
data = llm.complete_json(ESTIMATE_SYSTEM_PROMPT, estimate_in.description)
|
|
except llm.LLMUnavailableError:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail="KI-Schätzung gerade nicht erreichbar. Bitte später erneut versuchen.",
|
|
)
|
|
|
|
try:
|
|
return NutritionEstimateResponse(**data)
|
|
except (ValidationError, TypeError):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_502_BAD_GATEWAY,
|
|
detail="KI-Antwort war unbrauchbar. Bitte erneut versuchen oder Werte manuell eintragen.",
|
|
)
|
|
|
|
|
|
@router.get("/search", response_model=list[ProductSearchResult])
|
|
def search_products(
|
|
current_user: CurrentUserDep,
|
|
session: SessionDep,
|
|
q: str = Query(default=""),
|
|
limit: int = Query(default=20, ge=1, le=50),
|
|
):
|
|
query = select(Product)
|
|
if q:
|
|
query = query.where(Product.name.ilike(f"%{q}%"))
|
|
products = [
|
|
product
|
|
for product in session.exec(query).all()
|
|
if product_barcode_accessible(session, product.barcode, current_user.id)
|
|
]
|
|
|
|
recent_logs = session.exec(
|
|
select(FoodLog)
|
|
.where(FoodLog.user_id == current_user.id)
|
|
.order_by(FoodLog.timestamp.desc())
|
|
.limit(500)
|
|
).all()
|
|
last_used: dict[str, FoodLog] = {}
|
|
for log in recent_logs:
|
|
if log.barcode and log.barcode not in last_used:
|
|
last_used[log.barcode] = log
|
|
|
|
def sort_key(product: Product):
|
|
log = last_used.get(product.barcode)
|
|
if log:
|
|
return (0, -log.timestamp.timestamp())
|
|
return (1, product.name.lower())
|
|
|
|
products.sort(key=sort_key)
|
|
|
|
results = []
|
|
for product in products[:limit]:
|
|
log = last_used.get(product.barcode)
|
|
results.append(
|
|
ProductSearchResult(
|
|
**product.model_dump(),
|
|
last_amount_g=log.amount_g if log else None,
|
|
is_dish=product.barcode.startswith("dish-"),
|
|
)
|
|
)
|
|
return results
|
|
|
|
|
|
@router.get("/{barcode}", response_model=ProductRead)
|
|
async def get_product(barcode: str, current_user: CurrentUserDep, session: SessionDep):
|
|
product = session.exec(select(Product).where(Product.barcode == barcode)).first()
|
|
if product:
|
|
if not product_barcode_accessible(session, barcode, current_user.id):
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Produkt nicht gefunden")
|
|
return product
|
|
|
|
product = await _fetch_from_open_food_facts(barcode)
|
|
if product is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Produkt nicht gefunden")
|
|
|
|
session.add(product)
|
|
session.commit()
|
|
session.refresh(product)
|
|
return product
|