"""BAN YARO — Ernährungs-Routes""" import logging from fastapi import APIRouter, Depends, HTTPException, Request from pydantic import BaseModel, Field from typing import Optional from database import db from auth import get_current_user import ki as ki_module router = APIRouter() logger = logging.getLogger(__name__) # ------------------------------------------------------------------ # Schemas # ------------------------------------------------------------------ class FutterProfilUpdate(BaseModel): futter_typ: Optional[str] = Field(None, max_length=50) # trocken|nass|barf|mix marke: Optional[str] = Field(None, max_length=200) kcal_tag: Optional[int] = None portionen: Optional[int] = None notizen: Optional[str] = Field(None, max_length=5000) class KiBeratungRequest(BaseModel): frage: str = Field(..., min_length=3, max_length=2000) dog_name: Optional[str] = Field(None, max_length=80) rasse: Optional[str] = Field(None, max_length=80) alter: Optional[str] = Field(None, max_length=50) gewicht: Optional[float] = None aktiv: Optional[bool] = None # ------------------------------------------------------------------ # Hilfsfunktion: Zugriffsprüfung # ------------------------------------------------------------------ def _check_dog_access(conn, dog_id: int, user_id: int): row = conn.execute( "SELECT id FROM dogs WHERE id=? AND user_id=?", (dog_id, user_id) ).fetchone() if not row: raise HTTPException(404, "Hund nicht gefunden.") # ------------------------------------------------------------------ # GET /dogs/{dog_id}/ernaehrung # ------------------------------------------------------------------ @router.get("/{dog_id}/ernaehrung") async def get_ernaehrung(dog_id: int, user=Depends(get_current_user)): with db() as conn: _check_dog_access(conn, dog_id, user["id"]) row = conn.execute( "SELECT * FROM futter_profil WHERE dog_id=?", (dog_id,) ).fetchone() if not row: return {} return dict(row) # ------------------------------------------------------------------ # PUT /dogs/{dog_id}/ernaehrung # ------------------------------------------------------------------ @router.put("/{dog_id}/ernaehrung") async def put_ernaehrung(dog_id: int, body: FutterProfilUpdate, user=Depends(get_current_user)): with db() as conn: _check_dog_access(conn, dog_id, user["id"]) existing = conn.execute( "SELECT id FROM futter_profil WHERE dog_id=?", (dog_id,) ).fetchone() if existing: conn.execute(""" UPDATE futter_profil SET futter_typ=COALESCE(?, futter_typ), marke=COALESCE(?, marke), kcal_tag=COALESCE(?, kcal_tag), portionen=COALESCE(?, portionen), notizen=COALESCE(?, notizen), updated_at=datetime('now') WHERE dog_id=? """, (body.futter_typ, body.marke, body.kcal_tag, body.portionen, body.notizen, dog_id)) else: conn.execute(""" INSERT INTO futter_profil (dog_id, futter_typ, marke, kcal_tag, portionen, notizen) VALUES (?, ?, ?, ?, ?, ?) """, (dog_id, body.futter_typ, body.marke, body.kcal_tag, body.portionen or 2, body.notizen)) row = conn.execute( "SELECT * FROM futter_profil WHERE dog_id=?", (dog_id,) ).fetchone() return dict(row) # ------------------------------------------------------------------ # POST /dogs/{dog_id}/ernaehrung/ki-beratung # ------------------------------------------------------------------ @router.post("/{dog_id}/ernaehrung/ki-beratung") async def ki_ernaehrung(dog_id: int, body: KiBeratungRequest, request: Request, user=Depends(get_current_user)): if not body.frage or len(body.frage.strip()) < 3: raise HTTPException(400, "Bitte stelle eine Frage.") if len(body.frage) > 800: raise HTTPException(400, "Frage zu lang (max. 800 Zeichen).") with db() as conn: _check_dog_access(conn, dog_id, user["id"]) dog_name = body.dog_name or "unbekannt" rasse = body.rasse or "unbekannt" alter = body.alter or "unbekannt" gewicht = f"{body.gewicht} kg" if body.gewicht else "unbekannt" aktiv_str = "aktiv" if body.aktiv else "normal aktiv" system = ( "Du bist Ernährungsberater für Hunde. " "Antworte immer auf Deutsch, kurz und praktisch. " "Keine unnötigen Füllsätze. " "Weise bei ernsthaften Gesundheitsfragen immer auf den Tierarzt hin. " "Stelle keine medizinischen Diagnosen." ) prompt = ( f"Hund: {dog_name}, Rasse: {rasse}, Alter: {alter}, " f"Gewicht: {gewicht}, Aktivität: {aktiv_str}.\n\n" f"Frage: {body.frage.strip()}\n\n" "Antworte konkret und praktisch, maximal 200 Wörter." ) try: antwort = await ki_module.complete( prompt=prompt, system=system, max_tokens=500, requires_premium=False, user_id=user["id"], ) return {"antwort": antwort} except ki_module.KIUnavailableError as e: raise HTTPException(503, str(e)) except Exception: raise HTTPException(500, "KI momentan nicht verfügbar.") # ================================================================== # FUTTER-VERTRÄGLICHKEIT # ================================================================== REAKTION_TYPEN = { # Positiv "verdauung_gut": {"label": "Gute Verdauung", "kategorie": "positiv", "fenster_h": 8}, "energie_hoch": {"label": "Viel Energie", "kategorie": "positiv", "fenster_h": 12}, "fell_glaenzend": {"label": "Glänzendes Fell", "kategorie": "positiv", "fenster_h": 336}, # 2 Wochen # Gastrointestinal "erbrechen": {"label": "Erbrechen", "kategorie": "gastro_negativ", "fenster_h": 6}, "durchfall": {"label": "Durchfall", "kategorie": "gastro_negativ", "fenster_h": 8}, "blaehungen": {"label": "Blähungen", "kategorie": "gastro_negativ", "fenster_h": 6}, "weicher_stuhl": {"label": "Weicher Stuhl", "kategorie": "gastro_negativ", "fenster_h": 8}, "appetitlosigkeit":{"label": "Appetitlosigkeit", "kategorie": "gastro_negativ", "fenster_h": 12}, # Haut & Fell "juckreiz": {"label": "Juckreiz / Kratzen", "kategorie": "haut_negativ", "fenster_h": 72}, "haarausfall": {"label": "Haarausfall", "kategorie": "haut_negativ", "fenster_h": 336}, "stumpfes_fell": {"label": "Stumpfes Fell", "kategorie": "haut_negativ", "fenster_h": 336}, "schuppenbildung":{"label": "Schuppenbildung", "kategorie": "haut_negativ", "fenster_h": 168}, "roetungen": {"label": "Hautrötungen / Entzündung", "kategorie": "haut_negativ", "fenster_h": 72}, "pfotenlecken": {"label": "Pfoten lecken (chronisch)", "kategorie": "haut_negativ", "fenster_h": 168}, "ohrentzuendung": {"label": "Ohrentzündung", "kategorie": "haut_negativ", "fenster_h": 168}, "fettiges_fell": {"label": "Fettiges Fell / Seborrhö", "kategorie": "haut_negativ", "fenster_h": 336}, # Allgemeinbefinden "schlappheit": {"label": "Schlappheit / Apathie", "kategorie": "allgemein_negativ", "fenster_h": 12}, "nervositaet": {"label": "Nervosität / Unruhe", "kategorie": "allgemein_negativ", "fenster_h": 12}, "viel_trinken": {"label": "Ungewöhnlich viel trinken", "kategorie": "allgemein_negativ", "fenster_h": 24}, "sonstiges": {"label": "Sonstiges", "kategorie": "sonstiges", "fenster_h": 24}, } _POSITIV_KAT = {"positiv"} _NEGATIV_KAT = {"gastro_negativ", "haut_negativ", "allgemein_negativ"} _HAUT_HINWEIS = "Haut- & Fell-Symptome wie {label} entwickeln sich typischerweise über Wochen. Mindestens 4–6 Wochen Beobachtung empfohlen — auch nach einem Futterwechsel dauert eine Besserung 2–6 Wochen." _GASTRO_HINWEIS = "Magen-Darm-Symptome wie {label} treten meist innerhalb weniger Stunden auf. Wenn sie häufig wiederkehren, ist ein Tierarztbesuch empfohlen." class FutterEintragCreate(BaseModel): datum: str = Field(..., max_length=32) uhrzeit: str = Field(..., max_length=20) futter_name: str = Field(..., max_length=200) futter_typ: Optional[str] = Field("trockenfutter", max_length=50) menge_g: Optional[int] = None notiz: Optional[str] = Field(None, max_length=2000) class ReaktionCreate(BaseModel): datum: str = Field(..., max_length=32) uhrzeit: str = Field(..., max_length=20) reaktion_typ: str = Field(..., max_length=100) intensitaet: Optional[int] = 3 notiz: Optional[str] = Field(None, max_length=2000) # ------------------------------------------------------------------ # POST /dogs/{dog_id}/futter # ------------------------------------------------------------------ @router.post("/{dog_id}/futter") async def create_futter_eintrag(dog_id: int, body: FutterEintragCreate, user=Depends(get_current_user)): with db() as conn: _check_dog_access(conn, dog_id, user["id"]) cur = conn.execute(""" INSERT INTO futter_eintraege (dog_id, datum, uhrzeit, futter_name, futter_typ, menge_g, notiz) VALUES (?, ?, ?, ?, ?, ?, ?) """, (dog_id, body.datum, body.uhrzeit, body.futter_name, body.futter_typ or "trockenfutter", body.menge_g, body.notiz)) row = conn.execute( "SELECT * FROM futter_eintraege WHERE id=?", (cur.lastrowid,) ).fetchone() return dict(row) # ------------------------------------------------------------------ # GET /dogs/{dog_id}/futter # ------------------------------------------------------------------ @router.get("/{dog_id}/futter") async def list_futter_eintraege(dog_id: int, user=Depends(get_current_user)): with db() as conn: _check_dog_access(conn, dog_id, user["id"]) rows = conn.execute(""" SELECT * FROM futter_eintraege WHERE dog_id=? ORDER BY datum DESC, uhrzeit DESC LIMIT 50 """, (dog_id,)).fetchall() return [dict(r) for r in rows] # ------------------------------------------------------------------ # DELETE /dogs/{dog_id}/futter/{entry_id} # ------------------------------------------------------------------ @router.delete("/{dog_id}/futter/{entry_id}") async def delete_futter_eintrag(dog_id: int, entry_id: int, user=Depends(get_current_user)): with db() as conn: _check_dog_access(conn, dog_id, user["id"]) result = conn.execute( "DELETE FROM futter_eintraege WHERE id=? AND dog_id=?", (entry_id, dog_id) ) if result.rowcount == 0: raise HTTPException(404, "Eintrag nicht gefunden.") return {"ok": True} # ------------------------------------------------------------------ # POST /dogs/{dog_id}/futter/reaktion # ------------------------------------------------------------------ @router.post("/{dog_id}/futter/reaktion") async def create_reaktion(dog_id: int, body: ReaktionCreate, user=Depends(get_current_user)): with db() as conn: _check_dog_access(conn, dog_id, user["id"]) cur = conn.execute(""" INSERT INTO futter_reaktionen (dog_id, datum, uhrzeit, reaktion_typ, intensitaet, notiz) VALUES (?, ?, ?, ?, ?, ?) """, (dog_id, body.datum, body.uhrzeit, body.reaktion_typ, body.intensitaet or 3, body.notiz)) row = conn.execute( "SELECT * FROM futter_reaktionen WHERE id=?", (cur.lastrowid,) ).fetchone() return dict(row) # ------------------------------------------------------------------ # GET /dogs/{dog_id}/futter/reaktionen # ------------------------------------------------------------------ @router.get("/{dog_id}/futter/reaktionen") async def list_reaktionen(dog_id: int, user=Depends(get_current_user)): with db() as conn: _check_dog_access(conn, dog_id, user["id"]) rows = conn.execute(""" SELECT * FROM futter_reaktionen WHERE dog_id=? ORDER BY datum DESC, uhrzeit DESC LIMIT 50 """, (dog_id,)).fetchall() return [dict(r) for r in rows] # ------------------------------------------------------------------ # DELETE /dogs/{dog_id}/futter/reaktion/{react_id} # ------------------------------------------------------------------ @router.delete("/{dog_id}/futter/reaktion/{react_id}") async def delete_reaktion(dog_id: int, react_id: int, user=Depends(get_current_user)): with db() as conn: _check_dog_access(conn, dog_id, user["id"]) result = conn.execute( "DELETE FROM futter_reaktionen WHERE id=? AND dog_id=?", (react_id, dog_id) ) if result.rowcount == 0: raise HTTPException(404, "Reaktion nicht gefunden.") return {"ok": True} # ------------------------------------------------------------------ # GET /dogs/{dog_id}/futter/analyse # ------------------------------------------------------------------ @router.get("/{dog_id}/futter/analyse") async def futter_analyse(dog_id: int, user=Depends(get_current_user)): from datetime import datetime with db() as conn: _check_dog_access(conn, dog_id, user["id"]) eintraege = conn.execute( "SELECT * FROM futter_eintraege WHERE dog_id=? ORDER BY datum, uhrzeit", (dog_id,) ).fetchall() reaktionen = conn.execute( "SELECT * FROM futter_reaktionen WHERE dog_id=? ORDER BY datum, uhrzeit", (dog_id,) ).fetchall() def parse_ts(datum, uhrzeit): try: return datetime.fromisoformat(f"{datum}T{uhrzeit}") except Exception: return None # futter_name → {typ, mahlzeiten, positiv, negativ, kategorien: {kat: count}} futter_stats: dict = {} for e in eintraege: name = e["futter_name"] if name not in futter_stats: futter_stats[name] = { "name": name, "typ": e["futter_typ"], "mahlzeiten": 0, "positiv": 0, "negativ": 0, "kategorien": {}, } futter_stats[name]["mahlzeiten"] += 1 for r in reaktionen: r_ts = parse_ts(r["datum"], r["uhrzeit"]) if not r_ts: continue r_typ = r["reaktion_typ"] meta = REAKTION_TYPEN.get(r_typ, {"kategorie": "sonstiges", "fenster_h": 24}) kat = meta["kategorie"] fenster = meta["fenster_h"] # Mindestfenster 1h, maximales Fenster wie angegeben min_h = 1 for e in eintraege: e_ts = parse_ts(e["datum"], e["uhrzeit"]) if not e_ts: continue diff = (r_ts - e_ts).total_seconds() / 3600 if min_h <= diff <= fenster: name = e["futter_name"] if name not in futter_stats: continue if kat in _POSITIV_KAT: futter_stats[name]["positiv"] += 1 elif kat in _NEGATIV_KAT: futter_stats[name]["negativ"] += 1 # Kategorie-Zähler futter_stats[name]["kategorien"][kat] = \ futter_stats[name]["kategorien"].get(kat, 0) + 1 result_futter = [] for stats in futter_stats.values(): positiv = stats["positiv"] negativ = stats["negativ"] total = positiv + negativ if total == 0: score = 50 status = "neu" else: raw = (positiv - negativ * 2) / max(1, total) # raw liegt zwischen -2 und 1 → normieren auf 0-100 score = int(max(0, min(100, (raw + 2) / 3 * 100))) if score >= 60: status = "gut" elif score >= 30: status = "neutral" else: status = "problematisch" result_futter.append({ "name": stats["name"], "typ": stats["typ"], "mahlzeiten": stats["mahlzeiten"], "positiv": positiv, "negativ": negativ, "score": score, "status": status, "kategorien": stats["kategorien"], }) # Sortierung: problematisch → neutral → gut → neu, dann nach Score ORDER = {"problematisch": 0, "neutral": 1, "gut": 2, "neu": 3} result_futter.sort(key=lambda x: (ORDER.get(x["status"], 9), -x["score"])) # Hinweis ableiten: erstes problematisches Futter mit Haut/Gastro-Symptomen hinweis = None for f in result_futter: if f["status"] != "problematisch": continue kats = f["kategorien"] if kats.get("haut_negativ", 0) > 0: # Häufigstes Haut-Symptom finden haut_rxn = [ r["reaktion_typ"] for r in reaktionen if REAKTION_TYPEN.get(r["reaktion_typ"], {}).get("kategorie") == "haut_negativ" ] label = REAKTION_TYPEN.get(haut_rxn[0], {}).get("label", "Haut-Symptome") if haut_rxn else "Haut-Symptome" hinweis = _HAUT_HINWEIS.format(label=label) break if kats.get("gastro_negativ", 0) > 0: gastro_rxn = [ r["reaktion_typ"] for r in reaktionen if REAKTION_TYPEN.get(r["reaktion_typ"], {}).get("kategorie") == "gastro_negativ" ] label = REAKTION_TYPEN.get(gastro_rxn[0], {}).get("label", "Magen-Darm-Symptome") if gastro_rxn else "Magen-Darm-Symptome" hinweis = _GASTRO_HINWEIS.format(label=label) break # Kategorien-Übersicht über alle Reaktionen kategorien_gesamt: dict = {} for r in reaktionen: kat = REAKTION_TYPEN.get(r["reaktion_typ"], {}).get("kategorie", "sonstiges") kategorien_gesamt[kat] = kategorien_gesamt.get(kat, 0) + 1 return { "eintraege_count": len(eintraege), "reaktionen_count": len(reaktionen), "futter": result_futter, "kategorien": kategorien_gesamt, "hinweis": hinweis, }