518 lines
18 KiB
Python
518 lines
18 KiB
Python
"""
|
|
Exercises Router - v2.0 (Clean-Room Rebuild)
|
|
|
|
Komplett neu gebaut nach EXERCISES_API_SPEC.md v1.2
|
|
KEIN Legacy-Code aus v1 - nur M:N Relations, keine JSONB-Felder für Kataloge
|
|
"""
|
|
import json
|
|
import logging
|
|
from typing import Optional
|
|
|
|
from fastapi import APIRouter, HTTPException, Depends, Query
|
|
from pydantic import BaseModel, Field
|
|
|
|
from db import get_db, get_cursor, r2d
|
|
from auth import require_auth
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(prefix="/api", tags=["exercises"])
|
|
|
|
|
|
# ============================================================================
|
|
# Pydantic Models
|
|
# ============================================================================
|
|
|
|
class ExerciseCreate(BaseModel):
|
|
# Basis-Felder
|
|
title: str = Field(..., min_length=3, max_length=300)
|
|
summary: Optional[str] = None
|
|
goal: str = Field(..., min_length=10, max_length=5000)
|
|
execution: str = Field(..., min_length=10, max_length=10000)
|
|
preparation: Optional[str] = None
|
|
trainer_notes: Optional[str] = None
|
|
|
|
# Dauer & Gruppengröße
|
|
duration_min: Optional[int] = None
|
|
duration_max: Optional[int] = None
|
|
group_size_min: Optional[int] = None
|
|
group_size_max: Optional[int] = None
|
|
|
|
# Equipment (Liste von Strings)
|
|
equipment: list[str] = []
|
|
|
|
# M:N Relations (Liste von {id: int, is_primary: bool})
|
|
focus_areas_multi: list[dict] = []
|
|
training_styles_multi: list[dict] = []
|
|
target_groups_multi: list[dict] = []
|
|
age_groups: list[str] = [] # ["Kinder", "Teenager"] aus Katalog
|
|
|
|
# Skills (Liste von {skill_id: int, is_primary: bool, intensity: str, required_level: str, target_level: str})
|
|
skills: list[dict] = []
|
|
|
|
# Sichtbarkeit & Status
|
|
visibility: str = "private"
|
|
status: str = "draft"
|
|
club_id: Optional[int] = None
|
|
|
|
|
|
class ExerciseUpdate(BaseModel):
|
|
# Alle Felder optional für Partial Update
|
|
title: Optional[str] = Field(None, min_length=3, max_length=300)
|
|
summary: Optional[str] = None
|
|
goal: Optional[str] = Field(None, min_length=10, max_length=5000)
|
|
execution: Optional[str] = Field(None, min_length=10, max_length=10000)
|
|
preparation: Optional[str] = None
|
|
trainer_notes: Optional[str] = None
|
|
duration_min: Optional[int] = None
|
|
duration_max: Optional[int] = None
|
|
group_size_min: Optional[int] = None
|
|
group_size_max: Optional[int] = None
|
|
equipment: Optional[list[str]] = None
|
|
focus_areas_multi: Optional[list[dict]] = None
|
|
training_styles_multi: Optional[list[dict]] = None
|
|
target_groups_multi: Optional[list[dict]] = None
|
|
age_groups: Optional[list[str]] = None
|
|
skills: Optional[list[dict]] = None
|
|
visibility: Optional[str] = None
|
|
status: Optional[str] = None
|
|
club_id: Optional[int] = None
|
|
|
|
|
|
# ============================================================================
|
|
# Helper Functions
|
|
# ============================================================================
|
|
|
|
def enrich_exercise_detail(exercise_id: int, cur) -> dict:
|
|
"""
|
|
Lädt alle M:N Relations für eine Übung und gibt ein vollständiges
|
|
Exercise-Objekt zurück (wie in API-Spec GET /exercises/{id}).
|
|
"""
|
|
# Basis-Exercise
|
|
cur.execute(
|
|
"""SELECT e.*, p.name as creator_name, c.name as club_name
|
|
FROM exercises e
|
|
LEFT JOIN profiles p ON e.created_by = p.id
|
|
LEFT JOIN clubs c ON e.club_id = c.id
|
|
WHERE e.id = %s""",
|
|
(exercise_id,)
|
|
)
|
|
row = cur.fetchone()
|
|
if not row:
|
|
return None
|
|
|
|
exercise = r2d(row)
|
|
|
|
# Equipment JSONB → List
|
|
if exercise.get("equipment"):
|
|
exercise["equipment"] = exercise["equipment"] if isinstance(exercise["equipment"], list) else []
|
|
else:
|
|
exercise["equipment"] = []
|
|
|
|
# Focus Areas (M:N)
|
|
cur.execute(
|
|
"""SELECT efa.id, efa.focus_area_id, fa.name, fa.abbreviation, fa.color, fa.icon, efa.is_primary
|
|
FROM exercise_focus_areas efa
|
|
JOIN focus_areas fa ON efa.focus_area_id = fa.id
|
|
WHERE efa.exercise_id = %s
|
|
ORDER BY efa.is_primary DESC, fa.name""",
|
|
(exercise_id,)
|
|
)
|
|
exercise["focus_areas"] = [r2d(r) for r in cur.fetchall()]
|
|
|
|
# Training Styles (M:N)
|
|
cur.execute(
|
|
"""SELECT ets.id, ets.style_direction_id as training_style_id, sd.name, sd.abbreviation, ets.is_primary
|
|
FROM exercise_style_directions ets
|
|
JOIN style_directions sd ON ets.style_direction_id = sd.id
|
|
WHERE ets.exercise_id = %s
|
|
ORDER BY ets.is_primary DESC, sd.name""",
|
|
(exercise_id,)
|
|
)
|
|
exercise["training_styles"] = [r2d(r) for r in cur.fetchall()]
|
|
|
|
# Target Groups (M:N)
|
|
cur.execute(
|
|
"""SELECT etg.id, etg.target_group_id, tg.name, tg.description, etg.is_primary
|
|
FROM exercise_target_groups etg
|
|
JOIN target_groups tg ON etg.target_group_id = tg.id
|
|
WHERE etg.exercise_id = %s
|
|
ORDER BY etg.is_primary DESC, tg.name""",
|
|
(exercise_id,)
|
|
)
|
|
exercise["target_groups"] = [r2d(r) for r in cur.fetchall()]
|
|
|
|
# Age Groups (M:N) - direkt als VARCHAR gespeichert
|
|
cur.execute(
|
|
"""SELECT age_group
|
|
FROM exercise_age_groups
|
|
WHERE exercise_id = %s
|
|
ORDER BY age_group""",
|
|
(exercise_id,)
|
|
)
|
|
exercise["age_groups"] = [r["age_group"] for r in cur.fetchall()]
|
|
|
|
# Skills (M:N) mit Levels und Intensity
|
|
cur.execute(
|
|
"""SELECT es.id, es.skill_id, s.name as skill_name, s.category as skill_category,
|
|
es.is_primary, es.intensity, es.required_level, es.target_level, es.ai_suggested
|
|
FROM exercise_skills es
|
|
JOIN skills s ON es.skill_id = s.id
|
|
WHERE es.exercise_id = %s
|
|
ORDER BY es.is_primary DESC, s.name""",
|
|
(exercise_id,)
|
|
)
|
|
exercise["skills"] = [r2d(r) for r in cur.fetchall()]
|
|
|
|
# Variants (1:N) - mit Progression
|
|
cur.execute(
|
|
"""SELECT id, variant_name, description, execution_changes,
|
|
duration_min, duration_max, equipment_changes, difficulty_adjustment,
|
|
progression_level, sequence_order, prerequisite_variant_id
|
|
FROM exercise_variants
|
|
WHERE exercise_id = %s
|
|
ORDER BY progression_level, sequence_order""",
|
|
(exercise_id,)
|
|
)
|
|
exercise["variants"] = [r2d(r) for r in cur.fetchall()]
|
|
|
|
# Media (1:N)
|
|
cur.execute(
|
|
"""SELECT id, media_type, file_path, file_size, mime_type, original_filename,
|
|
embed_url, embed_platform, title, description, sort_order, is_primary, context
|
|
FROM exercise_media
|
|
WHERE exercise_id = %s
|
|
ORDER BY sort_order, id""",
|
|
(exercise_id,)
|
|
)
|
|
exercise["media"] = [r2d(r) for r in cur.fetchall()]
|
|
|
|
return exercise
|
|
|
|
|
|
def assign_exercise_relations(cur, conn, exercise_id: int, data: dict):
|
|
"""
|
|
Weist M:N Relations für eine Übung zu.
|
|
Löscht alte Zuordnungen und legt neue an (REPLACE-Logik).
|
|
"""
|
|
# Focus Areas
|
|
if "focus_areas_multi" in data:
|
|
cur.execute("DELETE FROM exercise_focus_areas WHERE exercise_id = %s", (exercise_id,))
|
|
for fa in data["focus_areas_multi"]:
|
|
cur.execute(
|
|
"""INSERT INTO exercise_focus_areas (exercise_id, focus_area_id, is_primary)
|
|
VALUES (%s, %s, %s)""",
|
|
(exercise_id, fa["focus_area_id"], fa.get("is_primary", False))
|
|
)
|
|
|
|
# Training Styles
|
|
if "training_styles_multi" in data:
|
|
cur.execute("DELETE FROM exercise_style_directions WHERE exercise_id = %s", (exercise_id,))
|
|
for ts in data["training_styles_multi"]:
|
|
cur.execute(
|
|
"""INSERT INTO exercise_style_directions (exercise_id, style_direction_id, is_primary)
|
|
VALUES (%s, %s, %s)""",
|
|
(exercise_id, ts["training_style_id"], ts.get("is_primary", False))
|
|
)
|
|
|
|
# Target Groups
|
|
if "target_groups_multi" in data:
|
|
cur.execute("DELETE FROM exercise_target_groups WHERE exercise_id = %s", (exercise_id,))
|
|
for tg in data["target_groups_multi"]:
|
|
cur.execute(
|
|
"""INSERT INTO exercise_target_groups (exercise_id, target_group_id, is_primary)
|
|
VALUES (%s, %s, %s)""",
|
|
(exercise_id, tg["target_group_id"], tg.get("is_primary", False))
|
|
)
|
|
|
|
# Age Groups (direkt als VARCHAR, CHECK constraint validiert)
|
|
if "age_groups" in data:
|
|
cur.execute("DELETE FROM exercise_age_groups WHERE exercise_id = %s", (exercise_id,))
|
|
for age_group_name in data["age_groups"]:
|
|
try:
|
|
cur.execute(
|
|
"INSERT INTO exercise_age_groups (exercise_id, age_group) VALUES (%s, %s)",
|
|
(exercise_id, age_group_name)
|
|
)
|
|
except Exception as e:
|
|
logger.warning("Age Group '%s' ungültig: %s", age_group_name, e)
|
|
|
|
# Skills
|
|
if "skills" in data:
|
|
cur.execute("DELETE FROM exercise_skills WHERE exercise_id = %s", (exercise_id,))
|
|
for skill in data["skills"]:
|
|
cur.execute(
|
|
"""INSERT INTO exercise_skills
|
|
(exercise_id, skill_id, is_primary, intensity, required_level, target_level, ai_suggested)
|
|
VALUES (%s, %s, %s, %s, %s, %s, %s)""",
|
|
(
|
|
exercise_id,
|
|
skill["skill_id"],
|
|
skill.get("is_primary", False),
|
|
skill.get("intensity"),
|
|
skill.get("required_level"),
|
|
skill.get("target_level"),
|
|
skill.get("ai_suggested", False),
|
|
)
|
|
)
|
|
|
|
conn.commit()
|
|
|
|
|
|
# ============================================================================
|
|
# Endpoints
|
|
# ============================================================================
|
|
|
|
@router.get("/exercises")
|
|
def list_exercises(
|
|
focus_area: Optional[int] = Query(default=None),
|
|
visibility: Optional[str] = Query(default=None),
|
|
status: Optional[str] = Query(default=None),
|
|
skill_id: Optional[int] = Query(default=None),
|
|
search: Optional[str] = Query(default=None),
|
|
limit: int = Query(default=50, ge=1, le=100),
|
|
offset: int = Query(default=0, ge=0),
|
|
session: dict = Depends(require_auth),
|
|
):
|
|
"""
|
|
Liste aller Übungen mit Filtern.
|
|
Lightweight Response (ohne M:N Details, nur IDs und Namen).
|
|
"""
|
|
profile_id = session["profile_id"]
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
|
|
# WHERE-Bedingungen
|
|
where = ["1=1"]
|
|
params = []
|
|
|
|
# Visibility Filter (private nur für Owner)
|
|
where.append("(e.visibility = 'official' OR e.visibility = 'club' OR e.created_by = %s)")
|
|
params.append(profile_id)
|
|
|
|
if visibility:
|
|
where.append("e.visibility = %s")
|
|
params.append(visibility)
|
|
|
|
if status:
|
|
where.append("e.status = %s")
|
|
params.append(status)
|
|
|
|
# Focus Area Filter (M:N Join)
|
|
if focus_area:
|
|
where.append("EXISTS (SELECT 1 FROM exercise_focus_areas efa WHERE efa.exercise_id = e.id AND efa.focus_area_id = %s)")
|
|
params.append(focus_area)
|
|
|
|
# Skill Filter (M:N Join)
|
|
if skill_id:
|
|
where.append("EXISTS (SELECT 1 FROM exercise_skills es WHERE es.exercise_id = e.id AND es.skill_id = %s)")
|
|
params.append(skill_id)
|
|
|
|
# Volltext-Suche (tsvector)
|
|
if search:
|
|
where.append("e.search_vector @@ plainto_tsquery('german', %s)")
|
|
params.append(search)
|
|
|
|
# Query
|
|
query = f"""
|
|
SELECT e.id, e.title, e.summary, e.visibility, e.status,
|
|
e.created_by, p.name as creator_name,
|
|
e.club_id, c.name as club_name,
|
|
e.created_at, e.updated_at
|
|
FROM exercises e
|
|
LEFT JOIN profiles p ON e.created_by = p.id
|
|
LEFT JOIN clubs c ON e.club_id = c.id
|
|
WHERE {' AND '.join(where)}
|
|
ORDER BY e.updated_at DESC
|
|
LIMIT %s OFFSET %s
|
|
"""
|
|
params.extend([limit, offset])
|
|
|
|
cur.execute(query, params)
|
|
rows = cur.fetchall()
|
|
|
|
return [r2d(r) for r in rows]
|
|
|
|
|
|
@router.get("/exercises/{exercise_id}")
|
|
def get_exercise(
|
|
exercise_id: int,
|
|
session: dict = Depends(require_auth),
|
|
):
|
|
"""
|
|
Exercise Detail mit allen M:N Relations (vollständig enriched).
|
|
"""
|
|
profile_id = session["profile_id"]
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
exercise = enrich_exercise_detail(exercise_id, cur)
|
|
|
|
if not exercise:
|
|
raise HTTPException(status_code=404, detail="Übung nicht gefunden")
|
|
|
|
# Permission Check (private nur für Owner)
|
|
if exercise["visibility"] == "private" and exercise["created_by"] != profile_id:
|
|
raise HTTPException(status_code=403, detail="Keine Berechtigung für diese Übung")
|
|
|
|
return exercise
|
|
|
|
|
|
@router.post("/exercises", status_code=201)
|
|
def create_exercise(
|
|
body: ExerciseCreate,
|
|
session: dict = Depends(require_auth),
|
|
):
|
|
"""
|
|
Erstellt eine neue Übung mit allen M:N Relations.
|
|
"""
|
|
profile_id = session["profile_id"]
|
|
|
|
# Validierung
|
|
if body.status not in ("draft", "in_review", "approved", "archived"):
|
|
raise HTTPException(status_code=400, detail="Ungültiger Status")
|
|
if body.visibility not in ("private", "club", "official"):
|
|
raise HTTPException(status_code=400, detail="Ungültige Visibility")
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
|
|
# Equipment als JSONB
|
|
equipment_json = json.dumps(body.equipment) if body.equipment else None
|
|
|
|
# INSERT
|
|
cur.execute(
|
|
"""INSERT INTO exercises
|
|
(title, summary, goal, execution, preparation, trainer_notes,
|
|
duration_min, duration_max, group_size_min, group_size_max,
|
|
equipment, visibility, status, created_by, club_id)
|
|
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
|
|
RETURNING id""",
|
|
(
|
|
body.title, body.summary, body.goal, body.execution,
|
|
body.preparation, body.trainer_notes,
|
|
body.duration_min, body.duration_max,
|
|
body.group_size_min, body.group_size_max,
|
|
equipment_json,
|
|
body.visibility, body.status, profile_id, body.club_id,
|
|
)
|
|
)
|
|
row = cur.fetchone()
|
|
exercise_id = row['id'] if isinstance(row, dict) else row[0]
|
|
conn.commit()
|
|
|
|
# M:N Relations zuweisen
|
|
data = body.dict()
|
|
assign_exercise_relations(cur, conn, exercise_id, data)
|
|
|
|
# Vollständiges Objekt zurückgeben
|
|
exercise = enrich_exercise_detail(exercise_id, cur)
|
|
|
|
return exercise
|
|
|
|
|
|
@router.put("/exercises/{exercise_id}")
|
|
def update_exercise(
|
|
exercise_id: int,
|
|
body: ExerciseUpdate,
|
|
session: dict = Depends(require_auth),
|
|
):
|
|
"""
|
|
Aktualisiert eine Übung (Partial Update).
|
|
Nur Owner darf editieren.
|
|
"""
|
|
profile_id = session["profile_id"]
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
|
|
# Existiert die Übung?
|
|
cur.execute("SELECT created_by FROM exercises WHERE id = %s", (exercise_id,))
|
|
row = cur.fetchone()
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="Übung nicht gefunden")
|
|
|
|
# Permission Check
|
|
if row[0] != profile_id:
|
|
raise HTTPException(status_code=403, detail="Nur der Ersteller darf editieren")
|
|
|
|
# UPDATE (nur gesetzte Felder)
|
|
fields = []
|
|
params = []
|
|
|
|
data = body.dict(exclude_unset=True)
|
|
|
|
# Basis-Felder
|
|
for field in ["title", "summary", "goal", "execution", "preparation", "trainer_notes",
|
|
"duration_min", "duration_max", "group_size_min", "group_size_max",
|
|
"visibility", "status", "club_id"]:
|
|
if field in data and data[field] is not None:
|
|
fields.append(f"{field} = %s")
|
|
params.append(data[field])
|
|
|
|
# Equipment (JSONB)
|
|
if "equipment" in data:
|
|
fields.append("equipment = %s")
|
|
params.append(json.dumps(data["equipment"]) if data["equipment"] else None)
|
|
|
|
# UPDATE ausführen (wenn Basis-Felder geändert wurden)
|
|
if fields:
|
|
fields.append("updated_at = NOW()")
|
|
params.append(exercise_id)
|
|
query = f"UPDATE exercises SET {', '.join(fields)} WHERE id = %s"
|
|
cur.execute(query, params)
|
|
conn.commit()
|
|
|
|
# M:N Relations aktualisieren (wenn angegeben)
|
|
assign_exercise_relations(cur, conn, exercise_id, data)
|
|
|
|
# Vollständiges Objekt zurückgeben
|
|
exercise = enrich_exercise_detail(exercise_id, cur)
|
|
|
|
return exercise
|
|
|
|
|
|
@router.delete("/exercises/{exercise_id}")
|
|
def delete_exercise(
|
|
exercise_id: int,
|
|
session: dict = Depends(require_auth),
|
|
):
|
|
"""
|
|
Löscht eine Übung.
|
|
Nur Owner oder Admin darf löschen.
|
|
"""
|
|
profile_id = session["profile_id"]
|
|
role = session.get("role")
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
|
|
# Existiert die Übung?
|
|
cur.execute("SELECT created_by FROM exercises WHERE id = %s", (exercise_id,))
|
|
row = cur.fetchone()
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="Übung nicht gefunden")
|
|
|
|
# Permission Check
|
|
if row[0] != profile_id and role not in ("admin", "superadmin"):
|
|
raise HTTPException(status_code=403, detail="Nur Ersteller oder Admin darf löschen")
|
|
|
|
# Prüfen ob Übung in Trainingseinheiten verwendet wird
|
|
cur.execute(
|
|
"SELECT COUNT(*) FROM exercise_block_items WHERE exercise_id = %s",
|
|
(exercise_id,)
|
|
)
|
|
count = cur.fetchone()[0]
|
|
if count > 0:
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail=f"Übung wird in {count} Block-Item(s) verwendet und kann nicht gelöscht werden"
|
|
)
|
|
|
|
# DELETE (Cascade löscht M:N Zuordnungen automatisch)
|
|
cur.execute("DELETE FROM exercises WHERE id = %s", (exercise_id,))
|
|
conn.commit()
|
|
|
|
return {"ok": True}
|