rene
1ff66a7083
PYDANTIC max_length (38 Routen, ~400 Field-Constraints): Schützt vor DoS durch Riesen-Payloads (10MB Thread-Titel etc.). Pragmatische Limits: - Titel/Name: 200 · Beschreibung/Body: 10000 · Notiz: 5000 - Email: 254 (RFC 5321) · URL: 500 · Slug/Kategorie: 100 - Hund-Name/Rasse: 80 · Hund-Bio: 2000 Top-betroffen: forum.py, diary.py, health.py, dogs.py, expenses.py, notes.py, auth.py, profile.py. Manuelle len()-Checks in profile, chat, ki entfernt (jetzt durch Field abgedeckt). PYTEST COVERAGE (+19 Tests, 37 grün + 1 xfail): - test_security.py: require_owner (Places GET/PATCH/DELETE mit Fremduser → 403), JWT-Blacklist (Logout invalidiert Token), Login-Lockout (5 Fehlversuche → 429 + Retry-After Header) - test_race.py: Invoice-Counter (20 parallele Threads, alle unique), Founder-Number (atomare Vergabe, voll bei 100) - test_validation.py: Forum-Titel 30k Zeichen → 422, Diary-Text 50k → 422 (verifiziert Pydantic max_length-Sweep) A11Y (Tap-Targets ≥44×44 + Dark-Mode-Kontrast): - #header-user-btn 36→44px, .header-back 40→44, .header-menu-btn 40→44 - dog-profile Wrapped-Slider Prev/Next 40→44 - forum-Lightbox Close 40→44 - --c-text-muted Light: #B0A090 (2.37:1 FAIL) → #7F6B58 (4.74:1 PASS) - --c-text-muted Dark: #806A58 (3.58:1 FAIL) → #A08878 (5.46:1 PASS) - Branding-Farben unangetastet
447 lines
19 KiB
Python
447 lines
19 KiB
Python
"""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,
|
||
}
|