Enhance Maturity Matrix Tools and API Integration
Some checks failed
Deploy Development / deploy (push) Successful in 40s
Test Suite / pytest-backend (push) Failing after 2s
Test Suite / lint-backend (push) Successful in 0s
Test Suite / build-frontend (push) Successful in 13s
Test Suite / k6 /health Baseline (push) Successful in 33s
Test Suite / playwright-tests (push) Successful in 1m19s

- Added new functionality for exporting and importing matrix editor data in JSON and CSV formats within the MaturityMatrixToolsAdmin component.
- Updated the API utility functions to support matrix editor exports and imports, enhancing the backend communication for Superadmin tasks.
- Refactored the client API to streamline request handling and improve code clarity.
- Included new UI elements for file upload and download actions, improving user experience in managing matrix data.
This commit is contained in:
Lars 2026-05-22 11:12:50 +02:00
parent cdeddc7cec
commit d58db3d5dd
6 changed files with 1089 additions and 11 deletions

View File

@ -193,7 +193,7 @@ def read_root():
return out return out
# Register routers # Register routers
from routers import auth, profiles, exercises, exercise_progression_graphs, clubs, club_memberships, club_join_requests, admin_users, platform_media_storage, media_assets, skills, skill_profiles, training_planning, dashboard, training_modules, training_framework_programs, catalogs, maturity_models, matrix_stack_bundle, import_wiki, import_wiki_admin, legal_documents, content_reports, ai_prompts_admin, ai_skill_retrieval_admin from routers import auth, profiles, exercises, exercise_progression_graphs, clubs, club_memberships, club_join_requests, admin_users, platform_media_storage, media_assets, skills, skill_profiles, training_planning, dashboard, training_modules, training_framework_programs, catalogs, maturity_models, matrix_stack_bundle, matrix_editor, import_wiki, import_wiki_admin, legal_documents, content_reports, ai_prompts_admin, ai_skill_retrieval_admin
app.include_router(auth.router) app.include_router(auth.router)
app.include_router(profiles.router) app.include_router(profiles.router)
@ -216,6 +216,7 @@ app.include_router(training_framework_programs.router)
app.include_router(catalogs.router) app.include_router(catalogs.router)
app.include_router(maturity_models.router) app.include_router(maturity_models.router)
app.include_router(matrix_stack_bundle.router) app.include_router(matrix_stack_bundle.router)
app.include_router(matrix_editor.router)
app.include_router(import_wiki.router) app.include_router(import_wiki.router)
app.include_router(import_wiki_admin.router) app.include_router(import_wiki_admin.router)
app.include_router(legal_documents.router) app.include_router(legal_documents.router)

View File

