Lars
756263bad4
- Introduced role-based upload limits for exercise media, allowing users to upload up to 1 MB while admins and superadmins can upload up to 1024 MB. - Updated the media upload logic to enforce these limits, enhancing the flexibility and control over media management in the application.
1371 lines
50 KiB
Python
1371 lines
50 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"])
|
||
|
||
# Kanonische Fähigkeitsstufen 1–5 (Übung ↔ Skill-Zeile), siehe Migration 029
|
||
_CANONICAL_SKILL_LEVELS = frozenset(
|
||
{"basis", "grundlagen", "aufbau", "fortgeschritten", "optimierung"}
|
||
)
|
||
_LEGACY_SKILL_LEVEL_SLUG = {
|
||
"einsteiger": "basis",
|
||
"experte": "optimierung",
|
||
"1": "basis",
|
||
"2": "grundlagen",
|
||
"3": "aufbau",
|
||
"4": "fortgeschritten",
|
||
"5": "optimierung",
|
||
}
|
||
|
||
# SQL: numerischer Rang aus target_level (fallback required_level) für Filter
|
||
_EXERCISE_SKILL_LEVEL_RANK_SQL = """
|
||
CASE COALESCE(
|
||
NULLIF(TRIM(LOWER(es.target_level::text)), ''),
|
||
NULLIF(TRIM(LOWER(es.required_level::text)), '')
|
||
)
|
||
WHEN 'basis' THEN 1
|
||
WHEN 'grundlagen' THEN 2
|
||
WHEN 'aufbau' THEN 3
|
||
WHEN 'fortgeschritten' THEN 4
|
||
WHEN 'optimierung' THEN 5
|
||
WHEN 'einsteiger' THEN 1
|
||
WHEN 'experte' THEN 5
|
||
WHEN '1' THEN 1
|
||
WHEN '2' THEN 2
|
||
WHEN '3' THEN 3
|
||
WHEN '4' THEN 4
|
||
WHEN '5' THEN 5
|
||
ELSE NULL END
|
||
""".strip()
|
||
|
||
|
||
def normalize_exercise_skill_level(value) -> Optional[str]:
|
||
"""Wandelt Legacy-/Zahlencodes in kanonische Slugs; ungültig → None."""
|
||
if value is None:
|
||
return None
|
||
s = str(value).strip().lower()
|
||
if not s:
|
||
return None
|
||
if s in _CANONICAL_SKILL_LEVELS:
|
||
return s
|
||
return _LEGACY_SKILL_LEVEL_SLUG.get(s)
|
||
|
||
MEDIA_ROOT = Path(os.getenv("MEDIA_ROOT", str(Path(__file__).resolve().parent.parent / "media")))
|
||
MAX_EXERCISE_MEDIA = 10
|
||
# Upload-Limits (Übungs-Medien): Trainer wie bisher kleiner; Admin/Superadmin höheres Limit für große Videos
|
||
_MAX_UPLOAD_MB_USER = max(1, int(os.getenv("EXERCISE_MEDIA_MAX_UPLOAD_MB", "50")))
|
||
_MAX_UPLOAD_MB_ADMIN = max(_MAX_UPLOAD_MB_USER, int(os.getenv("EXERCISE_MEDIA_ADMIN_MAX_UPLOAD_MB", "1024")))
|
||
MAX_UPLOAD_BYTES_USER = _MAX_UPLOAD_MB_USER * 1024 * 1024
|
||
MAX_UPLOAD_BYTES_ADMIN = _MAX_UPLOAD_MB_ADMIN * 1024 * 1024
|
||
ALLOWED_UPLOAD_MIMES = frozenset(
|
||
{"image/jpeg", "image/png", "image/gif", "video/mp4", "application/pdf"}
|
||
)
|
||
|
||
|
||
def _upload_limit_bytes(session: dict) -> int:
|
||
role = session.get("role") or ""
|
||
if role in ("admin", "superadmin"):
|
||
return MAX_UPLOAD_BYTES_ADMIN
|
||
return MAX_UPLOAD_BYTES_USER
|
||
|
||
|
||
# ============================================================================
|
||
# 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] = []
|
||
training_types_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
|
||
training_types_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]
|
||
|
||
|
||
class ExerciseVariantCreate(BaseModel):
|
||
variant_name: str = Field(..., min_length=3, max_length=200)
|
||
description: Optional[str] = None
|
||
execution_changes: Optional[str] = None
|
||
duration_min: Optional[int] = None
|
||
duration_max: Optional[int] = None
|
||
equipment_changes: Optional[list[str]] = None
|
||
difficulty_adjustment: Optional[str] = Field(None, max_length=50)
|
||
progression_level: int = Field(default=1, ge=1, le=10)
|
||
sequence_order: Optional[int] = None
|
||
prerequisite_variant_id: Optional[int] = None
|
||
|
||
|
||
class ExerciseVariantUpdate(BaseModel):
|
||
variant_name: Optional[str] = Field(None, min_length=3, max_length=200)
|
||
description: Optional[str] = None
|
||
execution_changes: Optional[str] = None
|
||
duration_min: Optional[int] = None
|
||
duration_max: Optional[int] = None
|
||
equipment_changes: Optional[list[str]] = None
|
||
difficulty_adjustment: Optional[str] = Field(None, max_length=50)
|
||
progression_level: Optional[int] = Field(None, ge=1, le=10)
|
||
sequence_order: Optional[int] = None
|
||
prerequisite_variant_id: Optional[int] = None
|
||
|
||
|
||
class ExerciseVariantsReorder(BaseModel):
|
||
variant_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 _variant_equipment_json(changes: Optional[list]) -> str:
|
||
return json.dumps(changes if changes else [])
|
||
|
||
|
||
def _normalize_variant_equipment_list(val) -> list:
|
||
if val is None:
|
||
return []
|
||
if isinstance(val, list):
|
||
return val
|
||
if isinstance(val, str):
|
||
try:
|
||
return json.loads(val)
|
||
except Exception:
|
||
return []
|
||
return []
|
||
|
||
|
||
def _validate_variant_prerequisite(cur, exercise_id: int, prereq_id: Optional[int]) -> None:
|
||
if prereq_id is None:
|
||
return
|
||
cur.execute(
|
||
"SELECT 1 FROM exercise_variants WHERE id = %s AND exercise_id = %s",
|
||
(prereq_id, exercise_id),
|
||
)
|
||
if not cur.fetchone():
|
||
raise HTTPException(status_code=400, detail="Voraussetzungs-Variante gehört nicht zu dieser Übung")
|
||
|
||
|
||
def _fetch_variant_row(cur, exercise_id: int, variant_id: int) -> dict:
|
||
cur.execute(
|
||
"""SELECT id, variant_name, description, execution_changes,
|
||
duration_min, duration_max, equipment_changes, difficulty_adjustment,
|
||
progression_level, sequence_order, prerequisite_variant_id, created_at
|
||
FROM exercise_variants WHERE id = %s AND exercise_id = %s""",
|
||
(variant_id, exercise_id),
|
||
)
|
||
row = cur.fetchone()
|
||
if not row:
|
||
raise HTTPException(status_code=404, detail="Variante nicht gefunden")
|
||
return r2d(row)
|
||
|
||
|
||
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()]
|
||
|
||
# Trainingsstil (Breitensport / Leistungssport …) — exercise_training_types
|
||
cur.execute(
|
||
"""SELECT ett.id, ett.training_type_id, tt.name, tt.abbreviation, ett.is_primary
|
||
FROM exercise_training_types ett
|
||
JOIN training_types tt ON ett.training_type_id = tt.id
|
||
WHERE ett.exercise_id = %s
|
||
ORDER BY ett.is_primary DESC, tt.sort_order NULLS LAST, tt.name""",
|
||
(exercise_id,),
|
||
)
|
||
exercise["training_types"] = [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()]
|
||
for sk in exercise["skills"]:
|
||
sk["required_level"] = normalize_exercise_skill_level(sk.get("required_level"))
|
||
sk["target_level"] = normalize_exercise_skill_level(sk.get("target_level"))
|
||
|
||
# Variants (1:N) - mit Progression (Reihenfolge: sequence_order, dann progression_level)
|
||
cur.execute(
|
||
"""SELECT id, variant_name, description, execution_changes,
|
||
duration_min, duration_max, equipment_changes, difficulty_adjustment,
|
||
progression_level, sequence_order, prerequisite_variant_id, created_at
|
||
FROM exercise_variants
|
||
WHERE exercise_id = %s
|
||
ORDER BY sequence_order NULLS LAST, progression_level, id""",
|
||
(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 (Stilrichtungen, z. B. Shotokan)
|
||
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))
|
||
)
|
||
|
||
# Trainingsstil (Breitensport, Leistungssport, …)
|
||
if "training_types_multi" in data:
|
||
cur.execute("DELETE FROM exercise_training_types WHERE exercise_id = %s", (exercise_id,))
|
||
for tt in data["training_types_multi"]:
|
||
cur.execute(
|
||
"""INSERT INTO exercise_training_types (exercise_id, training_type_id, is_primary)
|
||
VALUES (%s, %s, %s)""",
|
||
(exercise_id, tt["training_type_id"], tt.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"),
|
||
normalize_exercise_skill_level(skill.get("required_level")),
|
||
normalize_exercise_skill_level(skill.get("target_level")),
|
||
skill.get("ai_suggested", False),
|
||
)
|
||
)
|
||
|
||
conn.commit()
|
||
|
||
|
||
# ============================================================================
|
||
# Endpoints
|
||
# ============================================================================
|
||
|
||
def _merge_ids(multi: list[int], single: Optional[int]) -> list[int]:
|
||
"""Liste aus wiederholten Query-Parametern plus optional einem Legacy-Einzelfilter (ohne Duplikate)."""
|
||
seen: set[int] = set()
|
||
out: list[int] = []
|
||
for x in list(multi or []):
|
||
xi = int(x)
|
||
if xi not in seen:
|
||
seen.add(xi)
|
||
out.append(xi)
|
||
if single is not None:
|
||
xi = int(single)
|
||
if xi not in seen:
|
||
out.append(xi)
|
||
return out
|
||
|
||
|
||
def _merge_str_any(multi: list[str], single: Optional[str]) -> list[str]:
|
||
seen = set()
|
||
out = []
|
||
for x in list(multi or []):
|
||
s = str(x).strip()
|
||
if not s or s in seen:
|
||
continue
|
||
seen.add(s)
|
||
out.append(s)
|
||
if single is not None and str(single).strip():
|
||
s = str(single).strip()
|
||
if s not in seen:
|
||
out.append(s)
|
||
return out
|
||
|
||
|
||
@router.get("/exercises")
|
||
def list_exercises(
|
||
focus_area_ids: list[int] = Query(default=[], description="ODER: mind. einer dieser Fokusbereiche"),
|
||
focus_area: Optional[int] = Query(default=None, description="Einzel-ID (Legacy), wird mit focus_area_ids kombiniert"),
|
||
visibility_any: list[str] = Query(default=[], description="ODER: eine dieser Sichtbarkeiten"),
|
||
visibility: Optional[str] = Query(default=None, description="Einzel (Legacy)"),
|
||
status_any: list[str] = Query(default=[], description="ODER: einer dieser Statuswerte"),
|
||
status: Optional[str] = Query(default=None, description="Einzel (Legacy)"),
|
||
skill_ids: list[int] = Query(default=[], description="ODER: mind. eine dieser Fähigkeiten"),
|
||
skill_id: Optional[int] = Query(default=None, description="Einzel (Legacy)"),
|
||
style_direction_ids: list[int] = Query(default=[], description="ODER: mind. eine Stilrichtung"),
|
||
style_direction_id: Optional[int] = Query(default=None, description="Einzel (Legacy)"),
|
||
training_type_ids: list[int] = Query(default=[], description="ODER: mind. ein Trainingsstil"),
|
||
training_type_id: Optional[int] = Query(default=None, description="Einzel (Legacy)"),
|
||
target_group_ids: list[int] = Query(default=[], description="ODER: mind. eine Zielgruppe"),
|
||
target_group_id: Optional[int] = Query(default=None, description="Einzel (Legacy)"),
|
||
skill_min_level: Optional[int] = Query(default=None, ge=1, le=5),
|
||
skill_max_level: Optional[int] = Query(default=None, ge=1, le=5),
|
||
search: Optional[str] = Query(default=None),
|
||
ai_search: Optional[str] = Query(
|
||
default=None,
|
||
description="Platzhalter KI-Suche: derzeit gleiche Volltextlogik wie search (später Embeddings/Reranking)",
|
||
),
|
||
limit: int = Query(default=50, ge=1, le=100),
|
||
offset: int = Query(default=0, ge=0),
|
||
include_variants: bool = Query(
|
||
default=False,
|
||
description="Wenn true: Feld variants[] pro Übung (id, variant_name, sequence_order) für Planung/UI",
|
||
),
|
||
session: dict = Depends(require_auth),
|
||
):
|
||
"""
|
||
Liste aller Übungen mit Filtern.
|
||
Lightweight Response (ohne M:N Details, nur IDs und Namen).
|
||
Optional include_variants für Variantenauswahl in der Trainingsplanung.
|
||
"""
|
||
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)
|
||
|
||
vis_list = _merge_str_any(visibility_any, visibility)
|
||
if vis_list:
|
||
ph = ",".join(["%s"] * len(vis_list))
|
||
where.append(f"e.visibility IN ({ph})")
|
||
params.extend(vis_list)
|
||
|
||
st_list = _merge_str_any(status_any, status)
|
||
if st_list:
|
||
ph = ",".join(["%s"] * len(st_list))
|
||
where.append(f"e.status IN ({ph})")
|
||
params.extend(st_list)
|
||
|
||
fa_ids = _merge_ids(focus_area_ids, focus_area)
|
||
if fa_ids:
|
||
ph = ",".join(["%s"] * len(fa_ids))
|
||
where.append(
|
||
f"EXISTS (SELECT 1 FROM exercise_focus_areas efa WHERE efa.exercise_id = e.id AND efa.focus_area_id IN ({ph}))"
|
||
)
|
||
params.extend(fa_ids)
|
||
|
||
sk_ids = _merge_ids(skill_ids, skill_id)
|
||
if sk_ids:
|
||
ph = ",".join(["%s"] * len(sk_ids))
|
||
where.append(
|
||
f"EXISTS (SELECT 1 FROM exercise_skills es WHERE es.exercise_id = e.id AND es.skill_id IN ({ph}))"
|
||
)
|
||
params.extend(sk_ids)
|
||
|
||
sd_ids = _merge_ids(style_direction_ids, style_direction_id)
|
||
if sd_ids:
|
||
ph = ",".join(["%s"] * len(sd_ids))
|
||
where.append(
|
||
"EXISTS (SELECT 1 FROM exercise_style_directions esd "
|
||
f"WHERE esd.exercise_id = e.id AND esd.style_direction_id IN ({ph}))"
|
||
)
|
||
params.extend(sd_ids)
|
||
|
||
tt_ids = _merge_ids(training_type_ids, training_type_id)
|
||
if tt_ids:
|
||
ph = ",".join(["%s"] * len(tt_ids))
|
||
where.append(
|
||
"EXISTS (SELECT 1 FROM exercise_training_types ett "
|
||
f"WHERE ett.exercise_id = e.id AND ett.training_type_id IN ({ph}))"
|
||
)
|
||
params.extend(tt_ids)
|
||
|
||
tg_ids = _merge_ids(target_group_ids, target_group_id)
|
||
if tg_ids:
|
||
ph = ",".join(["%s"] * len(tg_ids))
|
||
where.append(
|
||
"EXISTS (SELECT 1 FROM exercise_target_groups etg "
|
||
f"WHERE etg.exercise_id = e.id AND etg.target_group_id IN ({ph}))"
|
||
)
|
||
params.extend(tg_ids)
|
||
|
||
if skill_min_level is not None or skill_max_level is not None:
|
||
lo = skill_min_level if skill_min_level is not None else 1
|
||
hi = skill_max_level if skill_max_level is not None else 5
|
||
if lo > hi:
|
||
lo, hi = hi, lo
|
||
where.append(
|
||
"EXISTS (SELECT 1 FROM exercise_skills es WHERE es.exercise_id = e.id AND ("
|
||
+ _EXERCISE_SKILL_LEVEL_RANK_SQL
|
||
+ ") BETWEEN %s AND %s)"
|
||
)
|
||
params.extend([lo, hi])
|
||
|
||
# Volltext (tsvector); ai_search gleiche Engine, bei zwei Begriffen ODER-Verknüpfung
|
||
s1 = (search or "").strip()
|
||
s2 = (ai_search or "").strip()
|
||
if s1 and s2 and s1 != s2:
|
||
where.append(
|
||
"(e.search_vector @@ plainto_tsquery('german', %s) "
|
||
"OR e.search_vector @@ plainto_tsquery('german', %s))"
|
||
)
|
||
params.extend([s1, s2])
|
||
elif s1 or s2:
|
||
qtext = s1 or s2
|
||
where.append("e.search_vector @@ plainto_tsquery('german', %s)")
|
||
params.append(qtext)
|
||
|
||
variants_sql = ""
|
||
if include_variants:
|
||
variants_sql = """,
|
||
(
|
||
SELECT COALESCE(
|
||
json_agg(
|
||
json_build_object(
|
||
'id', ev.id,
|
||
'variant_name', ev.variant_name,
|
||
'sequence_order', ev.sequence_order
|
||
)
|
||
ORDER BY ev.sequence_order NULLS LAST, ev.progression_level, ev.id
|
||
),
|
||
'[]'::json
|
||
)
|
||
FROM exercise_variants ev
|
||
WHERE ev.exercise_id = e.id
|
||
) AS variants"""
|
||
|
||
# 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
|
||
{variants_sql}
|
||
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
|
||
if include_variants:
|
||
v = d.get("variants")
|
||
if isinstance(v, str):
|
||
try:
|
||
d["variants"] = json.loads(v)
|
||
except Exception:
|
||
d["variants"] = []
|
||
elif v is None:
|
||
d["variants"] = []
|
||
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}
|
||
|
||
|
||
# --- Übungsvarianten (EXERCISES_API_SPEC.md) ---
|
||
|
||
|
||
@router.put("/exercises/{exercise_id}/variants/reorder")
|
||
def reorder_exercise_variants(
|
||
exercise_id: int,
|
||
body: ExerciseVariantsReorder,
|
||
session: dict = Depends(require_auth),
|
||
):
|
||
profile_id = session["profile_id"]
|
||
|
||
if len(body.variant_ids) != len(set(body.variant_ids)):
|
||
raise HTTPException(status_code=400, detail="variant_ids dürfen keine Duplikate enthalten")
|
||
|
||
with get_db() as conn:
|
||
cur = get_cursor(conn)
|
||
_assert_can_edit_exercise(cur, exercise_id, profile_id)
|
||
|
||
cur.execute(
|
||
"SELECT id FROM exercise_variants WHERE exercise_id = %s",
|
||
(exercise_id,),
|
||
)
|
||
existing = [r["id"] for r in cur.fetchall()]
|
||
if set(existing) != set(body.variant_ids):
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail="variant_ids müssen alle Varianten dieser Übung genau einmal enthalten",
|
||
)
|
||
|
||
for pos, vid in enumerate(body.variant_ids):
|
||
cur.execute(
|
||
"""UPDATE exercise_variants SET sequence_order = %s
|
||
WHERE id = %s AND exercise_id = %s""",
|
||
(pos + 1, vid, exercise_id),
|
||
)
|
||
conn.commit()
|
||
|
||
return {"ok": True, "reordered": len(body.variant_ids)}
|
||
|
||
|
||
@router.post("/exercises/{exercise_id}/variants", status_code=201)
|
||
def create_exercise_variant(
|
||
exercise_id: int,
|
||
body: ExerciseVariantCreate,
|
||
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)
|
||
_validate_variant_prerequisite(cur, exercise_id, body.prerequisite_variant_id)
|
||
|
||
eq_json = _variant_equipment_json(body.equipment_changes)
|
||
seq = body.sequence_order
|
||
if seq is None:
|
||
cur.execute(
|
||
"""SELECT COALESCE(MAX(sequence_order), 0) + 1 AS n
|
||
FROM exercise_variants WHERE exercise_id = %s""",
|
||
(exercise_id,),
|
||
)
|
||
seq = cur.fetchone()["n"]
|
||
|
||
desc = (body.description or "").strip() or None
|
||
exec_ch = (body.execution_changes or "").strip() or None
|
||
diff = (body.difficulty_adjustment or "").strip() or None
|
||
|
||
cur.execute(
|
||
"""INSERT INTO exercise_variants (
|
||
exercise_id, variant_name, description, execution_changes,
|
||
duration_min, duration_max, equipment_changes, difficulty_adjustment,
|
||
progression_level, sequence_order, prerequisite_variant_id
|
||
) VALUES (%s,%s,%s,%s,%s,%s,%s::jsonb,%s,%s,%s,%s)
|
||
RETURNING id""",
|
||
(
|
||
exercise_id,
|
||
body.variant_name.strip(),
|
||
desc,
|
||
exec_ch,
|
||
body.duration_min,
|
||
body.duration_max,
|
||
eq_json,
|
||
diff,
|
||
body.progression_level,
|
||
seq,
|
||
body.prerequisite_variant_id,
|
||
),
|
||
)
|
||
new_id = cur.fetchone()["id"]
|
||
row = _fetch_variant_row(cur, exercise_id, new_id)
|
||
conn.commit()
|
||
|
||
return row
|
||
|
||
|
||
@router.put("/exercises/{exercise_id}/variants/{variant_id}")
|
||
def update_exercise_variant(
|
||
exercise_id: int,
|
||
variant_id: int,
|
||
body: ExerciseVariantUpdate,
|
||
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)
|
||
old = _fetch_variant_row(cur, exercise_id, variant_id)
|
||
|
||
if "variant_name" in data and data["variant_name"] is not None:
|
||
old["variant_name"] = data["variant_name"].strip()
|
||
if "description" in data:
|
||
old["description"] = (data["description"] or "").strip() or None
|
||
if "execution_changes" in data:
|
||
old["execution_changes"] = (data["execution_changes"] or "").strip() or None
|
||
if "duration_min" in data:
|
||
old["duration_min"] = data["duration_min"]
|
||
if "duration_max" in data:
|
||
old["duration_max"] = data["duration_max"]
|
||
if "equipment_changes" in data:
|
||
old["equipment_changes"] = _normalize_variant_equipment_list(data["equipment_changes"])
|
||
if "difficulty_adjustment" in data:
|
||
old["difficulty_adjustment"] = (data["difficulty_adjustment"] or "").strip() or None
|
||
if "progression_level" in data and data["progression_level"] is not None:
|
||
old["progression_level"] = data["progression_level"]
|
||
if "sequence_order" in data:
|
||
old["sequence_order"] = data["sequence_order"]
|
||
if "prerequisite_variant_id" in data:
|
||
old["prerequisite_variant_id"] = data["prerequisite_variant_id"]
|
||
|
||
prereq = old.get("prerequisite_variant_id")
|
||
if prereq == variant_id:
|
||
raise HTTPException(status_code=400, detail="Variante kann nicht ihre eigene Voraussetzung sein")
|
||
_validate_variant_prerequisite(cur, exercise_id, prereq)
|
||
|
||
eq_db = _variant_equipment_json(_normalize_variant_equipment_list(old.get("equipment_changes")))
|
||
|
||
cur.execute(
|
||
"""UPDATE exercise_variants SET
|
||
variant_name = %s,
|
||
description = %s,
|
||
execution_changes = %s,
|
||
duration_min = %s,
|
||
duration_max = %s,
|
||
equipment_changes = %s::jsonb,
|
||
difficulty_adjustment = %s,
|
||
progression_level = %s,
|
||
sequence_order = %s,
|
||
prerequisite_variant_id = %s
|
||
WHERE id = %s AND exercise_id = %s""",
|
||
(
|
||
old["variant_name"],
|
||
old.get("description"),
|
||
old.get("execution_changes"),
|
||
old.get("duration_min"),
|
||
old.get("duration_max"),
|
||
eq_db,
|
||
old.get("difficulty_adjustment"),
|
||
old.get("progression_level"),
|
||
old.get("sequence_order"),
|
||
old.get("prerequisite_variant_id"),
|
||
variant_id,
|
||
exercise_id,
|
||
),
|
||
)
|
||
row = _fetch_variant_row(cur, exercise_id, variant_id)
|
||
conn.commit()
|
||
|
||
return row
|
||
def delete_exercise_variant(
|
||
exercise_id: int,
|
||
variant_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)
|
||
_fetch_variant_row(cur, exercise_id, variant_id)
|
||
|
||
cur.execute(
|
||
"SELECT COUNT(*) AS c FROM exercise_variants WHERE prerequisite_variant_id = %s",
|
||
(variant_id,),
|
||
)
|
||
cnt = int(cur.fetchone()["c"])
|
||
if cnt > 0:
|
||
raise HTTPException(
|
||
status_code=409,
|
||
detail="Variante ist Voraussetzung anderer Varianten — zuerst dort ändern oder entfernen",
|
||
)
|
||
|
||
cur.execute(
|
||
"DELETE FROM exercise_variants WHERE id = %s AND exercise_id = %s",
|
||
(variant_id, 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()
|
||
max_upload = _upload_limit_bytes(session)
|
||
if len(raw) > max_upload:
|
||
raise HTTPException(
|
||
status_code=413,
|
||
detail=f"Datei zu groß (max. {max_upload // (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}
|