363 lines
14 KiB
Python
363 lines
14 KiB
Python
"""BAN YARO — KI Routes"""
|
|
from fastapi import APIRouter, Depends, HTTPException, Request, UploadFile, File
|
|
from pydantic import BaseModel
|
|
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
|
|
rasse: Optional[str] = None
|
|
alter: Optional[str] = None
|
|
|
|
|
|
@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.")
|
|
if len(req.problem) > 1000:
|
|
raise HTTPException(400, "Beschreibung zu lang (max. 1000 Zeichen).")
|
|
|
|
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=False,
|
|
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
|
|
dog_id: Optional[int] = None
|
|
dog_name: Optional[str] = None
|
|
rasse: Optional[str] = None
|
|
|
|
|
|
@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.")
|
|
if len(req.symptom) > 1000:
|
|
raise HTTPException(400, "Beschreibung zu lang (max. 1000 Zeichen).")
|
|
|
|
# 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
|
|
rasse: Optional[str] = None
|
|
alter: Optional[int] = None
|
|
mode: str = "tomorrow" # "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="claude-opus-4-7",
|
|
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,
|
|
}
|