from urllib.parse import urlencode from fastapi import APIRouter, Depends, Form, Request from fastapi.responses import RedirectResponse from anode_client import anode_request from auth import require_admin from templating import templates # Workspace-Creator neu (Konzept 07.10.2026, siehe # platform-docs/roadmap/workspace-creator-neu.md): eigener Hauptbereich # "Workspaces" für root mit den Reitern Katalog, Workspaces und Vorlagen. # Schritt A: Katalog (nur lesend), ersetzt die frühere "Flottenmatrix". router = APIRouter(prefix="/admin/workspaces") def _liste(result: dict, schluessel: str) -> list[dict]: return result.get(schluessel, []) if result.get("success") else [] def _eintrag( name: str, description: str | None, icon_url: str | None, distros: list[dict], eintrag_id: str | None = None, pflege: dict | None = None, ) -> dict: # "pflege": was die Katalogseite zum Bearbeiten braucht (Art, Key, # Kategorie, ...); der Assistent und die Workspace-Seite ignorieren es. return { "id": eintrag_id, "name": name, "description": description, "icon_url": icon_url, "distros": distros, "distro_keys": [d["key"] for d in distros if d["verfuegbar"]], "pflege": pflege or {}, } def _nach_kategorie(kategorien: list[dict], eintraege_mit_kategorie: list[tuple[str | None, dict]]) -> list[dict]: """Gruppiert (kategorie_id, eintrag)-Paare in der Reihenfolge der Kategorien (sort_order); leere Kategorien entfallen, Einträge ohne Kategorie landen am Ende unter "Ohne Kategorie".""" nach_id: dict[str | None, list[dict]] = {} for kategorie_id, eintrag in eintraege_mit_kategorie: nach_id.setdefault(kategorie_id, []).append(eintrag) gruppen = [ {"name": k["name"], "description": k.get("description"), "eintraege": sorted(nach_id[k["id"]], key=lambda e: e["name"].lower())} for k in kategorien if k["id"] in nach_id ] bekannte_ids = {k["id"] for k in kategorien} ohne = [e for kid, eintraege in nach_id.items() if kid not in bekannte_ids for e in eintraege] if ohne: gruppen.append({"name": "Ohne Kategorie", "description": None, "eintraege": sorted(ohne, key=lambda e: e["name"].lower())}) return gruppen def katalog_aufbauen( kategorien: list[dict], merkmale: list[dict], webapp_templates: list[dict], backends: list[dict], blueprints: list[dict], icon_basis: str = "/admin/workspaces", ) -> list[dict]: """ Drei Bereiche (Apps, Webapps, Konfigurationsrichtlinien), darin Kategorien, darin Einträge mit Distro-Verfügbarkeit. Ob ein Merkmal auf einer Distro geht, ergibt sich wie bei der früheren Flottenmatrix aus den Blueprints (Merkmal × Distro → Ansible-Rolle). Webapps laufen im Browser und sind damit auf jeder Distro verfügbar. icon_basis ist der Pfad der Icon-Proxys (Admin oder Kundenportal, seit 23.10.2026). """ # Rolle je (Merkmal, Distro) - die Katalogseite zeigt sie im Tooltip # des Distro-Kennzeichens (früher eigene Seite "Distro-Abdeckung"). rollen_je_merkmal: dict[str, dict[str, str]] = {} for bp in blueprints: rollen_je_merkmal.setdefault(bp["merkmal"], {})[bp["backend"]] = bp.get("ansible_role") def distro_liste(verfuegbar, rollen: dict[str, str] | None = None) -> list[dict]: rollen = rollen or {} return [ {"key": b["key"], "name": b["name"], "version": b.get("version"), "verfuegbar": b["key"] in verfuegbar, "rolle": rollen.get(b["key"])} for b in backends ] art_je_kategorie = {k["id"]: k.get("art", "app") for k in kategorien} app_kategorien = [k for k in kategorien if k.get("art", "app") == "app"] richtlinien_kategorien = [k for k in kategorien if k.get("art") == "richtlinie"] apps, richtlinien = [], [] for m in merkmale: rollen = rollen_je_merkmal.get(m["key"], {}) eintrag = _eintrag( m["name"], m.get("description"), f"{icon_basis}{m['icon_path']}" if m.get("icon_path") else None, distro_liste(set(rollen), rollen), m["id"], {"art": "merkmal", "key": m["key"], "kategorie_id": m.get("kategorie_id"), "im_auftragskatalog": bool(m.get("im_auftragskatalog"))}, ) ziel = richtlinien if art_je_kategorie.get(m.get("kategorie_id")) == "richtlinie" else apps ziel.append((m.get("kategorie_id"), eintrag)) alle_distros = {b["key"] for b in backends} webapps = [ ( t.get("kategorie_id"), _eintrag( t["name"], t.get("description"), f"{icon_basis}/webapp-icons/{t['icon_path'].split('/')[-1]}" if t.get("icon_path") else None, distro_liste(alle_distros), t["id"], {"art": "webapp", "url": t.get("url"), "kategorie_id": t.get("kategorie_id")}, ), ) for t in webapp_templates ] bereiche = [ ("apps", "Apps", "📦", app_kategorien, apps), ("webapps", "Webapps", "🌐", app_kategorien, webapps), ("richtlinien", "Konfigurationsrichtlinien", "🛡️", richtlinien_kategorien, richtlinien), ] return [ { "key": key, "name": name, "symbol": symbol, "anzahl": len(eintraege), "kategorien": _nach_kategorie(kats, eintraege), } for key, name, symbol, kats, eintraege in bereiche ] @router.get("") def index(user: dict = Depends(require_admin)): return RedirectResponse("/admin/workspaces/katalog", status_code=303) @router.get("/vorlagen") def vorlagen(user: dict = Depends(require_admin)): return RedirectResponse("/admin/workspaces/katalog", status_code=303) # Bereich "Datenbank" (/admin/technik) ist in "Workspaces" aufgegangen # (Schritt B, 07.10.2026) - alte Lesezeichen und Links leiten weiter. # Bewusst 307/308 statt 303 für POST, damit ein noch offenes altes Formular # nicht stillschweigend zu einem GET wird. ALTE_TECHNIK_PFADE = [ ("webapp-templates", "vorlagen/webapp-vorlagen"), ("webapp-icons", "webapp-icons"), ("merkmal-icons", "merkmal-icons"), ("merkmale", "vorlagen/merkmale"), ("kategorien", "vorlagen/kategorien"), ("backends", "vorlagen/distros"), ("workspaces", "liste"), ] def neuer_pfad_fuer_technik(rest: str) -> str: rest = rest.strip("/") for alt, neu in ALTE_TECHNIK_PFADE: if rest == alt or rest.startswith(alt + "/"): return "/admin/workspaces/" + neu + rest[len(alt):] return "/admin/workspaces" technik_weiterleitung = APIRouter(prefix="/admin/technik") @technik_weiterleitung.api_route("", methods=["GET", "POST"]) @technik_weiterleitung.api_route("/{rest:path}", methods=["GET", "POST"]) def alte_technik_adresse(request: Request, rest: str = ""): status = 303 if request.method == "GET" else 307 return RedirectResponse(neuer_pfad_fuer_technik(rest), status_code=status) def katalog_laden(mit_kategorien: bool = False, icon_basis: str = "/admin/workspaces") -> tuple: """(bereiche, backends) - gemeinsame Datengrundlage für Katalog und Assistent; mit_kategorien hängt die Kategorienliste an (für die Bearbeiten-Formulare der Katalogseite).""" kategorien = _liste(anode_request("GET", "/api/v1/kategorien"), "kategorien") merkmale = _liste(anode_request("GET", "/api/v1/merkmale"), "merkmale") webapp_templates = _liste(anode_request("GET", "/api/v1/webapp-templates"), "webapp_templates") backends = _liste(anode_request("GET", "/api/v1/backends"), "backends") blueprints = _liste(anode_request("GET", "/api/v1/blueprints"), "blueprints") bereiche = katalog_aufbauen(kategorien, merkmale, webapp_templates, backends, blueprints, icon_basis) if mit_kategorien: return bereiche, backends, kategorien return bereiche, backends # GUI-Runde 22.10.2026: "Übersicht", "Merkmale bearbeiten" und # "Distro-Abdeckung" sind in dieser einen Seite aufgegangen, am 23.10.2026 # auch "Kategorien" und "Webapp-Vorlagen" (aufklappbar unter den Einträgen). @router.get("/katalog") def katalog(request: Request, user: dict = Depends(require_admin)): bereiche, backends, kategorien = katalog_laden(mit_kategorien=True) # Für den Abschnitt Webapp-Vorlagen (GUI-Runde 23.10.2026). webapp_templates = _liste(anode_request("GET", "/api/v1/webapp-templates"), "webapp_templates") return templates.TemplateResponse( request, "admin/workspaces_katalog.html", { "user": user, "bereiche": bereiche, "backends": backends, "kategorien": kategorien, "app_kategorien": [k for k in kategorien if k.get("art", "app") == "app"], "webapp_templates": webapp_templates, }, ) # --- Assistent "Workspace erstellen" (Schritt D, 07.10.2026) --- # Einstieg vereinfachen, Schritt 5 (25.10.2026): Kurzcharakteristik je # Distribution als Entscheidungshilfe (B15). "nur_admin": für Kunden noch # nicht wählbar (Debian-Image ohne Desktop, siehe Memory desktop-je-distro). DISTRO_INFO = { "mint-image": { "text": "Einfacher Umstieg von Windows: Startmenü und Taskleiste wie gewohnt. Tuxflotte-Standard für Schulen.", }, "ubuntu-image": { "text": "Wenn ein Anschluss an ein Active Directory geplant ist. Bedienung etwas anders als Windows; persönliche Konten (Personalisierung) gibt es hier noch nicht.", }, "debian-image": { "text": "Für ältere, schwache Geräte: schlank und sehr stabil.", "nur_admin": "Noch in Vorbereitung: derzeit ohne grafische Oberfläche.", }, } def assistent_seite( request: Request, user, werte: dict, fehler: str | None = None, *, ist_admin: bool = True, assistent_ziel: str = "/admin/workspaces/neu", icon_basis: str = "/admin/workspaces", standorte: list[dict] | None = None, eingerichtete_backend_ids: set[str] | None = None, ): """Assistent für Admin und (seit 23.10.2026) Kundenportal - der Kunde wählt keinen Eigentümer, der Workspace gehört immer seiner Organisation. Seit Schritt 5 (25.10.2026) wählt der Kunde zuerst die Distribution und am Ende einen Standort/Zweck (standorte); eingerichtete_backend_ids sagt, für welche Distros seine Organisation schon installieren kann.""" bereiche, backends = katalog_laden(icon_basis=icon_basis) organisationen = _liste(anode_request("GET", "/api/v1/organizations"), "organizations") if ist_admin else [] backends = [ { **b, "info": DISTRO_INFO.get(b["key"], {}).get("text"), "gesperrt": None if ist_admin else DISTRO_INFO.get(b["key"], {}).get("nur_admin"), "eingerichtet": None if eingerichtete_backend_ids is None else b["id"] in eingerichtete_backend_ids, } for b in backends ] return templates.TemplateResponse( request, "admin/workspace_assistent.html" if ist_admin else "workspace_assistent.html", { "user": user, "ist_admin": ist_admin, "assistent_ziel": assistent_ziel, "bereiche": bereiche, "backends": backends, "organisationen": organisationen, "standorte": standorte, "werte": werte, "fehler": fehler, }, status_code=400 if fehler else 200, ) @router.get("/neu") def assistent(request: Request, user: dict = Depends(require_admin)): return assistent_seite(request, user, {"merkmal_ids": [], "webapp_template_ids": []}) @router.post("/neu") async def assistent_speichern(request: Request, user: dict = Depends(require_admin)): form = await request.form() werte = { "name": (form.get("name") or "").strip(), "description": (form.get("description") or "").strip(), "organization_id": form.get("organization_id") or "", "backend_id": form.get("backend_id") or "", "merkmal_ids": form.getlist("merkmal_ids"), "webapp_template_ids": form.getlist("webapp_template_ids"), } if not werte["name"]: return assistent_seite(request, user, werte, "Bitte einen Namen angeben.") if not werte["backend_id"]: return assistent_seite(request, user, werte, "Bitte eine Distro wählen.") result = anode_request( "POST", "/api/v1/workspaces/assistent", json={ "name": werte["name"], "description": werte["description"] or None, "organization_id": werte["organization_id"] or None, "backend_id": werte["backend_id"], "merkmal_ids": werte["merkmal_ids"], "webapp_template_ids": werte["webapp_template_ids"], }, ) if not result.get("success"): return assistent_seite(request, user, werte, result.get("message", "Workspace konnte nicht angelegt werden.")) return RedirectResponse(f"/admin/workspaces/liste/{result['workspace']['id']}", status_code=303) # --- Workspace-Detail und -Verwaltung (Schritt E, 07.10.2026) --- # # Inhalte ändern sich nur im Entwurf (Server erzwingt das). Die Seite zeigt # die veröffentlichte Version nur lesend, im Entwurf einen Editor wie im # Assistenten plus Wirkungsvorschau und Veröffentlichen. def _zurueck(workspace_id: str, **meldung) -> RedirectResponse: ziel = f"/admin/workspaces/liste/{workspace_id}" if meldung: ziel += "?" + urlencode(meldung) return RedirectResponse(ziel, status_code=303) def _ergebnis(workspace_id: str, result: dict, erfolg: str | None = None) -> RedirectResponse: if not result.get("success"): return _zurueck(workspace_id, fehler=result.get("message", "Das hat nicht geklappt.")) return _zurueck(workspace_id, hinweis=erfolg) if erfolg else _zurueck(workspace_id) def detail_kontext( request: Request, workspace_id: str, workspace: dict, *, ist_admin: bool, darf_bearbeiten: bool, icon_basis: str, ) -> dict: """Alles, was templates/_workspace_detail.html braucht - für Admin und (seit 23.10.2026) Kundenportal. Ohne Bearbeitungsrecht (fremde Vorlage) keine Nutzung und keine Wirkung: die nennen andere Organisationen.""" ist_entwurf = darf_bearbeiten and bool(workspace.get("version") and workspace["version"]["status"] == "entwurf") bereiche, backends = katalog_laden(icon_basis=icon_basis) ausgewaehlt = {m["id"] for m in workspace["merkmale"]} distro_key = next((b["key"] for b in backends if workspace["backend"] and b["id"] == workspace["backend"]["id"]), None) webapps = _liste(anode_request("GET", f"/api/v1/workspaces/{workspace_id}/webapps"), "webapps") vorlagen_verfuegbar = ( _liste(anode_request("GET", f"/api/v1/workspaces/{workspace_id}/webapp-templates/available"), "webapp_templates") if ist_entwurf else [] ) wirkung = None if ist_entwurf: w = anode_request("GET", f"/api/v1/workspaces/{workspace_id}/wirkung") wirkung = w.get("wirkung") if w.get("success") else None nutzung = {} if darf_bearbeiten: nutzung = (wirkung or {}).get("nutzung") or anode_request( "GET", f"/api/v1/workspaces/{workspace_id}/nutzung" ).get("nutzung", {}) # Kopie-Prüfung (zweistufig: erst prüfen, dann anlegen) kopie = None if request.query_params.get("kopie_backend"): k = anode_request( "GET", f"/api/v1/workspaces/{workspace_id}/kopie-pruefung", params={"backend_id": request.query_params["kopie_backend"]}, ) kopie = { "name": request.query_params.get("kopie_name", ""), "backend_id": request.query_params["kopie_backend"], "organization_id": request.query_params.get("kopie_org", "") if ist_admin else "", "weggefallen": k.get("weggefallen", []) if k.get("success") else [], } return { "workspace": workspace, "workspace_id": workspace_id, "ist_entwurf": ist_entwurf, "bereiche": [b for b in bereiche if b["key"] != "webapps"], "ausgewaehlt": ausgewaehlt, "distro_key": distro_key, "webapps": webapps, "vorlagen_verfuegbar": vorlagen_verfuegbar, "wirkung": wirkung, "nutzung": nutzung, "versionen": _liste(anode_request("GET", f"/api/v1/workspaces/{workspace_id}/versionen"), "versionen"), "backends": backends, "kundengruppen": _liste(anode_request("GET", "/api/v1/kundengruppen"), "kundengruppen") if ist_admin else [], "organisationen": _liste(anode_request("GET", "/api/v1/organizations"), "organizations") if ist_admin else [], "kopie": kopie, "fehler": request.query_params.get("fehler"), "hinweis": request.query_params.get("hinweis"), "ist_admin": ist_admin, "darf_bearbeiten": darf_bearbeiten, "icon_basis": icon_basis, } @router.get("/liste/{workspace_id}") def workspace_detail(request: Request, workspace_id: str, user: dict = Depends(require_admin)): basis = {"user": user, "ws_basis": "/admin/workspaces/liste", "ws_liste_url": "/admin/workspaces/liste"} result = anode_request("GET", f"/api/v1/workspaces/{workspace_id}") workspace = result.get("workspace") if result.get("success") else None if workspace is None: return templates.TemplateResponse( request, "admin/workspace_detail.html", {**basis, "workspace": None}, status_code=404 ) return templates.TemplateResponse( request, "admin/workspace_detail.html", {**basis, **detail_kontext(request, workspace_id, workspace, ist_admin=True, darf_bearbeiten=True, icon_basis="/admin/workspaces")}, ) @router.post("/liste/{workspace_id}/entwurf") def entwurf_anlegen(workspace_id: str, user: dict = Depends(require_admin)): return _ergebnis(workspace_id, anode_request("POST", f"/api/v1/workspaces/{workspace_id}/entwurf")) @router.post("/liste/{workspace_id}/entwurf/verwerfen") def entwurf_verwerfen(workspace_id: str, user: dict = Depends(require_admin)): return _ergebnis( workspace_id, anode_request("DELETE", f"/api/v1/workspaces/{workspace_id}/entwurf"), "Entwurf verworfen." ) @router.post("/liste/{workspace_id}/veroeffentlichen") def veroeffentlichen(workspace_id: str, notiz: str = Form(default=""), user: dict = Depends(require_admin)): result = anode_request( "POST", f"/api/v1/workspaces/{workspace_id}/veroeffentlichen", json={"notiz": notiz.strip() or None} ) return _ergebnis(workspace_id, result, f"Version {result.get('nummer')} veröffentlicht.") @router.post("/liste/{workspace_id}/merkmale") async def workspace_merkmale_speichern(request: Request, workspace_id: str, user: dict = Depends(require_admin)): """Diff-basiertes Sammel-Speichern in den Entwurf (die Einzel-Endpunkte auf anode lehnen ab, wenn es keinen Entwurf gibt oder ein Merkmal auf der Distro fehlt - die erste Ablehnung wird angezeigt).""" form = await request.form() fehler = merkmale_diff_speichern(workspace_id, set(form.getlist("merkmal_ids"))) if fehler: return _ergebnis(workspace_id, fehler) return _zurueck(workspace_id, hinweis="Entwurf gespeichert.") def merkmale_diff_speichern(workspace_id: str, ausgewaehlt: set[str]) -> dict | None: """Gewählte Merkmale in den Entwurf übernehmen; liefert die erste Ablehnung von anode oder None.""" result = anode_request("GET", f"/api/v1/workspaces/{workspace_id}") workspace = result.get("workspace") if result.get("success") else None aktuell = {m["id"] for m in workspace["merkmale"]} if workspace else set() for merkmal_id in ausgewaehlt - aktuell: r = anode_request("POST", f"/api/v1/workspaces/{workspace_id}/merkmale/{merkmal_id}") if not r.get("success"): return r for merkmal_id in aktuell - ausgewaehlt: r = anode_request("DELETE", f"/api/v1/workspaces/{workspace_id}/merkmale/{merkmal_id}") if not r.get("success"): return r return None @router.post("/liste/{workspace_id}/webapps") def webapp_hinzufuegen( workspace_id: str, name: str = Form(...), url: str = Form(...), auf_desktop: bool = Form(default=False), user: dict = Depends(require_admin), ): return _ergebnis( workspace_id, anode_request( "POST", f"/api/v1/workspaces/{workspace_id}/webapps", json={"name": name, "url": url, "auf_desktop": auf_desktop}, ), ) @router.post("/liste/{workspace_id}/webapps/{webapp_id}/desktop") def webapp_desktop( workspace_id: str, webapp_id: str, auf_desktop: bool = Form(default=False), user: dict = Depends(require_admin), ): return _ergebnis( workspace_id, anode_request( "PUT", f"/api/v1/workspaces/{workspace_id}/webapps/{webapp_id}/desktop", json={"auf_desktop": auf_desktop}, ), ) @router.post("/liste/{workspace_id}/webapps/vorlage") def webapp_aus_vorlage(workspace_id: str, template_id: str = Form(...), user: dict = Depends(require_admin)): return _ergebnis( workspace_id, anode_request("POST", f"/api/v1/workspaces/{workspace_id}/webapp-templates/{template_id}/activate"), ) @router.post("/liste/{workspace_id}/webapps/{webapp_id}/entfernen") def webapp_entfernen(workspace_id: str, webapp_id: str, user: dict = Depends(require_admin)): return _ergebnis(workspace_id, anode_request("DELETE", f"/api/v1/workspaces/{workspace_id}/webapps/{webapp_id}")) @router.post("/liste/{workspace_id}/standardbrowser") def standardbrowser_setzen(workspace_id: str, browser: str = Form(default=""), user: dict = Depends(require_admin)): return _ergebnis( workspace_id, anode_request("PUT", f"/api/v1/workspaces/{workspace_id}/standardbrowser", json={"browser": browser or None}), "Standardbrowser gespeichert.", ) @router.post("/liste/{workspace_id}/einstellungen") def einstellungen( workspace_id: str, name: str = Form(...), description: str = Form(default=""), user: dict = Depends(require_admin) ): return _ergebnis( workspace_id, anode_request("PUT", f"/api/v1/workspaces/{workspace_id}", json={"name": name, "description": description or None}), "Gespeichert.", ) @router.post("/liste/{workspace_id}/vorlage") async def vorlage(request: Request, workspace_id: str, user: dict = Depends(require_admin)): form = await request.form() result = anode_request( "PUT", f"/api/v1/workspaces/{workspace_id}/vorlage", json={ "ist_vorlage": form.get("ist_vorlage") == "ja", "sichtbarkeit": form.get("sichtbarkeit") or "global", "kundengruppe_ids": form.getlist("kundengruppe_ids"), "organization_ids": form.getlist("organization_ids"), }, ) return _ergebnis(workspace_id, result, "Vorlagen-Einstellung gespeichert.") @router.post("/liste/{workspace_id}/kopie") def kopie_anlegen( workspace_id: str, name: str = Form(...), backend_id: str = Form(...), organization_id: str = Form(default=""), user: dict = Depends(require_admin), ): result = anode_request( "POST", f"/api/v1/workspaces/{workspace_id}/kopie", json={"name": name, "backend_id": backend_id, "organization_id": organization_id or None}, ) if not result.get("success"): return _ergebnis(workspace_id, result) neu = result["workspace"] hinweis = "Kopie angelegt." if neu.get("weggefallen"): hinweis += " Auf der neuen Distro weggefallen: " + ", ".join(neu["weggefallen"]) return _zurueck(neu["id"], hinweis=hinweis) @router.post("/liste/{workspace_id}/loeschen") def loeschen(workspace_id: str, bestaetigt: str = Form(default=""), user: dict = Depends(require_admin)): if bestaetigt != "ja": return _zurueck(workspace_id, fehler="Bitte das Löschen bestätigen.") result = anode_request("DELETE", f"/api/v1/workspaces/{workspace_id}") if not result.get("success"): return _ergebnis(workspace_id, result) return RedirectResponse("/admin/workspaces/liste?" + urlencode({"hinweis": "Workspace gelöscht."}), status_code=303)