Lars
43c6abce4a
Problem: Hard-codierte Werte (Fokusbereich, Trainingscharakter) + fehlende Dimensionen (Stil, Fähigkeiten-Matrix) + keine Rollen-basierte Sichtbarkeit Lösung: Dynamische Kataloge mit Admin-CRUD Migration 007_exercise_catalogs.sql: - focus_areas (statt hard-coded 'karate', 'selbstverteidigung', 'gewaltschutz') - training_styles (NEU: Shotokan, Goju-Ryu, Wado-Ryu, etc. mit Hierarchie) - training_characters (statt hard-coded 'grundlage', 'aufbau', etc.) - skill_categories (Matrix: Kategorien → Einzelfähigkeiten) - trainer_focus_areas (Zuordnung: Trainer → Fokusbereiche) - exercises erweitert: training_style_id, training_character_id, focus_area_id - skills erweitert: category_id, parent_skill_id, level, sort_order - Seed-Daten für alle Kataloge Backend (routers/catalogs.py): - CRUD für focus_areas (admin only) - CRUD für training_styles (admin only, mit parent_style_id) - CRUD für training_characters (admin only) - CRUD für skill_categories (admin only, mit parent_category_id) - CRUD für trainer_focus_areas (admin: assign, trainer: read own) - Alle mit status-Filter (active/inactive) Backend (routers/exercises.py): - CREATE/UPDATE erweitert um training_style_id, training_character_id, focus_area_id - Legacy-Felder (focus_area text, training_character text) bleiben parallel Backend (main.py): - catalogs Router registriert Nächster Schritt: Frontend-UI (Admin-Kataloge + Exercise-Formular-Update)
345 lines
12 KiB
Python
345 lines
12 KiB
Python
"""
|
|
Exercise Management Endpoints for Shinkan Jinkendo
|
|
|
|
Handles CRUD operations for exercises (Übungen).
|
|
"""
|
|
from typing import Optional
|
|
from fastapi import APIRouter, HTTPException, Depends, Query
|
|
|
|
from db import get_db, get_cursor, r2d
|
|
from auth import require_auth
|
|
|
|
router = APIRouter(prefix="/api", tags=["exercises"])
|
|
|
|
|
|
# ── List Exercises ────────────────────────────────────────────────────────
|
|
@router.get("/exercises")
|
|
def list_exercises(
|
|
focus_area: Optional[str] = Query(default=None),
|
|
visibility: Optional[str] = Query(default=None),
|
|
status: Optional[str] = Query(default=None),
|
|
skill_id: Optional[int] = Query(default=None),
|
|
session=Depends(require_auth)
|
|
):
|
|
"""
|
|
List exercises with optional filters.
|
|
|
|
Filters:
|
|
- focus_area: karate, selbstverteidigung, gewaltschutz
|
|
- visibility: private, club, official
|
|
- status: draft, in_review, approved, archived
|
|
- skill_id: Filter by associated skill
|
|
"""
|
|
profile_id = session['profile_id']
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
|
|
# Base query
|
|
query = """
|
|
SELECT DISTINCT e.*,
|
|
p.name as creator_name,
|
|
c.name as club_name,
|
|
tm.name as primary_method_name
|
|
FROM exercises e
|
|
LEFT JOIN profiles p ON e.created_by = p.id
|
|
LEFT JOIN clubs c ON e.club_id = c.id
|
|
LEFT JOIN training_methods tm ON e.primary_method_id = tm.id
|
|
"""
|
|
|
|
# Add skill join if filtering by skill
|
|
if skill_id:
|
|
query += " LEFT JOIN exercise_skills es ON e.id = es.exercise_id"
|
|
|
|
# Build WHERE clause
|
|
where = []
|
|
params = []
|
|
|
|
# Visibility filter: show own private + club + official
|
|
where.append("(e.visibility = 'official' OR e.visibility = 'club' OR e.created_by = %s)")
|
|
params.append(profile_id)
|
|
|
|
if focus_area:
|
|
where.append("e.focus_area = %s")
|
|
params.append(focus_area)
|
|
|
|
if visibility:
|
|
where.append("e.visibility = %s")
|
|
params.append(visibility)
|
|
|
|
if status:
|
|
where.append("e.status = %s")
|
|
params.append(status)
|
|
|
|
if skill_id:
|
|
where.append("es.skill_id = %s")
|
|
params.append(skill_id)
|
|
|
|
if where:
|
|
query += " WHERE " + " AND ".join(where)
|
|
|
|
query += " ORDER BY e.created_at DESC"
|
|
|
|
cur.execute(query, params)
|
|
rows = cur.fetchall()
|
|
return [r2d(r) for r in rows]
|
|
|
|
|
|
# ── Get Exercise ──────────────────────────────────────────────────────────
|
|
@router.get("/exercises/{exercise_id}")
|
|
def get_exercise(exercise_id: int, session=Depends(require_auth)):
|
|
"""Get exercise by ID with associated skills and variants."""
|
|
profile_id = session['profile_id']
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
|
|
# Get exercise
|
|
cur.execute("""
|
|
SELECT e.*,
|
|
p.name as creator_name,
|
|
c.name as club_name,
|
|
tm.name as primary_method_name
|
|
FROM exercises e
|
|
LEFT JOIN profiles p ON e.created_by = p.id
|
|
LEFT JOIN clubs c ON e.club_id = c.id
|
|
LEFT JOIN training_methods tm ON e.primary_method_id = tm.id
|
|
WHERE e.id = %s
|
|
""", (exercise_id,))
|
|
exercise = cur.fetchone()
|
|
|
|
if not exercise:
|
|
raise HTTPException(404, "Übung nicht gefunden")
|
|
|
|
exercise = r2d(exercise)
|
|
|
|
# Check visibility
|
|
if exercise['visibility'] == 'private' and exercise['created_by'] != profile_id:
|
|
raise HTTPException(403, "Keine Berechtigung")
|
|
|
|
# Get associated skills
|
|
cur.execute("""
|
|
SELECT es.*, s.name as skill_name, s.category as skill_category
|
|
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()]
|
|
|
|
# Get variants
|
|
cur.execute("""
|
|
SELECT * FROM exercise_variants
|
|
WHERE exercise_id = %s
|
|
ORDER BY created_at
|
|
""", (exercise_id,))
|
|
exercise['variants'] = [r2d(r) for r in cur.fetchall()]
|
|
|
|
# Get media
|
|
cur.execute("""
|
|
SELECT * FROM exercise_media
|
|
WHERE exercise_id = %s
|
|
ORDER BY sort_order, created_at
|
|
""", (exercise_id,))
|
|
exercise['media'] = [r2d(r) for r in cur.fetchall()]
|
|
|
|
return exercise
|
|
|
|
|
|
# ── Create Exercise ───────────────────────────────────────────────────────
|
|
@router.post("/exercises")
|
|
def create_exercise(data: dict, session=Depends(require_auth)):
|
|
"""Create new exercise."""
|
|
profile_id = session['profile_id']
|
|
|
|
# Required fields
|
|
title = data.get('title')
|
|
goal = data.get('goal')
|
|
execution = data.get('execution')
|
|
|
|
if not title or not goal or not execution:
|
|
raise HTTPException(400, "Titel, Ziel und Durchführung sind Pflichtfelder")
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
|
|
# Insert exercise
|
|
cur.execute("""
|
|
INSERT INTO exercises (
|
|
title, summary, goal, execution, preparation, trainer_notes,
|
|
equipment, duration_min, duration_max, group_size_min, group_size_max,
|
|
age_groups, focus_area, secondary_areas, training_character,
|
|
primary_method_id, secondary_method_ids,
|
|
training_style_id, training_character_id, focus_area_id,
|
|
visibility, status, created_by, club_id
|
|
) VALUES (
|
|
%s, %s, %s, %s, %s, %s,
|
|
%s, %s, %s, %s, %s,
|
|
%s, %s, %s, %s,
|
|
%s, %s,
|
|
%s, %s, %s,
|
|
%s, %s, %s, %s
|
|
) RETURNING id
|
|
""", (
|
|
title,
|
|
data.get('summary'),
|
|
goal,
|
|
execution,
|
|
data.get('preparation'),
|
|
data.get('trainer_notes'),
|
|
data.get('equipment'), # JSONB
|
|
data.get('duration_min'),
|
|
data.get('duration_max'),
|
|
data.get('group_size_min'),
|
|
data.get('group_size_max'),
|
|
data.get('age_groups'), # JSONB
|
|
data.get('focus_area'), # Legacy
|
|
data.get('secondary_areas'), # JSONB
|
|
data.get('training_character'), # Legacy
|
|
data.get('primary_method_id'),
|
|
data.get('secondary_method_ids'), # JSONB
|
|
data.get('training_style_id'), # NEU
|
|
data.get('training_character_id'), # NEU
|
|
data.get('focus_area_id'), # NEU
|
|
data.get('visibility', 'private'),
|
|
data.get('status', 'draft'),
|
|
profile_id,
|
|
data.get('club_id')
|
|
))
|
|
|
|
exercise_id = cur.fetchone()['id']
|
|
|
|
# Add skills if provided
|
|
if data.get('skills'):
|
|
for skill in data['skills']:
|
|
cur.execute("""
|
|
INSERT INTO exercise_skills (
|
|
exercise_id, skill_id, is_primary, intensity,
|
|
development_contribution, required_level, target_level
|
|
) VALUES (%s, %s, %s, %s, %s, %s, %s)
|
|
""", (
|
|
exercise_id,
|
|
skill['skill_id'],
|
|
skill.get('is_primary', False),
|
|
skill.get('intensity'),
|
|
skill.get('development_contribution'),
|
|
skill.get('required_level'),
|
|
skill.get('target_level')
|
|
))
|
|
|
|
conn.commit()
|
|
|
|
return get_exercise(exercise_id, session)
|
|
|
|
|
|
# ── Update Exercise ───────────────────────────────────────────────────────
|
|
@router.put("/exercises/{exercise_id}")
|
|
def update_exercise(exercise_id: int, data: dict, session=Depends(require_auth)):
|
|
"""Update exercise."""
|
|
profile_id = session['profile_id']
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
|
|
# Check ownership
|
|
cur.execute("SELECT created_by FROM exercises WHERE id = %s", (exercise_id,))
|
|
row = cur.fetchone()
|
|
if not row:
|
|
raise HTTPException(404, "Übung nicht gefunden")
|
|
|
|
if row['created_by'] != profile_id:
|
|
raise HTTPException(403, "Keine Berechtigung")
|
|
|
|
# Update exercise
|
|
cur.execute("""
|
|
UPDATE exercises SET
|
|
title = %s, summary = %s, goal = %s, execution = %s,
|
|
preparation = %s, trainer_notes = %s, equipment = %s,
|
|
duration_min = %s, duration_max = %s,
|
|
group_size_min = %s, group_size_max = %s,
|
|
age_groups = %s, focus_area = %s, secondary_areas = %s,
|
|
training_character = %s, primary_method_id = %s,
|
|
secondary_method_ids = %s,
|
|
training_style_id = %s, training_character_id = %s, focus_area_id = %s,
|
|
visibility = %s, status = %s,
|
|
club_id = %s, updated_at = NOW()
|
|
WHERE id = %s
|
|
""", (
|
|
data.get('title'),
|
|
data.get('summary'),
|
|
data.get('goal'),
|
|
data.get('execution'),
|
|
data.get('preparation'),
|
|
data.get('trainer_notes'),
|
|
data.get('equipment'),
|
|
data.get('duration_min'),
|
|
data.get('duration_max'),
|
|
data.get('group_size_min'),
|
|
data.get('group_size_max'),
|
|
data.get('age_groups'),
|
|
data.get('focus_area'), # Legacy
|
|
data.get('secondary_areas'),
|
|
data.get('training_character'), # Legacy
|
|
data.get('primary_method_id'),
|
|
data.get('secondary_method_ids'),
|
|
data.get('training_style_id'), # NEU
|
|
data.get('training_character_id'), # NEU
|
|
data.get('focus_area_id'), # NEU
|
|
data.get('visibility'),
|
|
data.get('status'),
|
|
data.get('club_id'),
|
|
exercise_id
|
|
))
|
|
|
|
# Update skills if provided
|
|
if 'skills' in data:
|
|
# Delete existing skills
|
|
cur.execute("DELETE FROM exercise_skills WHERE exercise_id = %s", (exercise_id,))
|
|
|
|
# Add new skills
|
|
for skill in data['skills']:
|
|
cur.execute("""
|
|
INSERT INTO exercise_skills (
|
|
exercise_id, skill_id, is_primary, intensity,
|
|
development_contribution, required_level, target_level
|
|
) VALUES (%s, %s, %s, %s, %s, %s, %s)
|
|
""", (
|
|
exercise_id,
|
|
skill['skill_id'],
|
|
skill.get('is_primary', False),
|
|
skill.get('intensity'),
|
|
skill.get('development_contribution'),
|
|
skill.get('required_level'),
|
|
skill.get('target_level')
|
|
))
|
|
|
|
conn.commit()
|
|
|
|
return get_exercise(exercise_id, session)
|
|
|
|
|
|
# ── Delete Exercise ───────────────────────────────────────────────────────
|
|
@router.delete("/exercises/{exercise_id}")
|
|
def delete_exercise(exercise_id: int, session=Depends(require_auth)):
|
|
"""Delete exercise (only owner or admin)."""
|
|
profile_id = session['profile_id']
|
|
role = session.get('role')
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
|
|
# Check ownership or admin
|
|
cur.execute("SELECT created_by FROM exercises WHERE id = %s", (exercise_id,))
|
|
row = cur.fetchone()
|
|
if not row:
|
|
raise HTTPException(404, "Übung nicht gefunden")
|
|
|
|
if row['created_by'] != profile_id and role not in ['admin', 'superadmin']:
|
|
raise HTTPException(403, "Keine Berechtigung")
|
|
|
|
# Delete (CASCADE handles skills, variants, media)
|
|
cur.execute("DELETE FROM exercises WHERE id = %s", (exercise_id,))
|
|
conn.commit()
|
|
|
|
return {"ok": True}
|