""" 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"}