Lars
cb11e39201
- Introduced new API endpoints for managing exercise media, including upload, update, delete, and reorder functionalities. - Updated the exercise creation and update logic to ensure goal and execution fields are validated and normalized. - Refactored frontend components to support the new exercise media features, including a dedicated import section for complete stack files. - Removed the deprecated ExercisesPage component and replaced it with a more modular structure for exercise management. - Incremented database schema version to 20260427028 and updated changelog to reflect these changes.
869 lines
31 KiB
Python
869 lines
31 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 hashlib
|
|
import json
|
|
import logging
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
from fastapi import APIRouter, HTTPException, Depends, Query, UploadFile, File, Form
|
|
from pydantic import BaseModel, Field, model_validator
|
|
|
|
from db import get_db, get_cursor, r2d
|
|
from auth import require_auth
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(prefix="/api", tags=["exercises"])
|
|
|
|
MEDIA_ROOT = Path(os.getenv("MEDIA_ROOT", str(Path(__file__).resolve().parent.parent / "media")))
|
|
MAX_EXERCISE_MEDIA = 10
|
|
MAX_UPLOAD_BYTES = 50 * 1024 * 1024
|
|
ALLOWED_UPLOAD_MIMES = frozenset(
|
|
{"image/jpeg", "image/png", "image/gif", "video/mp4", "application/pdf"}
|
|
)
|
|
|
|
|
|
# ============================================================================
|
|
# Pydantic Models
|
|
# ============================================================================
|
|
|
|
class ExerciseCreate(BaseModel):
|
|
# Basis-Felder (goal/execution: DB-Constraint mind. eines; Wiki oft nur eines)
|
|
title: str = Field(..., min_length=3, max_length=300)
|
|
summary: Optional[str] = None
|
|
goal: Optional[str] = Field(None, max_length=5000)
|
|
execution: Optional[str] = Field(None, 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
|
|
|
|
@model_validator(mode="after")
|
|
def normalize_goal_execution(self):
|
|
g = (self.goal or "").strip() or None
|
|
e = (self.execution or "").strip() or None
|
|
if not g and not e:
|
|
raise ValueError("Mindestens eines der Felder Ziel oder Durchführung ist erforderlich")
|
|
self.goal = g
|
|
self.execution = e
|
|
return self
|
|
|
|
|
|
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, max_length=5000)
|
|
execution: Optional[str] = Field(None, 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
|
|
|
|
@model_validator(mode="after")
|
|
def normalize_goal_execution(self):
|
|
if self.goal is not None:
|
|
self.goal = self.goal.strip() or None
|
|
if self.execution is not None:
|
|
self.execution = self.execution.strip() or None
|
|
return self
|
|
|
|
|
|
class ExerciseMediaUpdate(BaseModel):
|
|
title: Optional[str] = None
|
|
description: Optional[str] = None
|
|
is_primary: Optional[bool] = None
|
|
context: Optional[str] = None
|
|
|
|
|
|
class ExerciseMediaReorder(BaseModel):
|
|
media_ids: list[int]
|
|
|
|
|
|
# ============================================================================
|
|
# Helper Functions
|
|
# ============================================================================
|
|
|
|
def _row_created_by(row) -> int:
|
|
if row is None:
|
|
return None
|
|
if isinstance(row, dict):
|
|
return row.get("created_by")
|
|
return row[0]
|
|
|
|
|
|
def _ensure_media_dirs():
|
|
sub = MEDIA_ROOT / "exercises"
|
|
sub.mkdir(parents=True, exist_ok=True)
|
|
return sub
|
|
|
|
|
|
def _detect_embed_platform(url: str) -> Optional[str]:
|
|
if not url:
|
|
return None
|
|
u = url.lower()
|
|
if "youtube.com" in u or "youtu.be" in u:
|
|
return "youtube"
|
|
if "vimeo.com" in u:
|
|
return "vimeo"
|
|
if "instagram.com" in u:
|
|
return "instagram"
|
|
if "tiktok.com" in u:
|
|
return "tiktok"
|
|
return None
|
|
|
|
|
|
def _assert_can_edit_exercise(cur, exercise_id: int, profile_id: int):
|
|
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")
|
|
if _row_created_by(row) != profile_id:
|
|
raise HTTPException(status_code=403, detail="Nur der Ersteller darf diese Übung bearbeiten")
|
|
|
|
|
|
def _count_exercise_media(cur, exercise_id: int) -> int:
|
|
cur.execute("SELECT COUNT(*) AS c FROM exercise_media WHERE exercise_id = %s", (exercise_id,))
|
|
r = cur.fetchone()
|
|
return int(r["c"] if isinstance(r, dict) else r[0])
|
|
|
|
|
|
def _abs_media_path(file_path_db: str) -> Optional[Path]:
|
|
if not file_path_db or file_path_db.startswith("http"):
|
|
return None
|
|
rel = file_path_db.lstrip("/")
|
|
if rel.startswith("media/"):
|
|
rel = rel[len("media/") :]
|
|
p = MEDIA_ROOT / rel
|
|
try:
|
|
p.resolve().relative_to(MEDIA_ROOT.resolve())
|
|
except ValueError:
|
|
return None
|
|
return p
|
|
|
|
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 (primary_focus_name für Listen-Ansicht gemäß Spec-Beispiel „focus_area“-Label)
|
|
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,
|
|
(
|
|
SELECT fa.name FROM exercise_focus_areas efa
|
|
JOIN focus_areas fa ON fa.id = efa.focus_area_id
|
|
WHERE efa.exercise_id = e.id
|
|
ORDER BY efa.is_primary DESC NULLS LAST, fa.name ASC
|
|
LIMIT 1
|
|
) AS primary_focus_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 {' AND '.join(where)}
|
|
ORDER BY e.updated_at DESC
|
|
LIMIT %s OFFSET %s
|
|
"""
|
|
params.extend([limit, offset])
|
|
|
|
cur.execute(query, params)
|
|
rows = cur.fetchall()
|
|
|
|
out = []
|
|
for r in rows:
|
|
d = r2d(r)
|
|
pfn = d.get("primary_focus_name")
|
|
d["focus_area"] = pfn
|
|
out.append(d)
|
|
return out
|
|
|
|
|
|
@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_created_by(row) != 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_created_by(row) != 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 Block-Items verwendet wird
|
|
cur.execute(
|
|
"SELECT COUNT(*) AS cnt FROM exercise_block_items WHERE exercise_id = %s",
|
|
(exercise_id,)
|
|
)
|
|
crow = cur.fetchone()
|
|
count = crow["cnt"] if isinstance(crow, dict) else crow[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}
|
|
|
|
|
|
# --- Medien (MEDIA_UPLOAD_SPEC.md / EXERCISES_API_SPEC.md) ---
|
|
|
|
|
|
def _fetch_media_row(cur, exercise_id: int, media_id: int) -> Optional[dict]:
|
|
cur.execute(
|
|
"""SELECT id, exercise_id, media_type, file_path, file_size, mime_type, original_filename,
|
|
embed_url, embed_platform, title, description, sort_order, is_primary, context, created_at
|
|
FROM exercise_media WHERE id = %s AND exercise_id = %s""",
|
|
(media_id, exercise_id),
|
|
)
|
|
row = cur.fetchone()
|
|
return r2d(row) if row else None
|
|
|
|
|
|
@router.post("/exercises/{exercise_id}/media", status_code=201)
|
|
async def upload_exercise_media(
|
|
exercise_id: int,
|
|
session: dict = Depends(require_auth),
|
|
file: Optional[UploadFile] = File(None),
|
|
embed_url: Optional[str] = Form(None),
|
|
media_type: str = Form(...),
|
|
title: str = Form(""),
|
|
description: str = Form(""),
|
|
context: str = Form("ablauf"),
|
|
is_primary: bool = Form(False),
|
|
):
|
|
profile_id = session["profile_id"]
|
|
if media_type not in ("image", "video", "document", "sketch"):
|
|
raise HTTPException(status_code=400, detail="Ungültiger media_type")
|
|
if context not in ("ablauf", "detail", "trainer_hint"):
|
|
raise HTTPException(status_code=400, detail="Ungültiger context")
|
|
|
|
emb = (embed_url or "").strip() or None
|
|
has_file = file is not None and bool(file.filename)
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
_assert_can_edit_exercise(cur, exercise_id, profile_id)
|
|
|
|
if _count_exercise_media(cur, exercise_id) >= MAX_EXERCISE_MEDIA:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Maximal {MAX_EXERCISE_MEDIA} Medien pro Übung",
|
|
)
|
|
|
|
if has_file and emb:
|
|
raise HTTPException(status_code=400, detail="Entweder Datei oder embed_url, nicht beides")
|
|
if not has_file and not emb:
|
|
raise HTTPException(status_code=400, detail="Datei oder embed_url erforderlich")
|
|
|
|
sort_sql = (
|
|
"COALESCE((SELECT MAX(sort_order) + 1 FROM exercise_media WHERE exercise_id = %s), 1)"
|
|
)
|
|
|
|
if emb:
|
|
platform = _detect_embed_platform(emb)
|
|
if not platform:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Ungültige Embed-URL (erlaubt: YouTube, Vimeo, Instagram, TikTok)",
|
|
)
|
|
cur.execute(
|
|
f"""INSERT INTO exercise_media (
|
|
exercise_id, media_type, file_path, file_size, mime_type, original_filename,
|
|
embed_url, embed_platform, title, description, context, is_primary, sort_order
|
|
) VALUES (
|
|
%s, %s, NULL, NULL, NULL, NULL, %s, %s, %s, %s, %s, %s, {sort_sql}
|
|
)
|
|
RETURNING id, exercise_id, media_type, file_path, file_size, mime_type, original_filename,
|
|
embed_url, embed_platform, title, description, sort_order, is_primary, context, created_at""",
|
|
(
|
|
exercise_id,
|
|
media_type,
|
|
emb,
|
|
platform,
|
|
title or None,
|
|
description or None,
|
|
context,
|
|
is_primary,
|
|
exercise_id,
|
|
),
|
|
)
|
|
else:
|
|
raw = await file.read()
|
|
if len(raw) > MAX_UPLOAD_BYTES:
|
|
raise HTTPException(
|
|
status_code=413,
|
|
detail=f"Datei zu groß (max. {MAX_UPLOAD_BYTES // (1024 * 1024)} MB)",
|
|
)
|
|
mime = file.content_type or ""
|
|
if mime not in ALLOWED_UPLOAD_MIMES:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Dateityp nicht erlaubt: {mime or 'unbekannt'}",
|
|
)
|
|
ext = Path(file.filename or "").suffix[:12] if file.filename else ""
|
|
if not ext and mime == "image/jpeg":
|
|
ext = ".jpg"
|
|
elif not ext and mime == "image/png":
|
|
ext = ".png"
|
|
digest = hashlib.sha256(raw).hexdigest()[:12]
|
|
fname = f"{digest}_{exercise_id}{ext}"
|
|
dest_dir = _ensure_media_dirs()
|
|
dest_path = dest_dir / fname
|
|
dest_path.write_bytes(raw)
|
|
db_path = f"/media/exercises/{fname}"
|
|
cur.execute(
|
|
f"""INSERT INTO exercise_media (
|
|
exercise_id, media_type, file_path, file_size, mime_type, original_filename,
|
|
embed_url, embed_platform, title, description, context, is_primary, sort_order
|
|
) VALUES (
|
|
%s, %s, %s, %s, %s, %s, NULL, NULL, %s, %s, %s, %s, {sort_sql}
|
|
)
|
|
RETURNING id, exercise_id, media_type, file_path, file_size, mime_type, original_filename,
|
|
embed_url, embed_platform, title, description, sort_order, is_primary, context, created_at""",
|
|
(
|
|
exercise_id,
|
|
media_type,
|
|
db_path,
|
|
len(raw),
|
|
mime,
|
|
file.filename,
|
|
title or None,
|
|
description or None,
|
|
context,
|
|
is_primary,
|
|
exercise_id,
|
|
),
|
|
)
|
|
row = cur.fetchone()
|
|
conn.commit()
|
|
return r2d(row)
|
|
|
|
|
|
@router.put("/exercises/{exercise_id}/media/reorder")
|
|
def reorder_exercise_media(
|
|
exercise_id: int,
|
|
body: ExerciseMediaReorder,
|
|
session: dict = Depends(require_auth),
|
|
):
|
|
profile_id = session["profile_id"]
|
|
ids = body.media_ids
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
_assert_can_edit_exercise(cur, exercise_id, profile_id)
|
|
cur.execute(
|
|
"SELECT id FROM exercise_media WHERE exercise_id = %s ORDER BY sort_order, id",
|
|
(exercise_id,),
|
|
)
|
|
existing = [r["id"] if isinstance(r, dict) else r[0] for r in cur.fetchall()]
|
|
if set(ids) != set(existing) or len(ids) != len(existing):
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="media_ids unvollständig oder gehören nicht zu dieser Übung",
|
|
)
|
|
for i, mid in enumerate(ids, start=1):
|
|
cur.execute(
|
|
"UPDATE exercise_media SET sort_order = %s, updated_at = NOW() WHERE id = %s AND exercise_id = %s",
|
|
(i, mid, exercise_id),
|
|
)
|
|
conn.commit()
|
|
return {"ok": True, "reordered": len(ids)}
|
|
|
|
|
|
@router.put("/exercises/{exercise_id}/media/{media_id}")
|
|
def update_exercise_media(
|
|
exercise_id: int,
|
|
media_id: int,
|
|
body: ExerciseMediaUpdate,
|
|
session: dict = Depends(require_auth),
|
|
):
|
|
profile_id = session["profile_id"]
|
|
data = body.dict(exclude_unset=True)
|
|
if not data:
|
|
raise HTTPException(status_code=400, detail="Keine Felder zum Aktualisieren")
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
_assert_can_edit_exercise(cur, exercise_id, profile_id)
|
|
if not _fetch_media_row(cur, exercise_id, media_id):
|
|
raise HTTPException(status_code=404, detail="Medium nicht gefunden")
|
|
if "context" in data and data["context"] not in ("ablauf", "detail", "trainer_hint", None):
|
|
raise HTTPException(status_code=400, detail="Ungültiger context")
|
|
|
|
fields = []
|
|
params = []
|
|
for k in ("title", "description", "is_primary", "context"):
|
|
if k in data:
|
|
fields.append(f"{k} = %s")
|
|
params.append(data[k])
|
|
if fields:
|
|
fields.append("updated_at = NOW()")
|
|
params.extend([media_id, exercise_id])
|
|
cur.execute(
|
|
f"UPDATE exercise_media SET {', '.join(fields)} WHERE id = %s AND exercise_id = %s",
|
|
params,
|
|
)
|
|
conn.commit()
|
|
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, created_at
|
|
FROM exercise_media WHERE id = %s""",
|
|
(media_id,),
|
|
)
|
|
return r2d(cur.fetchone())
|
|
|
|
|
|
@router.delete("/exercises/{exercise_id}/media/{media_id}")
|
|
def delete_exercise_media(
|
|
exercise_id: int,
|
|
media_id: int,
|
|
session: dict = Depends(require_auth),
|
|
):
|
|
profile_id = session["profile_id"]
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
_assert_can_edit_exercise(cur, exercise_id, profile_id)
|
|
cur.execute(
|
|
"""SELECT file_path FROM exercise_media WHERE id = %s AND exercise_id = %s""",
|
|
(media_id, exercise_id),
|
|
)
|
|
row = cur.fetchone()
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="Medium nicht gefunden")
|
|
fp = row["file_path"] if isinstance(row, dict) else row[0]
|
|
cur.execute(
|
|
"DELETE FROM exercise_media WHERE id = %s AND exercise_id = %s",
|
|
(media_id, exercise_id),
|
|
)
|
|
conn.commit()
|
|
|
|
abs_p = _abs_media_path(fp) if fp else None
|
|
if abs_p and abs_p.is_file():
|
|
try:
|
|
abs_p.unlink()
|
|
except OSError as e:
|
|
logger.warning("Medien-Datei konnte nicht gelöscht werden: %s", e)
|
|
|
|
return {"ok": True}
|