""" % (html.escape(title), _CSS_TAG, extra_head, nav_html, who, body_html)
# ── Ersteinrichtung (neuer Mandant) ────────────────────────────────────────────
@app.route("/setup/new", methods=["GET", "POST"])
def setup_new():
# Die Ersteinrichtung ist ein einmaliger Vorgang: sobald irgendein
# Mandant existiert, ist die Registrierung neuer Organisationen ueber
# diese Route dauerhaft gesperrt (Nutzer-Vorgabe 2026-07-13: "/setup/new
# darf nur einmal funktionieren"). Weitere Mandanten koennen danach nur
# noch durch einen Administrator ueber die Datenbank/ein internes
# Verfahren angelegt werden, nicht mehr per offenem Self-Service-Formular.
if db.any_tenant_exists():
return redirect(url_for("login", already_setup="1"))
error = None
# Vorbelegung: bei GET die Standardwerte, bei einem fehlgeschlagenen POST
# die vom Nutzer eingegebenen Werte (ausser Passwoerter -- die bleiben aus
# Sicherheitsgruenden leer und muessen erneut eingegeben werden).
firma = ""
email = ""
dsb_name = ""
dsb_email = ""
retention_tickets = "1095"
retention_audit = "1825"
sla_antwort = "480"
sla_loesung = "2880"
if request.method == "POST":
firma = request.form.get("firma", "").strip()
email = request.form.get("email", "").strip().lower()
pw = request.form.get("password", "")
pw2 = request.form.get("password2", "")
dsb_name = request.form.get("dsb_name", "").strip()
dsb_email = request.form.get("dsb_email", "").strip()
retention_tickets = request.form.get("retention_tickets_days", "1095").strip()
retention_audit = request.form.get("retention_audit_days", "1825").strip()
sla_antwort = request.form.get("sla_antwort_minuten", "480").strip()
sla_loesung = request.form.get("sla_loesung_minuten", "2880").strip()
if not firma:
error = "Firmenname ist erforderlich."
elif not email or "@" not in email:
error = "Bitte eine gueltige E-Mail-Adresse angeben."
elif len(pw) < 10:
error = "Das Passwort muss mindestens 10 Zeichen lang sein."
elif pw != pw2:
error = "Die Passwoerter stimmen nicht ueberein."
elif db.email_exists(email):
error = "Diese E-Mail-Adresse ist bereits registriert."
else:
try:
tenant_id = db.create_tenant(
firma, dsb_name or None, dsb_email or None,
int(retention_tickets or 1095), int(retention_audit or 1825),
int(sla_antwort or 480), int(sla_loesung or 2880),
)
user_id = db.create_user(tenant_id, email, generate_password_hash(pw), role="admin")
db.create_service(
tenant_id, "AES -- Autonomous Engineering System",
"Automatisierte Software-Entwicklung: Projekte, Phasen, "
"LLM-gestuetzte Agenten, Auto-Fix und Release-Pipeline.",
"Entwicklung", True, AES_DASHBOARD_URL,
)
db.log_audit(tenant_id, user_id, "tenant_created", "tenant", str(tenant_id),
{"firma": firma}, _client_ip())
db.log_audit(tenant_id, user_id, "user_created", "user", str(user_id),
{"email": email, "role": "admin"}, _client_ip())
return redirect(url_for("login", registered="1"))
except Exception as e:
error = "Fehler bei der Einrichtung: %s" % html.escape(str(e))
err_html = "
%s
" % error if error else ""
return """
Neue Organisation einrichten -- ITSM%s
""" % (
_CSS_TAG, err_html,
html.escape(firma), html.escape(email),
html.escape(sla_antwort), html.escape(sla_loesung),
html.escape(dsb_name), html.escape(dsb_email),
html.escape(retention_tickets), html.escape(retention_audit),
)
# ── Login/Logout ────────────────────────────────────────────────────────────────
@app.route("/login", methods=["GET", "POST"])
def login():
error = None
notice = None
if request.args.get("registered"):
notice = "Organisation angelegt. Bitte melde dich mit deinem Admin-Konto an."
elif request.args.get("already_setup"):
notice = "Die Ersteinrichtung wurde bereits abgeschlossen. Bitte melde dich mit deinem bestehenden Konto an."
if request.method == "POST":
email = request.form.get("email", "").strip().lower()
pw = request.form.get("password", "")
user = db.get_user_by_email(email)
if user and check_password_hash(user["password_hash"], pw):
tenant = db.get_tenant(user["tenant_id"])
session["user_id"] = user["id"]
session["email"] = user["email"]
session["tenant_id"] = user["tenant_id"]
session["tenant_name"] = tenant["name"] if tenant else ""
session["role"] = user["role"]
db.touch_last_login(user["id"])
db.log_audit(user["tenant_id"], user["id"], "login_success", "user", str(user["id"]), None, _client_ip())
return redirect(request.args.get("next") or url_for("tickets_route"))
db.log_audit(user["tenant_id"] if user else None, user["id"] if user else None,
"login_failed", "user", email, None, _client_ip())
error = "E-Mail oder Passwort falsch."
err_html = "
%s
" % html.escape(error) if error else ""
notice_html = "
%s
" % html.escape(notice) if notice else ""
# Der Registrierungs-Link wird nur angezeigt, solange die Ersteinrichtung
# noch nicht abgeschlossen ist -- danach ist /setup/new gesperrt, ein
# Link dorthin waere ein toter Verweis (Nutzer-Vorgabe 2026-07-13).
setup_link_html = ('
') if not db.any_tenant_exists() else ""
return """
Anmelden -- ITSM%s
""" % (_CSS_TAG, err_html, notice_html, setup_link_html)
@app.route("/logout")
def logout():
u = _current_user()
if u:
db.log_audit(u["tenant_id"], u["id"], "logout", "user", str(u["id"]), None, _client_ip())
session.clear()
return redirect(url_for("login"))
# ── Tickets: Berechnung + Rendering ────────────────────────────────────────────
def _is_overdue(t):
if t["status"] in ("Geloest", "Geschlossen"):
return False
created = t["created_at"]
if created.tzinfo is None:
created = created.replace(tzinfo=_dt.timezone.utc)
age_min = (_dt.datetime.now(_dt.timezone.utc) - created).total_seconds() / 60
return age_min > (t.get("sla_loesung_minuten") or 999999)
def _kpis(tickets):
total = len(tickets)
offen = sum(1 for t in tickets if t["status"] == "Offen")
bearbeitung = sum(1 for t in tickets if t["status"] == "In Bearbeitung")
warten = sum(1 for t in tickets if t["status"] == "Warten")
ueberfaellig = sum(1 for t in tickets if _is_overdue(t))
now = _dt.datetime.now(_dt.timezone.utc)
geloest_7d = 0
for t in tickets:
if t["status"] in ("Geloest", "Geschlossen"):
upd = t["updated_at"]
if upd.tzinfo is None:
upd = upd.replace(tzinfo=_dt.timezone.utc)
if (now - upd).days <= 7:
geloest_7d += 1
return {"Gesamt": total, "Offen": offen, "In Bearbeitung": bearbeitung,
"Warten auf Input": warten, "Ueberfaellig": ueberfaellig, "Geloest (7 Tage)": geloest_7d}
def _status_class(s):
return {"Offen": "offen", "In Bearbeitung": "inbearbeitung", "Warten": "warten",
"Geloest": "geloest", "Geschlossen": "geschlossen"}.get(s, "offen")
def _prio_class(p):
return {"Kritisch": "krit", "Hoch": "hoch", "Mittel": "mittel", "Niedrig": "niedrig"}.get(p, "mittel")
def _ticket_row(t):
status = "ueberfaellig" if _is_overdue(t) else _status_class(t["status"])
status_label = "Ueberfaellig" if _is_overdue(t) else t["status"]
return """
' % (html.escape(label), value)
def _tickets_body(tenant_id, filter_status=None, filter_kategorie=None):
tickets = db.list_tickets(tenant_id)
kpis = _kpis(tickets)
kpi_html = "".join(_kpi_tile(k, v) for k, v in kpis.items())
shown = tickets
if filter_kategorie:
shown = [t for t in shown if t["kategorie"] == filter_kategorie]
tabs = [("", "Alle"), ("Offen", "Offen"), ("In Bearbeitung", "In Bearbeitung"),
("Warten", "Warten"), ("Geloest", "Geloest"), ("Ueberfaellig", "Ueberfaellig")]
base = "/probleme" if filter_kategorie == "Problem" else (
"/aenderungen" if filter_kategorie == "Change" else (
"/releases" if filter_kategorie == "Release" else "/tickets"))
tabs_html = "".join(
'%s' % (
base, ("?status=" + s) if s else "", "active" if (s or "") == (filter_status or "") else "", label)
for s, label in tabs)
if filter_status == "Ueberfaellig":
shown = [t for t in shown if _is_overdue(t)]
elif filter_status:
shown = [t for t in shown if t["status"] == filter_status]
rows = "".join(_ticket_row(t) for t in shown) or '
Keine Tickets.
'
services = db.list_services(tenant_id)
svc_opts = "".join('' % (s["id"], html.escape(s["name"])) for s in services)
new_form = """
Neues Ticket
""" % (
"".join('' % (c, c) for c in CATEGORIES),
"".join('' % (p, p) for p in PRIORITIES), svc_opts)
return """
Tickets
%s
%s
%s
Ticket-Nr
Titel (Service)
Status
Prioritaet
Kategorie
Zugewiesen an
Aktualisiert
Fortschritt
%s
""" % (kpi_html, new_form, tabs_html, rows)
@app.route("/tickets")
def tickets_route():
r = _require_login()
if r:
return r
return _shell("/tickets", "Tickets", _tickets_body(session["tenant_id"], request.args.get("status")))
@app.route("/probleme")
def probleme_route():
r = _require_login()
if r:
return r
return _shell("/probleme", "Probleme", _tickets_body(session["tenant_id"], request.args.get("status"), "Problem"))
@app.route("/aenderungen")
def aenderungen_route():
r = _require_login()
if r:
return r
return _shell("/aenderungen", "Aenderungen", _tickets_body(session["tenant_id"], request.args.get("status"), "Change"))
@app.route("/releases")
def releases_route():
r = _require_login()
if r:
return r
return _shell("/releases", "Releases", _tickets_body(session["tenant_id"], request.args.get("status"), "Release"))
@app.route("/tickets/new", methods=["POST"])
def tickets_new_route():
r = _require_login()
if r:
return r
tenant_id = session["tenant_id"]
service_id = request.form.get("service_id") or None
tid, ticket_nr = db.create_ticket(
tenant_id,
request.form.get("titel", "").strip() or "(ohne Titel)",
request.form.get("beschreibung", "").strip(),
int(service_id) if service_id else None,
request.form.get("prioritaet", "Mittel"),
request.form.get("kategorie", "Task"),
session.get("email"),
session["user_id"],
480, 2880,
session.get("email"),
)
db.log_audit(tenant_id, session["user_id"], "ticket_created", "ticket", ticket_nr, None, _client_ip())
return redirect(url_for("tickets_route"))
@app.route("/tickets//status", methods=["POST"])
def ticket_status_route(ticket_id):
r = _require_login()
if r:
return r
tenant_id = session["tenant_id"]
status = request.form.get("status", "Offen")
db.update_ticket_status(tenant_id, ticket_id, status, session.get("email"))
db.log_audit(tenant_id, session["user_id"], "ticket_status_changed", "ticket", str(ticket_id),
{"status": status}, _client_ip())
return jsonify({"ok": True})
@app.route("/api/tickets/")
def api_ticket(ticket_id):
r = _require_login()
if r:
return r
ticket = db.get_ticket(session["tenant_id"], ticket_id)
if not ticket:
abort(404)
return jsonify(_jsonable(ticket))
# ── Service-Katalog ─────────────────────────────────────────────────────────────
def _service_live_status(endpoint):
"""Prueft die Live-Erreichbarkeit eines gebuchten Service-Endpunkts (z.B. AES-Dashboard).
Generisch fuer alle Services mit Endpoint, nicht nur AES. Timeout kurz gehalten,
damit ein haengender/ausgefallener Service den Katalog nicht blockiert.
Rueckgabe: True (erreichbar) / False (nicht erreichbar) / None (kein Endpoint hinterlegt)."""
if not endpoint:
return None
try:
req = urllib.request.Request(endpoint, method="GET")
with urllib.request.urlopen(req, timeout=4) as resp:
return 200 <= resp.status < 400
except Exception:
return False
def _services_body(tenant_id):
services = db.list_services(tenant_id)
cards = []
for s in services:
if not s["gebucht"]:
badge = 'nicht gebucht'
elif s.get("endpoint"):
live = _service_live_status(s["endpoint"])
if live is True:
badge = 'gebucht · erreichbar'
elif live is False:
badge = 'gebucht · nicht erreichbar'
else:
badge = 'gebucht'
else:
badge = 'gebucht'
link = ('
' + "".join(cards) + add_form
@app.route("/services")
def services_route():
r = _require_login()
if r:
return r
return _shell("/services", "Service-Katalog", _services_body(session["tenant_id"]))
@app.route("/services/new", methods=["POST"])
def services_new_route():
r = _require_login()
if r:
return r
tenant_id = session["tenant_id"]
sid = db.create_service(
tenant_id, request.form.get("name", "").strip(), request.form.get("beschreibung", "").strip(),
"Sonstiges", True, request.form.get("endpoint", "").strip(),
)
db.log_audit(tenant_id, session["user_id"], "service_created", "service", str(sid), None, _client_ip())
return redirect(url_for("services_route"))
# ── Administration: Einstellungen + Audit-Log ──────────────────────────────────
@app.route("/admin", methods=["GET", "POST"])
def admin_route():
r = _require_admin()
if r:
return r
tenant_id = session["tenant_id"]
notice = ""
if request.method == "POST":
db.update_tenant_settings(
tenant_id,
request.form.get("dsb_name", "").strip() or None,
request.form.get("dsb_email", "").strip() or None,
int(request.form.get("retention_tickets_days") or 1095),
int(request.form.get("retention_audit_days") or 1825),
int(request.form.get("sla_antwort_minuten") or 480),
int(request.form.get("sla_loesung_minuten") or 2880),
)
db.log_audit(tenant_id, session["user_id"], "tenant_settings_updated", "tenant", str(tenant_id), None, _client_ip())
notice = "
Gespeichert.
"
if request.args.get("cleanup"):
notice += "
Bereinigung durchgefuehrt -- Details im Audit-Log.
"
t = db.get_tenant(tenant_id)
body = """
Einstellungen
%s
%s
Aufbewahrungsfrist-Bereinigung
Laeuft automatisch alle 24 Stunden im Hintergrund: geloeste/geschlossene Tickets und Audit-Log-Eintraege, die aelter als die oben hinterlegten Fristen sind, werden entfernt. Offene/laufende Tickets werden nie geloescht.
""" % (
html.escape(t["name"]), notice, t["sla_antwort_minuten"], t["sla_loesung_minuten"],
html.escape(t["dsb_name"] or ""), html.escape(t["dsb_email"] or ""),
t["retention_tickets_days"], t["retention_audit_days"])
return _shell("/admin", "Einstellungen", body)
@app.route("/admin/retention/run", methods=["POST"])
def admin_retention_run():
r = _require_admin()
if r:
return r
db.run_retention_cleanup()
return redirect(url_for("admin_route", cleanup="1"))
@app.route("/admin/audit")
def admin_audit_route():
r = _require_admin()
if r:
return r
entries = db.list_audit(session["tenant_id"], limit=300)
rows = "".join(
"
%s
%s
%s
%s %s
" % (
e["zeit"].strftime("%Y-%m-%d %H:%M:%S"), html.escape(e["user_email"] or "(unbekannt)"),
html.escape(e["aktion"]), html.escape(e["entity_typ"] or ""), html.escape(e["entity_id"] or ""))
for e in entries
) or "
Keine Eintraege.
"
body = """
Audit-Log
Alle sicherheitsrelevanten Aktionen dieses Mandanten (Anmeldungen, Aenderungen). Aufbewahrung gemaess Einstellungen.
Zeitpunkt (UTC)
Benutzer
Aktion
Objekt
%s
""" % rows
return _shell("/admin/audit", "Audit-Log", body)
# ── Administration: Benutzerverwaltung ─────────────────────────────────────────
def _role_pill(role):
return '%s' % (html.escape(role), html.escape(role))
def _users_body(tenant_id, notice="", form_vals=None):
# form_vals: bei einem fehlgeschlagenen POST die eingegebenen Werte (ausser
# Passwoerter), damit das Formular nicht komplett geleert wird (Nutzer-
# Vorgabe 2026-07-13, analog zum gleichen Fix bei /setup/new).
fv = form_vals or {}
users = db.list_users(tenant_id)
rows = []
for u in users:
status_pill = ('aktiv'
if u["active"] else
'gesperrt')
toggle_label = "Sperren" if u["active"] else "Entsperren"
role_opts = "".join(
'' % (r, "selected" if r == u["role"] else "", r)
for r in ("admin", "agent")
)
name = " ".join(x for x in (u["vorname"], u["nachname"]) if x) or "--"
rows.append("""
")
open_attr = " open" if (notice or fv) else ""
add_form = """+ Neuen Benutzer
Neuen Benutzer anlegen
Manuelle Verwaltung. Eine AD/LDAP-Anbindung ist als naechste Ausbaustufe vorgesehen (Spalte "Quelle" zeigt dann "ldap" statt "local").
""" % (
open_attr,
html.escape(fv.get("email", "")),
"selected" if fv.get("role") == "agent" or not fv.get("role") else "",
"selected" if fv.get("role") == "admin" else "",
html.escape(fv.get("vorname", "")), html.escape(fv.get("nachname", "")),
html.escape(fv.get("telefon", "")), html.escape(fv.get("abteilung", "")),
html.escape(fv.get("adresse", "")),
)
return '
Benutzerverwaltung
' + notice + table + add_form
@app.route("/admin/users", methods=["GET", "POST"])
def admin_users_route():
r = _require_admin()
if r:
return r
tenant_id = session["tenant_id"]
notice = ""
form_vals = {}
if request.method == "POST":
email = request.form.get("email", "").strip().lower()
role = request.form.get("role", "agent").strip()
if role not in ("admin", "agent"):
role = "agent"
pw = request.form.get("password", "")
pw2 = request.form.get("password2", "")
vorname = request.form.get("vorname", "").strip()
nachname = request.form.get("nachname", "").strip()
telefon = request.form.get("telefon", "").strip()
abteilung = request.form.get("abteilung", "").strip()
adresse = request.form.get("adresse", "").strip()
form_vals = {"email": email, "role": role, "vorname": vorname, "nachname": nachname,
"telefon": telefon, "abteilung": abteilung, "adresse": adresse}
if not email or "@" not in email:
notice = "
Bitte eine gueltige E-Mail-Adresse angeben.
"
elif len(pw) < 10:
notice = "
Das Passwort muss mindestens 10 Zeichen lang sein.
"
elif pw != pw2:
notice = "
Die Passwoerter stimmen nicht ueberein.
"
elif db.email_exists(email):
notice = "
Diese E-Mail-Adresse ist bereits registriert.
"
else:
uid = db.create_user(tenant_id, email, generate_password_hash(pw), role=role,
vorname=vorname or None, nachname=nachname or None,
telefon=telefon or None, abteilung=abteilung or None,
adresse=adresse or None)
db.log_audit(tenant_id, session["user_id"], "user_created", "user", str(uid),
{"email": email, "role": role, "auth_source": "local"}, _client_ip())
notice = "
Benutzer angelegt.
"
form_vals = {}
return _shell("/admin/users", "Benutzerverwaltung", _users_body(tenant_id, notice, form_vals))
@app.route("/admin/users//edit", methods=["GET", "POST"])
def admin_users_edit_route(user_id):
r = _require_admin()
if r:
return r
tenant_id = session["tenant_id"]
u = db.get_user(tenant_id, user_id)
if not u:
abort(404)
error = None
ok_msg = None
if request.method == "POST":
email = request.form.get("email", "").strip().lower()
vorname = request.form.get("vorname", "").strip()
nachname = request.form.get("nachname", "").strip()
telefon = request.form.get("telefon", "").strip()
abteilung = request.form.get("abteilung", "").strip()
adresse = request.form.get("adresse", "").strip()
if not email or "@" not in email:
error = "Bitte eine gueltige E-Mail-Adresse angeben."
elif db.email_exists_excluding(email, user_id):
error = "Diese E-Mail-Adresse wird bereits von einem anderen Benutzer verwendet."
else:
old_email = u["email"]
if email != old_email:
db.update_user_email(tenant_id, user_id, email)
db.log_audit(tenant_id, session["user_id"], "user_email_changed", "user", str(user_id),
{"alt": old_email, "neu": email}, _client_ip())
if user_id == session["user_id"]:
session["email"] = email
db.update_user_profile(tenant_id, user_id, vorname or None, nachname or None,
telefon or None, abteilung or None, adresse or None)
db.log_audit(tenant_id, session["user_id"], "user_profile_updated", "user", str(user_id),
None, _client_ip())
u = dict(u)
u.update({"email": email, "vorname": vorname, "nachname": nachname, "telefon": telefon,
"abteilung": abteilung, "adresse": adresse})
ok_msg = "Gespeichert."
err_html = "
%s
" % html.escape(error) if error else ""
ok_html = ("
%s
" % html.escape(ok_msg)) if ok_msg else ""
body = """
Benutzerdetails
%s%s
Zurueck zur Benutzerverwaltung""" % (
err_html, ok_html, html.escape(u["email"]),
html.escape(u["vorname"] or ""), html.escape(u["nachname"] or ""),
html.escape(u["telefon"] or ""), html.escape(u["abteilung"] or ""),
html.escape(u["adresse"] or ""),
)
return _shell("/admin/users", "Benutzerdetails", body)
@app.route("/admin/users//role", methods=["POST"])
def admin_users_role_route(user_id):
r = _require_admin()
if r:
return r
tenant_id = session["tenant_id"]
role = request.form.get("role", "agent").strip()
if role not in ("admin", "agent"):
return redirect(url_for("admin_users_route"))
db.set_user_role(tenant_id, user_id, role)
db.log_audit(tenant_id, session["user_id"], "user_role_changed", "user", str(user_id),
{"role": role}, _client_ip())
return redirect(url_for("admin_users_route"))
@app.route("/admin/users//active", methods=["POST"])
def admin_users_active_route(user_id):
r = _require_admin()
if r:
return r
tenant_id = session["tenant_id"]
current = db.get_user(tenant_id, user_id)
if not current:
return redirect(url_for("admin_users_route"))
new_active = not current["active"]
if user_id == session["user_id"] and not new_active:
return redirect(url_for("admin_users_route")) # eigenes Konto nicht selbst sperren
db.set_user_active(tenant_id, user_id, new_active)
db.log_audit(tenant_id, session["user_id"],
"user_activated" if new_active else "user_deactivated",
"user", str(user_id), None, _client_ip())
return redirect(url_for("admin_users_route"))
# ── Wissensdatenbank ────────────────────────────────────────────────────────────
def _kb_list_body(tenant_id):
kategorie = request.args.get("kategorie") or None
q = request.args.get("q") or None
articles = db.list_kb_articles(tenant_id, kategorie=kategorie, q=q)
tabs = "".join(
'%s' % (
html.escape(k), "active" if kategorie == k else "", html.escape(k))
for k in KB_KATEGORIEN
)
tabs = 'Alle' % ("active" if not kategorie else "") + tabs
rows = []
for a in articles:
tag_pills = "".join('%s' % html.escape(t.strip())
for t in (a["tags"] or "").split(",") if t.strip())
rows.append("""
' % (tabs, search_form, table))
@app.route("/wissen")
def wissen_route():
r = _require_login()
if r:
return r
return _shell("/wissen", "Wissensdatenbank", _kb_list_body(session["tenant_id"]))
@app.route("/wissen/new", methods=["GET", "POST"])
def wissen_new_route():
r = _require_login()
if r:
return r
tenant_id = session["tenant_id"]
error = None
titel, kategorie, inhalt, tags = "", "Allgemein", "", ""
if request.method == "POST":
titel = request.form.get("titel", "").strip()
kategorie = request.form.get("kategorie", "Allgemein").strip()
inhalt = request.form.get("inhalt", "").strip()
tags = request.form.get("tags", "").strip()
if not titel:
error = "Titel ist erforderlich."
elif not inhalt:
error = "Inhalt darf nicht leer sein."
else:
aid = db.create_kb_article(tenant_id, titel, kategorie, inhalt, tags, session["user_id"])
db.log_audit(tenant_id, session["user_id"], "kb_article_created", "kb_article", str(aid),
{"titel": titel}, _client_ip())
return redirect(url_for("wissen_detail_route", article_id=aid))
kat_opts = "".join('' % (k, "selected" if k == kategorie else "", k)
for k in KB_KATEGORIEN)
err_html = "
%s
" % html.escape(error) if error else ""
body = """
Neuer Wissensartikel
%s
""" % (err_html, html.escape(titel), kat_opts, html.escape(inhalt), html.escape(tags))
return _shell("/wissen", "Neuer Wissensartikel", body)
@app.route("/wissen/")
def wissen_detail_route(article_id):
r = _require_login()
if r:
return r
a = db.get_kb_article(session["tenant_id"], article_id)
if not a:
abort(404)
tag_pills = "".join('%s' % html.escape(t.strip())
for t in (a["tags"] or "").split(",") if t.strip())
body = """
%s
%s · %s · aktualisiert %s
%s
%s
BearbeitenZurueck zur Uebersicht""" % (
html.escape(a["titel"]), html.escape(a["kategorie"]), html.escape(a["autor_email"] or ""),
a["updated_at"].strftime("%Y-%m-%d %H:%M"), html.escape(a["inhalt"]), tag_pills, a["id"])
return _shell("/wissen", a["titel"], body)
@app.route("/wissen//edit", methods=["GET", "POST"])
def wissen_edit_route(article_id):
r = _require_login()
if r:
return r
tenant_id = session["tenant_id"]
a = db.get_kb_article(tenant_id, article_id)
if not a:
abort(404)
error = None
if request.method == "POST":
titel = request.form.get("titel", "").strip()
kategorie = request.form.get("kategorie", "Allgemein").strip()
inhalt = request.form.get("inhalt", "").strip()
tags = request.form.get("tags", "").strip()
if not titel or not inhalt:
error = "Titel und Inhalt duerfen nicht leer sein."
else:
db.update_kb_article(tenant_id, article_id, titel, kategorie, inhalt, tags)
db.log_audit(tenant_id, session["user_id"], "kb_article_updated", "kb_article", str(article_id),
{"titel": titel}, _client_ip())
return redirect(url_for("wissen_detail_route", article_id=article_id))
a = dict(a)
a.update({"titel": titel, "kategorie": kategorie, "inhalt": inhalt, "tags": tags})
kat_opts = "".join('' % (k, "selected" if k == a["kategorie"] else "", k)
for k in KB_KATEGORIEN)
err_html = "
%s
" % html.escape(error) if error else ""
body = """
Artikel bearbeiten
%s
""" % (err_html, html.escape(a["titel"]), kat_opts, html.escape(a["inhalt"]), html.escape(a["tags"] or ""), article_id)
return _shell("/wissen", "Artikel bearbeiten", body)
@app.route("/wissen//delete", methods=["POST"])
def wissen_delete_route(article_id):
r = _require_login()
if r:
return r
tenant_id = session["tenant_id"]
db.delete_kb_article(tenant_id, article_id)
db.log_audit(tenant_id, session["user_id"], "kb_article_deleted", "kb_article", str(article_id), None, _client_ip())
return redirect(url_for("wissen_route"))
# ── CMDB (Configuration Management Database) ───────────────────────────────────
def _ci_status_pill(status):
cls = {"Aktiv": "geloest", "Inaktiv": "geschlossen", "Wartung": "bearbeitung", "Ausgemustert": "ueberfaellig"}.get(status, "offen")
return '%s' % (cls, html.escape(status))
def _cmdb_list_body(tenant_id):
ci_typ = request.args.get("typ") or None
cis = db.list_cis(tenant_id, ci_typ=ci_typ)
tabs = 'Alle' % ("active" if not ci_typ else "")
tabs += "".join('%s' % (t, "active" if ci_typ == t else "", t) for t in CI_TYPES)
rows = []
for c in cis:
rows.append("""
' % (tabs, table))
@app.route("/assets")
def assets_route():
r = _require_login()
if r:
return r
return _shell("/assets", "CMDB", _cmdb_list_body(session["tenant_id"]))
@app.route("/assets/new", methods=["GET", "POST"])
def assets_new_route():
r = _require_login()
if r:
return r
tenant_id = session["tenant_id"]
error = None
name, ci_typ, status, beschreibung, attr_text = "", "Server", "Aktiv", "", ""
if request.method == "POST":
name = request.form.get("name", "").strip()
ci_typ = request.form.get("ci_typ", "Sonstiges").strip()
status = request.form.get("status", "Aktiv").strip()
beschreibung = request.form.get("beschreibung", "").strip()
attr_text = request.form.get("attribute", "")
if not name:
error = "Name ist erforderlich."
else:
cid = db.create_ci(tenant_id, name, ci_typ, status, beschreibung, _text_to_attrs(attr_text))
db.log_audit(tenant_id, session["user_id"], "ci_created", "ci", str(cid),
{"name": name, "ci_typ": ci_typ}, _client_ip())
return redirect(url_for("assets_detail_route", ci_id=cid))
typ_opts = "".join('' % (t, "selected" if t == ci_typ else "", t) for t in CI_TYPES)
status_opts = "".join('' % (s, "selected" if s == status else "", s) for s in CI_STATUS)
err_html = "
%s
" % html.escape(error) if error else ""
body = """
Neues Configuration Item
%s
""" % (err_html, html.escape(name), typ_opts, status_opts, html.escape(beschreibung), html.escape(attr_text))
return _shell("/assets", "Neues Configuration Item", body)
@app.route("/assets/", methods=["GET", "POST"])
def assets_detail_route(ci_id):
r = _require_login()
if r:
return r
tenant_id = session["tenant_id"]
error = None
if request.method == "POST":
name = request.form.get("name", "").strip()
ci_typ = request.form.get("ci_typ", "Sonstiges").strip()
status = request.form.get("status", "Aktiv").strip()
beschreibung = request.form.get("beschreibung", "").strip()
attr_text = request.form.get("attribute", "")
if not name:
error = "Name ist erforderlich."
else:
db.update_ci(tenant_id, ci_id, name, ci_typ, status, beschreibung, _text_to_attrs(attr_text))
db.log_audit(tenant_id, session["user_id"], "ci_updated", "ci", str(ci_id), {"name": name}, _client_ip())
c = db.get_ci(tenant_id, ci_id)
if not c:
abort(404)
typ_opts = "".join('' % (t, "selected" if t == c["ci_typ"] else "", t) for t in CI_TYPES)
status_opts = "".join('' % (s, "selected" if s == c["status"] else "", s) for s in CI_STATUS)
err_html = "
%s
" % html.escape(error) if error else ""
other_cis = [x for x in db.list_cis(tenant_id) if x["id"] != ci_id]
ci_opts = "".join('' % (x["id"], html.escape(x["name"]), x["ci_typ"]) for x in other_cis)
rel_typ_opts = "".join('' % (t, t) for t in REL_TYPES)
rel_out_rows = "".join(
"
%s %s
" % (
html.escape(r["beziehungs_typ"]), html.escape(r["to_name"]), ci_id, r["id"])
for r in c["rel_out"]
) or "
Keine ausgehenden Beziehungen.
"
rel_in_rows = "".join(
"
%s %s(eingehend)
" % (
html.escape(r["beziehungs_typ"]), html.escape(r["from_name"]))
for r in c["rel_in"]
) or "
Keine eingehenden Beziehungen.
"
body = """
%s
%s
Beziehungen
Ausgehend
%s
Eingehend
%s
Neue Beziehung
""" % (
html.escape(c["name"]), err_html, html.escape(c["name"]), typ_opts, status_opts,
html.escape(c["beschreibung"] or ""), _attrs_to_text(c["attribute"]),
rel_out_rows, rel_in_rows, ci_id, rel_typ_opts, ci_opts,
)
return _shell("/assets", c["name"], body)
@app.route("/assets//relationships", methods=["POST"])
def assets_relationship_add_route(ci_id):
r = _require_login()
if r:
return r
tenant_id = session["tenant_id"]
to_ci_id = request.form.get("to_ci_id", type=int)
beziehungs_typ = request.form.get("beziehungs_typ", "haengt ab von").strip()
if to_ci_id and to_ci_id != ci_id:
rid = db.create_ci_relationship(tenant_id, ci_id, to_ci_id, beziehungs_typ)
db.log_audit(tenant_id, session["user_id"], "ci_relationship_created", "ci_relationship", str(rid),
{"from": ci_id, "to": to_ci_id, "typ": beziehungs_typ}, _client_ip())
return redirect(url_for("assets_detail_route", ci_id=ci_id))
@app.route("/assets//relationships//delete", methods=["POST"])
def assets_relationship_delete_route(ci_id, rel_id):
r = _require_login()
if r:
return r
tenant_id = session["tenant_id"]
db.delete_ci_relationship(tenant_id, rel_id)
db.log_audit(tenant_id, session["user_id"], "ci_relationship_deleted", "ci_relationship", str(rel_id), None, _client_ip())
return redirect(url_for("assets_detail_route", ci_id=ci_id))
@app.route("/")
def root_route():
if not db.any_tenant_exists():
return redirect(url_for("setup_new"))
if session.get("user_id"):
return redirect(url_for("tickets_route"))
return redirect(url_for("login"))
@app.route("/health")
def health_route():
try:
with db.get_conn() as conn:
with conn.cursor() as cur:
cur.execute("SELECT 1")
return jsonify({"status": "ok", "db": "ok"})
except Exception as e:
return jsonify({"status": "error", "db": str(e)}), 500
db.init_db()
def _retention_loop():
"""Hintergrund-Thread: bereinigt abgelaufene Daten periodisch (siehe
db.run_retention_cleanup). Laeuft einmal kurz nach dem Start und danach im
festen Intervall. Der Postgres-Advisory-Lock in run_retention_cleanup()
schuetzt davor, dass mehrere Worker/Prozesse gleichzeitig loeschen."""
time.sleep(30)
while True:
try:
db.run_retention_cleanup()
except Exception as e:
print("[retention] Fehler bei der Bereinigung:", e, flush=True)
time.sleep(RETENTION_INTERVAL_SECONDS)
threading.Thread(target=_retention_loop, daemon=True).start()
if __name__ == "__main__":
app.run(host="0.0.0.0", port=8090)