Lars
b752883392
All checks were successful
Deploy Development / deploy (push) Successful in 41s
Test Suite / pytest-backend (push) Successful in 23s
Test Suite / lint-backend (push) Successful in 0s
Test Suite / build-frontend (push) Successful in 7s
Test Suite / playwright-tests (push) Successful in 25s
- Updated PostgreSQL binding in docker-compose to restrict access to localhost only. - Implemented a new API endpoint for secure media file delivery, requiring authentication via token. - Enhanced governance checks for exercise media access, ensuring only authorized users can retrieve files. - Updated frontend components to utilize the new media file access method, improving user experience while maintaining security. - Documented changes in production readiness audit and access layer endpoint audit for clarity on security enhancements.
1321 lines
48 KiB
Python
1321 lines
48 KiB
Python
"""
|
|
Reifegradmodelle / Fähigkeitsmatrix (kontextbezogen)
|
|
|
|
Kontext zu Fokusbereich, Stilrichtung, Zielgruppe: jeweils M:N (leer = gilt überall).
|
|
|
|
Lesen: Liste & resolve für alle authentifizierten Nutzer; GET eines Modells nach ID nur Portal-Admin (Admin-UI).
|
|
|
|
Schreiben: admin, superadmin.
|
|
"""
|
|
from datetime import datetime, timezone
|
|
from typing import Any, Dict, List, Optional, Sequence
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from fastapi.responses import JSONResponse
|
|
|
|
from auth import require_auth
|
|
from db import get_db, get_cursor, r2d
|
|
|
|
router = APIRouter(prefix="/api", tags=["maturity_models"])
|
|
|
|
|
|
def _require_admin(session: dict) -> None:
|
|
role = session.get("role")
|
|
if role not in ("admin", "superadmin"):
|
|
raise HTTPException(403, "Nur Administratoren dürfen Reifegradmodelle verwalten")
|
|
|
|
|
|
def _base_maturity_model(cur, model_id: int) -> Optional[Dict[str, Any]]:
|
|
cur.execute("SELECT * FROM maturity_models WHERE id = %s", (model_id,))
|
|
row = cur.fetchone()
|
|
return r2d(row) if row else None
|
|
|
|
|
|
def _attach_context(cur, m: Dict[str, Any]) -> Dict[str, Any]:
|
|
mid = m["id"]
|
|
cur.execute(
|
|
"""
|
|
SELECT fa.id, fa.name, fa.abbreviation, mfa.is_primary
|
|
FROM maturity_model_focus_areas mfa
|
|
JOIN focus_areas fa ON fa.id = mfa.focus_area_id
|
|
WHERE mfa.maturity_model_id = %s
|
|
ORDER BY mfa.is_primary DESC NULLS LAST, fa.sort_order, fa.name
|
|
""",
|
|
(mid,),
|
|
)
|
|
m["focus_areas"] = [r2d(r) for r in cur.fetchall()]
|
|
|
|
cur.execute(
|
|
"""
|
|
SELECT sd.id, sd.name, msd.is_primary
|
|
FROM maturity_model_style_directions msd
|
|
JOIN style_directions sd ON sd.id = msd.style_direction_id
|
|
WHERE msd.maturity_model_id = %s
|
|
ORDER BY msd.is_primary DESC NULLS LAST, sd.name
|
|
""",
|
|
(mid,),
|
|
)
|
|
m["style_directions"] = [r2d(r) for r in cur.fetchall()]
|
|
|
|
cur.execute(
|
|
"""
|
|
SELECT tg.id, tg.name, mtg.is_primary
|
|
FROM maturity_model_target_groups mtg
|
|
JOIN target_groups tg ON tg.id = mtg.target_group_id
|
|
WHERE mtg.maturity_model_id = %s
|
|
ORDER BY mtg.is_primary DESC NULLS LAST, tg.name
|
|
""",
|
|
(mid,),
|
|
)
|
|
m["target_groups"] = [r2d(r) for r in cur.fetchall()]
|
|
|
|
m["focus_area_name"] = m["focus_areas"][0]["name"] if m["focus_areas"] else None
|
|
m["style_direction_name"] = m["style_directions"][0]["name"] if m["style_directions"] else None
|
|
m["target_group_name"] = m["target_groups"][0]["name"] if m["target_groups"] else None
|
|
return m
|
|
|
|
|
|
def _normalize_id_list(raw: Any) -> Optional[List[int]]:
|
|
if raw is None:
|
|
return None
|
|
if isinstance(raw, (str, bytes)):
|
|
raise HTTPException(400, "Erwarte Liste von IDs")
|
|
if not isinstance(raw, Sequence):
|
|
raise HTTPException(400, "Erwarte Liste von IDs")
|
|
out: List[int] = []
|
|
for x in raw:
|
|
if x is None:
|
|
continue
|
|
out.append(int(x))
|
|
return out
|
|
|
|
|
|
def _write_context_junctions(cur, model_id: int, data: Dict[str, Any]) -> None:
|
|
fa = data.get("focus_area_ids")
|
|
if fa is None and data.get("focus_area_id") is not None:
|
|
fa = [data.get("focus_area_id")]
|
|
if fa is not None:
|
|
fa = _normalize_id_list(fa)
|
|
cur.execute("DELETE FROM maturity_model_focus_areas WHERE maturity_model_id = %s", (model_id,))
|
|
for i, fid in enumerate(fa or []):
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO maturity_model_focus_areas (maturity_model_id, focus_area_id, is_primary)
|
|
VALUES (%s, %s, %s)
|
|
""",
|
|
(model_id, fid, i == 0),
|
|
)
|
|
|
|
sd = data.get("style_direction_ids")
|
|
if sd is None and data.get("style_direction_id") is not None:
|
|
sd = [data.get("style_direction_id")]
|
|
if sd is not None:
|
|
sd = _normalize_id_list(sd)
|
|
cur.execute("DELETE FROM maturity_model_style_directions WHERE maturity_model_id = %s", (model_id,))
|
|
for i, sid in enumerate(sd or []):
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO maturity_model_style_directions (maturity_model_id, style_direction_id, is_primary)
|
|
VALUES (%s, %s, %s)
|
|
""",
|
|
(model_id, sid, i == 0),
|
|
)
|
|
|
|
tg = data.get("target_group_ids")
|
|
if tg is None and data.get("target_group_id") is not None:
|
|
tg = [data.get("target_group_id")]
|
|
if tg is not None:
|
|
tg = _normalize_id_list(tg)
|
|
cur.execute("DELETE FROM maturity_model_target_groups WHERE maturity_model_id = %s", (model_id,))
|
|
for i, tid in enumerate(tg or []):
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO maturity_model_target_groups (maturity_model_id, target_group_id, is_primary)
|
|
VALUES (%s, %s, %s)
|
|
""",
|
|
(model_id, tid, i == 0),
|
|
)
|
|
|
|
|
|
def _load_full_model(cur, model_id: int) -> Dict[str, Any]:
|
|
base = _base_maturity_model(cur, model_id)
|
|
if not base:
|
|
raise HTTPException(404, "Reifegradmodell nicht gefunden")
|
|
_attach_context(cur, base)
|
|
|
|
cur.execute(
|
|
"""
|
|
SELECT * FROM model_levels
|
|
WHERE maturity_model_id = %s
|
|
ORDER BY sort_order ASC, level_number ASC
|
|
""",
|
|
(model_id,),
|
|
)
|
|
levels = [r2d(r) for r in cur.fetchall()]
|
|
|
|
cur.execute(
|
|
"""
|
|
SELECT ms.*, s.name AS skill_name, s.status AS skill_status,
|
|
sc.name AS skill_subcategory_name, sc.slug AS skill_subcategory_slug,
|
|
mc.name AS skill_main_category_name, mc.slug AS skill_main_category_slug
|
|
FROM model_skills ms
|
|
JOIN skills s ON s.id = ms.skill_id
|
|
LEFT JOIN skill_categories sc ON s.category_id = sc.id
|
|
LEFT JOIN skill_main_categories mc ON s.main_category_id = mc.id
|
|
WHERE ms.maturity_model_id = %s
|
|
ORDER BY ms.sort_order ASC, mc.sort_order NULLS LAST, sc.sort_order NULLS LAST, s.name ASC
|
|
""",
|
|
(model_id,),
|
|
)
|
|
model_skills = [r2d(r) for r in cur.fetchall()]
|
|
|
|
cur.execute(
|
|
"""
|
|
SELECT msl.*, s.name AS skill_name
|
|
FROM model_skill_levels msl
|
|
JOIN skills s ON s.id = msl.skill_id
|
|
WHERE msl.maturity_model_id = %s
|
|
ORDER BY s.name ASC, msl.level_number ASC
|
|
""",
|
|
(model_id,),
|
|
)
|
|
skill_levels = [r2d(r) for r in cur.fetchall()]
|
|
|
|
return {
|
|
**base,
|
|
"levels": levels,
|
|
"model_skills": model_skills,
|
|
"skill_levels": skill_levels,
|
|
}
|
|
|
|
|
|
def _insert_default_levels(cur, model_id: int, level_count: int) -> None:
|
|
for i in range(1, level_count + 1):
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO model_levels (maturity_model_id, level_number, name, description, sort_order)
|
|
VALUES (%s, %s, %s, %s, %s)
|
|
""",
|
|
(model_id, i, f"Stufe {i}", None, i),
|
|
)
|
|
|
|
|
|
def _replace_levels(cur, model_id: int, level_count: int, levels: List[Dict[str, Any]]) -> None:
|
|
if len(levels) != level_count:
|
|
raise HTTPException(
|
|
400,
|
|
f"Anzahl der Stufen-Definitionen ({len(levels)}) muss level_count ({level_count}) entsprechen",
|
|
)
|
|
seen = set()
|
|
for lev in levels:
|
|
num = int(lev.get("level_number") or 0)
|
|
if num < 1 or num > level_count:
|
|
raise HTTPException(400, f"Ungültige level_number: {num}")
|
|
if num in seen:
|
|
raise HTTPException(400, f"Doppelte level_number: {num}")
|
|
seen.add(num)
|
|
if seen != set(range(1, level_count + 1)):
|
|
raise HTTPException(400, "level_number muss lückenlos 1..level_count abdecken")
|
|
|
|
cur.execute("DELETE FROM model_levels WHERE maturity_model_id = %s", (model_id,))
|
|
for lev in sorted(levels, key=lambda x: int(x.get("level_number"))):
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO model_levels (maturity_model_id, level_number, name, description, sort_order)
|
|
VALUES (%s, %s, %s, %s, %s)
|
|
""",
|
|
(
|
|
model_id,
|
|
int(lev["level_number"]),
|
|
(lev.get("name") or "").strip() or f"Stufe {lev['level_number']}",
|
|
lev.get("description"),
|
|
int(lev.get("sort_order") or lev["level_number"]),
|
|
),
|
|
)
|
|
|
|
|
|
def _dim_matches(items: List[Dict[str, Any]], query_id: Optional[int]) -> bool:
|
|
if not items:
|
|
return True
|
|
if query_id is None:
|
|
return True
|
|
return any(int(x["id"]) == int(query_id) for x in items)
|
|
|
|
|
|
def _dim_score(items: List[Dict[str, Any]], query_id: Optional[int]) -> int:
|
|
if not items:
|
|
return 0
|
|
if query_id is not None and any(int(x["id"]) == int(query_id) for x in items):
|
|
return 1
|
|
return 0
|
|
|
|
|
|
def _binding_model_active(cur, maturity_model_id: int) -> bool:
|
|
b = _base_maturity_model(cur, maturity_model_id)
|
|
return bool(b and b.get("status") == "active")
|
|
|
|
|
|
def _binding_matches_query_dims(
|
|
style_direction_id: Optional[int],
|
|
training_type_id: Optional[int],
|
|
b_style: Any,
|
|
b_tt: Any,
|
|
) -> bool:
|
|
"""
|
|
Prüft, ob eine Binding-Zeile zur Anfrage passt.
|
|
|
|
Nicht gesetzte Spalten der Zeile gelten als „egal“ (Wildcard):
|
|
- Nur Fokus: Stilrichtung und Trainingsstil NULL → gilt für alle Stile/Trainingsstile unter diesem Fokus.
|
|
- Fokus + Stilrichtung: Trainingsstil NULL → gilt nur für diese Stilrichtung, für jeden Trainingsstil.
|
|
- Fokus + Trainingsstil: Stilrichtung NULL → gilt nur für diesen Trainingsstil, für jede Stilrichtung.
|
|
- Alle drei gesetzt → ausschließlich diese Kombination.
|
|
|
|
In der Anfrage fehlende Dimensionen schließen Zeilen aus, die diese Dimension festgelegt haben
|
|
(z. B. Fokus+Trainingsstil-Binding zählt nicht, wenn kein Trainingsstil angefragt wird).
|
|
"""
|
|
if b_style is not None:
|
|
if style_direction_id is None or int(b_style) != int(style_direction_id):
|
|
return False
|
|
if b_tt is not None:
|
|
if training_type_id is None or int(b_tt) != int(training_type_id):
|
|
return False
|
|
return True
|
|
|
|
|
|
def _binding_dim_count(b_style: Any, b_tt: Any) -> int:
|
|
n = 0
|
|
if b_style is not None:
|
|
n += 1
|
|
if b_tt is not None:
|
|
n += 1
|
|
return n
|
|
|
|
|
|
def _focus_has_any_bindings(cur, focus_area_id: int) -> bool:
|
|
cur.execute(
|
|
"""
|
|
SELECT 1 FROM maturity_model_context_bindings
|
|
WHERE focus_area_id = %s
|
|
LIMIT 1
|
|
""",
|
|
(focus_area_id,),
|
|
)
|
|
return cur.fetchone() is not None
|
|
|
|
|
|
def _resolve_binding_model_ids(
|
|
cur,
|
|
focus_area_id: int,
|
|
style_direction_id: Optional[int],
|
|
training_type_id: Optional[int],
|
|
) -> List[int]:
|
|
"""
|
|
Alle Bindings zum Fokus, die zur Abfrage passen (inkl. Fokus+Trainingsstil ohne Stilrichtung),
|
|
sortiert nach Spezifität (weniger spezifisch zuerst). Gleiche model_id nur einmal.
|
|
Nur aktive Modelle. Leere Liste → Legacy-Resolve.
|
|
"""
|
|
cur.execute(
|
|
"""
|
|
SELECT id, maturity_model_id, style_direction_id, training_type_id
|
|
FROM maturity_model_context_bindings
|
|
WHERE focus_area_id = %s
|
|
""",
|
|
(focus_area_id,),
|
|
)
|
|
rows = [r2d(r) for r in cur.fetchall()]
|
|
matching: List[Dict[str, Any]] = []
|
|
for r in rows:
|
|
if _binding_matches_query_dims(
|
|
style_direction_id,
|
|
training_type_id,
|
|
r.get("style_direction_id"),
|
|
r.get("training_type_id"),
|
|
):
|
|
matching.append(r)
|
|
if not matching:
|
|
return []
|
|
matching.sort(
|
|
key=lambda r: (
|
|
_binding_dim_count(r.get("style_direction_id"), r.get("training_type_id")),
|
|
int(r["id"]),
|
|
)
|
|
)
|
|
out: List[int] = []
|
|
for r in matching:
|
|
mid = int(r["maturity_model_id"])
|
|
if not _binding_model_active(cur, mid):
|
|
continue
|
|
if mid not in out:
|
|
out.append(mid)
|
|
return out
|
|
|
|
|
|
def _merge_loaded_models(loaded: List[Dict[str, Any]]) -> Dict[str, Any]:
|
|
"""Überlagert Zelltexte in Reihenfolge der Kette; Zeilen kommen aus Basis + fehlende Fähigkeiten aus Overlays."""
|
|
if not loaded:
|
|
raise ValueError("merge: keine Modelle")
|
|
base = loaded[0]
|
|
base_lc = int(base.get("level_count") or 5)
|
|
base_levels = base.get("levels") or []
|
|
|
|
skill_rows: Dict[int, Dict[str, Any]] = {}
|
|
order: List[int] = []
|
|
for ms in base.get("model_skills") or []:
|
|
sid = int(ms["skill_id"])
|
|
skill_rows[sid] = dict(ms)
|
|
order.append(sid)
|
|
|
|
for fm in loaded[1:]:
|
|
for ms in fm.get("model_skills") or []:
|
|
sid = int(ms["skill_id"])
|
|
if sid not in skill_rows:
|
|
row = dict(ms)
|
|
row["maturity_model_id"] = base["id"]
|
|
skill_rows[sid] = row
|
|
order.append(sid)
|
|
|
|
merged_model_skills = [skill_rows[sid] for sid in order]
|
|
|
|
cell_map: Dict[tuple, Dict[str, Any]] = {}
|
|
for fm in loaded:
|
|
for sl in fm.get("skill_levels") or []:
|
|
ln = int(sl["level_number"])
|
|
if ln < 1 or ln > base_lc:
|
|
continue
|
|
sid = int(sl["skill_id"])
|
|
key = (sid, ln)
|
|
cell_map[key] = {
|
|
"maturity_model_id": base["id"],
|
|
"skill_id": sid,
|
|
"level_number": ln,
|
|
"description": sl.get("description") or "",
|
|
"observable_criteria": sl.get("observable_criteria"),
|
|
"example_exercise_hints": sl.get("example_exercise_hints"),
|
|
"ai_generated": sl.get("ai_generated"),
|
|
"skill_name": sl.get("skill_name"),
|
|
}
|
|
|
|
for sid, ln in list(cell_map.keys()):
|
|
row = cell_map[(sid, ln)]
|
|
if not row.get("skill_name"):
|
|
row["skill_name"] = skill_rows.get(sid, {}).get("skill_name")
|
|
|
|
merged_skill_levels = list(cell_map.values())
|
|
merged_skill_levels.sort(key=lambda x: (x.get("skill_name") or "", x["level_number"]))
|
|
|
|
out = {
|
|
**base,
|
|
"model_skills": merged_model_skills,
|
|
"skill_levels": merged_skill_levels,
|
|
"levels": base_levels,
|
|
"level_count": base_lc,
|
|
"resolution": {
|
|
"merged": len(loaded) > 1,
|
|
"source_model_ids": [int(m["id"]) for m in loaded],
|
|
"binding_strategy": "specificity_merge",
|
|
},
|
|
}
|
|
return out
|
|
|
|
|
|
def _legacy_resolve_pick_model_id(
|
|
cur,
|
|
focus_area_id: Optional[int],
|
|
style_direction_id: Optional[int],
|
|
target_group_id: Optional[int],
|
|
) -> Optional[int]:
|
|
cur.execute("SELECT * FROM maturity_models WHERE status = 'active' ORDER BY id")
|
|
rows = [r2d(r) for r in cur.fetchall()]
|
|
enriched: List[Dict[str, Any]] = []
|
|
for m in rows:
|
|
_attach_context(cur, m)
|
|
enriched.append(m)
|
|
|
|
def ok(m: Dict[str, Any]) -> bool:
|
|
if not _dim_matches(m.get("focus_areas") or [], focus_area_id):
|
|
return False
|
|
if not _dim_matches(m.get("style_directions") or [], style_direction_id):
|
|
return False
|
|
if not _dim_matches(m.get("target_groups") or [], target_group_id):
|
|
return False
|
|
return True
|
|
|
|
def score(m: Dict[str, Any]) -> int:
|
|
return (
|
|
_dim_score(m.get("focus_areas") or [], focus_area_id)
|
|
+ _dim_score(m.get("style_directions") or [], style_direction_id)
|
|
+ _dim_score(m.get("target_groups") or [], target_group_id)
|
|
)
|
|
|
|
candidates = [m for m in enriched if ok(m)]
|
|
if not candidates:
|
|
return None
|
|
best = max(candidates, key=lambda m: (score(m), m.get("id") or 0))
|
|
return int(best["id"])
|
|
|
|
|
|
@router.get("/maturity-models/resolve")
|
|
def resolve_maturity_model(
|
|
focus_area_id: Optional[int] = Query(default=None),
|
|
style_direction_id: Optional[int] = Query(default=None),
|
|
target_group_id: Optional[int] = Query(default=None),
|
|
training_type_id: Optional[int] = Query(
|
|
default=None,
|
|
description="Trainingsstil (training_types, z. B. Breitensport); auch ohne Stilrichtung nutzbar.",
|
|
),
|
|
session: dict = Depends(require_auth),
|
|
):
|
|
"""
|
|
Liefert die Fähigkeitsmatrix zum Kontext.
|
|
|
|
**Priorität 1:** `maturity_model_context_bindings`: alle passenden Zeilen zum Fokus (z. B. nur Fokus,
|
|
Fokus+Stilrichtung, Fokus+Trainingsstil, alle drei) werden nach Spezifität gemerged; spezifischere
|
|
Zuordnungen überschreiben Zelltexte gleicher Fähigkeit/Stufe.
|
|
|
|
**Priorität 2 (Legacy):** Ein aktives Modell per M:N am Modell (Zielgruppe unverändert).
|
|
Nur wenn für den Fokusbereich **keine** Einträge in `maturity_model_context_bindings` existieren —
|
|
sonst würde Legacy den Trainingsstil ignorieren und Kontext-Bindings unterlaufen.
|
|
"""
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
if focus_area_id is not None:
|
|
chain = _resolve_binding_model_ids(
|
|
cur,
|
|
int(focus_area_id),
|
|
style_direction_id,
|
|
training_type_id,
|
|
)
|
|
if chain:
|
|
loaded = [_load_full_model(cur, mid) for mid in chain]
|
|
return _merge_loaded_models(loaded)
|
|
if _focus_has_any_bindings(cur, int(focus_area_id)):
|
|
return None
|
|
|
|
mid = _legacy_resolve_pick_model_id(
|
|
cur, focus_area_id, style_direction_id, target_group_id
|
|
)
|
|
if mid is None:
|
|
return None
|
|
return _load_full_model(cur, mid)
|
|
|
|
|
|
@router.get("/maturity-models")
|
|
def list_maturity_models(
|
|
status: Optional[str] = Query(default=None),
|
|
focus_area_id: Optional[int] = Query(default=None),
|
|
session: dict = Depends(require_auth),
|
|
):
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
q = "SELECT * FROM maturity_models mm WHERE 1=1"
|
|
params: List[Any] = []
|
|
if status:
|
|
q += " AND mm.status = %s"
|
|
params.append(status)
|
|
if focus_area_id is not None:
|
|
q += """ AND (
|
|
EXISTS (
|
|
SELECT 1 FROM maturity_model_focus_areas mfa
|
|
WHERE mfa.maturity_model_id = mm.id
|
|
AND mfa.focus_area_id = %s
|
|
)
|
|
OR NOT EXISTS (
|
|
SELECT 1 FROM maturity_model_focus_areas mfa2
|
|
WHERE mfa2.maturity_model_id = mm.id
|
|
)
|
|
)"""
|
|
params.append(focus_area_id)
|
|
q += " ORDER BY mm.name ASC"
|
|
cur.execute(q, params)
|
|
rows = [r2d(r) for r in cur.fetchall()]
|
|
for m in rows:
|
|
_attach_context(cur, m)
|
|
return rows
|
|
|
|
|
|
@router.get("/maturity-models/{model_id}")
|
|
def get_maturity_model(model_id: int, session: dict = Depends(require_auth)):
|
|
_require_admin(session)
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
return _load_full_model(cur, model_id)
|
|
|
|
|
|
@router.post("/maturity-models")
|
|
def create_maturity_model(data: Dict[str, Any], session: dict = Depends(require_auth)):
|
|
_require_admin(session)
|
|
name = (data.get("name") or "").strip()
|
|
if not name:
|
|
raise HTTPException(400, "Name ist Pflichtfeld")
|
|
|
|
level_count = int(data.get("level_count") or 5)
|
|
if level_count < 3 or level_count > 10:
|
|
raise HTTPException(400, "level_count muss zwischen 3 und 10 liegen")
|
|
|
|
profile_id = session.get("profile_id")
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
try:
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO maturity_models (
|
|
name, description,
|
|
level_count, status, version,
|
|
created_by, club_id,
|
|
import_source, import_id
|
|
) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)
|
|
RETURNING id
|
|
""",
|
|
(
|
|
name,
|
|
data.get("description"),
|
|
level_count,
|
|
data.get("status") or "draft",
|
|
data.get("version") or "1.0",
|
|
profile_id,
|
|
data.get("club_id"),
|
|
data.get("import_source"),
|
|
data.get("import_id"),
|
|
),
|
|
)
|
|
mid = cur.fetchone()["id"]
|
|
except Exception as e:
|
|
if "unique" in str(e).lower() or "duplicate" in str(e).lower():
|
|
raise HTTPException(409, "Konflikt: import_id oder Name bereits vergeben") from e
|
|
raise
|
|
|
|
if any(
|
|
k in data
|
|
for k in (
|
|
"focus_area_ids",
|
|
"style_direction_ids",
|
|
"target_group_ids",
|
|
"focus_area_id",
|
|
"style_direction_id",
|
|
"target_group_id",
|
|
)
|
|
):
|
|
_write_context_junctions(cur, mid, data)
|
|
|
|
levels = data.get("levels")
|
|
if levels:
|
|
_replace_levels(cur, mid, level_count, levels)
|
|
else:
|
|
_insert_default_levels(cur, mid, level_count)
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
return _load_full_model(cur, mid)
|
|
|
|
|
|
@router.put("/maturity-models/{model_id}")
|
|
def update_maturity_model(model_id: int, data: Dict[str, Any], session: dict = Depends(require_auth)):
|
|
_require_admin(session)
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cur.execute("SELECT * FROM maturity_models WHERE id = %s", (model_id,))
|
|
row = cur.fetchone()
|
|
if not row:
|
|
raise HTTPException(404, "Reifegradmodell nicht gefunden")
|
|
current = r2d(row)
|
|
|
|
level_count = int(current["level_count"])
|
|
if "level_count" in data and data["level_count"] is not None:
|
|
level_count = int(data["level_count"])
|
|
if level_count < 3 or level_count > 10:
|
|
raise HTTPException(400, "level_count muss zwischen 3 und 10 liegen")
|
|
|
|
if level_count != int(current["level_count"]):
|
|
if "levels" not in data or data["levels"] is None:
|
|
raise HTTPException(
|
|
400,
|
|
"Stufenanzahl ändern nur zusammen mit vollständiger levels-Liste (Stufen-Editor).",
|
|
)
|
|
|
|
if level_count < int(current["level_count"]):
|
|
cur.execute(
|
|
"DELETE FROM model_skill_levels WHERE maturity_model_id = %s AND level_number > %s",
|
|
(model_id, level_count),
|
|
)
|
|
|
|
sets: List[str] = []
|
|
vals: List[Any] = []
|
|
for key in ("name", "description", "status", "version"):
|
|
if key in data:
|
|
sets.append(f"{key} = %s")
|
|
vals.append(data[key])
|
|
|
|
sets.append("level_count = %s")
|
|
vals.append(level_count)
|
|
sets.append("updated_at = NOW()")
|
|
vals.append(model_id)
|
|
|
|
cur.execute(
|
|
f"UPDATE maturity_models SET {', '.join(sets)} WHERE id = %s",
|
|
tuple(vals),
|
|
)
|
|
|
|
if any(
|
|
k in data
|
|
for k in (
|
|
"focus_area_ids",
|
|
"style_direction_ids",
|
|
"target_group_ids",
|
|
"focus_area_id",
|
|
"style_direction_id",
|
|
"target_group_id",
|
|
)
|
|
):
|
|
_write_context_junctions(cur, model_id, data)
|
|
|
|
if "levels" in data and data["levels"] is not None:
|
|
_replace_levels(cur, model_id, level_count, data["levels"])
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
return _load_full_model(cur, model_id)
|
|
|
|
|
|
@router.delete("/maturity-models/{model_id}")
|
|
def delete_maturity_model(model_id: int, session: dict = Depends(require_auth)):
|
|
role = session.get("role")
|
|
if role != "superadmin":
|
|
raise HTTPException(403, "Nur Superadmins dürfen Reifegradmodelle löschen")
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cur.execute("DELETE FROM maturity_models WHERE id = %s RETURNING id", (model_id,))
|
|
if not cur.fetchone():
|
|
raise HTTPException(404, "Reifegradmodell nicht gefunden")
|
|
return {"ok": True}
|
|
|
|
|
|
@router.post("/maturity-models/{model_id}/skills")
|
|
def add_model_skill(model_id: int, data: Dict[str, Any], session: dict = Depends(require_auth)):
|
|
_require_admin(session)
|
|
skill_id = data.get("skill_id")
|
|
if not skill_id:
|
|
raise HTTPException(400, "skill_id ist Pflicht")
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
if not _base_maturity_model(cur, model_id):
|
|
raise HTTPException(404, "Reifegradmodell nicht gefunden")
|
|
cur.execute("SELECT id FROM skills WHERE id = %s", (skill_id,))
|
|
if not cur.fetchone():
|
|
raise HTTPException(404, "Fähigkeit nicht gefunden")
|
|
try:
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO model_skills (maturity_model_id, skill_id, sort_order, relevance)
|
|
VALUES (%s, %s, %s, %s)
|
|
RETURNING id
|
|
""",
|
|
(
|
|
model_id,
|
|
skill_id,
|
|
int(data.get("sort_order") or 0),
|
|
data.get("relevance"),
|
|
),
|
|
)
|
|
except Exception as e:
|
|
if "unique" in str(e).lower():
|
|
raise HTTPException(409, "Fähigkeit ist bereits im Modell") from e
|
|
raise
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
return _load_full_model(cur, model_id)
|
|
|
|
|
|
@router.delete("/maturity-models/{model_id}/skills/{skill_id}")
|
|
def remove_model_skill(model_id: int, skill_id: int, session: dict = Depends(require_auth)):
|
|
_require_admin(session)
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cur.execute(
|
|
"DELETE FROM model_skills WHERE maturity_model_id = %s AND skill_id = %s RETURNING id",
|
|
(model_id, skill_id),
|
|
)
|
|
if not cur.fetchone():
|
|
raise HTTPException(404, "Zuordnung nicht gefunden")
|
|
cur.execute(
|
|
"DELETE FROM model_skill_levels WHERE maturity_model_id = %s AND skill_id = %s",
|
|
(model_id, skill_id),
|
|
)
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
return _load_full_model(cur, model_id)
|
|
|
|
|
|
@router.put("/maturity-models/{model_id}/skill-levels")
|
|
def upsert_model_skill_levels(model_id: int, data: Dict[str, Any], session: dict = Depends(require_auth)):
|
|
_require_admin(session)
|
|
entries: List[Dict[str, Any]] = data.get("entries") or []
|
|
if not entries:
|
|
raise HTTPException(400, "entries darf nicht leer sein")
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cur.execute("SELECT level_count FROM maturity_models WHERE id = %s", (model_id,))
|
|
row = cur.fetchone()
|
|
if not row:
|
|
raise HTTPException(404, "Reifegradmodell nicht gefunden")
|
|
level_count = int(row["level_count"])
|
|
|
|
for e in entries:
|
|
sid = e.get("skill_id")
|
|
ln = int(e.get("level_number") or 0)
|
|
desc = (e.get("description") or "").strip()
|
|
if not sid or ln < 1 or ln > level_count:
|
|
raise HTTPException(400, "Ungültige skill_id oder level_number")
|
|
|
|
if not desc:
|
|
cur.execute(
|
|
"""
|
|
DELETE FROM model_skill_levels
|
|
WHERE maturity_model_id = %s AND skill_id = %s AND level_number = %s
|
|
""",
|
|
(model_id, sid, ln),
|
|
)
|
|
continue
|
|
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO model_skill_levels (
|
|
maturity_model_id, skill_id, level_number,
|
|
description, observable_criteria, example_exercise_hints
|
|
) VALUES (%s,%s,%s,%s,%s,%s)
|
|
ON CONFLICT (maturity_model_id, skill_id, level_number)
|
|
DO UPDATE SET
|
|
description = EXCLUDED.description,
|
|
observable_criteria = EXCLUDED.observable_criteria,
|
|
example_exercise_hints = EXCLUDED.example_exercise_hints,
|
|
updated_at = NOW()
|
|
""",
|
|
(
|
|
model_id,
|
|
sid,
|
|
ln,
|
|
desc,
|
|
e.get("observable_criteria"),
|
|
e.get("example_exercise_hints"),
|
|
),
|
|
)
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
return _load_full_model(cur, model_id)
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# Kontext-Bindings (Fokus / Stilrichtung / Trainingsstil) — hierarchisch
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
@router.get("/maturity-model-context-bindings")
|
|
def list_maturity_model_context_bindings(session: dict = Depends(require_auth)):
|
|
role = session.get("role")
|
|
if role not in ("admin", "superadmin"):
|
|
raise HTTPException(403, "Nur Administratoren")
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cur.execute(
|
|
"""
|
|
SELECT b.id, b.maturity_model_id, mm.name AS maturity_model_name, mm.status AS model_status,
|
|
b.focus_area_id, fa.name AS focus_area_name,
|
|
b.style_direction_id, sd.name AS style_direction_name,
|
|
b.training_type_id, tt.name AS training_type_name
|
|
FROM maturity_model_context_bindings b
|
|
JOIN maturity_models mm ON mm.id = b.maturity_model_id
|
|
JOIN focus_areas fa ON fa.id = b.focus_area_id
|
|
LEFT JOIN style_directions sd ON sd.id = b.style_direction_id
|
|
LEFT JOIN training_types tt ON tt.id = b.training_type_id
|
|
ORDER BY fa.sort_order NULLS LAST, fa.name,
|
|
(b.style_direction_id IS NULL) DESC,
|
|
sd.name NULLS LAST,
|
|
(b.training_type_id IS NULL) DESC,
|
|
tt.name NULLS LAST
|
|
"""
|
|
)
|
|
return [r2d(r) for r in cur.fetchall()]
|
|
|
|
|
|
@router.post("/maturity-model-context-bindings")
|
|
def upsert_maturity_model_context_binding(data: Dict[str, Any], session: dict = Depends(require_auth)):
|
|
_require_admin(session)
|
|
mid = data.get("maturity_model_id")
|
|
fa = data.get("focus_area_id")
|
|
if mid is None or fa is None:
|
|
raise HTTPException(400, "maturity_model_id und focus_area_id sind Pflichtfelder")
|
|
mid = int(mid)
|
|
fa = int(fa)
|
|
sd_raw = data.get("style_direction_id")
|
|
tt_raw = data.get("training_type_id")
|
|
sd: Optional[int] = int(sd_raw) if sd_raw is not None else None
|
|
tt: Optional[int] = int(tt_raw) if tt_raw is not None else None
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
if not _base_maturity_model(cur, mid):
|
|
raise HTTPException(404, "Reifegradmodell nicht gefunden")
|
|
cur.execute("SELECT id FROM focus_areas WHERE id = %s", (fa,))
|
|
if not cur.fetchone():
|
|
raise HTTPException(404, "Fokusbereich nicht gefunden")
|
|
if sd is not None:
|
|
cur.execute("SELECT id FROM style_directions WHERE id = %s", (sd,))
|
|
if not cur.fetchone():
|
|
raise HTTPException(404, "Stilrichtung nicht gefunden")
|
|
if tt is not None:
|
|
cur.execute("SELECT id FROM training_types WHERE id = %s", (tt,))
|
|
if not cur.fetchone():
|
|
raise HTTPException(404, "Trainingsstil nicht gefunden")
|
|
|
|
if sd is None and tt is None:
|
|
cur.execute(
|
|
"""
|
|
DELETE FROM maturity_model_context_bindings
|
|
WHERE focus_area_id = %s AND style_direction_id IS NULL AND training_type_id IS NULL
|
|
""",
|
|
(fa,),
|
|
)
|
|
elif sd is not None and tt is None:
|
|
cur.execute(
|
|
"""
|
|
DELETE FROM maturity_model_context_bindings
|
|
WHERE focus_area_id = %s AND style_direction_id = %s AND training_type_id IS NULL
|
|
""",
|
|
(fa, sd),
|
|
)
|
|
elif sd is None and tt is not None:
|
|
cur.execute(
|
|
"""
|
|
DELETE FROM maturity_model_context_bindings
|
|
WHERE focus_area_id = %s AND style_direction_id IS NULL AND training_type_id = %s
|
|
""",
|
|
(fa, tt),
|
|
)
|
|
else:
|
|
cur.execute(
|
|
"""
|
|
DELETE FROM maturity_model_context_bindings
|
|
WHERE focus_area_id = %s AND style_direction_id = %s AND training_type_id = %s
|
|
""",
|
|
(fa, sd, tt),
|
|
)
|
|
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO maturity_model_context_bindings (
|
|
maturity_model_id, focus_area_id, style_direction_id, training_type_id
|
|
)
|
|
VALUES (%s, %s, %s, %s)
|
|
RETURNING id
|
|
""",
|
|
(mid, fa, sd, tt),
|
|
)
|
|
new_id = cur.fetchone()["id"]
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cur.execute(
|
|
"""
|
|
SELECT b.id, b.maturity_model_id, mm.name AS maturity_model_name, mm.status AS model_status,
|
|
b.focus_area_id, fa.name AS focus_area_name,
|
|
b.style_direction_id, sd.name AS style_direction_name,
|
|
b.training_type_id, tt.name AS training_type_name
|
|
FROM maturity_model_context_bindings b
|
|
JOIN maturity_models mm ON mm.id = b.maturity_model_id
|
|
JOIN focus_areas fa ON fa.id = b.focus_area_id
|
|
LEFT JOIN style_directions sd ON sd.id = b.style_direction_id
|
|
LEFT JOIN training_types tt ON tt.id = b.training_type_id
|
|
WHERE b.id = %s
|
|
""",
|
|
(new_id,),
|
|
)
|
|
row = cur.fetchone()
|
|
return r2d(row) if row else {"id": new_id}
|
|
|
|
|
|
@router.delete("/maturity-model-context-bindings/{binding_id}")
|
|
def delete_maturity_model_context_binding(binding_id: int, session: dict = Depends(require_auth)):
|
|
_require_admin(session)
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cur.execute(
|
|
"DELETE FROM maturity_model_context_bindings WHERE id = %s RETURNING id",
|
|
(binding_id,),
|
|
)
|
|
if not cur.fetchone():
|
|
raise HTTPException(404, "Zuordnung nicht gefunden")
|
|
return {"ok": True}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# Export / Import (JSON)
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
def _strip_model_for_export(full: Dict[str, Any]) -> Dict[str, Any]:
|
|
ms_out: List[Dict[str, Any]] = []
|
|
for ms in full.get("model_skills") or []:
|
|
ms_out.append(
|
|
{
|
|
"skill_id": ms.get("skill_id"),
|
|
"sort_order": ms.get("sort_order", 0),
|
|
"relevance": ms.get("relevance"),
|
|
"skill_name": ms.get("skill_name"),
|
|
"skill_subcategory_name": ms.get("skill_subcategory_name"),
|
|
"skill_subcategory_slug": ms.get("skill_subcategory_slug"),
|
|
"skill_main_category_name": ms.get("skill_main_category_name"),
|
|
"skill_main_category_slug": ms.get("skill_main_category_slug"),
|
|
}
|
|
)
|
|
sl_out: List[Dict[str, Any]] = []
|
|
for sl in full.get("skill_levels") or []:
|
|
sl_out.append(
|
|
{
|
|
"skill_id": sl.get("skill_id"),
|
|
"level_number": sl.get("level_number"),
|
|
"description": sl.get("description") or "",
|
|
"observable_criteria": sl.get("observable_criteria"),
|
|
"skill_name": sl.get("skill_name"),
|
|
}
|
|
)
|
|
lv_out: List[Dict[str, Any]] = []
|
|
for lv in full.get("levels") or []:
|
|
lv_out.append(
|
|
{
|
|
"level_number": lv.get("level_number"),
|
|
"name": lv.get("name"),
|
|
"description": lv.get("description"),
|
|
"sort_order": lv.get("sort_order"),
|
|
}
|
|
)
|
|
return {
|
|
"id": full.get("id"),
|
|
"name": full.get("name"),
|
|
"description": full.get("description"),
|
|
"level_count": full.get("level_count"),
|
|
"status": full.get("status"),
|
|
"version": full.get("version"),
|
|
"levels": lv_out,
|
|
"model_skills": ms_out,
|
|
"skill_levels": sl_out,
|
|
}
|
|
|
|
|
|
def _delete_bindings_for_model(cur, model_id: int) -> None:
|
|
cur.execute(
|
|
"DELETE FROM maturity_model_context_bindings WHERE maturity_model_id = %s",
|
|
(model_id,),
|
|
)
|
|
|
|
|
|
def _insert_binding_row(cur, model_id: int, fa: int, sd: Optional[int], tt: Optional[int]) -> None:
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO maturity_model_context_bindings (
|
|
maturity_model_id, focus_area_id, style_direction_id, training_type_id
|
|
) VALUES (%s, %s, %s, %s)
|
|
""",
|
|
(model_id, fa, sd, tt),
|
|
)
|
|
|
|
|
|
def _apply_import_bindings(cur, model_id: int, binds: List[Dict[str, Any]]) -> None:
|
|
for b in binds:
|
|
fa = int(b["focus_area_id"])
|
|
sd_raw = b.get("style_direction_id")
|
|
tt_raw = b.get("training_type_id")
|
|
sd: Optional[int] = int(sd_raw) if sd_raw is not None else None
|
|
tt: Optional[int] = int(tt_raw) if tt_raw is not None else None
|
|
cur.execute("SELECT id FROM focus_areas WHERE id = %s", (fa,))
|
|
if not cur.fetchone():
|
|
continue
|
|
if sd is not None:
|
|
cur.execute("SELECT id FROM style_directions WHERE id = %s", (sd,))
|
|
if not cur.fetchone():
|
|
continue
|
|
if tt is not None:
|
|
cur.execute("SELECT id FROM training_types WHERE id = %s", (tt,))
|
|
if not cur.fetchone():
|
|
continue
|
|
if sd is None and tt is None:
|
|
cur.execute(
|
|
"""
|
|
DELETE FROM maturity_model_context_bindings
|
|
WHERE focus_area_id = %s AND style_direction_id IS NULL AND training_type_id IS NULL
|
|
""",
|
|
(fa,),
|
|
)
|
|
elif sd is not None and tt is None:
|
|
cur.execute(
|
|
"""
|
|
DELETE FROM maturity_model_context_bindings
|
|
WHERE focus_area_id = %s AND style_direction_id = %s AND training_type_id IS NULL
|
|
""",
|
|
(fa, sd),
|
|
)
|
|
elif sd is None and tt is not None:
|
|
cur.execute(
|
|
"""
|
|
DELETE FROM maturity_model_context_bindings
|
|
WHERE focus_area_id = %s AND style_direction_id IS NULL AND training_type_id = %s
|
|
""",
|
|
(fa, tt),
|
|
)
|
|
else:
|
|
cur.execute(
|
|
"""
|
|
DELETE FROM maturity_model_context_bindings
|
|
WHERE focus_area_id = %s AND style_direction_id = %s AND training_type_id = %s
|
|
""",
|
|
(fa, sd, tt),
|
|
)
|
|
_insert_binding_row(cur, model_id, fa, sd, tt)
|
|
|
|
|
|
def _apply_import_model_payload(cur, model_id: int, m: Dict[str, Any]) -> None:
|
|
cur.execute(
|
|
"DELETE FROM model_skill_levels WHERE maturity_model_id = %s", (model_id,)
|
|
)
|
|
cur.execute("DELETE FROM model_skills WHERE maturity_model_id = %s", (model_id,))
|
|
cur.execute("DELETE FROM model_levels WHERE maturity_model_id = %s", (model_id,))
|
|
|
|
lc = int(m.get("level_count") or 5)
|
|
if lc < 3 or lc > 10:
|
|
raise HTTPException(400, "level_count muss zwischen 3 und 10 liegen")
|
|
|
|
cur.execute(
|
|
"""
|
|
UPDATE maturity_models SET
|
|
level_count = %s,
|
|
name = COALESCE(%s, name),
|
|
description = COALESCE(%s, description),
|
|
status = COALESCE(%s, status),
|
|
version = COALESCE(%s, version),
|
|
updated_at = NOW()
|
|
WHERE id = %s
|
|
""",
|
|
(
|
|
lc,
|
|
m.get("name"),
|
|
m.get("description"),
|
|
m.get("status"),
|
|
m.get("version"),
|
|
model_id,
|
|
),
|
|
)
|
|
|
|
levels = m.get("levels") or []
|
|
if not levels:
|
|
_insert_default_levels(cur, model_id, lc)
|
|
else:
|
|
for lev in levels:
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO model_levels (maturity_model_id, level_number, name, description, sort_order)
|
|
VALUES (%s, %s, %s, %s, %s)
|
|
""",
|
|
(
|
|
model_id,
|
|
int(lev["level_number"]),
|
|
lev.get("name") or f"Stufe {lev['level_number']}",
|
|
lev.get("description"),
|
|
int(lev.get("sort_order") or lev["level_number"]),
|
|
),
|
|
)
|
|
|
|
for ms in m.get("model_skills") or []:
|
|
sid = int(ms["skill_id"])
|
|
cur.execute("SELECT id FROM skills WHERE id = %s", (sid,))
|
|
if not cur.fetchone():
|
|
raise HTTPException(400, f"Unbekannte skill_id {sid} (Fähigkeit fehlt in dieser Datenbank)")
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO model_skills (maturity_model_id, skill_id, sort_order, relevance)
|
|
VALUES (%s, %s, %s, %s)
|
|
""",
|
|
(model_id, sid, int(ms.get("sort_order") or 0), ms.get("relevance")),
|
|
)
|
|
|
|
for sl in m.get("skill_levels") or []:
|
|
sid = int(sl["skill_id"])
|
|
ln = int(sl["level_number"])
|
|
if ln < 1 or ln > lc:
|
|
continue
|
|
desc = (sl.get("description") or "").strip()
|
|
if not desc:
|
|
continue
|
|
cur.execute("SELECT id FROM skills WHERE id = %s", (sid,))
|
|
if not cur.fetchone():
|
|
raise HTTPException(400, f"skill_levels: unbekannte skill_id {sid}")
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO model_skill_levels (
|
|
maturity_model_id, skill_id, level_number, description, observable_criteria
|
|
) VALUES (%s, %s, %s, %s, %s)
|
|
ON CONFLICT (maturity_model_id, skill_id, level_number)
|
|
DO UPDATE SET
|
|
description = EXCLUDED.description,
|
|
observable_criteria = EXCLUDED.observable_criteria,
|
|
updated_at = NOW()
|
|
""",
|
|
(model_id, sid, ln, desc, sl.get("observable_criteria")),
|
|
)
|
|
|
|
|
|
@router.get("/maturity-models/{model_id}/export")
|
|
def export_maturity_model_bundle(model_id: int, session: dict = Depends(require_auth)):
|
|
_require_admin(session)
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
if not _base_maturity_model(cur, model_id):
|
|
raise HTTPException(404, "Reifegradmodell nicht gefunden")
|
|
full = _load_full_model(cur, model_id)
|
|
cur.execute(
|
|
"""
|
|
SELECT focus_area_id, style_direction_id, training_type_id
|
|
FROM maturity_model_context_bindings
|
|
WHERE maturity_model_id = %s
|
|
""",
|
|
(model_id,),
|
|
)
|
|
binds = [r2d(r) for r in cur.fetchall()]
|
|
|
|
bundle = {
|
|
"kind": "shinkan.maturity_model.v1",
|
|
"export_version": 1,
|
|
"exported_at": datetime.now(timezone.utc).isoformat(),
|
|
"model": _strip_model_for_export(full),
|
|
"context_bindings_for_model": binds,
|
|
}
|
|
fname = f"reifegradmodell-{model_id}.json"
|
|
return JSONResponse(
|
|
content=bundle,
|
|
headers={"Content-Disposition": f'attachment; filename="{fname}"'},
|
|
)
|
|
|
|
|
|
@router.get("/maturity-models/export-resolved")
|
|
def export_resolved_maturity_bundle(
|
|
focus_area_id: int = Query(..., description="Fokusbereich (Pflicht)"),
|
|
style_direction_id: Optional[int] = Query(default=None),
|
|
training_type_id: Optional[int] = Query(default=None),
|
|
session: dict = Depends(require_auth),
|
|
):
|
|
"""Exportiert die per Bindings aufgelöste Matrix (inkl. Merge) als JSON."""
|
|
_require_admin(session)
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
chain = _resolve_binding_model_ids(
|
|
cur, int(focus_area_id), style_direction_id, training_type_id
|
|
)
|
|
if chain:
|
|
loaded = [_load_full_model(cur, mid) for mid in chain]
|
|
merged = _merge_loaded_models(loaded)
|
|
else:
|
|
if _focus_has_any_bindings(cur, int(focus_area_id)):
|
|
raise HTTPException(
|
|
404,
|
|
"Kein Reifegradmodell für diese Kontext-Kombination (es gibt Bindings für diesen Fokus, "
|
|
"aber keine passende Zeile).",
|
|
)
|
|
mid = _legacy_resolve_pick_model_id(
|
|
cur, focus_area_id, style_direction_id, None
|
|
)
|
|
if mid is None:
|
|
raise HTTPException(
|
|
404,
|
|
"Kein Modell für diesen Kontext (keine Bindings für den Fokus und kein Legacy-Treffer).",
|
|
)
|
|
merged = _load_full_model(cur, mid)
|
|
|
|
bundle = {
|
|
"kind": "shinkan.maturity_matrix_resolved.v1",
|
|
"export_version": 1,
|
|
"exported_at": datetime.now(timezone.utc).isoformat(),
|
|
"resolve_params": {
|
|
"focus_area_id": focus_area_id,
|
|
"style_direction_id": style_direction_id,
|
|
"training_type_id": training_type_id,
|
|
},
|
|
"matrix": _strip_model_for_export(merged),
|
|
"resolution": merged.get("resolution"),
|
|
}
|
|
return JSONResponse(
|
|
content=bundle,
|
|
headers={
|
|
"Content-Disposition": 'attachment; filename="faehigkeitsmatrix-aufgeloest.json"'
|
|
},
|
|
)
|
|
|
|
|
|
@router.post("/maturity-models/import")
|
|
def import_maturity_model_bundle(data: Dict[str, Any], session: dict = Depends(require_auth)):
|
|
_require_admin(session)
|
|
kind = data.get("kind")
|
|
if kind not in ("shinkan.maturity_model.v1", "shinkan.maturity_matrix_resolved.v1"):
|
|
raise HTTPException(
|
|
400,
|
|
"kind muss shinkan.maturity_model.v1 oder shinkan.maturity_matrix_resolved.v1 sein",
|
|
)
|
|
mode = (data.get("mode") or "create").strip().lower()
|
|
if mode not in ("create", "replace"):
|
|
raise HTTPException(400, "mode muss create oder replace sein")
|
|
|
|
mpart = data.get("model") or data.get("matrix") or {}
|
|
name = (mpart.get("name") or "").strip()
|
|
if mode == "create" and not name:
|
|
raise HTTPException(400, "Modellname fehlt")
|
|
|
|
import_bindings = bool(data.get("import_bindings", True))
|
|
profile_id = session.get("profile_id")
|
|
replace_id = data.get("replace_model_id")
|
|
if mode == "replace":
|
|
if not replace_id:
|
|
raise HTTPException(400, "replace_model_id erforderlich bei mode=replace")
|
|
replace_id = int(replace_id)
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
if mode == "create":
|
|
lc = int(mpart.get("level_count") or 5)
|
|
if lc < 3 or lc > 10:
|
|
raise HTTPException(400, "level_count ungültig")
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO maturity_models (
|
|
name, description, level_count, status, version, created_by
|
|
) VALUES (%s, %s, %s, %s, %s, %s)
|
|
RETURNING id
|
|
""",
|
|
(
|
|
name,
|
|
mpart.get("description"),
|
|
lc,
|
|
mpart.get("status") or "draft",
|
|
mpart.get("version") or "1.0",
|
|
profile_id,
|
|
),
|
|
)
|
|
new_id = int(cur.fetchone()["id"])
|
|
_apply_import_model_payload(cur, new_id, mpart)
|
|
if import_bindings and kind == "shinkan.maturity_model.v1":
|
|
binds = data.get("context_bindings_for_model") or []
|
|
if binds:
|
|
_apply_import_bindings(cur, new_id, binds)
|
|
return {"ok": True, "id": new_id, "mode": "create"}
|
|
|
|
if not _base_maturity_model(cur, replace_id):
|
|
raise HTTPException(404, "replace_model_id nicht gefunden")
|
|
_apply_import_model_payload(cur, replace_id, mpart)
|
|
if import_bindings and kind == "shinkan.maturity_model.v1":
|
|
binds = data.get("context_bindings_for_model") or []
|
|
_delete_bindings_for_model(cur, replace_id)
|
|
if binds:
|
|
_apply_import_bindings(cur, replace_id, binds)
|
|
return {"ok": True, "id": replace_id, "mode": "replace"}
|