@ -0,0 +1,726 @@
"""
Superadmin Export/Import für zentrale Pflege der Fähigkeitsmatrix.
Fokus: Beschreibungen und Gewichtungen (skills.importance, model_skills.relevance,
skill_level_definitions, model_skill_levels) flaches, bearbeitbares Format.
Kein Vereinsbezug require_auth + is_superadmin; kein TenantContext.
"""
from __future__ import annotations
import csv
import io
import re
import uuid
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional, Tuple
from fastapi import APIRouter, Body, Depends, HTTPException, Query
from fastapi.encoders import jsonable_encoder
from fastapi.responses import JSONResponse, PlainTextResponse
from auth import require_auth
from club_tenancy import is_superadmin
from db import get_db, get_cursor, r2d
router = APIRouter(prefix="/api/admin/matrix-editor", tags=["admin_matrix_editor"])
KIND_V1 = "shinkan.matrix_editor.v1"
SKILLS_CSV_FIELDS = [
"skill_key",
"main_category",
"subcategory",
"skill_name",
"description",
"importance",
"karate_relevance",
"relevance_level",
"level_1",
"level_2",
"level_3",
"level_4",
"level_5",
]
MATRIX_CSV_FIELDS = [
"model_key",
"model_name",
"skill_key",
"skill_name",
"relevance",
"sort_order",
"level_number",
"level_label",
"description",
"observable_criteria",
]
def _require_superadmin(session: dict) -> dict:
role = (session.get("role") or "").strip().lower()
if not is_superadmin(role):
raise HTTPException(status_code=403, detail="Nur Superadmins")
return session
def _slugify_label(text: str) -> str:
t = (text or "").strip().lower()
t = re.sub(r"[^a-z0-9äöüß]+", "_", t, flags=re.IGNORECASE)
t = re.sub(r"_+", "_", t).strip("_")
return t[:48] or "gruppe"
def _skill_key(main_slug: Optional[str], cat_slug: Optional[str], name: str) -> str:
return f"{main_slug or ''}|{cat_slug or ''}|{(name or '').strip()}"
def _model_key(import_id: Optional[str], name: str) -> str:
iid = (import_id or "").strip()
if iid:
return iid
return _slugify_label(name)
def _parse_skill_key(raw: str) -> Tuple[str, str, str]:
parts = (raw or "").split("|", 2)
while len(parts) < 3:
parts.append("")
return parts[0], parts[1], parts[2]
def _csv_response(text: str, filename: str) -> PlainTextResponse:
return PlainTextResponse(
content=text,
media_type="text/csv; charset=utf-8",
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
)
def _rows_to_csv(fieldnames: List[str], rows: List[Dict[str, Any]]) -> str:
buf = io.StringIO()
writer = csv.DictWriter(buf, fieldnames=fieldnames, extrasaction="ignore", lineterminator="\n")
writer.writeheader()
for row in rows:
writer.writerow({k: row.get(k, "") for k in fieldnames})
return buf.getvalue()
def _load_editor_payload(cur) -> Dict[str, Any]:
cur.execute(
"""
SELECT s.id, s.name, s.description, s.importance, s.karate_relevance, s.relevance_level,
mc.slug AS main_category_slug, mc.name AS main_category_name,
sc.slug AS category_slug, sc.name AS category_name
FROM skills s
LEFT JOIN skill_main_categories mc ON s.main_category_id = mc.id
LEFT JOIN skill_categories sc ON s.category_id = sc.id
ORDER BY mc.sort_order NULLS LAST, sc.sort_order NULLS LAST, s.sort_order NULLS LAST, s.name
"""
)
skill_rows = [r2d(r) for r in cur.fetchall()]
skill_ids = [int(r["id"]) for r in skill_rows]
level_defs: Dict[int, Dict[int, str]] = {}
if skill_ids:
cur.execute(
"""
SELECT skill_id, level, description
FROM skill_level_definitions
WHERE skill_id = ANY(%s)
ORDER BY skill_id, level
""",
(skill_ids,),
)
for r in cur.fetchall():
sid = int(r["skill_id"])
level_defs.setdefault(sid, {})[int(r["level"])] = r.get("description") or ""
skills_out: List[Dict[str, Any]] = []
skills_csv: List[Dict[str, Any]] = []
for s in skill_rows:
sid = int(s["id"])
main_slug = s.get("main_category_slug")
cat_slug = s.get("category_slug")
name = (s.get("name") or "").strip()
key = _skill_key(main_slug, cat_slug, name)
defs = level_defs.get(sid, {})
level_map = {str(i): defs.get(i, "") for i in range(1, 6)}
entry = {
"skill_key": key,
"main_category": s.get("main_category_name") or "",
"subcategory": s.get("category_name") or "",
"skill_name": name,
"description": s.get("description") or "",
"importance": s.get("importance"),
"karate_relevance": s.get("karate_relevance"),
"relevance_level": s.get("relevance_level"),
"level_definitions": level_map,
}
skills_out.append(entry)
csv_row = {
"skill_key": key,
"main_category": entry["main_category"],
"subcategory": entry["subcategory"],
"skill_name": name,
"description": entry["description"],
"importance": entry["importance"] if entry["importance"] is not None else "",
"karate_relevance": entry["karate_relevance"] or "",
"relevance_level": entry["relevance_level"] if entry["relevance_level"] is not None else "",
}
for i in range(1, 6):
csv_row[f"level_{i}"] = level_map.get(str(i), "")
skills_csv.append(csv_row)
cur.execute("SELECT id, name, import_id, level_count, status FROM maturity_models ORDER BY name")
models_raw = [r2d(r) for r in cur.fetchall()]
models_out: List[Dict[str, Any]] = []
matrix_csv: List[Dict[str, Any]] = []
for m in models_raw:
mid = int(m["id"])
mkey = _model_key(m.get("import_id"), m.get("name") or "")
mname = (m.get("name") or "").strip()
cur.execute(
"""
SELECT level_number, name, description
FROM model_levels
WHERE maturity_model_id = %s
ORDER BY sort_order ASC, level_number ASC
""",
(mid,),
)
levels = [r2d(r) for r in cur.fetchall()]
level_label = {int(lv["level_number"]): lv.get("name") or "" for lv in levels}
cur.execute(
"""
SELECT ms.skill_id, ms.sort_order, ms.relevance, s.name AS skill_name,
mc.slug AS main_category_slug, sc.slug AS category_slug
FROM model_skills ms
JOIN skills s ON s.id = ms.skill_id
LEFT JOIN skill_main_categories mc ON s.main_category_id = mc.id
LEFT JOIN skill_categories sc ON s.category_id = sc.id
WHERE ms.maturity_model_id = %s
ORDER BY ms.sort_order ASC, ms.id ASC
""",
(mid,),
)
model_skills = [r2d(r) for r in cur.fetchall()]
cur.execute(
"""
SELECT msl.skill_id, msl.level_number, msl.description, msl.observable_criteria,
s.name AS skill_name, mc.slug AS main_category_slug, sc.slug AS category_slug
FROM model_skill_levels msl
JOIN skills s ON s.id = msl.skill_id
LEFT JOIN skill_main_categories mc ON s.main_category_id = mc.id
LEFT JOIN skill_categories sc ON s.category_id = sc.id
WHERE msl.maturity_model_id = %s
ORDER BY msl.skill_id, msl.level_number
""",
(mid,),
)
skill_levels = [r2d(r) for r in cur.fetchall()]
ms_by_skill: Dict[int, Dict[str, Any]] = {int(ms["skill_id"]): ms for ms in model_skills}
cells_by_skill: Dict[int, List[Dict[str, Any]]] = {}
for sl in skill_levels:
sid = int(sl["skill_id"])
cells_by_skill.setdefault(sid, []).append(
{
"level_number": int(sl["level_number"]),
"description": sl.get("description") or "",
"observable_criteria": sl.get("observable_criteria"),
}
)
matrix_rows: List[Dict[str, Any]] = []
seen_skills = set()
for ms in model_skills:
sid = int(ms["skill_id"])
seen_skills.add(sid)
skey = _skill_key(ms.get("main_category_slug"), ms.get("category_slug"), ms.get("skill_name") or "")
matrix_rows.append(
{
"skill_key": skey,
"skill_name": ms.get("skill_name") or "",
"relevance": ms.get("relevance"),
"sort_order": int(ms.get("sort_order") or 0),
"cells": cells_by_skill.get(sid, []),
}
)
for cell in cells_by_skill.get(sid, []):
ln = int(cell["level_number"])
matrix_csv.append(
{
"model_key": mkey,
"model_name": mname,
"skill_key": skey,
"skill_name": ms.get("skill_name") or "",
"relevance": ms.get("relevance") or "",
"sort_order": int(ms.get("sort_order") or 0),
"level_number": ln,
"level_label": level_label.get(ln, ""),
"description": cell.get("description") or "",
"observable_criteria": cell.get("observable_criteria") or "",
}
)
for sid, cells in cells_by_skill.items():
if sid in seen_skills:
continue
sample = cells[0] if cells else {}
cur.execute(
"""
SELECT s.name, mc.slug AS main_category_slug, sc.slug AS category_slug
FROM skills s
LEFT JOIN skill_main_categories mc ON s.main_category_id = mc.id
LEFT JOIN skill_categories sc ON s.category_id = sc.id
WHERE s.id = %s
""",
(sid,),
)
sk = r2d(cur.fetchone()) or {}
skey = _skill_key(sk.get("main_category_slug"), sk.get("category_slug"), sk.get("name") or "")
matrix_rows.append(
{
"skill_key": skey,
"skill_name": sk.get("name") or "",
"relevance": None,
"sort_order": 0,
"cells": cells,
}
)
for cell in cells:
ln = int(cell["level_number"])
matrix_csv.append(
{
"model_key": mkey,
"model_name": mname,
"skill_key": skey,
"skill_name": sk.get("name") or "",
"relevance": "",
"sort_order": 0,
"level_number": ln,
"level_label": level_label.get(ln, ""),
"description": cell.get("description") or "",
"observable_criteria": cell.get("observable_criteria") or "",
}
)
models_out.append(
{
"model_key": mkey,
"model_name": mname,
"level_count": int(m.get("level_count") or 5),
"status": m.get("status"),
"level_labels": [
{
"level_number": int(lv["level_number"]),
"name": lv.get("name") or "",
"description": lv.get("description"),
}
for lv in levels
],
"matrix_rows": matrix_rows,
}
)
return {
"skills": skills_out,
"skills_csv_rows": skills_csv,
"maturity_models": models_out,
"matrix_csv_rows": matrix_csv,
}
def _build_skill_lookup(cur) -> Dict[str, int]:
cur.execute(
"""
SELECT s.id, s.name, mc.slug AS main_category_slug, sc.slug AS category_slug
FROM skills s
LEFT JOIN skill_main_categories mc ON s.main_category_id = mc.id
LEFT JOIN skill_categories sc ON s.category_id = sc.id
"""
)
out: Dict[str, int] = {}
for r in cur.fetchall():
row = r2d(r)
key = _skill_key(row.get("main_category_slug"), row.get("category_slug"), row.get("name") or "")
out[key] = int(row["id"])
return out
def _build_model_lookup(cur) -> Dict[str, int]:
cur.execute("SELECT id, name, import_id FROM maturity_models")
out: Dict[str, int] = {}
for r in cur.fetchall():
row = r2d(r)
key = _model_key(row.get("import_id"), row.get("name") or "")
out[key] = int(row["id"])
return out
def _resolve_skill_id(
cur,
skill_lookup: Dict[str, int],
*,
skill_key: str,
main_category: str = "",
subcategory: str = "",
skill_name: str = "",
) -> Optional[int]:
key = (skill_key or "").strip()
if key and key in skill_lookup:
return skill_lookup[key]
if key:
main_slug, cat_slug, name = _parse_skill_key(key)
if name and _skill_key(main_slug, cat_slug, name) in skill_lookup:
return skill_lookup[_skill_key(main_slug, cat_slug, name)]
name = (skill_name or "").strip()
if not name:
return None
cur.execute(
"""
SELECT s.id
FROM skills s
LEFT JOIN skill_main_categories mc ON s.main_category_id = mc.id
LEFT JOIN skill_categories sc ON s.category_id = sc.id
WHERE s.name = %s
AND COALESCE(mc.name, '') = COALESCE(%s, '')
AND COALESCE(sc.name, '') = COALESCE(%s, '')
LIMIT 1
""",
(name, (main_category or "").strip() or None, (subcategory or "").strip() or None),
)
row = cur.fetchone()
if row:
return int(row["id"])
cur.execute("SELECT id FROM skills WHERE name = %s LIMIT 1", (name,))
row = cur.fetchone()
return int(row["id"]) if row else None
def _optional_int(val: Any) -> Optional[int]:
if val is None or val == "":
return None
try:
return int(val)
except (TypeError, ValueError):
return None
def _apply_skills_import(cur, skills: List[Dict[str, Any]], skill_lookup: Dict[str, int], warnings: List[str]) -> int:
updated = 0
for row in skills:
sid = _resolve_skill_id(
cur,
skill_lookup,
skill_key=str(row.get("skill_key") or ""),
main_category=str(row.get("main_category") or ""),
subcategory=str(row.get("subcategory") or ""),
skill_name=str(row.get("skill_name") or ""),
)
if not sid:
warnings.append(f"Skill nicht gefunden: {row.get('skill_key') or row.get('skill_name')}")
continue
importance = row.get("importance")
if importance == "":
importance = None
relevance_level = row.get("relevance_level")
if relevance_level == "":
relevance_level = None
cur.execute(
"""
UPDATE skills SET
description = COALESCE(%s, description),
importance = COALESCE(%s, importance),
karate_relevance = COALESCE(%s, karate_relevance),
relevance_level = COALESCE(%s, relevance_level),
updated_at = NOW()
WHERE id = %s
""",
(
row.get("description"),
importance,
row.get("karate_relevance") if row.get("karate_relevance") != "" else None,
relevance_level,
sid,
),
)
updated += 1
defs = row.get("level_definitions")
if isinstance(defs, dict):
for lvl_str, desc in defs.items():
try:
lvl = int(lvl_str)
except (TypeError, ValueError):
continue
if lvl < 1 or lvl > 10:
continue
text = (desc or "").strip()
if not text:
continue
cur.execute(
"""
INSERT INTO skill_level_definitions (skill_id, level, description)
VALUES (%s, %s, %s)
ON CONFLICT (skill_id, level) DO UPDATE SET
description = EXCLUDED.description,
updated_at = NOW()
""",
(sid, lvl, text),
)
else:
for i in range(1, 6):
col = row.get(f"level_{i}")
if col is None or str(col).strip() == "":
continue
cur.execute(
"""
INSERT INTO skill_level_definitions (skill_id, level, description)
VALUES (%s, %s, %s)
ON CONFLICT (skill_id, level) DO UPDATE SET
description = EXCLUDED.description,
updated_at = NOW()
""",
(sid, i, str(col).strip()),
)
return updated
def _apply_matrix_import(
cur,
models: List[Dict[str, Any]],
model_lookup: Dict[str, int],
skill_lookup: Dict[str, int],
warnings: List[str],
) -> int:
cells_updated = 0
for model in models:
mkey = str(model.get("model_key") or "").strip()
mid = model_lookup.get(mkey)
if not mid and model.get("model_name"):
mid = model_lookup.get(_model_key(None, str(model.get("model_name"))))
if not mid:
warnings.append(f"Reifegradmodell nicht gefunden: {mkey or model.get('model_name')}")
continue
rows = model.get("matrix_rows") or []
if not rows and model.get("cells"):
rows = [model]
for mrow in rows:
sid = _resolve_skill_id(
cur,
skill_lookup,
skill_key=str(mrow.get("skill_key") or ""),
skill_name=str(mrow.get("skill_name") or ""),
)
if not sid:
warnings.append(f"Matrix-Zeile: Skill nicht gefunden ({mrow.get('skill_key')})")
continue
relevance = mrow.get("relevance")
if relevance is not None and str(relevance).strip() != "":
sort_order = _optional_int(mrow.get("sort_order")) or 0
cur.execute(
"""
INSERT INTO model_skills (maturity_model_id, skill_id, sort_order, relevance)
VALUES (%s, %s, %s, %s)
ON CONFLICT (maturity_model_id, skill_id) DO UPDATE SET
sort_order = EXCLUDED.sort_order,
relevance = EXCLUDED.relevance
""",
(mid, sid, sort_order, str(relevance).strip()),
)
cells = mrow.get("cells") or []
for cell in cells:
ln = _optional_int(cell.get("level_number"))
if ln is None:
continue
desc = (cell.get("description") or "").strip()
if not desc:
continue
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 = COALESCE(EXCLUDED.observable_criteria, model_skill_levels.observable_criteria),
updated_at = NOW()
""",
(
mid,
sid,
ln,
desc,
cell.get("observable_criteria") if cell.get("observable_criteria") else None,
),
)
cells_updated += 1
return cells_updated
def _parse_csv_text(text: str) -> List[Dict[str, str]]:
reader = csv.DictReader(io.StringIO(text))
return [dict(row) for row in reader]
def _skills_from_csv_rows(rows: List[Dict[str, str]]) -> List[Dict[str, Any]]:
out: List[Dict[str, Any]] = []
for row in rows:
if not any(str(v).strip() for v in row.values()):
continue
entry: Dict[str, Any] = {
"skill_key": row.get("skill_key", ""),
"main_category": row.get("main_category", ""),
"subcategory": row.get("subcategory", ""),
"skill_name": row.get("skill_name", ""),
"description": row.get("description", ""),
"importance": _optional_int(row.get("importance")),
"karate_relevance": row.get("karate_relevance", ""),
"relevance_level": _optional_int(row.get("relevance_level")),
}
for i in range(1, 6):
entry[f"level_{i}"] = row.get(f"level_{i}", "")
out.append(entry)
return out
def _matrix_from_csv_rows(rows: List[Dict[str, str]]) -> List[Dict[str, Any]]:
by_model: Dict[str, Dict[str, Any]] = {}
row_index: Dict[Tuple[str, str], Dict[str, Any]] = {}
for row in rows:
if not any(str(v).strip() for v in row.values()):
continue
mkey = (row.get("model_key") or "").strip()
if not mkey:
continue
model = by_model.setdefault(
mkey,
{"model_key": mkey, "model_name": row.get("model_name") or "", "matrix_rows": []},
)
skey = (row.get("skill_key") or "").strip()
row_key = (mkey, skey or (row.get("skill_name") or "").strip())
mrow = row_index.get(row_key)
if not mrow:
mrow = {
"skill_key": skey,
"skill_name": row.get("skill_name") or "",
"relevance": row.get("relevance") or "",
"sort_order": _optional_int(row.get("sort_order")) or 0,
"cells": [],
}
row_index[row_key] = mrow
model["matrix_rows"].append(mrow)
ln = _optional_int(row.get("level_number"))
if ln is None:
continue
desc = (row.get("description") or "").strip()
if not desc:
continue
mrow["cells"].append(
{
"level_number": ln,
"description": desc,
"observable_criteria": row.get("observable_criteria") or "",
}
)
return list(by_model.values())
@router.get("/export")
def export_matrix_editor(
format: str = Query("json", pattern="^(json|csv_skills|csv_matrix)$"),
session: dict = Depends(require_auth),
):
_require_superadmin(session)
export_uid = str(uuid.uuid4())
with get_db() as conn:
cur = get_cursor(conn)
payload = _load_editor_payload(cur)
if format == "csv_skills":
csv_text = _rows_to_csv(SKILLS_CSV_FIELDS, payload["skills_csv_rows"])
return _csv_response(csv_text, f"faehigkeiten-katalog-{export_uid[:8]}.csv")
if format == "csv_matrix":
csv_text = _rows_to_csv(MATRIX_CSV_FIELDS, payload["matrix_csv_rows"])
return _csv_response(csv_text, f"faehigkeitsmatrix-zellen-{export_uid[:8]}.csv")
bundle = {
"kind": KIND_V1,
"export_version": 1,
"bundle_export_id": export_uid,
"exported_at": datetime.now(timezone.utc).isoformat(),
"skills": payload["skills"],
"maturity_models": payload["maturity_models"],
}
return JSONResponse(
content=jsonable_encoder(bundle),
headers={
"Content-Disposition": f'attachment; filename="matrix-editor-{export_uid[:8]}.json"'
},
)
@router.post("/import")
def import_matrix_editor(
data: Dict[str, Any] = Body(...),
session: dict = Depends(require_auth),
):
_require_superadmin(session)
warnings: List[str] = []
skills: List[Dict[str, Any]] = []
models: List[Dict[str, Any]] = []
kind = data.get("kind")
if kind == KIND_V1:
skills = list(data.get("skills") or [])
models = list(data.get("maturity_models") or [])
elif kind == "shinkan.matrix_editor.csv":
part = (data.get("part") or "").strip().lower()
csv_text = data.get("csv_text") or ""
if not csv_text.strip():
raise HTTPException(400, "csv_text fehlt")
rows = _parse_csv_text(csv_text)
if part == "skills":
skills = _skills_from_csv_rows(rows)
elif part == "matrix":
models = _matrix_from_csv_rows(rows)
else:
raise HTTPException(400, "part muss skills oder matrix sein")
else:
raise HTTPException(400, f"Unbekanntes Format (kind={kind!r})")
if not skills and not models:
raise HTTPException(400, "Keine importierbaren Daten")
with get_db() as conn:
cur = get_cursor(conn)
skill_lookup = _build_skill_lookup(cur)
model_lookup = _build_model_lookup(cur)
skills_updated = _apply_skills_import(cur, skills, skill_lookup, warnings) if skills else 0
cells_updated = _apply_matrix_import(cur, models, model_lookup, skill_lookup, warnings) if models else 0
return {
"ok": True,
"skills_updated": skills_updated,
"matrix_cells_updated": cells_updated,
"warnings": warnings,
}

View File

@ -0,0 +1,146 @@
"""GET/POST /api/admin/matrix-editor — Superadmin Export/Import."""
from __future__ import annotations
import os
from unittest.mock import MagicMock, patch
import pytest
from fastapi.testclient import TestClient
os.environ.setdefault("SKIP_DB_MIGRATE", "1")
from auth import require_auth
from main import app
@pytest.fixture
def client() -> TestClient:
return TestClient(app)
@pytest.fixture(autouse=True)
def _clear_overrides():
yield
app.dependency_overrides.pop(require_auth, None)
def test_matrix_editor_export_requires_superadmin(client: TestClient) -> None:
def _admin():
return {"profile_id": 1, "role": "admin"}
app.dependency_overrides[require_auth] = _admin
r = client.get("/api/admin/matrix-editor/export", headers={"X-Auth-Token": "t"})
assert r.status_code == 403
@patch("routers.matrix_editor._load_editor_payload")
def test_matrix_editor_export_json_ok(mock_load, client: TestClient) -> None:
def _superadmin():
return {"profile_id": 1, "role": "superadmin"}
app.dependency_overrides[require_auth] = _superadmin
mock_load.return_value = {
"skills": [{"skill_key": "a|b|Test", "skill_name": "Test", "description": "x", "level_definitions": {}}],
"skills_csv_rows": [],
"maturity_models": [],
"matrix_csv_rows": [],
}
mock_cm = MagicMock()
mock_cm.__enter__.return_value = MagicMock()
mock_cm.__exit__.return_value = False
with patch("routers.matrix_editor.get_db", return_value=mock_cm), patch(
"routers.matrix_editor.get_cursor", return_value=MagicMock()
):
r = client.get("/api/admin/matrix-editor/export", headers={"X-Auth-Token": "t"})
assert r.status_code == 200
body = r.json()
assert body["kind"] == "shinkan.matrix_editor.v1"
assert len(body["skills"]) == 1
@patch("routers.matrix_editor._load_editor_payload")
def test_matrix_editor_export_csv_skills(mock_load, client: TestClient) -> None:
def _superadmin():
return {"profile_id": 1, "role": "superadmin"}
app.dependency_overrides[require_auth] = _superadmin
mock_load.return_value = {
"skills": [],
"skills_csv_rows": [
{
"skill_key": "main|sub|Skill A",
"main_category": "Main",
"subcategory": "Sub",
"skill_name": "Skill A",
"description": "Desc",
"importance": 3,
"karate_relevance": "",
"relevance_level": "",
"level_1": "L1",
"level_2": "",
"level_3": "",
"level_4": "",
"level_5": "",
}
],
"maturity_models": [],
"matrix_csv_rows": [],
}
mock_cm = MagicMock()
mock_cm.__enter__.return_value = MagicMock()
mock_cm.__exit__.return_value = False
with patch("routers.matrix_editor.get_db", return_value=mock_cm), patch(
"routers.matrix_editor.get_cursor", return_value=MagicMock()
):
r = client.get(
"/api/admin/matrix-editor/export?format=csv_skills",
headers={"X-Auth-Token": "t"},
)
assert r.status_code == 200
assert "skill_key" in r.text
assert "Skill A" in r.text
@patch("routers.matrix_editor._apply_matrix_import", return_value=0)
@patch("routers.matrix_editor._apply_skills_import", return_value=1)
@patch("routers.matrix_editor._build_model_lookup", return_value={})
@patch("routers.matrix_editor._build_skill_lookup", return_value={"main|sub|Skill A": 42})
def test_matrix_editor_import_json_ok(
mock_skill_lookup,
mock_model_lookup,
mock_apply_skills,
mock_apply_matrix,
client: TestClient,
) -> None:
def _superadmin():
return {"profile_id": 1, "role": "superadmin"}
app.dependency_overrides[require_auth] = _superadmin
mock_cm = MagicMock()
mock_conn = MagicMock()
mock_cm.__enter__.return_value = mock_conn
mock_cm.__exit__.return_value = False
with patch("routers.matrix_editor.get_db", return_value=mock_cm), patch(
"routers.matrix_editor.get_cursor", return_value=MagicMock()
):
r = client.post(
"/api/admin/matrix-editor/import",
headers={"X-Auth-Token": "t", "Content-Type": "application/json"},
json={
"kind": "shinkan.matrix_editor.v1",
"skills": [{"skill_key": "main|sub|Skill A", "description": "Neu"}],
},
)
assert r.status_code == 200
body = r.json()
assert body["ok"] is True
assert body["skills_updated"] == 1

View File

@ -13,13 +13,10 @@ export function mergeActiveClubHeader(headers = {}) {
if (cid && /^\d+$/.test(String(cid).trim())) { if (cid && /^\d+$/.test(String(cid).trim())) {
return { ...headers, 'X-Active-Club-Id': String(cid).trim() } return { ...headers, 'X-Active-Club-Id': String(cid).trim() }
} }
return { ...headers } return headers
} }
/** async function _fetchWithAuth(endpoint, options = {}) {
* Generischer API-Aufruf inkl. X-Auth-Token und X-Active-Club-Id.
*/
export async function request(endpoint, options = {}) {
const token = localStorage.getItem('authToken') const token = localStorage.getItem('authToken')
const method = (options.method || 'GET').toUpperCase() const method = (options.method || 'GET').toUpperCase()
@ -65,7 +62,7 @@ export async function request(endpoint, options = {}) {
throw new Error(snippet ? `HTTP ${response.status}: ${snippet}` : `HTTP ${response.status}`) throw new Error(snippet ? `HTTP ${response.status}: ${snippet}` : `HTTP ${response.status}`)
} }
return response.json() return response
} catch (e) { } catch (e) {
if (e instanceof TypeError && (e.message === 'Failed to fetch' || e.message.includes('fetch'))) { if (e instanceof TypeError && (e.message === 'Failed to fetch' || e.message.includes('fetch'))) {
const hint = const hint =
@ -77,3 +74,22 @@ export async function request(endpoint, options = {}) {
throw e throw e
} }
} }
/**
* Generischer API-Aufruf inkl. X-Auth-Token und X-Active-Club-Id.
*/
export async function request(endpoint, options = {}) {
const response = await _fetchWithAuth(endpoint, options)
return response.json()
}
/** Text-Download (z. B. CSV-Export) mit gleicher Auth wie request(). */
export async function requestText(endpoint, options = {}) {
const response = await _fetchWithAuth(endpoint, options)
const disposition = response.headers.get('Content-Disposition') || ''
const match = disposition.match(/filename="([^"]+)"/)
return {
text: await response.text(),
filename: match ? match[1] : null,
}
}

View File

@ -13,6 +13,16 @@ function downloadJson(obj, filename) {
URL.revokeObjectURL(url) URL.revokeObjectURL(url)
} }
function downloadText(text, filename, mime = 'text/csv;charset=utf-8') {
const blob = new Blob([text], { type: mime })
const url = URL.createObjectURL(blob)
const a = document.createElement('a')
a.href = url
a.download = filename
a.click()
URL.revokeObjectURL(url)
}
function groupModelSkills(model) { function groupModelSkills(model) {
if (!model?.model_skills?.length) return [] if (!model?.model_skills?.length) return []
const groups = new Map() const groups = new Map()
@ -56,6 +66,8 @@ export default function MaturityMatrixToolsAdmin() {
const [stackConfirmText, setStackConfirmText] = useState('') const [stackConfirmText, setStackConfirmText] = useState('')
const [stackLoading, setStackLoading] = useState(false) const [stackLoading, setStackLoading] = useState(false)
const [stackImportLoading, setStackImportLoading] = useState(false) const [stackImportLoading, setStackImportLoading] = useState(false)
const [editorLoading, setEditorLoading] = useState(false)
const [editorImportLoading, setEditorImportLoading] = useState(false)
useEffect(() => { useEffect(() => {
let cancelled = false let cancelled = false
@ -144,6 +156,94 @@ export default function MaturityMatrixToolsAdmin() {
} }
} }
async function handleExportEditorJson() {
setError('')
setMessage('')
setEditorLoading(true)
try {
const bundle = await api.exportMatrixEditorBundle()
const name = `matrix-editor-${(bundle.bundle_export_id || 'export').slice(0, 8)}.json`
downloadJson(bundle, name)
setMessage('Editor-Export (JSON) heruntergeladen.')
} catch (err) {
setError(err.message || String(err))
} finally {
setEditorLoading(false)
}
}
async function handleExportEditorCsv(part) {
setError('')
setMessage('')
setEditorLoading(true)
try {
const { text, filename } = await api.exportMatrixEditorCsv(part)
downloadText(
text,
filename || (part === 'skills' ? 'faehigkeiten-katalog.csv' : 'faehigkeitsmatrix-zellen.csv')
)
setMessage(part === 'skills' ? 'Katalog-CSV heruntergeladen.' : 'Matrix-CSV heruntergeladen.')
} catch (err) {
setError(err.message || String(err))
} finally {
setEditorLoading(false)
}
}
async function handleImportEditorJson(e) {
const file = e.target.files?.[0]
if (!file) return
setError('')
setMessage('')
setEditorImportLoading(true)
try {
const data = JSON.parse(await file.text())
if (data.kind !== 'shinkan.matrix_editor.v1') {
setError('Erwartet wird kind: shinkan.matrix_editor.v1')
return
}
const res = await api.importMatrixEditorBundle(data)
const w = res.warnings || []
setMessage(
`Import OK: ${res.skills_updated || 0} Fähigkeit(en), ${res.matrix_cells_updated || 0} Matrix-Zelle(n) aktualisiert.` +
(w.length ? ` ${w.length} Hinweis(e) in der Konsole.` : '')
)
if (w.length) console.warn('matrix_editor import warnings', w)
} catch (err) {
setError(err.message || String(err))
} finally {
setEditorImportLoading(false)
e.target.value = ''
}
}
async function handleImportEditorCsv(e, part) {
const file = e.target.files?.[0]
if (!file) return
setError('')
setMessage('')
setEditorImportLoading(true)
try {
const csv_text = await file.text()
const res = await api.importMatrixEditorBundle({
kind: 'shinkan.matrix_editor.csv',
part,
csv_text
})
const w = res.warnings || []
setMessage(
`CSV-Import (${part}) OK: ${res.skills_updated || 0} Fähigkeit(en), ${res.matrix_cells_updated || 0} Matrix-Zelle(n) aktualisiert.` +
(w.length ? ` ${w.length} Hinweis(e) in der Konsole.` : '')
)
if (w.length) console.warn('matrix_editor csv import warnings', w)
} catch (err) {
setError(err.message || String(err))
} finally {
setEditorImportLoading(false)
e.target.value = ''
}
}
async function handleExportStack() { async function handleExportStack() {
setError('') setError('')
setMessage('') setMessage('')
@ -234,9 +334,8 @@ export default function MaturityMatrixToolsAdmin() {
return ( return (
<div className="admin-matrix-tools"> <div className="admin-matrix-tools">
<p className="admin-matrix-tools__intro muted"> <p className="admin-matrix-tools__intro muted">
Matrix nach Kontext auflösen, hierarchisch nach Hauptkategorie und Kategorie darstellen, sowie JSON Zentral Beschreibungen und Gewichtungen pflegen (Superadmin), Matrix nach Kontext anzeigen, sowie
exportieren oder importieren (gespeichertes Modell inkl. optional Kontext-Bindings, aufgelöste Matrix, oder{' '} vollständige JSON-Stacks für Test Prod.
<strong>Komplett-Stack</strong> mit Fähigkeitskatalog und allen Reifegradmodellen für Test Prod).
</p> </p>
{error ? ( {error ? (
@ -246,6 +345,76 @@ export default function MaturityMatrixToolsAdmin() {
) : null} ) : null}
{message ? <p className="muted admin-matrix-tools__msg">{message}</p> : null} {message ? <p className="muted admin-matrix-tools__msg">{message}</p> : null}
<section className="card admin-matrix-tools__section">
<h2 className="admin-matrix-tools__h2">Zentral bearbeiten (Export / Import)</h2>
<p className="muted admin-matrix-tools__hint">
Flaches Format für Excel oder JSON-Editor: Fähigkeits-Beschreibungen,{' '}
<code className="admin-bindings__code">importance</code> (Gewichtung 15), globale Stufen-Texte (
<code className="admin-bindings__code">level_1</code> <code className="admin-bindings__code">level_5</code>
), Matrix-Zelltexte und Zeilen-Relevanz pro Reifegradmodell. Import aktualisiert bestehende Einträge
es werden keine neuen Fähigkeiten oder Modelle angelegt.
</p>
<div className="admin-matrix-tools__actions">
<button
type="button"
className="btn btn-primary"
disabled={editorLoading || editorImportLoading}
onClick={handleExportEditorJson}
>
{editorLoading ? 'Export…' : 'JSON exportieren'}
</button>
<button
type="button"
className="btn btn-secondary"
disabled={editorLoading || editorImportLoading}
onClick={() => handleExportEditorCsv('skills')}
>
Katalog als CSV
</button>
<button
type="button"
className="btn btn-secondary"
disabled={editorLoading || editorImportLoading}
onClick={() => handleExportEditorCsv('matrix')}
>
Matrix-Zellen als CSV
</button>
</div>
<div className="admin-matrix-tools__io-grid admin-matrix-tools__io-grid--editor">
<div>
<h3 className="admin-matrix-tools__h3">JSON importieren</h3>
<label className="form-label">Datei (<code className="admin-bindings__code">shinkan.matrix_editor.v1</code>)</label>
<input
type="file"
accept="application/json,.json"
disabled={editorImportLoading}
onChange={handleImportEditorJson}
/>
</div>
<div>
<h3 className="admin-matrix-tools__h3">CSV importieren</h3>
<p className="muted admin-matrix-tools__hint">
Nur geänderte Datei hochladen Katalog-CSV oder Matrix-CSV getrennt.
</p>
<label className="form-label">Katalog-CSV</label>
<input
type="file"
accept="text/csv,.csv"
disabled={editorImportLoading}
onChange={(e) => handleImportEditorCsv(e, 'skills')}
/>
<label className="form-label admin-matrix-tools__btn-mt">Matrix-CSV</label>
<input
type="file"
accept="text/csv,.csv"
disabled={editorImportLoading}
onChange={(e) => handleImportEditorCsv(e, 'matrix')}
/>
</div>
</div>
{editorImportLoading ? <p className="muted admin-matrix-tools__msg">Import läuft</p> : null}
</section>
<section className="card admin-matrix-tools__section"> <section className="card admin-matrix-tools__section">
<h2 className="admin-matrix-tools__h2">Kontext und Anzeige</h2> <h2 className="admin-matrix-tools__h2">Kontext und Anzeige</h2>
<div className="admin-matrix-tools__filters"> <div className="admin-matrix-tools__filters">

View File

@ -4,7 +4,7 @@
* Zentrale API-Kommunikation mit automatischer Token-Injektion * Zentrale API-Kommunikation mit automatischer Token-Injektion
*/ */
import { request, ACTIVE_CLUB_STORAGE_KEY } from '../api/client.js' import { request, ACTIVE_CLUB_STORAGE_KEY, requestText } from '../api/client.js'
import * as exercises from '../api/exercises.js' import * as exercises from '../api/exercises.js'
import * as planning from '../api/planning.js' import * as planning from '../api/planning.js'
import * as skillProfiles from '../api/skillProfiles.js' import * as skillProfiles from '../api/skillProfiles.js'
@ -523,6 +523,23 @@ export async function importMatrixStackBundle(payload) {
}) })
} }
/** Superadmin: flacher Export für zentrale Pflege (Beschreibungen, Gewichtungen) */
export async function exportMatrixEditorBundle() {
return request('/api/admin/matrix-editor/export')
}
export async function exportMatrixEditorCsv(part) {
const format = part === 'skills' ? 'csv_skills' : 'csv_matrix'
return requestText(`/api/admin/matrix-editor/export?format=${format}`)
}
export async function importMatrixEditorBundle(payload) {
return request('/api/admin/matrix-editor/import', {
method: 'POST',
body: JSON.stringify(payload)
})
}
// Style Directions (formerly Training Styles) // Style Directions (formerly Training Styles)
export async function listStyleDirections(filters = {}) { export async function listStyleDirections(filters = {}) {
const query = new URLSearchParams(filters).toString() const query = new URLSearchParams(filters).toString()
@ -874,6 +891,9 @@ export const api = {
exportResolvedMaturityBundle, exportResolvedMaturityBundle,
exportMatrixStackBundle, exportMatrixStackBundle,
importMatrixStackBundle, importMatrixStackBundle,
exportMatrixEditorBundle,
exportMatrixEditorCsv,
importMatrixEditorBundle,
resolveMaturityModel, resolveMaturityModel,
getMaturityModel, getMaturityModel,
createMaturityModel, createMaturityModel,