"""BAN YARO — Übungs- & Trainingsfortschritt""" from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel, Field from typing import Optional import datetime import ki from database import db from auth import get_current_user, require_admin from cache import ttl_cache router = APIRouter() # ------------------------------------------------------------------ # Alle Übungen aus DB (öffentlich, kein Auth) # Statische Daten → 1h TTL-Cache. Wird in update_exercise() invalidiert. # ------------------------------------------------------------------ @ttl_cache(ttl=3600) def _load_exercises_by_tab() -> dict: import json as _json CAT_TO_TAB = { 'Grundkommando': 'grundkommandos', 'Trick': 'tricks', 'Problemverhalten': 'problemverhalten', 'Mentale Auslastung': 'mentale-auslastung', 'Hundesport': 'hundesport', 'Koerperpflege': 'koerperpflege', 'Körperpflege': 'koerperpflege', 'Welpe Basics': 'welpe-basics', } with db() as conn: rows = conn.execute(""" SELECT exercise_id, js_exercise_id, name, kategorie, schwierigkeit, alter_ab, dauer, beschreibung, schritte, tipp FROM training_exercises ORDER BY kategorie, name """).fetchall() by_tab: dict = {} for r in rows: tab = CAT_TO_TAB.get(r['kategorie'], r['kategorie'].lower().replace(' ', '-')) by_tab.setdefault(tab, []).append({ 'exercise_id': r['exercise_id'], 'js_exercise_id': r['js_exercise_id'], 'name': r['name'], 'kategorie': tab, 'schwierigkeit': r['schwierigkeit'] or 'mittel', 'alter': r['alter_ab'], 'dauer': r['dauer'], 'beschreibung': r['beschreibung'], 'schritte': _json.loads(r['schritte'] or '[]'), 'tipp': r['tipp'], }) return by_tab @router.get("/exercises") async def get_exercises(): """Alle Übungen aus der DB, gruppiert nach Tab-ID (1h-Cache).""" return _load_exercises_by_tab() # ------------------------------------------------------------------ # Admin: Übung bearbeiten (beschreibung / schritte / tipp) # ------------------------------------------------------------------ class ExerciseUpdate(BaseModel): beschreibung: Optional[str] = Field(None, max_length=10000) schritte: Optional[str] = Field(None, max_length=10000) # JSON-String: '["Schritt 1", ...]' tipp: Optional[str] = Field(None, max_length=5000) @router.put("/exercises/{exercise_id}") async def update_exercise(exercise_id: int, body: ExerciseUpdate, _=Depends(require_admin)): """Partial update von beschreibung/schritte/tipp einer Übung (nur Admin).""" with db() as conn: row = conn.execute( "SELECT id FROM training_exercises WHERE id=?", (exercise_id,) ).fetchone() if not row: raise HTTPException(status_code=404, detail="Übung nicht gefunden") fields, vals = [], [] if body.beschreibung is not None: fields.append("beschreibung=?"); vals.append(body.beschreibung) if body.schritte is not None: fields.append("schritte=?"); vals.append(body.schritte) if body.tipp is not None: fields.append("tipp=?"); vals.append(body.tipp) if not fields: return {"ok": True, "updated": 0} vals.append(exercise_id) conn.execute(f"UPDATE training_exercises SET {', '.join(fields)} WHERE id=?", vals) # Cache invalidieren, damit der Admin-Edit sofort sichtbar wird _load_exercises_by_tab.cache_clear() return {"ok": True, "updated": len(fields)} # ------------------------------------------------------------------ # Übungs-Status # ------------------------------------------------------------------ class ProgressUpdate(BaseModel): exercise_id: str = Field(..., max_length=200) status: Optional[str] = Field(None, max_length=50) dog_id: Optional[int] = None @router.get("/progress") async def get_progress(dog_id: Optional[int] = None, user=Depends(get_current_user)): uid = user["id"] with db() as conn: if dog_id: rows = conn.execute( "SELECT exercise_id, status, updated_at FROM exercise_progress WHERE dog_id=?", (dog_id,) ).fetchall() else: rows = conn.execute( "SELECT exercise_id, status, updated_at FROM exercise_progress WHERE user_id=?", (uid,) ).fetchall() return [dict(r) for r in rows] @router.post("/progress") async def upsert_progress(body: ProgressUpdate, user=Depends(get_current_user)): uid = user["id"] with db() as conn: if body.dog_id: conn.execute(""" INSERT INTO exercise_progress (user_id, dog_id, exercise_id, status) VALUES (?,?,?,?) ON CONFLICT(dog_id, exercise_id) DO UPDATE SET status=excluded.status, updated_at=datetime('now') """, (uid, body.dog_id, body.exercise_id, body.status)) else: conn.execute(""" INSERT INTO exercise_progress (user_id, exercise_id, status) VALUES (?,?,?) ON CONFLICT(dog_id, exercise_id) DO UPDATE SET status=excluded.status, updated_at=datetime('now') """, (uid, body.exercise_id, body.status)) return {"ok": True} # ------------------------------------------------------------------ # Trainingsplan-Checkboxen # ------------------------------------------------------------------ class PlanProgress(BaseModel): item_key: str = Field(..., max_length=200) checked: bool dog_id: Optional[int] = None @router.get("/plan-progress") async def get_plan_progress(dog_id: Optional[int] = None, user=Depends(get_current_user)): uid = user["id"] with db() as conn: if dog_id: rows = conn.execute( "SELECT item_key, checked FROM training_plan_progress WHERE dog_id=?", (dog_id,) ).fetchall() else: rows = conn.execute( "SELECT item_key, checked FROM training_plan_progress WHERE user_id=?", (uid,) ).fetchall() return [dict(r) for r in rows] @router.post("/plan-progress") async def upsert_plan_progress(body: PlanProgress, user=Depends(get_current_user)): uid = user["id"] with db() as conn: if body.checked: conn.execute(""" INSERT OR REPLACE INTO training_plan_progress (user_id, dog_id, item_key, checked) VALUES (?,?,?,1) """, (uid, body.dog_id, body.item_key)) else: conn.execute( "DELETE FROM training_plan_progress WHERE dog_id=? AND item_key=?", (body.dog_id, body.item_key) ) return {"ok": True} # ------------------------------------------------------------------ # Empfehlungen (rule-based) # ------------------------------------------------------------------ GRUNDKOMMANDOS_ORDER = ['Sitz', 'Platz', 'Bleib', 'Hier / Komm', 'Fuß', 'Aus / Lass es', 'Warte'] TRICKS_FIRST = ['Pfote / Schütteln', 'Dreh', 'Auf die Decke', 'Nasenarbeit / Suchen'] @router.get("/suggestions") async def get_suggestions(dog_id: Optional[int] = None, user=Depends(get_current_user)): uid = user["id"] with db() as conn: if dog_id: rows = conn.execute( "SELECT exercise_id, status FROM exercise_progress WHERE dog_id=?", (dog_id,) ).fetchall() else: rows = conn.execute( "SELECT exercise_id, status FROM exercise_progress WHERE user_id=?", (uid,) ).fetchall() progress = {r["exercise_id"]: r["status"] for r in rows} def key(name): return f"grundkommandos_{name.replace(' ', '_').replace('/', '')}" def tkey(name): return f"tricks_{name.replace(' ', '_').replace('/', '')}" suggestions = [] # Noch-nicht Übungen — direkte Hilfe stuck = [n for n in GRUNDKOMMANDOS_ORDER if progress.get(key(n)) == 'noch-nicht'] if stuck: suggestions.append({ "type": "help", "icon": "warning", "title": f'\u201e{stuck[0]}\u201c klappt noch nicht', "text": "Mach einen Schritt zurück: Kürzere Einheiten, mehr Leckerlis, weniger Ablenkung. Schau dir die Trainingsgrundlagen an.", "action_tab": "grundkommandos", "action_name": stuck[0], }) # Manchmal-Übungen — intensivieren almost = [n for n in GRUNDKOMMANDOS_ORDER if progress.get(key(n)) == 'manchmal'] if almost: suggestions.append({ "type": "boost", "icon": "fire", "title": f'Fast da: \u201e{almost[0]}\u201c', "text": "Du bist auf dem richtigen Weg! Übe täglich 3–5 Minuten — dann sitzt es bald.", "action_tab": "grundkommandos", "action_name": almost[0], }) # Nächste ungestartete Grundkommando grundk_mastered = [n for n in GRUNDKOMMANDOS_ORDER if progress.get(key(n)) == 'sitzt'] next_up = next((n for n in GRUNDKOMMANDOS_ORDER if progress.get(key(n)) in (None, 'noch-nicht') and n not in stuck), None) if len(grundk_mastered) == len(GRUNDKOMMANDOS_ORDER): # Alle Grundkommandos gemeistert → Tricks empfehlen next_trick = next((n for n in TRICKS_FIRST if progress.get(tkey(n)) is None), None) if next_trick: suggestions.append({ "type": "next", "icon": "star", "title": "Alle Grundkommandos sitzen! 🎉", "text": f'Zeit für Tricks! Starte mit \u201e{next_trick}\u201c \u2014 Spaß garantiert.', "action_tab": "tricks", "action_name": next_trick, }) elif next_up and not almost: suggestions.append({ "type": "next", "icon": "arrow-right", "title": f"Bereit für den nächsten Schritt?", "text": f'Starte jetzt mit \u201e{next_up}\u201c. {"Du hast bereits " + str(len(grundk_mastered)) + " Grundkommandos gemeistert!" if grundk_mastered else "Es ist die perfekte Basis für alles Weitere."}', "action_tab": "grundkommandos", "action_name": next_up, }) elif not progress: # Noch gar kein Fortschritt suggestions.append({ "type": "start", "icon": "flag", "title": "Womit fangen wir an?", "text": "\u201eSitz\u201c ist das erste Kommando für jeden Hund \u2014 einfach, schnell erlernt und die Basis für alles.", "action_tab": "grundkommandos", "action_name": "Sitz", }) return suggestions[:2] # max 2 Empfehlungen # ------------------------------------------------------------------ # Training: Session-Protokoll # ------------------------------------------------------------------ STIMMUNGS_LABELS = { "aufmerksam": "aufmerksam", "muede": "müde", "abgelenkt": "abgelenkt", "super": "super motiviert", } TRAINING_BADGES = [ ("training_first", 1, "Erste Trainingseinheit", "Ihr habt gemeinsam die erste Einheit abgeschlossen \U0001f43e"), ("training_5", 5, "5 Einheiten", "5 Trainingseinheiten \u2014 ihr seid dabei!"), ("training_10", 10, "10 Einheiten", "10 Einheiten! {name} macht gro\u00dfartige Fortschritte."), ("training_25", 25, "25 Einheiten", "25 Einheiten \u2014 eine echte Trainingspartnerschaft!"), ("training_50", 50, "50 Einheiten", "50 Einheiten! {name} und du seid ein echtes Team."), ("training_top_5", 5, "5 Top-Trainings", "5 Top-Trainings \u2014 {name} ist ein Schnellerner!"), ] def _check_badges(conn, user_id: int, dog_name: str) -> list: """Prüft und vergibt Trainings-Badges. Gibt neu verdiente Badges zurück.""" total = conn.execute( "SELECT COUNT(*) FROM training_sessions WHERE user_id=?", (user_id,) ).fetchone()[0] top_count = conn.execute( "SELECT COUNT(*) FROM training_sessions WHERE user_id=? AND ist_top=1", (user_id,) ).fetchone()[0] existing = { r[0] for r in conn.execute( "SELECT badge_id FROM user_badges WHERE user_id=?", (user_id,) ).fetchall() } new_badges = [] for badge_id, threshold, title, desc_tpl in TRAINING_BADGES: if badge_id in existing: continue count = top_count if badge_id == "training_top_5" else total if count >= threshold: desc = desc_tpl.format(name=dog_name) conn.execute( "INSERT OR IGNORE INTO user_badges (user_id, badge_id) VALUES (?,?)", (user_id, badge_id) ) new_badges.append({"id": badge_id, "title": title, "desc": desc}) return new_badges class SessionCreate(BaseModel): dog_id: int exercise_id: str = Field(..., max_length=200) exercise_name: str = Field(..., max_length=200) datum: Optional[str] = Field(None, max_length=32) wiederholungen: int = 1 erfolgsquote: int = 50 hund_stimmung: Optional[str] = Field("aufmerksam", max_length=50) zufriedenheit: Optional[int] = 3 notiz: Optional[str] = Field(None, max_length=2000) tagebuch_eintrag: bool = False # ignoriert — Training hat eigenes Protokoll @router.post("/sessions") async def log_session(body: SessionCreate, user=Depends(get_current_user)): uid = user["id"] with db() as conn: # Hund-Zugehörigkeit prüfen dog = conn.execute( "SELECT id, name FROM dogs WHERE id=? AND user_id=?", (body.dog_id, uid) ).fetchone() if not dog: raise HTTPException(404, "Hund nicht gefunden.") dog_name = dog["name"] datum = body.datum or datetime.date.today().isoformat() ist_top = int(body.erfolgsquote >= 80 and body.zufriedenheit >= 4) cur = conn.execute( """ INSERT INTO training_sessions (user_id, dog_id, exercise_id, exercise_name, datum, wiederholungen, erfolgsquote, hund_stimmung, zufriedenheit, notiz, ist_top) VALUES (?,?,?,?,?,?,?,?,?,?,?) """, (uid, body.dog_id, body.exercise_id, body.exercise_name, datum, body.wiederholungen, body.erfolgsquote, body.hund_stimmung, body.zufriedenheit, body.notiz, ist_top) ) session_id = cur.lastrowid # Badges prüfen new_badges = _check_badges(conn, uid, dog_name) session = { "id": session_id, "user_id": uid, "dog_id": body.dog_id, "exercise_id": body.exercise_id, "exercise_name": body.exercise_name, "datum": datum, "wiederholungen": body.wiederholungen, "erfolgsquote": body.erfolgsquote, "hund_stimmung": body.hund_stimmung, "zufriedenheit": body.zufriedenheit, "notiz": body.notiz, "ist_top": bool(ist_top), } return { "session": session, "ist_top": bool(ist_top), "badges": new_badges, } @router.get("/sessions") async def get_sessions( dog_id: int, limit: int = 50, offset: int = 0, user=Depends(get_current_user) ): uid = user["id"] with db() as conn: dog = conn.execute( "SELECT id FROM dogs WHERE id=? AND user_id=?", (dog_id, uid) ).fetchone() if not dog: raise HTTPException(404, "Hund nicht gefunden.") rows = conn.execute( """ SELECT * FROM training_sessions WHERE user_id=? AND dog_id=? ORDER BY datum DESC, created_at DESC LIMIT ? OFFSET ? """, (uid, dog_id, limit, offset) ).fetchall() return [dict(r) for r in rows] @router.get("/calendar") async def get_calendar( dog_id: int, year: Optional[int] = None, month: Optional[int] = None, user=Depends(get_current_user) ): uid = user["id"] today = datetime.date.today() year = year or today.year month = month or today.month month_str = f"{year:04d}-{month:02d}" with db() as conn: dog = conn.execute( "SELECT id FROM dogs WHERE id=? AND user_id=?", (dog_id, uid) ).fetchone() if not dog: raise HTTPException(404, "Hund nicht gefunden.") rows = conn.execute( """ SELECT datum, COUNT(*) AS count, MAX(ist_top) AS top FROM training_sessions WHERE user_id=? AND dog_id=? AND datum LIKE ? GROUP BY datum """, (uid, dog_id, month_str + "-%") ).fetchall() days = { r["datum"]: {"count": r["count"], "top": bool(r["top"])} for r in rows } return {"days": days} @router.get("/stats") async def get_stats(dog_id: int, user=Depends(get_current_user)): uid = user["id"] with db() as conn: dog = conn.execute( "SELECT id FROM dogs WHERE id=? AND user_id=?", (dog_id, uid) ).fetchone() if not dog: raise HTTPException(404, "Hund nicht gefunden.") total_sessions = conn.execute( "SELECT COUNT(*) FROM training_sessions WHERE user_id=? AND dog_id=?", (uid, dog_id) ).fetchone()[0] total_exercises = conn.execute( "SELECT COUNT(DISTINCT exercise_id) FROM training_sessions WHERE user_id=? AND dog_id=?", (uid, dog_id) ).fetchone()[0] avg_row = conn.execute( "SELECT AVG(erfolgsquote) FROM training_sessions WHERE user_id=? AND dog_id=?", (uid, dog_id) ).fetchone() avg_erfolgsquote = round(avg_row[0], 1) if avg_row[0] is not None else 0.0 best_row = conn.execute( """ SELECT exercise_name, AVG(erfolgsquote) AS avg_e FROM training_sessions WHERE user_id=? AND dog_id=? GROUP BY exercise_id ORDER BY avg_e DESC LIMIT 1 """, (uid, dog_id) ).fetchone() best_exercise = ( {"name": best_row["exercise_name"], "avg_erfolg": round(best_row["avg_e"], 1)} if best_row else None ) training_days = conn.execute( "SELECT COUNT(DISTINCT datum) FROM training_sessions WHERE user_id=? AND dog_id=?", (uid, dog_id) ).fetchone()[0] top_sessions = conn.execute( "SELECT COUNT(*) FROM training_sessions WHERE user_id=? AND dog_id=? AND ist_top=1", (uid, dog_id) ).fetchone()[0] today = datetime.date.today() week_start = (today - datetime.timedelta(days=today.weekday())).isoformat() this_week = conn.execute( "SELECT COUNT(*) FROM training_sessions WHERE user_id=? AND dog_id=? AND datum >= ?", (uid, dog_id, week_start) ).fetchone()[0] month_start = today.replace(day=1).isoformat() this_month = conn.execute( "SELECT COUNT(*) FROM training_sessions WHERE user_id=? AND dog_id=? AND datum >= ?", (uid, dog_id, month_start) ).fetchone()[0] # Streak: aufeinanderfolgende Tage bis heute day_rows = conn.execute( """ SELECT DISTINCT datum FROM training_sessions WHERE user_id=? AND dog_id=? ORDER BY datum DESC """, (uid, dog_id) ).fetchall() streak_days = 0 check = today for row in day_rows: d = datetime.date.fromisoformat(row["datum"]) if d == check: streak_days += 1 check -= datetime.timedelta(days=1) else: break return { "total_sessions": total_sessions, "total_exercises": total_exercises, "avg_erfolgsquote": avg_erfolgsquote, "best_exercise": best_exercise, "training_days": training_days, "top_sessions": top_sessions, "this_week": this_week, "this_month": this_month, "streak_days": streak_days, } # ------------------------------------------------------------------ # Virtueller Trainer — statistikbasierte Empfehlungen # ------------------------------------------------------------------ import math from collections import defaultdict def _recommend_reps(recent_avg: float, last_reps: int, last_mood: str) -> int: base = last_reps or 5 if recent_avg >= 80: reps = min(base + 1, 10) elif recent_avg >= 60: reps = base elif recent_avg < 40: reps = max(base - 2, 2) else: reps = max(base - 1, 3) if last_mood == "müde": reps = max(reps - 1, 2) elif last_mood == "super": reps = min(reps + 1, 10) return reps def _reason(type_: str, s: dict) -> str: days, recent, trend = s["days_since"], s["recent_avg"], s["trend"] if type_ == "üben": if trend == "declining": return f"Erfolgsquote fällt – zuletzt {round(recent)}%." if recent < 40: return f"Noch Luft nach oben: {round(recent)}% zuletzt." return f"Braucht Aufmerksamkeit: {round(recent)}% Erfolg." if type_ == "festigen": if trend == "improving": return f"Guter Fortschritt auf {round(recent)}% – dran bleiben." if days >= 3: return f"Seit {days} Tagen nicht geübt – kurz auffrischen." return f"Läuft gut ({round(recent)}%) – festigen." # entdecken return f"Seit {days} Tag{'en' if days != 1 else ''} nicht geübt." # Synthetische Stats wenn nur exercise_progress vorhanden (keine Sessions) _STATUS_SYNTHETIC = { 'noch-nicht': {'recent_avg': 25.0, 'trend': 'new', 'days_since': 5, 'suggested_reps': 3}, 'manchmal': {'recent_avg': 45.0, 'trend': 'stable', 'days_since': 7, 'suggested_reps': 5}, 'meistens': {'recent_avg': 72.0, 'trend': 'stable', 'days_since': 5, 'suggested_reps': 5}, 'sitzt': {'recent_avg': 92.0, 'trend': 'stable', 'days_since': 14, 'suggested_reps': 7}, } def _ex_key(tab: str, name: str) -> str: return f"{tab}_{name.replace(' ', '_').replace('/', '')}" def _parse_ex_name(exercise_id: str) -> str: parts = exercise_id.split("_", 1) if len(parts) < 2: return exercise_id return parts[1].replace("__", " / ").replace("_", " ") @router.get("/recommendations") async def get_recommendations(dog_id: int, user=Depends(get_current_user)): uid = user["id"] today = datetime.date.today() with db() as conn: dog = conn.execute( "SELECT id FROM dogs WHERE id=? AND user_id=?", (dog_id, uid) ).fetchone() if not dog: raise HTTPException(404, "Hund nicht gefunden.") rows = conn.execute( """ SELECT exercise_id, exercise_name, datum, wiederholungen, erfolgsquote, hund_stimmung FROM training_sessions WHERE user_id=? AND dog_id=? ORDER BY datum DESC, created_at DESC """, (uid, dog_id) ).fetchall() progress_rows = conn.execute( "SELECT exercise_id, status FROM exercise_progress WHERE user_id=?", (uid,) ).fetchall() progress = {r["exercise_id"]: r["status"] for r in progress_rows} # --- Session-basierte Stats --- by_ex = defaultdict(list) for r in rows: by_ex[r["exercise_id"]].append(dict(r)) stats = [] session_ids = set() for ex_id, sessions in by_ex.items(): session_ids.add(ex_id) last = sessions[0] last_date = datetime.date.fromisoformat(last["datum"]) days_since = (today - last_date).days all_success = [s["erfolgsquote"] for s in sessions] avg_success = sum(all_success) / len(all_success) recent_avg = sum(s["erfolgsquote"] for s in sessions[:3]) / len(sessions[:3]) older = sessions[3:6] if older: older_avg = sum(s["erfolgsquote"] for s in older) / len(older) diff = recent_avg - older_avg trend = "improving" if diff > 10 else ("declining" if diff < -10 else "stable") else: trend = "new" prognose = None if trend == "improving" and older: improvement = recent_avg - (sum(s["erfolgsquote"] for s in older) / len(older)) if improvement > 0 and recent_avg < 80: prognose = max(1, math.ceil((80 - recent_avg) / (improvement / 3))) stats.append({ "exercise_id": ex_id, "exercise_name": last["exercise_name"], "tab": ex_id.split("_")[0], "session_count": len(sessions), "avg_success": round(avg_success, 1), "recent_avg": round(recent_avg, 1), "trend": trend, "days_since": days_since, "suggested_reps": _recommend_reps(recent_avg, last["wiederholungen"], last["hund_stimmung"]), "prognose_sessions": prognose, "from_status": False, }) # --- Status-basierte synthetische Stats (Übungen ohne Sessions) --- for ex_id, status in progress.items(): if ex_id in session_ids or status not in _STATUS_SYNTHETIC: continue synth = _STATUS_SYNTHETIC[status] stats.append({ "exercise_id": ex_id, "exercise_name": _parse_ex_name(ex_id), "tab": ex_id.split("_")[0], "session_count": 0, "avg_success": synth["recent_avg"], "recent_avg": synth["recent_avg"], "trend": synth["trend"], "days_since": synth["days_since"], "suggested_reps": synth["suggested_reps"], "prognose_sessions": None, "from_status": True, }) if not stats: return {"recommendations": [], "all_stats": {}, "summary": { "total_exercises_trained": 0, "exercises_on_track": 0, "exercises_need_work": 0}} # --- Levelup: Tricks empfehlen wenn Grundlage stark --- grundk_ids = [_ex_key("grundkommandos", n) for n in GRUNDKOMMANDOS_ORDER] grundk_mastered = [eid for eid in grundk_ids if progress.get(eid) in ("sitzt", "meistens")] levelup_rec = None if len(grundk_mastered) >= 4: tricks_pairs = [(_ex_key("tricks", n), n) for n in TRICKS_FIRST] first_trick = next( ((eid, name) for eid, name in tricks_pairs if progress.get(eid) is None and eid not in session_ids), None ) if first_trick: eid, name = first_trick levelup_rec = { "exercise_id": eid, "exercise_name": name, "tab": "tricks", "session_count": 0, "avg_success": 0.0, "recent_avg": 0.0, "trend": "new", "days_since": 0, "suggested_reps": 5, "prognose_sessions": None, "from_status": True, "type": "levelup", "reason": f"{len(grundk_mastered)} Grundkommandos sitzen — Zeit für den nächsten Level!", } # --- Auswahl: üben / festigen / levelup-oder-entdecken --- def needs_work_score(s): score = min(s["days_since"], 7) * 0.3 + (100 - s["recent_avg"]) * 0.5 if s["trend"] == "declining": score += 20 return score not_today = [s for s in stats if s["days_since"] > 0] needs_work = sorted([s for s in not_today if s["recent_avg"] < 70], key=needs_work_score, reverse=True) festigen = sorted([s for s in not_today if s["recent_avg"] >= 70], key=lambda s: s["days_since"], reverse=True) entdecken = sorted([s for s in not_today if s["days_since"] >= 5], key=lambda s: s["days_since"], reverse=True) recs = [] used = set() for pool, type_ in [(needs_work, "üben"), (festigen, "festigen")]: for s in pool: if s["exercise_id"] not in used: recs.append({**s, "type": type_, "reason": _reason(type_, s)}) used.add(s["exercise_id"]) break # Slot 3: levelup hat Vorrang vor entdecken if levelup_rec and levelup_rec["exercise_id"] not in used: recs.append(levelup_rec) used.add(levelup_rec["exercise_id"]) else: for s in entdecken: if s["exercise_id"] not in used: recs.append({**s, "type": "entdecken", "reason": _reason("entdecken", s)}) used.add(s["exercise_id"]) break # Auffüllen if len(recs) < 3: for s in sorted(not_today, key=lambda s: s["days_since"], reverse=True): if len(recs) >= 3: break if s["exercise_id"] not in used: recs.append({**s, "type": "üben", "reason": _reason("üben", s)}) used.add(s["exercise_id"]) all_stats = { s["exercise_id"]: { "recent_avg": s["recent_avg"], "session_count": s["session_count"], "trend": s["trend"], "days_since": s["days_since"], } for s in stats } return { "recommendations": recs[:3], "all_stats": all_stats, "summary": { "total_exercises_trained": len(stats), "exercises_on_track": len([s for s in stats if s["recent_avg"] >= 70]), "exercises_need_work": len([s for s in stats if s["recent_avg"] < 60]), }, } class KiFeedbackRequest(BaseModel): dog_id: int KI_DAILY_LIMIT = 10 @router.post("/ki-feedback") async def ki_feedback(body: KiFeedbackRequest, user=Depends(get_current_user)): uid = user["id"] today = datetime.date.today().isoformat() with db() as conn: dog = conn.execute( "SELECT id, name, rasse, geburtstag FROM dogs WHERE id=? AND user_id=?", (body.dog_id, uid) ).fetchone() if not dog: raise HTTPException(404, "Hund nicht gefunden.") # Cache prüfen cache_row = conn.execute( "SELECT feedback, generated_at FROM training_ki_cache WHERE dog_id=?", (body.dog_id,) ).fetchone() if cache_row: generated_at = datetime.datetime.fromisoformat(cache_row["generated_at"]) age_hours = (datetime.datetime.utcnow() - generated_at).total_seconds() / 3600 # Neue Sessions seit letzter Generierung? new_since = conn.execute( "SELECT COUNT(*) FROM training_sessions WHERE dog_id=? AND created_at > ?", (body.dog_id, cache_row["generated_at"]) ).fetchone()[0] if age_hours < 6 and new_since == 0: used = conn.execute( "SELECT COALESCE(SUM(count),0) FROM ki_daily_calls WHERE user_id=? AND date=?", (uid, today) ).fetchone()[0] return {"feedback": cache_row["feedback"], "cached": True, "daily_used": used, "daily_limit": KI_DAILY_LIMIT} # Tageslimit prüfen (nur bei echtem API-Aufruf) conn.execute(""" CREATE TABLE IF NOT EXISTS ki_daily_calls ( user_id INTEGER NOT NULL, date TEXT NOT NULL, count INTEGER NOT NULL DEFAULT 0, source TEXT NOT NULL DEFAULT 'cloud', PRIMARY KEY (user_id, date, source) ) """) daily_used = conn.execute( "SELECT COALESCE(SUM(count),0) FROM ki_daily_calls WHERE user_id=? AND date=?", (uid, today) ).fetchone()[0] if daily_used >= KI_DAILY_LIMIT: raise HTTPException(429, f"Tages-Limit erreicht ({KI_DAILY_LIMIT} Anfragen/Tag). Morgen wieder verfügbar.") # Hund-Alter berechnen alter_str = "unbekannt" if dog["geburtstag"]: try: geb = datetime.date.fromisoformat(dog["geburtstag"]) alter_jahre = (datetime.date.today() - geb).days // 365 alter_str = f"{alter_jahre} Jahre" except Exception: pass # Letzte 20 Sessions laden sessions = conn.execute( """ SELECT datum, exercise_name, wiederholungen, erfolgsquote, hund_stimmung, zufriedenheit, notiz, ist_top FROM training_sessions WHERE dog_id=? ORDER BY datum DESC, created_at DESC LIMIT 20 """, (body.dog_id,) ).fetchall() # Übungsfortschritt laden progress_rows = conn.execute( "SELECT exercise_id, status FROM exercise_progress WHERE user_id=?", (uid,) ).fetchall() # Prompt zusammenbauen dog_name = dog["name"] dog_rasse = dog["rasse"] or "unbekannter Rasse" sessions_text = "\n".join( f" {r['datum']}: {r['exercise_name']}, " f"{r['wiederholungen']}x, {r['erfolgsquote']}% Erfolg, " f"Stimmung: {STIMMUNGS_LABELS.get(r['hund_stimmung'], r['hund_stimmung'])}, " f"Zufriedenheit: {r['zufriedenheit']}/5" + (f", Notiz: {r['notiz']}" if r["notiz"] else "") + (" [TOP]" if r["ist_top"] else "") for r in sessions ) or " (noch keine Sessions)" progress_text = "\n".join( f" {r['exercise_id']}: {r['status']}" for r in progress_rows ) or " (kein Fortschritt erfasst)" prompt = ( f"Hund: {dog_name}, {dog_rasse}, {alter_str}\n" f"Letzte Sessions:\n{sessions_text}\n" f"Übungsfortschritt:\n{progress_text}\n\n" "Antworte mit maximal 3 kurzen Abschnitten:\n" "1. Was gut läuft (1-2 Sätze, immer positiv beginnen)\n" "2. Konkrete Empfehlung (1-2 Sätze, spezifisch auf die Daten bezogen)\n" "3. Kleiner motivierender Abschluss (1 Satz)\n\n" "Sprich über den Hund, nicht über den Besitzer. Kein Druck, keine Forderungen." ) system = ( "Du bist ein einfühlsamer, positiver Hundetrainer. " "Analysiere die Trainingshistorie und gib kurzes, motivierendes Feedback auf Deutsch." ) try: feedback_text, ki_source = await ki.complete( prompt, system=system, max_tokens=400, requires_premium=False, user_is_premium=user.get("is_premium", False), return_source=True, ) except (ki.KIUnavailableError, ki.KIPremiumRequired) as e: raise HTTPException(503, str(e)) # Cache speichern + Tageszähler erhöhen with db() as conn: conn.execute( """ INSERT INTO training_ki_cache (dog_id, feedback, generated_at) VALUES (?,?,datetime('now')) ON CONFLICT(dog_id) DO UPDATE SET feedback=excluded.feedback, generated_at=excluded.generated_at """, (body.dog_id, feedback_text) ) conn.execute(""" INSERT INTO ki_daily_calls (user_id, date, count, source) VALUES (?, ?, 1, ?) ON CONFLICT(user_id, date, source) DO UPDATE SET count = count + 1 """, (uid, today, ki_source)) new_count = conn.execute( "SELECT COALESCE(SUM(count),0) FROM ki_daily_calls WHERE user_id=? AND date=?", (uid, today) ).fetchone()[0] return {"feedback": feedback_text, "cached": False, "daily_used": new_count, "daily_limit": KI_DAILY_LIMIT}