Lars
69b26fc928
- Updated role permissions to allow trainers and users to create clubs and training groups. - Modified database insertion logic to reflect the correct role for trainers during registration. - Enhanced frontend components to display appropriate messages and buttons based on user roles. - Improved user guidance in the Clubs and Training Planning pages, emphasizing the need for clubs and groups before planning training sessions.
836 lines
29 KiB
Python
836 lines
29 KiB
Python
"""
|
||
Training Planning – Trainingseinheiten, strukturierter Ablauf (Sektionen + Übungen/Notizen)
|
||
und wiederverwendbare Trainingsvorlagen (Sektions-Gliederung).
|
||
|
||
Governance (Vorlagen-rechte über Vereine/„offiziell“) kann später nachgezogen werden.
|
||
"""
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
from fastapi import APIRouter, Depends, HTTPException, Query
|
||
|
||
from db import get_db, get_cursor, r2d
|
||
from auth import require_auth
|
||
|
||
router = APIRouter(prefix="/api", tags=["training_planning"])
|
||
|
||
|
||
def _has_planning_role(role: Optional[str]) -> bool:
|
||
"""Kann Trainingseinheiten/Vorlagen anlegen (bis Governance: auch einfacher Account)."""
|
||
return role in ("admin", "superadmin", "trainer", "user")
|
||
|
||
|
||
def _optional_positive_int(val, field_name: str) -> Optional[int]:
|
||
if val is None or val == "":
|
||
return None
|
||
try:
|
||
i = int(val)
|
||
except (TypeError, ValueError):
|
||
raise HTTPException(status_code=400, detail=f"Ungültige {field_name}")
|
||
if i < 1:
|
||
raise HTTPException(status_code=400, detail=f"Ungültige {field_name}")
|
||
return i
|
||
|
||
|
||
def _validate_variant_for_exercise(cur, exercise_id: Optional[int], variant_id: Optional[int]):
|
||
if not variant_id:
|
||
return
|
||
if not exercise_id:
|
||
raise HTTPException(
|
||
status_code=400, detail="exercise_variant_id nur zusammen mit exercise_id erlaubt"
|
||
)
|
||
cur.execute(
|
||
"SELECT 1 FROM exercise_variants WHERE id = %s AND exercise_id = %s",
|
||
(variant_id, exercise_id),
|
||
)
|
||
if not cur.fetchone():
|
||
raise HTTPException(status_code=400, detail="Variante passt nicht zur gewählten Übung")
|
||
|
||
|
||
def _can_access_group_for_create(cur, group_id: int, profile_id: int, role: str) -> None:
|
||
cur.execute(
|
||
"SELECT trainer_id, co_trainer_ids FROM training_groups WHERE id = %s",
|
||
(group_id,),
|
||
)
|
||
group = cur.fetchone()
|
||
if not group:
|
||
raise HTTPException(status_code=404, detail="Trainingsgruppe nicht gefunden")
|
||
co_trainers = group["co_trainer_ids"] or []
|
||
if not _has_planning_role(role):
|
||
raise HTTPException(status_code=403, detail="Nur Trainer dürfen Trainingseinheiten erstellen")
|
||
if role not in ["admin", "superadmin"]:
|
||
if group["trainer_id"] != profile_id and profile_id not in co_trainers:
|
||
raise HTTPException(
|
||
status_code=403, detail="Nur der zuständige Trainer darf für diese Gruppe planen"
|
||
)
|
||
|
||
|
||
def _training_unit_guard_row(cur, unit_id: int) -> Dict[str, Any]:
|
||
cur.execute(
|
||
"""
|
||
SELECT tu.id, tu.created_by, tu.group_id, tu.plan_template_id,
|
||
tg.trainer_id, tg.co_trainer_ids
|
||
FROM training_units tu
|
||
LEFT JOIN training_groups tg ON tu.group_id = tg.id
|
||
WHERE tu.id = %s
|
||
""",
|
||
(unit_id,),
|
||
)
|
||
row = cur.fetchone()
|
||
if not row:
|
||
raise HTTPException(status_code=404, detail="Trainingseinheit nicht gefunden")
|
||
return r2d(row)
|
||
|
||
|
||
def _assert_training_unit_permission(
|
||
cur, unit_row: Dict[str, Any], profile_id: int, role: str
|
||
) -> None:
|
||
co_trainers = unit_row["co_trainer_ids"] or []
|
||
if role not in ["admin", "superadmin"]:
|
||
if (
|
||
unit_row["created_by"] != profile_id
|
||
and unit_row["trainer_id"] != profile_id
|
||
and profile_id not in co_trainers
|
||
):
|
||
raise HTTPException(status_code=403, detail="Keine Berechtigung")
|
||
|
||
|
||
def _assert_delete_training_unit(role: str, created_by: int, profile_id: int) -> None:
|
||
if role not in ["admin", "superadmin"] and created_by != profile_id:
|
||
raise HTTPException(status_code=403, detail="Keine Berechtigung")
|
||
|
||
|
||
def _fetch_sections(cur, unit_id: int) -> List[Dict[str, Any]]:
|
||
cur.execute(
|
||
"""
|
||
SELECT id, training_unit_id, order_index, title, guidance_notes, source_template_section_id
|
||
FROM training_unit_sections
|
||
WHERE training_unit_id = %s
|
||
ORDER BY order_index
|
||
""",
|
||
(unit_id,),
|
||
)
|
||
secs = []
|
||
for sec_row in cur.fetchall():
|
||
sec = r2d(sec_row)
|
||
cur.execute(
|
||
"""
|
||
SELECT tusi.*,
|
||
e.title AS exercise_title,
|
||
e.summary AS exercise_summary,
|
||
e.focus_area AS exercise_focus_area,
|
||
ev.variant_name AS exercise_variant_name
|
||
FROM training_unit_section_items tusi
|
||
LEFT JOIN exercises e ON tusi.exercise_id = e.id
|
||
LEFT JOIN exercise_variants ev ON tusi.exercise_variant_id = ev.id
|
||
WHERE tusi.section_id = %s
|
||
ORDER BY tusi.order_index
|
||
""",
|
||
(sec["id"],),
|
||
)
|
||
sec["items"] = [r2d(r) for r in cur.fetchall()]
|
||
secs.append(sec)
|
||
return secs
|
||
|
||
|
||
def _flatten_exercises_from_sections(unit: Dict[str, Any]) -> None:
|
||
flat: List[Dict[str, Any]] = []
|
||
for sec in sorted(unit.get("sections", []), key=lambda s: s.get("order_index", 0)):
|
||
for item in sorted(sec.get("items", []), key=lambda i: i.get("order_index", 0)):
|
||
if item.get("item_type") == "exercise":
|
||
flat.append(item)
|
||
unit["exercises"] = flat
|
||
|
||
|
||
def _hydrate_training_unit_payload(cur, unit: Dict[str, Any]) -> Dict[str, Any]:
|
||
uid = unit["id"]
|
||
unit["sections"] = _fetch_sections(cur, uid)
|
||
_flatten_exercises_from_sections(unit)
|
||
return unit
|
||
|
||
|
||
def _insert_section_items(cur, section_id: int, items_in: Optional[List[Any]], start_order: int = 0):
|
||
if items_in is None:
|
||
items_in = []
|
||
for i, raw in enumerate(items_in):
|
||
itype = raw.get("item_type")
|
||
if not itype:
|
||
itype = "exercise" if raw.get("exercise_id") else "note"
|
||
order_ix = raw.get("order_index")
|
||
if order_ix is None:
|
||
order_ix = start_order + i
|
||
if itype == "note":
|
||
body = raw.get("note_body")
|
||
if body is None:
|
||
body = ""
|
||
cur.execute(
|
||
"""
|
||
INSERT INTO training_unit_section_items (
|
||
section_id, order_index, item_type,
|
||
exercise_id, exercise_variant_id,
|
||
planned_duration_min, actual_duration_min,
|
||
notes, modifications, note_body
|
||
) VALUES (%s, %s, 'note',
|
||
NULL, NULL, NULL, NULL, NULL, NULL, %s
|
||
)
|
||
""",
|
||
(section_id, order_ix, body),
|
||
)
|
||
continue
|
||
|
||
eid = raw.get("exercise_id")
|
||
if not eid:
|
||
continue
|
||
eid = int(eid)
|
||
vid = _optional_positive_int(raw.get("exercise_variant_id"), "exercise_variant_id")
|
||
_validate_variant_for_exercise(cur, eid, vid)
|
||
cur.execute(
|
||
"""
|
||
INSERT INTO training_unit_section_items (
|
||
section_id, order_index, item_type,
|
||
exercise_id, exercise_variant_id,
|
||
planned_duration_min, actual_duration_min,
|
||
notes, modifications, note_body
|
||
) VALUES (%s, %s, 'exercise',
|
||
%s, %s, %s, %s, %s, %s, NULL
|
||
)
|
||
""",
|
||
(
|
||
section_id,
|
||
order_ix,
|
||
eid,
|
||
vid,
|
||
raw.get("planned_duration_min"),
|
||
raw.get("actual_duration_min"),
|
||
raw.get("notes"),
|
||
raw.get("modifications"),
|
||
),
|
||
)
|
||
|
||
|
||
def _replace_unit_sections(cur, unit_id: int, sections_in: List[Any]):
|
||
cur.execute("DELETE FROM training_unit_sections WHERE training_unit_id = %s", (unit_id,))
|
||
for si, sec in enumerate(sections_in):
|
||
title = (sec.get("title") or "").strip() or "Abschnitt"
|
||
order_ix = sec.get("order_index")
|
||
if order_ix is None:
|
||
order_ix = si
|
||
src_tsec = _optional_positive_int(sec.get("source_template_section_id"), "source_template_section_id")
|
||
cur.execute(
|
||
"""
|
||
INSERT INTO training_unit_sections (
|
||
training_unit_id, order_index, title, guidance_notes, source_template_section_id
|
||
) VALUES (%s, %s, %s, %s, %s)
|
||
RETURNING id
|
||
""",
|
||
(
|
||
unit_id,
|
||
order_ix,
|
||
title,
|
||
sec.get("guidance_notes"),
|
||
src_tsec,
|
||
),
|
||
)
|
||
sid = cur.fetchone()["id"]
|
||
_insert_section_items(cur, sid, sec.get("items"))
|
||
|
||
|
||
def _insert_sections_from_legacy_exercises(cur, unit_id: int, exercises_in: List[Any]):
|
||
if not exercises_in:
|
||
return
|
||
cur.execute(
|
||
"""
|
||
INSERT INTO training_unit_sections (training_unit_id, order_index, title, guidance_notes)
|
||
VALUES (%s, 0, %s, NULL)
|
||
RETURNING id
|
||
""",
|
||
(unit_id, "Übungen"),
|
||
)
|
||
sid = cur.fetchone()["id"]
|
||
slot = 0
|
||
filtered: List[Dict[str, Any]] = []
|
||
for ex in exercises_in:
|
||
eid = ex.get("exercise_id")
|
||
if not eid:
|
||
continue
|
||
eid = int(eid)
|
||
vid = _optional_positive_int(ex.get("exercise_variant_id"), "exercise_variant_id")
|
||
_validate_variant_for_exercise(cur, eid, vid)
|
||
filtered.append(
|
||
{
|
||
"item_type": "exercise",
|
||
"order_index": slot,
|
||
"exercise_id": eid,
|
||
"exercise_variant_id": vid,
|
||
"planned_duration_min": ex.get("planned_duration_min"),
|
||
"actual_duration_min": ex.get("actual_duration_min"),
|
||
"notes": ex.get("notes"),
|
||
"modifications": ex.get("modifications"),
|
||
}
|
||
)
|
||
slot += 1
|
||
_insert_section_items(cur, sid, filtered, start_order=0)
|
||
|
||
|
||
def _instantiate_from_template(cur, unit_id: int, template_id: int):
|
||
cur.execute(
|
||
"""
|
||
SELECT id, title, guidance_text
|
||
FROM training_plan_template_sections
|
||
WHERE template_id = %s
|
||
ORDER BY order_index
|
||
""",
|
||
(template_id,),
|
||
)
|
||
rows = cur.fetchall()
|
||
for row in rows:
|
||
r = r2d(row)
|
||
cur.execute(
|
||
"""
|
||
INSERT INTO training_unit_sections (
|
||
training_unit_id, order_index, title, guidance_notes, source_template_section_id
|
||
) VALUES (%s, (
|
||
SELECT COALESCE(MAX(order_index), -1) + 1 FROM training_unit_sections u2
|
||
WHERE u2.training_unit_id = %s
|
||
), %s, %s, %s)
|
||
""",
|
||
(unit_id, unit_id, r["title"], r["guidance_text"], r["id"]),
|
||
)
|
||
|
||
# Fallback: keine Sektionen in Vorlage → ein leerer Block
|
||
if not rows:
|
||
cur.execute(
|
||
"""
|
||
INSERT INTO training_unit_sections (training_unit_id, order_index, title, guidance_notes)
|
||
SELECT %s, 0, %s, NULL
|
||
WHERE NOT EXISTS (SELECT 1 FROM training_unit_sections WHERE training_unit_id = %s)
|
||
""",
|
||
(unit_id, "Hauptteil", unit_id),
|
||
)
|
||
|
||
|
||
def _template_access(cur, tid: int, profile_id: int, role: str) -> Dict[str, Any]:
|
||
cur.execute(
|
||
"""
|
||
SELECT *
|
||
FROM training_plan_templates
|
||
WHERE id = %s
|
||
""",
|
||
(tid,),
|
||
)
|
||
r = cur.fetchone()
|
||
if not r:
|
||
raise HTTPException(status_code=404, detail="Trainingsvorlage nicht gefunden")
|
||
row = r2d(r)
|
||
if role in ["admin", "superadmin"]:
|
||
return row
|
||
if row["created_by"] != profile_id:
|
||
raise HTTPException(status_code=403, detail="Keine Berechtigung für diese Vorlage")
|
||
return row
|
||
|
||
|
||
# ── Vorlagen ────────────────────────────────────────────────────────────
|
||
|
||
|
||
@router.get("/training-plan-templates")
|
||
def list_training_plan_templates(session=Depends(require_auth)):
|
||
profile_id = session["profile_id"]
|
||
role = session.get("role")
|
||
with get_db() as conn:
|
||
cur = get_cursor(conn)
|
||
if role in ["admin", "superadmin"]:
|
||
cur.execute(
|
||
"""
|
||
SELECT t.*,
|
||
(SELECT COUNT(*) FROM training_plan_template_sections s WHERE s.template_id = t.id)
|
||
AS sections_count
|
||
FROM training_plan_templates t
|
||
ORDER BY t.updated_at DESC NULLS LAST, t.name
|
||
"""
|
||
)
|
||
else:
|
||
cur.execute(
|
||
"""
|
||
SELECT t.*,
|
||
(SELECT COUNT(*) FROM training_plan_template_sections s WHERE s.template_id = t.id)
|
||
AS sections_count
|
||
FROM training_plan_templates t
|
||
WHERE t.created_by = %s
|
||
ORDER BY t.updated_at DESC NULLS LAST, t.name
|
||
""",
|
||
(profile_id,),
|
||
)
|
||
return [r2d(r) for r in cur.fetchall()]
|
||
|
||
|
||
@router.get("/training-plan-templates/{template_id}")
|
||
def get_training_plan_template(template_id: int, session=Depends(require_auth)):
|
||
profile_id = session["profile_id"]
|
||
role = session.get("role")
|
||
with get_db() as conn:
|
||
cur = get_cursor(conn)
|
||
row = _template_access(cur, template_id, profile_id, role)
|
||
cur.execute(
|
||
"""
|
||
SELECT *
|
||
FROM training_plan_template_sections
|
||
WHERE template_id = %s
|
||
ORDER BY order_index
|
||
""",
|
||
(template_id,),
|
||
)
|
||
row["sections"] = [r2d(r) for r in cur.fetchall()]
|
||
return row
|
||
|
||
|
||
@router.post("/training-plan-templates")
|
||
def create_training_plan_template(data: dict, session=Depends(require_auth)):
|
||
profile_id = session["profile_id"]
|
||
role = session.get("role")
|
||
if not _has_planning_role(role):
|
||
raise HTTPException(status_code=403, detail="Nur Trainer dürfen Vorlagen anlegen")
|
||
name = (data.get("name") or "").strip()
|
||
if not name:
|
||
raise HTTPException(status_code=400, detail="name ist Pflicht")
|
||
club_id = data.get("club_id")
|
||
sections_in = data.get("sections") or []
|
||
|
||
with get_db() as conn:
|
||
cur = get_cursor(conn)
|
||
cur.execute(
|
||
"""
|
||
INSERT INTO training_plan_templates (club_id, created_by, name, description)
|
||
VALUES (%s, %s, %s, %s)
|
||
RETURNING id
|
||
""",
|
||
(club_id, profile_id, name, data.get("description")),
|
||
)
|
||
tid = cur.fetchone()["id"]
|
||
for si, sec in enumerate(sections_in):
|
||
title = (sec.get("title") or "").strip() or f"Abschnitt {si + 1}"
|
||
order_ix = sec.get("order_index")
|
||
if order_ix is None:
|
||
order_ix = si
|
||
cur.execute(
|
||
"""
|
||
INSERT INTO training_plan_template_sections (template_id, order_index, title, guidance_text)
|
||
VALUES (%s, %s, %s, %s)
|
||
""",
|
||
(tid, order_ix, title, sec.get("guidance_text")),
|
||
)
|
||
conn.commit()
|
||
return get_training_plan_template(tid, session)
|
||
|
||
|
||
@router.put("/training-plan-templates/{template_id}")
|
||
def update_training_plan_template(template_id: int, data: dict, session=Depends(require_auth)):
|
||
profile_id = session["profile_id"]
|
||
role = session.get("role")
|
||
with get_db() as conn:
|
||
cur = get_cursor(conn)
|
||
_template_access(cur, template_id, profile_id, role)
|
||
fields = []
|
||
params: List[Any] = []
|
||
if "name" in data:
|
||
name = data.get("name")
|
||
name = name.strip() if isinstance(name, str) else ""
|
||
if not name:
|
||
raise HTTPException(status_code=400, detail="name ist Pflicht")
|
||
fields.append("name = %s")
|
||
params.append(name)
|
||
if "description" in data:
|
||
fields.append("description = %s")
|
||
params.append(data.get("description"))
|
||
if "club_id" in data:
|
||
fields.append("club_id = %s")
|
||
params.append(data.get("club_id"))
|
||
fields.append("updated_at = NOW()")
|
||
params.append(template_id)
|
||
cur.execute(
|
||
f"""
|
||
UPDATE training_plan_templates SET {", ".join(fields)}
|
||
WHERE id = %s
|
||
""",
|
||
tuple(params),
|
||
)
|
||
if "sections" in data:
|
||
cur.execute(
|
||
"DELETE FROM training_plan_template_sections WHERE template_id = %s", (template_id,)
|
||
)
|
||
sections_in = data["sections"] or []
|
||
for si, sec in enumerate(sections_in):
|
||
title = (sec.get("title") or "").strip() or f"Abschnitt {si + 1}"
|
||
order_ix = sec.get("order_index")
|
||
if order_ix is None:
|
||
order_ix = si
|
||
cur.execute(
|
||
"""
|
||
INSERT INTO training_plan_template_sections (template_id, order_index, title, guidance_text)
|
||
VALUES (%s, %s, %s, %s)
|
||
""",
|
||
(template_id, order_ix, title, sec.get("guidance_text")),
|
||
)
|
||
conn.commit()
|
||
return get_training_plan_template(template_id, session)
|
||
|
||
|
||
@router.delete("/training-plan-templates/{template_id}")
|
||
def delete_training_plan_template(template_id: int, session=Depends(require_auth)):
|
||
profile_id = session["profile_id"]
|
||
role = session.get("role")
|
||
with get_db() as conn:
|
||
cur = get_cursor(conn)
|
||
_template_access(cur, template_id, profile_id, role)
|
||
cur.execute("DELETE FROM training_plan_templates WHERE id = %s", (template_id,))
|
||
conn.commit()
|
||
return {"ok": True}
|
||
|
||
|
||
# ── Einheiten ─────────────────────────────────────────────────────────────
|
||
|
||
|
||
@router.get("/training-units")
|
||
def list_training_units(
|
||
group_id: Optional[int] = Query(default=None),
|
||
start_date: Optional[str] = Query(default=None),
|
||
end_date: Optional[str] = Query(default=None),
|
||
status: Optional[str] = Query(default=None),
|
||
session=Depends(require_auth),
|
||
):
|
||
profile_id = session["profile_id"]
|
||
role = session.get("role")
|
||
|
||
with get_db() as conn:
|
||
cur = get_cursor(conn)
|
||
|
||
query = """
|
||
SELECT tu.*,
|
||
tg.name as group_name,
|
||
tg.weekday as group_weekday,
|
||
c.name as club_name,
|
||
p.name as trainer_name
|
||
FROM training_units tu
|
||
LEFT JOIN training_groups tg ON tu.group_id = tg.id
|
||
LEFT JOIN clubs c ON tg.club_id = c.id
|
||
LEFT JOIN profiles p ON tu.created_by = p.id
|
||
"""
|
||
|
||
where = []
|
||
params = []
|
||
|
||
if role not in ["admin", "superadmin"]:
|
||
where.append("(tu.created_by = %s OR tg.trainer_id = %s)")
|
||
params.extend([profile_id, profile_id])
|
||
|
||
if group_id:
|
||
where.append("tu.group_id = %s")
|
||
params.append(group_id)
|
||
|
||
if start_date:
|
||
where.append("tu.planned_date >= %s")
|
||
params.append(start_date)
|
||
|
||
if end_date:
|
||
where.append("tu.planned_date <= %s")
|
||
params.append(end_date)
|
||
|
||
if status:
|
||
where.append("tu.status = %s")
|
||
params.append(status)
|
||
|
||
if where:
|
||
query += " WHERE " + " AND ".join(where)
|
||
|
||
query += " ORDER BY tu.planned_date DESC, tu.planned_time_start DESC"
|
||
|
||
cur.execute(query, params)
|
||
rows = cur.fetchall()
|
||
return [r2d(r) for r in rows]
|
||
|
||
|
||
@router.get("/training-units/{unit_id}")
|
||
def get_training_unit(unit_id: int, session=Depends(require_auth)):
|
||
profile_id = session["profile_id"]
|
||
role = session.get("role")
|
||
|
||
with get_db() as conn:
|
||
cur = get_cursor(conn)
|
||
|
||
cur.execute(
|
||
"""
|
||
SELECT tu.*,
|
||
tg.name as group_name,
|
||
tg.weekday as group_weekday,
|
||
tg.time_start as group_time_start,
|
||
tg.time_end as group_time_end,
|
||
tg.location as group_location,
|
||
c.name as club_name,
|
||
p.name as trainer_name
|
||
FROM training_units tu
|
||
LEFT JOIN training_groups tg ON tu.group_id = tg.id
|
||
LEFT JOIN clubs c ON tg.club_id = c.id
|
||
LEFT JOIN profiles p ON tu.created_by = p.id
|
||
WHERE tu.id = %s
|
||
""",
|
||
(unit_id,),
|
||
)
|
||
|
||
unit = cur.fetchone()
|
||
if not unit:
|
||
raise HTTPException(status_code=404, detail="Trainingseinheit nicht gefunden")
|
||
|
||
unit = r2d(unit)
|
||
|
||
cur.execute("SELECT trainer_id FROM training_groups WHERE id = %s", (unit["group_id"],))
|
||
group = cur.fetchone()
|
||
|
||
if role not in ["admin", "superadmin"]:
|
||
if unit["created_by"] != profile_id and (not group or group["trainer_id"] != profile_id):
|
||
raise HTTPException(status_code=403, detail="Keine Berechtigung")
|
||
|
||
_hydrate_training_unit_payload(cur, unit)
|
||
return unit
|
||
|
||
|
||
@router.post("/training-units")
|
||
def create_training_unit(data: dict, session=Depends(require_auth)):
|
||
profile_id = session["profile_id"]
|
||
role = session.get("role")
|
||
|
||
group_id = data.get("group_id")
|
||
planned_date = data.get("planned_date")
|
||
|
||
if not group_id or not planned_date:
|
||
raise HTTPException(status_code=400, detail="group_id und planned_date sind Pflichtfelder")
|
||
|
||
plan_template_id = _optional_positive_int(data.get("plan_template_id"), "plan_template_id")
|
||
|
||
with get_db() as conn:
|
||
cur = get_cursor(conn)
|
||
|
||
_can_access_group_for_create(cur, group_id, profile_id, role)
|
||
|
||
tpl_id_safe = None
|
||
if plan_template_id:
|
||
_template_access(cur, plan_template_id, profile_id, role)
|
||
tpl_id_safe = plan_template_id
|
||
|
||
cur.execute(
|
||
"""
|
||
INSERT INTO training_units (
|
||
group_id, planned_date, planned_time_start, planned_time_end,
|
||
planned_focus, status, notes, trainer_notes, created_by,
|
||
plan_template_id
|
||
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
|
||
RETURNING id
|
||
""",
|
||
(
|
||
group_id,
|
||
planned_date,
|
||
data.get("planned_time_start"),
|
||
data.get("planned_time_end"),
|
||
data.get("planned_focus"),
|
||
data.get("status", "planned"),
|
||
data.get("notes"),
|
||
data.get("trainer_notes"),
|
||
profile_id,
|
||
tpl_id_safe,
|
||
),
|
||
)
|
||
|
||
unit_id = cur.fetchone()["id"]
|
||
|
||
sections_in = data.get("sections")
|
||
exercises_in = data.get("exercises")
|
||
|
||
if sections_in is not None:
|
||
_replace_unit_sections(cur, unit_id, sections_in)
|
||
elif tpl_id_safe:
|
||
_instantiate_from_template(cur, unit_id, tpl_id_safe)
|
||
elif exercises_in is not None:
|
||
_insert_sections_from_legacy_exercises(cur, unit_id, exercises_in)
|
||
|
||
conn.commit()
|
||
|
||
return get_training_unit(unit_id, session)
|
||
|
||
|
||
@router.put("/training-units/{unit_id}")
|
||
def update_training_unit(unit_id: int, data: dict, session=Depends(require_auth)):
|
||
profile_id = session["profile_id"]
|
||
role = session.get("role")
|
||
|
||
with get_db() as conn:
|
||
cur = get_cursor(conn)
|
||
|
||
unit_row = _training_unit_guard_row(cur, unit_id)
|
||
_assert_training_unit_permission(cur, unit_row, profile_id, role)
|
||
|
||
tpl_upd = data.get("plan_template_id") if "plan_template_id" in data else None
|
||
tpl_id_val = None
|
||
if tpl_upd not in (None, ""):
|
||
tid = _optional_positive_int(tpl_upd, "plan_template_id")
|
||
if tid:
|
||
_template_access(cur, tid, profile_id, role)
|
||
tpl_id_val = tid
|
||
|
||
cur.execute(
|
||
"""
|
||
UPDATE training_units SET
|
||
planned_date = COALESCE(%s, planned_date),
|
||
planned_time_start = %s,
|
||
planned_time_end = %s,
|
||
planned_focus = %s,
|
||
actual_date = %s,
|
||
actual_time_start = %s,
|
||
actual_time_end = %s,
|
||
attendance_count = %s,
|
||
status = %s,
|
||
notes = %s,
|
||
trainer_notes = %s,
|
||
plan_template_id = COALESCE(%s, plan_template_id),
|
||
updated_at = NOW()
|
||
WHERE id = %s
|
||
""",
|
||
(
|
||
data.get("planned_date"),
|
||
data.get("planned_time_start"),
|
||
data.get("planned_time_end"),
|
||
data.get("planned_focus"),
|
||
data.get("actual_date"),
|
||
data.get("actual_time_start"),
|
||
data.get("actual_time_end"),
|
||
data.get("attendance_count"),
|
||
data.get("status"),
|
||
data.get("notes"),
|
||
data.get("trainer_notes"),
|
||
tpl_id_val,
|
||
unit_id,
|
||
),
|
||
)
|
||
|
||
content_handled = False
|
||
if data.get("reset_from_template"):
|
||
tid = tpl_id_val or unit_row.get("plan_template_id")
|
||
if not tid:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail="reset_from_template erfordert plan_template_id auf der Einheit oder im Request",
|
||
)
|
||
_template_access(cur, tid, profile_id, role)
|
||
cur.execute("DELETE FROM training_unit_sections WHERE training_unit_id = %s", (unit_id,))
|
||
cur.execute(
|
||
"UPDATE training_units SET plan_template_id = %s WHERE id = %s", (tid, unit_id)
|
||
)
|
||
_instantiate_from_template(cur, unit_id, tid)
|
||
content_handled = True
|
||
|
||
if not content_handled and "sections" in data:
|
||
_replace_unit_sections(cur, unit_id, data["sections"] or [])
|
||
elif not content_handled and "exercises" in data:
|
||
cur.execute("DELETE FROM training_unit_sections WHERE training_unit_id = %s", (unit_id,))
|
||
_insert_sections_from_legacy_exercises(cur, unit_id, data["exercises"] or [])
|
||
|
||
conn.commit()
|
||
|
||
return get_training_unit(unit_id, session)
|
||
|
||
|
||
@router.delete("/training-units/{unit_id}")
|
||
def delete_training_unit(unit_id: int, session=Depends(require_auth)):
|
||
profile_id = session["profile_id"]
|
||
role = session.get("role")
|
||
|
||
with get_db() as conn:
|
||
cur = get_cursor(conn)
|
||
|
||
cur.execute(
|
||
"SELECT created_by FROM training_units WHERE id = %s",
|
||
(unit_id,),
|
||
)
|
||
|
||
unit = cur.fetchone()
|
||
|
||
if not unit:
|
||
raise HTTPException(status_code=404, detail="Trainingseinheit nicht gefunden")
|
||
|
||
_assert_delete_training_unit(role, unit["created_by"], profile_id)
|
||
|
||
cur.execute("DELETE FROM training_units WHERE id = %s", (unit_id,))
|
||
conn.commit()
|
||
|
||
return {"ok": True}
|
||
|
||
|
||
@router.post("/training-units/quick-create")
|
||
def quick_create_training_unit(data: dict, session=Depends(require_auth)):
|
||
profile_id = session["profile_id"]
|
||
|
||
group_id = data.get("group_id")
|
||
planned_date = data.get("planned_date")
|
||
plan_template_id = _optional_positive_int(data.get("plan_template_id"), "plan_template_id")
|
||
|
||
if not group_id or not planned_date:
|
||
raise HTTPException(status_code=400, detail="group_id und planned_date sind Pflichtfelder")
|
||
|
||
with get_db() as conn:
|
||
cur = get_cursor(conn)
|
||
|
||
cur.execute(
|
||
"""
|
||
SELECT weekday, time_start, time_end, trainer_id, co_trainer_ids
|
||
FROM training_groups
|
||
WHERE id = %s
|
||
""",
|
||
(group_id,),
|
||
)
|
||
|
||
group = cur.fetchone()
|
||
|
||
if not group:
|
||
raise HTTPException(status_code=404, detail="Trainingsgruppe nicht gefunden")
|
||
|
||
role = session.get("role")
|
||
co_trainers = group["co_trainer_ids"] or []
|
||
|
||
if not _has_planning_role(role):
|
||
raise HTTPException(status_code=403, detail="Nur Trainer dürfen Trainingseinheiten erstellen")
|
||
|
||
if role not in ["admin", "superadmin"]:
|
||
if group["trainer_id"] != profile_id and profile_id not in co_trainers:
|
||
raise HTTPException(status_code=403, detail="Keine Berechtigung für diese Gruppe")
|
||
|
||
tpl_id_safe = None
|
||
if plan_template_id:
|
||
_template_access(cur, plan_template_id, profile_id, role)
|
||
tpl_id_safe = plan_template_id
|
||
|
||
cur.execute(
|
||
"""
|
||
INSERT INTO training_units (
|
||
group_id, planned_date,
|
||
planned_time_start, planned_time_end,
|
||
status, created_by, plan_template_id
|
||
) VALUES (%s, %s, %s, %s, %s, %s, %s)
|
||
RETURNING id
|
||
""",
|
||
(
|
||
group_id,
|
||
planned_date,
|
||
group["time_start"],
|
||
group["time_end"],
|
||
"planned",
|
||
profile_id,
|
||
tpl_id_safe,
|
||
),
|
||
)
|
||
|
||
unit_id = cur.fetchone()["id"]
|
||
|
||
if tpl_id_safe:
|
||
_instantiate_from_template(cur, unit_id, tpl_id_safe)
|
||
|
||
conn.commit()
|
||
|
||
return get_training_unit(unit_id, session)
|
||
|