"""BAN YARO — KI Routes""" from fastapi import APIRouter, Depends, HTTPException, Request, UploadFile, File from pydantic import BaseModel, Field from typing import Optional import ki as ki_module from auth import get_current_user from ratelimit import check as rl_check from database import db router = APIRouter() class TrainingRequest(BaseModel): problem: str = Field(..., min_length=10, max_length=1000) rasse: Optional[str] = Field(None, max_length=80) alter: Optional[str] = Field(None, max_length=50) @router.post("/training") async def ki_training(req: TrainingRequest, request: Request, user=Depends(get_current_user)): """KI-Trainingsberatung für individuelle Verhaltens- und Trainingsprobleme.""" rl_check(request, max_requests=10, window_seconds=3600, key="ki_training") if not req.problem or len(req.problem.strip()) < 10: raise HTTPException(400, "Bitte beschreibe das Problem genauer.") rasse = req.rasse or "unbekannt" alter = req.alter or "unbekannt" system = ( "Du bist ein erfahrener, zertifizierter Hundetrainer mit Schwerpunkt " "auf positiver Verstärkung und gewaltfreier Erziehung. " "Antworte immer auf Deutsch, konkret, verständlich und motivierend. " "Gib keine Ratschläge die Schmerz oder Zwang beinhalten. " "Wenn das Problem schwerwiegend ist (Aggression, starke Angst), " "empfehle professionellen Hundetrainer vor Ort zusätzlich." ) prompt = f"""Hund: {rasse}, {alter} alt. Problem: {req.problem.strip()} Bitte gib: 1. Eine kurze Einschätzung des Problems (1-2 Sätze) 2. 3-5 konkrete Trainingsschritte die ich heute starten kann 3. Was ich vermeiden sollte 4. Wann ich einen Profi hinzuziehen sollte (falls relevant) Schreibe klar und strukturiert, ohne unnötigen Fachjargon.""" try: result = await ki_module.complete( prompt=prompt, system=system, max_tokens=600, requires_premium=True, user_id=user["id"], ) return {"antwort": result} except ki_module.KIUnavailableError as e: raise HTTPException(503, str(e)) except Exception as e: raise HTTPException(500, "KI momentan nicht verfügbar.") # ------------------------------------------------------------------ # POST /ki/tierarzt — KI-Tierarztfragen # ------------------------------------------------------------------ class TierarztRequest(BaseModel): symptom: str = Field(..., min_length=5, max_length=1000) dog_id: Optional[int] = None dog_name: Optional[str] = Field(None, max_length=80) rasse: Optional[str] = Field(None, max_length=80) @router.post("/tierarzt") async def ki_tierarzt(req: TierarztRequest, request: Request, user=Depends(get_current_user)): """KI-Tierarztfragen: Symptombeschreibung → erste Einschätzung.""" if not req.symptom or len(req.symptom.strip()) < 5: raise HTTPException(400, "Bitte beschreibe das Symptom genauer.") # Rate-Limit: max 5 Anfragen pro User pro Tag with db() as conn: count = conn.execute( "SELECT COUNT(*) FROM ki_tierarzt_log " "WHERE user_id=? AND created_at >= datetime('now','-1 day')", (user["id"],) ).fetchone()[0] if count >= 5: raise HTTPException(429, "Tageslimit erreicht. Du kannst maximal 5 Tierarztfragen pro Tag stellen.") dog_name = req.dog_name or "unbekannt" rasse = req.rasse or "unbekannt" system = ( "Du bist ein erfahrener Tierarzt-Assistent für Hunde. " "Deine Aufgabe ist es, Hundebesitzern eine erste Orientierung zu geben — " "kein Ersatz für eine echte tierärztliche Untersuchung. " "Antworte immer auf Deutsch, klar und verständlich. " "Stelle keine medizinischen Diagnosen. " "Empfehle im Zweifel immer den Gang zum Tierarzt." ) prompt = f"""Hund: {dog_name}, Rasse: {rasse} Symptom: {req.symptom.strip()} Gib eine strukturierte, verständliche Einschätzung: 1. Mögliche Ursachen (2-3 wahrscheinlichste) 2. Was der Besitzer jetzt tun kann (Erstmaßnahmen) 3. Wann unbedingt zum Tierarzt (Dringlichkeit: beobachten / bald / sofort) Antworte auf Deutsch, klar und verständlich. Maximal 300 Wörter. Schreibe KEINE medizinischen Diagnosen und empfehle im Zweifel immer den Tierarzt.""" try: antwort = await ki_module.complete( prompt=prompt, system=system, max_tokens=600, requires_premium=False, user_id=user["id"], ) # Erfolg: Rate-Limit-Eintrag speichern with db() as conn: conn.execute( "INSERT INTO ki_tierarzt_log (user_id, dog_id) VALUES (?, ?)", (user["id"], req.dog_id) ) return {"antwort": antwort, "anfragen_heute": count + 1, "limit": 5} except ki_module.KIUnavailableError as e: raise HTTPException(503, str(e)) except HTTPException: raise except Exception: raise HTTPException(500, "KI momentan nicht verfügbar.") # ------------------------------------------------------------------ # Rate-Limit-Helfer für Rassen-Erkennung # ------------------------------------------------------------------ _RASSE_DAILY_LIMIT = 10 def _check_rasse_limit(user_id: int) -> int: """Gibt verbleibende Erkennungen zurück. Wirft HTTPException wenn Limit erreicht.""" with db() as conn: used = conn.execute( """SELECT COUNT(*) FROM ki_rasse_log WHERE user_id = ? AND created_at >= datetime('now', 'start of day')""", (user_id,) ).fetchone()[0] remaining = _RASSE_DAILY_LIMIT - used if remaining <= 0: raise HTTPException(429, f"Tageslimit erreicht ({_RASSE_DAILY_LIMIT} Erkennungen/Tag). Morgen wieder verfügbar.") return remaining def _log_rasse_request(user_id: int): with db() as conn: conn.execute( "INSERT INTO ki_rasse_log (user_id) VALUES (?)", (user_id,) ) # ------------------------------------------------------------------ # POST /ki/geburtstag — Geburtstags-Überraschungsideen (kostenlos für alle) # ------------------------------------------------------------------ class BirthdayRequest(BaseModel): dog_id: int name: str = Field(..., max_length=80) rasse: Optional[str] = Field(None, max_length=80) alter: Optional[int] = None mode: str = Field("tomorrow", max_length=20) # "tomorrow" | "today" @router.post("/geburtstag") async def ki_geburtstag(req: BirthdayRequest, request: Request, user=Depends(get_current_user)): """Kostenlose KI-Geburtstagsideen — kein Premium nötig, 1x/Tag, DB-gecacht.""" from datetime import date year = date.today().year mode = req.mode if req.mode in ("tomorrow", "today") else "tomorrow" # Aus DB-Cache zurückgeben wenn bereits generiert with db() as conn: cached = conn.execute( "SELECT content FROM bday_ki_cache WHERE dog_id=? AND year=? AND mode=?", (req.dog_id, year, mode) ).fetchone() if cached: return {"answer": cached["content"], "cached": True} name = req.name.strip()[:40] or "deinen Hund" rasse = req.rasse or None alter = req.alter rasse_str = f"({rasse})" if rasse else "" if mode == "today": # Aus Sicht des Hundes — was er sich für seinen Geburtstag vorstellt alter_str = f"{alter}. Geburtstag" if alter else "Geburtstag" system = ( "Du bist ein Hund und erzählst aus deiner eigenen Perspektive. " "Schreibe auf Deutsch, verspielt, liebevoll und mit Hundelogik. " "Verwende typische Hundegedanken: Fressen, Gassi, Schmusen, Spielen, Gerüche." ) prompt = ( f"Ich bin {name} {rasse_str} und heute ist mein {alter_str}! " f"Erzähl in meiner Stimme (als Hund), wie ich mir den perfekten Geburtstagstag vorgestellt habe — " f"von Morgen bis Abend. Was möchte ich erleben, fressen, riechen, spielen? " f"Ca. 150 Wörter, herzlich und humorvoll." ) else: # Überraschungsideen für morgen alter_str = f"{alter}. Geburtstag" if alter else "Geburtstag" system = ( "Du bist ein begeisterter Hundefreund mit vielen kreativen Ideen. " "Antworte auf Deutsch, herzlich, konkret und mit einer Prise Humor. " "Fokus auf praktische, umsetzbare Überraschungen." ) prompt = ( f"Morgen ist der {alter_str} von {name} {rasse_str}! " f"Was können wir {name} besonders gönnen? " f"Gib 5 konkrete, liebevolle Überraschungsideen — von einfach bis aufwendig, " f"jeweils mit einem Satz warum Hunde das lieben." ) try: answer = await ki_module.complete( system=system, prompt=prompt, max_tokens=600, requires_premium=False, user_id=user["id"], ) with db() as conn: conn.execute( "INSERT OR REPLACE INTO bday_ki_cache (dog_id, year, mode, content) VALUES (?,?,?,?)", (req.dog_id, year, mode, answer) ) return {"answer": answer, "cached": False} except ki_module.KIUnavailableError: raise HTTPException(503, "KI momentan nicht verfügbar.") # ------------------------------------------------------------------ # POST /ki/rasse-erkennung — Vision-basierte Rassenerkennung # ------------------------------------------------------------------ @router.post("/rasse-erkennung") async def ki_rasse_erkennung( request: Request, file: UploadFile = File(...), user=Depends(get_current_user), ): """Hunderassen per Foto erkennen (Claude Vision, max 5 MB, 10x/Tag).""" import base64 import json import re import anthropic # Dateigröße prüfen content = await file.read() if len(content) > 5 * 1024 * 1024: raise HTTPException(400, "Bild zu groß. Maximal 5 MB erlaubt.") # MIME-Typ prüfen ct = (file.content_type or "").lower() if not ct.startswith("image/"): raise HTTPException(400, "Nur Bilddateien erlaubt (JPG, PNG, WebP).") # MIME-Typ auf erlaubte Werte beschränken allowed_mimes = {"image/jpeg", "image/png", "image/webp", "image/gif"} mime_type = ct if ct in allowed_mimes else "image/jpeg" # Rate-Limit prüfen remaining_before = _check_rasse_limit(user["id"]) # Anthropic-Key zur Laufzeit prüfen (nicht nur beim Modulstart) import os as _os api_key = _os.getenv("ANTHROPIC_KEY") or ki_module.ANTHROPIC_KEY if not api_key: raise HTTPException(503, "KI-Bildanalyse ist momentan nicht verfügbar.") base64_data = base64.standard_b64encode(content).decode("utf-8") prompt_text = """Analysiere dieses Bild und erkenne die Hunderasse(n). Antworte NUR im folgenden JSON-Format (kein anderer Text): { "rassen": [ {"name": "Labrador Retriever", "sicherheit": 85, "beschreibung": "Kurze Begründung"}, {"name": "Golden Retriever", "sicherheit": 12, "beschreibung": "Falls Mischling"} ], "ist_hund": true, "hinweis": "Optionaler Hinweis z.B. bei Welpen oder schlechter Bildqualität" } Gib 1-3 Rassen nach Wahrscheinlichkeit sortiert an. Sicherheit in Prozent (0-100). Falls kein Hund erkennbar: ist_hund=false und leeres rassen-Array.""" try: def _sync_call(): client = anthropic.Anthropic(api_key=api_key) return client.messages.create( model=ki_module.VISION_MODEL, max_tokens=500, messages=[{ "role": "user", "content": [ { "type": "image", "source": { "type": "base64", "media_type": mime_type, "data": base64_data, } }, { "type": "text", "text": prompt_text, } ] }] ) import asyncio response = await asyncio.get_event_loop().run_in_executor(None, _sync_call) raw = response.content[0].text.strip() except anthropic.APIError as e: raise HTTPException(503, f"KI-Bildanalyse nicht verfügbar: {e}") except Exception as e: raise HTTPException(500, "Fehler bei der Bildanalyse.") # JSON parsen — Claude kann manchmal ```json ... ``` wrappen cleaned = re.sub(r"^```(?:json)?\s*|\s*```$", "", raw, flags=re.DOTALL).strip() try: parsed = json.loads(cleaned) except json.JSONDecodeError: raise HTTPException(500, "KI-Antwort konnte nicht verarbeitet werden.") # Usage loggen (erst nach erfolgreicher KI-Antwort) _log_rasse_request(user["id"]) remaining_after = remaining_before - 1 # Wiki-Slugs für erkannte Rassen nachschlagen rassen = parsed.get("rassen", []) if rassen: with db() as conn: for r in rassen: name = r.get("name", "") # Exakter Name-Match (case-insensitive) row = conn.execute( "SELECT slug FROM wiki_rassen WHERE LOWER(name) = LOWER(?)", (name,) ).fetchone() r["wiki_slug"] = row["slug"] if row else None return { "rassen": rassen, "ist_hund": parsed.get("ist_hund", False), "hinweis": parsed.get("hinweis") or None, "verbleibende_anfragen": remaining_after, } # ------------------------------------------------------------------ # POST /ki/abschied — Persönlicher Abschiedstext für verstorbenen Hund # ------------------------------------------------------------------ class AbschiedRequest(BaseModel): dog_id: int name: str = Field(..., max_length=80) rasse: Optional[str] = Field(None, max_length=80) km_total: Optional[float] = None diary_count: Optional[int] = None gemeinsam_tage: Optional[int] = None last_entry_titel: Optional[str] = Field(None, max_length=200) @router.post("/abschied") async def ki_abschied(req: AbschiedRequest, request: Request, user=Depends(get_current_user)): """Persönlicher Abschiedstext — einmalig generiert, DB-gecacht.""" with db() as conn: cached = conn.execute( "SELECT content FROM bday_ki_cache WHERE dog_id=? AND year=9999 AND mode='abschied'", (req.dog_id,) ).fetchone() if cached: return {"text": cached["content"], "cached": True} name = req.name.strip()[:40] rasse = req.rasse or "" km = f"{req.km_total:.0f} km" if req.km_total else None tage = f"{req.gemeinsam_tage} gemeinsame Tage" if req.gemeinsam_tage else None eintr = f"{req.diary_count} Tagebucheinträge" if req.diary_count else None stats_str = ", ".join(filter(None, [km, tage, eintr])) rasse_str = f" ({rasse})" if rasse else "" system = ( "Du bist ein einfühlsamer Begleiter für Menschen in Trauer um ihren Hund. " "Schreibe warmherzig, persönlich und respektvoll auf Deutsch. " "Keine Floskeln, kein Kitsch — echte Wärme. " "Erwähne die Statistiken natürlich eingebunden." ) prompt = ( f"{name}{rasse_str} ist über die Regenbogenbrücke gegangen. " f"Schreibe einen kurzen, persönlichen Abschiedstext (ca. 80–100 Wörter) " f"der die Verbundenheit würdigt. " f"Statistiken: {stats_str or 'nicht bekannt'}. " f"Sei warm, nicht sentimental überladen. Schließe mit einem hoffnungsvollen Gedanken." ) try: text = await ki_module.complete( system=system, prompt=prompt, max_tokens=300, requires_premium=False, user_id=user["id"], ) with db() as conn: conn.execute( "INSERT OR REPLACE INTO bday_ki_cache (dog_id, year, mode, content) VALUES (?,9999,'abschied',?)", (req.dog_id, text) ) return {"text": text, "cached": False} except Exception as e: raise HTTPException(503, str(e))