Lars
0dfc08459e
- Added logic to retrieve existing trainer notes if not provided during the update of training units. - Updated the TrainingCoachPage to include new controls for managing training sessions, including timer functionalities and navigation enhancements. - Improved user experience with clearer button labels and conditional rendering based on the training session state.
853 lines
29 KiB
Python
853 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,
|
||
(
|
||
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 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
|
||
|
||
trainer_notes_val = None
|
||
if "trainer_notes" not in data:
|
||
cur.execute(
|
||
"SELECT trainer_notes FROM training_units WHERE id = %s",
|
||
(unit_id,),
|
||
)
|
||
row_tn = cur.fetchone()
|
||
trainer_notes_val = row_tn["trainer_notes"] if row_tn else None
|
||
else:
|
||
trainer_notes_val = data.get("trainer_notes")
|
||
|
||
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"),
|
||
trainer_notes_val,
|
||
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)
|
||
|