import os import sqlite3 import io import zipfile from datetime import datetime, date from flask import Flask, render_template, request, redirect, url_for, g, flash, send_file from werkzeug.utils import secure_filename import updater BASE_DIR = os.path.dirname(os.path.abspath(__file__)) DB_PATH = os.path.join(BASE_DIR, "data", "bibliothek.db") BILDER_DIR = os.path.join(BASE_DIR, "data", "doku_bilder") ERLAUBTE_BILD_ENDUNGEN = {".png", ".jpg", ".jpeg", ".gif", ".webp"} app = Flask(__name__) app.secret_key = "server-bibliothek-lokal" # nur lokal genutzt, kein Login-System app.config["MAX_CONTENT_LENGTH"] = 15 * 1024 * 1024 # 15 MB pro Anfrage (Bild-Upload) # Wird bei jeder Aktualisierung von Claude hochgezaehlt/geaendert - damit # jederzeit im Fuss der Seitenleiste sichtbar ist, ob wirklich die neueste # Version laeuft (nicht nur auf der Festplatte liegt, sondern auch vom # gestarteten Programm geladen wurde). APP_VERSION = "2026-07-23 · r17 (Zugriffsbeschränkung lokal/Netzwerk + Zugriffsprotokoll)" @app.context_processor def _version_injizieren(): return {"app_version": APP_VERSION, "ist_lokal": ist_lokaler_zugriff()} # Nach wie vielen Tagen ohne Pruefung ein Tool als "bitte pruefen" markiert wird TAGE_BIS_WARNUNG = 30 # --------------------------------------------------------------------------- # Zugriffsbeschraenkung: voller Zugriff nur direkt am Server (z. B. per RDP, # im Browser "localhost" statt der Netzwerk-IP aufgerufen). Von jedem anderen # Rechner im Netzwerk aus ist NUR die eingeschraenkte Software-Anforderung # erreichbar - alles andere (Katalog verwalten, Protokolle einsehen usw.) # bleibt dem Server selbst vorbehalten. # --------------------------------------------------------------------------- OEFFENTLICHE_ENDPUNKTE = {"einrichten", "static"} def ist_lokaler_zugriff(): return request.remote_addr in ("127.0.0.1", "::1") @app.before_request def _zugriff_beschraenken(): if ist_lokaler_zugriff(): return # voller Zugriff, direkt am Server if request.endpoint in OEFFENTLICHE_ENDPUNKTE: return # eingeschraenkter Bereich bleibt aus dem Netzwerk erreichbar return render_template("zugriff_eingeschraenkt.html"), 403 def protokolliere_zugriff(aktion, angegebener_name=""): """Schreibt einen Eintrag ins Zugriffsprotokoll - IP-Adresse, per Reverse-DNS ermittelter Rechnername (falls aufloesbar) und der selbst angegebene Name. Hinweis: ein echter Windows-Benutzername laesst sich ueber reines HTTP ohne Domänen-/NTLM-Integration nicht zuverlaessig ermitteln, siehe ARCHITEKTUR.md.""" import socket ip = request.remote_addr try: rechnername = socket.gethostbyaddr(ip)[0] except Exception: rechnername = None db = get_db() db.execute( """INSERT INTO zugriffslog (zeitpunkt, ip_adresse, rechnername, angegebener_name, aktion, lokal) VALUES (?, ?, ?, ?, ?, ?)""", ( datetime.now().isoformat(timespec="seconds"), ip, rechnername, angegebener_name, aktion, 1 if ist_lokaler_zugriff() else 0, ), ) db.commit() # --------------------------------------------------------------------------- # Datenbank # --------------------------------------------------------------------------- def get_db(): if "db" not in g: os.makedirs(os.path.dirname(DB_PATH), exist_ok=True) g.db = sqlite3.connect(DB_PATH) g.db.row_factory = sqlite3.Row g.db.execute("PRAGMA foreign_keys = ON") return g.db @app.teardown_appcontext def close_db(exception=None): db = g.pop("db", None) if db is not None: db.close() def init_db(): os.makedirs(os.path.dirname(DB_PATH), exist_ok=True) conn = sqlite3.connect(DB_PATH) conn.executescript( """ CREATE TABLE IF NOT EXISTS software ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, kategorie TEXT, version TEXT, installer_pfad TEXT, install_parameter TEXT, dokumentation TEXT, letzte_pruefung TEXT, erstellt_am TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS zuweisungen ( id INTEGER PRIMARY KEY AUTOINCREMENT, mitarbeiter TEXT NOT NULL, rechnername TEXT, software_id INTEGER NOT NULL, installierte_version TEXT, installiert_von TEXT, installiert_am TEXT NOT NULL, notiz TEXT, FOREIGN KEY (software_id) REFERENCES software (id) ON DELETE CASCADE ); CREATE TABLE IF NOT EXISTS update_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, software_id INTEGER NOT NULL, zeitpunkt TEXT NOT NULL, alte_version TEXT, neue_version TEXT, erfolg INTEGER NOT NULL, meldung TEXT, FOREIGN KEY (software_id) REFERENCES software (id) ON DELETE CASCADE ); CREATE TABLE IF NOT EXISTS protokolle ( id INTEGER PRIMARY KEY AUTOINCREMENT, mitarbeiter TEXT NOT NULL, rechnername TEXT, datum TEXT NOT NULL, erstellt_von TEXT, notiz TEXT, erstellt_zeitpunkt TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS doku_bilder ( id INTEGER PRIMARY KEY AUTOINCREMENT, software_id INTEGER NOT NULL, dateiname TEXT NOT NULL, hochgeladen_am TEXT NOT NULL, FOREIGN KEY (software_id) REFERENCES software (id) ON DELETE CASCADE ); CREATE TABLE IF NOT EXISTS zugriffslog ( id INTEGER PRIMARY KEY AUTOINCREMENT, zeitpunkt TEXT NOT NULL, ip_adresse TEXT, rechnername TEXT, angegebener_name TEXT, aktion TEXT, lokal INTEGER NOT NULL ); """ ) # Migration: Spalten ergaenzen, falls die Datenbank aus einer aelteren # Version stammt (bestehende Daten bleiben dabei unangetastet). vorhandene_spalten = {row[1] for row in conn.execute("PRAGMA table_info(software)")} neue_spalten = { "auto_update": "INTEGER NOT NULL DEFAULT 0", "update_quelle": "TEXT", "update_wert": "TEXT", } for spalte, definition in neue_spalten.items(): if spalte not in vorhandene_spalten: conn.execute(f"ALTER TABLE software ADD COLUMN {spalte} {definition}") zuweisungen_spalten = {row[1] for row in conn.execute("PRAGMA table_info(zuweisungen)")} if "protokoll_id" not in zuweisungen_spalten: conn.execute("ALTER TABLE zuweisungen ADD COLUMN protokoll_id INTEGER") # Rueckwirkend: Eintraege aus der Zeit vor den Protokollen (protokoll_id # noch leer) zu je einem Protokoll pro Mitarbeiter/Rechner/Datum buendeln, # damit auch alte Testdaten in der neuen Ansicht auftauchen. verwaiste = conn.execute( "SELECT DISTINCT mitarbeiter, rechnername, installiert_am FROM zuweisungen WHERE protokoll_id IS NULL" ).fetchall() for mitarbeiter, rechnername, installiert_am in verwaiste: cur = conn.execute( "INSERT INTO protokolle (mitarbeiter, rechnername, datum, erstellt_zeitpunkt) VALUES (?, ?, ?, ?)", (mitarbeiter, rechnername, installiert_am, datetime.now().isoformat(timespec="seconds")), ) conn.execute( """UPDATE zuweisungen SET protokoll_id = ? WHERE protokoll_id IS NULL AND mitarbeiter = ? AND rechnername IS ? AND installiert_am = ?""", (cur.lastrowid, mitarbeiter, rechnername, installiert_am), ) conn.commit() conn.close() # --------------------------------------------------------------------------- # Hilfsfunktionen # --------------------------------------------------------------------------- def status_fuer(letzte_pruefung): """Ermittelt Status-Badge (text, css-klasse) anhand des letzten Pruefdatums.""" if not letzte_pruefung: return ("Nie geprueft", "status-unbekannt") try: d = datetime.strptime(letzte_pruefung, "%Y-%m-%d").date() except (ValueError, TypeError): return ("Nie geprueft", "status-unbekannt") tage = (date.today() - d).days if tage > TAGE_BIS_WARNUNG: return (f"Bitte pruefen ({tage} Tage)", "status-warnung") return (f"Aktuell (geprueft vor {tage} Tg.)", "status-ok") def heute_str(): return date.today().isoformat() def _ps_escape(text): """Escaped einen Text fuer die Verwendung in einem einfach gequoteten PowerShell-String (einzelne Hochkommas werden verdoppelt).""" return (text or "").replace("'", "''") # Fester Ordner auf dem Zielrechner: Downloads, entpackte portable Tools und # das lokale Mini-Protokoll landen IMMER hier - ein einziger, vorhersehbarer # Pfad, den man z. B. in Windows Defender / der Antivirensoftware als # Ausnahme eintragen kann, statt viele verstreute Einzelpfade. CLIENT_ORDNER = r"C:\ClientBibliothek" _BAT_LAUNCHER = ( "@echo off\r\n" "set ZIEL=C:\\ClientBibliothek\r\n" 'if not exist "%ZIEL%" mkdir "%ZIEL%"\r\n' 'xcopy "%~dp0*" "%ZIEL%\\" /E /I /Y >nul\r\n' "echo ============================================================\r\n" "echo Rechner-Einrichtung wird gestartet...\r\n" "echo Admin-Rechte werden bei Bedarf per Windows-Abfrage angefragt.\r\n" "echo Alle Dateien landen unter %ZIEL%\r\n" "echo ============================================================\r\n" "echo.\r\n" 'powershell -NoProfile -ExecutionPolicy Bypass -File "%ZIEL%\\einrichten.ps1"\r\n' "pause\r\n" ) def _erstelle_einrichtungs_skript(server_url, mitarbeiter, tools): """Baut den Inhalt der einrichten.ps1 zusammen: installiert jedes ausgewaehlte Tool aus der MITGELIEFERTEN Datei (liegt im 'tools'-Unterordner desselben Pakets, kein Netzwerkzugriff auf den Server noetig) und meldet das Ergebnis am Ende BEST-EFFORT als Protokoll an die Server-Bibliothek zurueck (falls Netzwerkzugriff vorhanden ist - schlaegt das fehl, bleibt trotzdem das lokale Protokoll erhalten).""" zeilen = [ '$ErrorActionPreference = "Stop"', f"$server = '{_ps_escape(server_url)}'", f"$mitarbeiter = '{_ps_escape(mitarbeiter)}'", f"$clientOrdner = '{_ps_escape(CLIENT_ORDNER)}'", "", "if (-not ([Security.Principal.WindowsPrincipal][Security.Principal.WindowsIdentity]::GetCurrent()).IsInRole([Security.Principal.WindowsBuiltInRole]::Administrator)) {", ' Start-Process powershell -Verb RunAs -ArgumentList "-NoProfile -ExecutionPolicy Bypass -File `"$PSCommandPath`""', " exit", "}", "", "$toolsOrdner = Join-Path $clientOrdner \"tools\"", "$paketeOrdner = Join-Path $clientOrdner \"pakete\"", "$protokollDatei = Join-Path $clientOrdner \"protokoll.log\"", "New-Item -ItemType Directory -Force -Path $paketeOrdner | Out-Null", "", "try {", " Add-MpPreference -ExclusionPath $clientOrdner -ErrorAction Stop", ' Write-Host "Defender-Ausnahme fuer $clientOrdner eingetragen." -ForegroundColor Green', "} catch {", ' Write-Host "Hinweis: Defender-Ausnahme konnte nicht automatisch eingetragen werden (evtl. andere Antivirensoftware oder per Richtlinie gesperrt)." -ForegroundColor Yellow', "}", "", "function Protokolliere($zeile) {", ' $eintrag = (Get-Date -Format "yyyy-MM-dd HH:mm") + " - $mitarbeiter - " + $zeile', " Add-Content -Path $protokollDatei -Value $eintrag", "}", "", 'Write-Host "============================================================"', 'Write-Host " Server-Bibliothek - Rechner-Einrichtung fuer $mitarbeiter"', f'Write-Host " Alle Dateien unter {_ps_escape(CLIENT_ORDNER)}"', 'Write-Host "============================================================"', 'Write-Host ""', "", "$erfolgreicheIds = @()", "", "function InstalliereTool($id, $name, $endung, $parameter) {", ' Write-Host "Installiere: $name ..." -ForegroundColor Cyan', ' $quelle = Join-Path $toolsOrdner ($id.ToString() + $endung)', " if (-not (Test-Path $quelle)) {", ' Write-Host (" FEHLER: Datei fehlt im Paket unter " + $quelle) -ForegroundColor Red', ' Protokolliere ($name + ": FEHLER, Datei fehlt im Paket")', " return $false", " }", " try {", ' if ($endung -eq ".msi") {', " Start-Process \"msiexec.exe\" -ArgumentList ('/i \"' + $quelle + '\" ' + $parameter) -Wait", " } else {", " Start-Process $quelle -ArgumentList $parameter -Wait", " }", " } catch {", ' Write-Host (" FEHLER bei der Installation von " + $name + ": " + $_) -ForegroundColor Red', ' Protokolliere ($name + ": FEHLER bei der Installation")', " return $false", " }", ' Protokolliere ($name + ": installiert")', ' Write-Host " fertig." -ForegroundColor Green', " return $true", "}", "", "function BereitstellenPortablesTool($id, $name) {", ' Write-Host "Portables Tool: $name ..." -ForegroundColor Cyan', ' $quelle = Join-Path $toolsOrdner ($id.ToString() + ".zip")', " if (-not (Test-Path $quelle)) {", ' Write-Host (" FEHLER: ZIP fehlt im Paket unter " + $quelle) -ForegroundColor Red', ' Protokolliere ($name + ": FEHLER, ZIP fehlt im Paket")', " return $false", " }", ' $zielZip = Join-Path $paketeOrdner ($name + ".zip")', " Copy-Item -Path $quelle -Destination $zielZip -Force", ' Protokolliere ($name + ": als ZIP bereitgestellt unter " + $zielZip)', ' Write-Host (" fertig, liegt als ZIP unter " + $zielZip) -ForegroundColor Green', ' Write-Host " (bewusst NICHT automatisch entpackt - bitte manuell an den Zielort entpacken)" -ForegroundColor DarkGray', " return $true", "}", "", ] for t in tools: if t["typ"] == "ordner": zeilen.append( f"if (BereitstellenPortablesTool {t['id']} '{_ps_escape(t['name'])}') " f"{{ $erfolgreicheIds += {t['id']} }}" ) else: zeilen.append( f"if (InstalliereTool {t['id']} '{_ps_escape(t['name'])}' '{t['endung']}' " f"'{_ps_escape(t['parameter'])}') {{ $erfolgreicheIds += {t['id']} }}" ) zeilen += [ "", 'Write-Host ""', 'Write-Host "Protokoll wird an den Server gemeldet (optional, braucht Netzwerkzugriff)..."', '$koerper = "mitarbeiter=" + [uri]::EscapeDataString($mitarbeiter)', '$koerper += "&rechnername=" + [uri]::EscapeDataString($env:COMPUTERNAME)', '$koerper += "&installiert_von=" + [uri]::EscapeDataString("Automatisches Einrichtungs-Skript")', '$koerper += "&installiert_am=" + (Get-Date -Format "yyyy-MM-dd")', "foreach ($id in $erfolgreicheIds) { $koerper += \"&software_ids=$id\" }", "try {", ' Invoke-RestMethod -Uri "$server/zuweisungen/neu" -Method POST -Body $koerper -ContentType "application/x-www-form-urlencoded" | Out-Null', ' Write-Host "Zentrales Protokoll auf dem Server aktualisiert." -ForegroundColor Green', "} catch {", ' Write-Host "Kein Netzwerkzugriff auf den Server - zentrales Protokoll nicht aktualisiert (lokales Protokoll unter protokoll.log ist trotzdem vollstaendig)." -ForegroundColor Yellow', "}", "", 'Write-Host ""', 'Write-Host "============================================================"', f'Write-Host (" Fertig! " + $erfolgreicheIds.Count + " von {len(tools)} Tools installiert.")', 'Write-Host "============================================================"', 'Read-Host "Enter druecken zum Beenden"', ] return "\r\n".join(zeilen) + "\r\n" @app.route("/zugriffslog") def zugriffslog(): db = get_db() eintraege = db.execute( "SELECT * FROM zugriffslog ORDER BY zeitpunkt DESC LIMIT 300" ).fetchall() return render_template("zugriffslog.html", eintraege=eintraege) @app.route("/einrichten", methods=["GET", "POST"]) def einrichten(): db = get_db() if request.method == "GET": protokolliere_zugriff("Seite aufgerufen") alle = db.execute( "SELECT id, name, kategorie, version, installer_pfad, install_parameter, auto_update FROM software " "ORDER BY kategorie COLLATE NOCASE, name COLLATE NOCASE" ).fetchall() installierbar = [] nicht_installierbar = [] for sw in alle: pfad = sw["installer_pfad"] if not pfad or pfad.lower().startswith(("http://", "https://")): if pfad: nicht_installierbar.append(sw["name"]) continue gefunden = updater.verwaltete_datei_pfad(pfad, sw["name"]) if gefunden: installierbar.append({ "id": sw["id"], "name": sw["name"], "kategorie": sw["kategorie"] or "Ohne Kategorie", "version": sw["version"], "typ": "installer", "endung": os.path.splitext(gefunden)[1].lower(), "parameter": sw["install_parameter"] or "", "installer_pfad": pfad, }) continue if not sw["auto_update"]: ordner = updater.effektiver_ablageordner(pfad, sw["name"]) if os.path.isdir(ordner) and os.listdir(ordner): installierbar.append({ "id": sw["id"], "name": sw["name"], "kategorie": sw["kategorie"] or "Ohne Kategorie", "version": sw["version"], "typ": "ordner", "endung": "", "parameter": "", "installer_pfad": pfad, }) continue nicht_installierbar.append(sw["name"]) if request.method == "POST": mitarbeiter = request.form.get("mitarbeiter", "").strip() ausgewaehlt_ids = set(request.form.getlist("software_ids")) ausgewaehlt = [t for t in installierbar if str(t["id"]) in ausgewaehlt_ids] if not mitarbeiter or not ausgewaehlt: flash("Bitte Mitarbeiter angeben und mindestens ein Tool auswaehlen.", "fehler") return redirect(url_for("einrichten")) server_url = request.host_url.rstrip("/") ps1_inhalt = _erstelle_einrichtungs_skript(server_url, mitarbeiter, ausgewaehlt) protokolliere_zugriff( "Paket heruntergeladen: " + ", ".join(t["name"] for t in ausgewaehlt), angegebener_name=mitarbeiter, ) speicher = io.BytesIO() with zipfile.ZipFile(speicher, "w", zipfile.ZIP_DEFLATED) as zf: zf.writestr("einrichten.ps1", ps1_inhalt) zf.writestr("Rechner-einrichten-starten.bat", _BAT_LAUNCHER) # Die echten Installer-Dateien/ZIPs direkt mit ins Paket packen, # damit auf dem Zielrechner beim Ausfuehren KEIN Netzwerkzugriff # auf den Server mehr noetig ist (nur die abschliessende # Protokoll-Rueckmeldung ist optional und best-effort). for t in ausgewaehlt: if t["typ"] == "installer": quelldatei = updater.verwaltete_datei_pfad(t["installer_pfad"], t["name"]) if quelldatei and os.path.isfile(quelldatei): zf.write(quelldatei, f"tools/{t['id']}{t['endung']}") else: ordner = updater.effektiver_ablageordner(t["installer_pfad"], t["name"]) zf.writestr(f"tools/{t['id']}.zip", _zippe_ordner_in_bytes(ordner)) speicher.seek(0) return send_file( speicher, as_attachment=True, download_name="rechner-einrichten.zip", mimetype="application/zip", ) return render_template( "einrichten.html", installierbar=installierbar, nicht_installierbar=nicht_installierbar ) def _zippe_ordner_in_bytes(ordner): """Packt einen Ordner (ohne '_sicherungen'-Unterordner) in ein ZIP im Arbeitsspeicher und gibt die rohen Bytes zurueck. Wird sowohl fuer den Download-Button als auch fuer das Einrichtungs-Paket verwendet.""" speicher = io.BytesIO() with zipfile.ZipFile(speicher, "w", zipfile.ZIP_DEFLATED) as zf: for wurzel, unterordner, dateien in os.walk(ordner): unterordner[:] = [o for o in unterordner if o != "_sicherungen"] for datei in dateien: voller_pfad = os.path.join(wurzel, datei) rel_pfad = os.path.relpath(voller_pfad, ordner) zf.write(voller_pfad, rel_pfad) return speicher.getvalue() def _ordner_fuer_installer_sicherstellen(installer_pfad, name): """Legt den Tool-eigenen Ordner fuer den angegebenen Installer-Pfad automatisch an, falls er noch nicht existiert. Damit muss nie manuell im Windows-Explorer ein Ordner erstellt werden.""" if not installer_pfad or installer_pfad.lower().startswith(("http://", "https://")): return ordner = updater.effektiver_ablageordner(installer_pfad, name) if ordner: try: os.makedirs(ordner, exist_ok=True) except OSError: pass # z. B. ungueltiges Laufwerk - beim Speichern nicht kritisch app.jinja_env.filters["status"] = status_fuer # --------------------------------------------------------------------------- # Routen: Dashboard / Software # --------------------------------------------------------------------------- @app.route("/") def dashboard(): suche = request.args.get("q", "").strip() db = get_db() if suche: rows = db.execute( """SELECT * FROM software WHERE name LIKE ? OR kategorie LIKE ? ORDER BY name COLLATE NOCASE""", (f"%{suche}%", f"%{suche}%"), ).fetchall() else: rows = db.execute("SELECT * FROM software ORDER BY name COLLATE NOCASE").fetchall() gesamt = len(rows) warnungen = sum(1 for r in rows if status_fuer(r["letzte_pruefung"])[1] == "status-warnung") unbekannt = sum(1 for r in rows if status_fuer(r["letzte_pruefung"])[1] == "status-unbekannt") auto_anzahl = sum(1 for r in rows if r["auto_update"]) return render_template( "dashboard.html", software_liste=rows, suche=suche, gesamt=gesamt, warnungen=warnungen, unbekannt=unbekannt, auto_anzahl=auto_anzahl, ) @app.route("/software/neu", methods=["GET", "POST"]) def software_neu(): if request.method == "POST": db = get_db() installer_pfad = request.form.get("installer_pfad", "").strip() auto_update = 1 if request.form.get("auto_update") else 0 db.execute( """INSERT INTO software (name, kategorie, version, installer_pfad, install_parameter, dokumentation, letzte_pruefung, erstellt_am, auto_update, update_quelle, update_wert) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", ( request.form["name"].strip(), request.form.get("kategorie", "").strip(), request.form.get("version", "").strip(), installer_pfad, request.form.get("install_parameter", "").strip(), request.form.get("dokumentation", "").strip(), heute_str(), datetime.now().isoformat(timespec="seconds"), auto_update, request.form.get("update_quelle", "").strip(), request.form.get("update_wert", "").strip(), ), ) db.commit() _ordner_fuer_installer_sicherstellen(installer_pfad, request.form["name"].strip()) flash(f"„{request.form['name']}“ wurde hinzugefuegt.", "erfolg") return redirect(url_for("dashboard")) return render_template("software_form.html", software=None) @app.route("/software/") def software_detail(software_id): db = get_db() sw = db.execute("SELECT * FROM software WHERE id = ?", (software_id,)).fetchone() if sw is None: flash("Dieses Tool wurde nicht gefunden.", "fehler") return redirect(url_for("dashboard")) verlauf = db.execute( """SELECT * FROM zuweisungen WHERE software_id = ? ORDER BY installiert_am DESC""", (software_id,), ).fetchall() update_verlauf = db.execute( """SELECT * FROM update_log WHERE software_id = ? ORDER BY zeitpunkt DESC LIMIT 15""", (software_id,), ).fetchall() bilder = db.execute( "SELECT * FROM doku_bilder WHERE software_id = ? ORDER BY hochgeladen_am", (software_id,), ).fetchall() return render_template( "software_detail.html", sw=sw, verlauf=verlauf, update_verlauf=update_verlauf, bilder=bilder ) @app.route("/software//bearbeiten", methods=["GET", "POST"]) def software_bearbeiten(software_id): db = get_db() sw = db.execute("SELECT * FROM software WHERE id = ?", (software_id,)).fetchone() if sw is None: flash("Dieses Tool wurde nicht gefunden.", "fehler") return redirect(url_for("dashboard")) if request.method == "POST": installer_pfad = request.form.get("installer_pfad", "").strip() auto_update = 1 if request.form.get("auto_update") else 0 db.execute( """UPDATE software SET name = ?, kategorie = ?, version = ?, installer_pfad = ?, install_parameter = ?, dokumentation = ?, auto_update = ?, update_quelle = ?, update_wert = ? WHERE id = ?""", ( request.form["name"].strip(), request.form.get("kategorie", "").strip(), request.form.get("version", "").strip(), installer_pfad, request.form.get("install_parameter", "").strip(), request.form.get("dokumentation", "").strip(), auto_update, request.form.get("update_quelle", "").strip(), request.form.get("update_wert", "").strip(), software_id, ), ) db.commit() _ordner_fuer_installer_sicherstellen(installer_pfad, request.form["name"].strip()) flash("Aenderungen gespeichert.", "erfolg") return redirect(url_for("software_detail", software_id=software_id)) return render_template("software_form.html", software=sw) @app.route("/software//pruefen", methods=["POST"]) def software_pruefen(software_id): db = get_db() db.execute( "UPDATE software SET letzte_pruefung = ? WHERE id = ?", (heute_str(), software_id), ) db.commit() flash("Als heute geprueft markiert.", "erfolg") return redirect(request.referrer or url_for("dashboard")) def _fuehre_update_check_aus(db, sw_row): """Fuehrt den Update-Check fuer eine Software aus, schreibt das Ergebnis ins Log und aktualisiert bei Erfolg Version/Pruefdatum. Gibt das UpdateResult zurueck.""" ergebnis = updater.pruefe_software( sw_row["name"], sw_row["update_quelle"], sw_row["update_wert"], sw_row["version"], sw_row["installer_pfad"] ) db.execute( """INSERT INTO update_log (software_id, zeitpunkt, alte_version, neue_version, erfolg, meldung) VALUES (?, ?, ?, ?, ?, ?)""", ( sw_row["id"], datetime.now().isoformat(timespec="seconds"), ergebnis.alte_version, ergebnis.neue_version, 1 if ergebnis.erfolg else 0, ergebnis.meldung, ), ) if ergebnis.erfolg: # Erfolgreich (egal ob neue Version geholt oder bereits aktuell) -> # Pruefdatum aktualisieren, damit der Status-Badge das widerspiegelt. neue_version = ergebnis.neue_version or sw_row["version"] db.execute( "UPDATE software SET version = ?, letzte_pruefung = ? WHERE id = ?", (neue_version, heute_str(), sw_row["id"]), ) # Bei Fehlschlag bewusst NICHT letzte_pruefung aktualisieren, damit das # Tool ueber den Status "Bitte pruefen" sichtbar bleibt. db.commit() return ergebnis @app.route("/software//update-jetzt", methods=["POST"]) def software_update_jetzt(software_id): db = get_db() sw = db.execute("SELECT * FROM software WHERE id = ?", (software_id,)).fetchone() if sw is None: flash("Dieses Tool wurde nicht gefunden.", "fehler") return redirect(url_for("dashboard")) if not sw["auto_update"] or not sw["update_quelle"]: flash("Fuer dieses Tool ist keine automatische Aktualisierung eingerichtet.", "fehler") return redirect(url_for("software_detail", software_id=software_id)) ergebnis = _fuehre_update_check_aus(db, sw) kategorie = "erfolg" if ergebnis.erfolg else "fehler" flash(f"{sw['name']}: {ergebnis.meldung}", kategorie) return redirect(url_for("software_detail", software_id=software_id)) @app.route("/updates/alle-jetzt", methods=["POST"]) def updates_alle_jetzt(): db = get_db() tools = db.execute("SELECT * FROM software WHERE auto_update = 1 AND update_quelle != ''").fetchall() if not tools: flash("Kein Tool hat aktuell die automatische Aktualisierung aktiviert.", "fehler") return redirect(url_for("dashboard")) erfolgreich, fehlgeschlagen, aktualisiert = 0, 0, 0 for sw in tools: ergebnis = _fuehre_update_check_aus(db, sw) if ergebnis.erfolg: erfolgreich += 1 if not ergebnis.unveraendert(): aktualisiert += 1 else: fehlgeschlagen += 1 flash( f"Update-Lauf fertig: {len(tools)} geprueft, {aktualisiert} aktualisiert, " f"{erfolgreich - aktualisiert} bereits aktuell, {fehlgeschlagen} fehlgeschlagen.", "erfolg" if fehlgeschlagen == 0 else "fehler", ) return redirect(url_for("dashboard")) @app.route("/software//loeschen", methods=["POST"]) def software_loeschen(software_id): db = get_db() db.execute("DELETE FROM software WHERE id = ?", (software_id,)) db.commit() flash("Tool wurde geloescht.", "erfolg") return redirect(url_for("dashboard")) # --------------------------------------------------------------------------- # Routen: Doku-Bilder # --------------------------------------------------------------------------- @app.route("/software//bild-hochladen", methods=["POST"]) def bild_hochladen(software_id): db = get_db() sw = db.execute("SELECT id FROM software WHERE id = ?", (software_id,)).fetchone() if sw is None: flash("Dieses Tool wurde nicht gefunden.", "fehler") return redirect(url_for("dashboard")) datei = request.files.get("bild") if not datei or not datei.filename: flash("Keine Datei ausgewaehlt.", "fehler") return redirect(url_for("software_detail", software_id=software_id)) endung = os.path.splitext(datei.filename)[1].lower() if endung not in ERLAUBTE_BILD_ENDUNGEN: flash("Nur Bilder erlaubt (PNG, JPG, GIF, WEBP).", "fehler") return redirect(url_for("software_detail", software_id=software_id)) ordner = os.path.join(BILDER_DIR, str(software_id)) os.makedirs(ordner, exist_ok=True) sicherer_name = secure_filename(datei.filename) or ("bild" + endung) dateiname = f"{datetime.now().strftime('%Y%m%d%H%M%S%f')}_{sicherer_name}" datei.save(os.path.join(ordner, dateiname)) db.execute( "INSERT INTO doku_bilder (software_id, dateiname, hochgeladen_am) VALUES (?, ?, ?)", (software_id, dateiname, datetime.now().isoformat(timespec="seconds")), ) db.commit() flash("Bild hochgeladen.", "erfolg") return redirect(url_for("software_detail", software_id=software_id)) @app.route("/software//bild/") def bild_anzeigen(software_id, bild_id): db = get_db() bild = db.execute( "SELECT * FROM doku_bilder WHERE id = ? AND software_id = ?", (bild_id, software_id) ).fetchone() if bild is None: return "Bild nicht gefunden.", 404 pfad = os.path.join(BILDER_DIR, str(software_id), bild["dateiname"]) if not os.path.isfile(pfad): return "Bild nicht gefunden.", 404 return send_file(pfad) @app.route("/software//bild//loeschen", methods=["POST"]) def bild_loeschen(software_id, bild_id): db = get_db() bild = db.execute( "SELECT * FROM doku_bilder WHERE id = ? AND software_id = ?", (bild_id, software_id) ).fetchone() if bild is not None: pfad = os.path.join(BILDER_DIR, str(software_id), bild["dateiname"]) if os.path.isfile(pfad): os.remove(pfad) db.execute("DELETE FROM doku_bilder WHERE id = ?", (bild_id,)) db.commit() flash("Bild geloescht.", "erfolg") return redirect(url_for("software_detail", software_id=software_id)) @app.errorhandler(413) def datei_zu_gross(e): flash("Die Datei ist zu gross (maximal 15 MB pro Bild).", "fehler") return redirect(request.referrer or url_for("dashboard")) # --------------------------------------------------------------------------- # Routen: Protokolle (ein Protokoll je Mitarbeiter/Rechner-Einrichtung) # --------------------------------------------------------------------------- @app.route("/protokolle") def protokolle(): db = get_db() mitarbeiter_filter = request.args.get("mitarbeiter", "").strip() rechner_filter = request.args.get("rechner", "").strip() query = """SELECT p.*, COUNT(z.id) AS anzahl_tools FROM protokolle p LEFT JOIN zuweisungen z ON z.protokoll_id = p.id WHERE 1=1""" params = [] if mitarbeiter_filter: query += " AND p.mitarbeiter LIKE ?" params.append(f"%{mitarbeiter_filter}%") if rechner_filter: query += " AND p.rechnername LIKE ?" params.append(f"%{rechner_filter}%") query += " GROUP BY p.id ORDER BY p.datum DESC, p.erstellt_zeitpunkt DESC" rows = db.execute(query, params).fetchall() return render_template( "protokolle.html", protokolle=rows, mitarbeiter_filter=mitarbeiter_filter, rechner_filter=rechner_filter, ) @app.route("/protokolle/") def protokoll_detail(protokoll_id): db = get_db() protokoll = db.execute("SELECT * FROM protokolle WHERE id = ?", (protokoll_id,)).fetchone() if protokoll is None: flash("Dieses Protokoll wurde nicht gefunden.", "fehler") return redirect(url_for("protokolle")) eintraege = db.execute( """SELECT z.*, s.name AS software_name FROM zuweisungen z JOIN software s ON s.id = z.software_id WHERE z.protokoll_id = ? ORDER BY s.name COLLATE NOCASE""", (protokoll_id,), ).fetchall() return render_template("protokoll_detail.html", protokoll=protokoll, eintraege=eintraege) @app.route("/protokolle//loeschen", methods=["POST"]) def protokoll_loeschen(protokoll_id): db = get_db() db.execute("DELETE FROM zuweisungen WHERE protokoll_id = ?", (protokoll_id,)) db.execute("DELETE FROM protokolle WHERE id = ?", (protokoll_id,)) db.commit() flash("Protokoll geloescht.", "erfolg") return redirect(url_for("protokolle")) @app.route("/zuweisungen/neu", methods=["GET", "POST"]) def zuweisung_neu(): db = get_db() protokoll_id = request.values.get("protokoll_id", type=int) bestehendes_protokoll = None if protokoll_id: bestehendes_protokoll = db.execute("SELECT * FROM protokolle WHERE id = ?", (protokoll_id,)).fetchone() if bestehendes_protokoll is None: flash("Dieses Protokoll wurde nicht gefunden.", "fehler") return redirect(url_for("protokolle")) if request.method == "POST": software_ids = request.form.getlist("software_ids") if not software_ids: flash("Bitte mindestens ein Tool auswaehlen.", "fehler") return redirect(request.url) installiert_von = request.form.get("installiert_von", "").strip() installiert_am = request.form.get("installiert_am") or heute_str() notiz = request.form.get("notiz", "").strip() if bestehendes_protokoll: ziel_protokoll_id = bestehendes_protokoll["id"] mitarbeiter = bestehendes_protokoll["mitarbeiter"] rechnername = bestehendes_protokoll["rechnername"] else: mitarbeiter = request.form["mitarbeiter"].strip() rechnername = request.form.get("rechnername", "").strip() cur = db.execute( """INSERT INTO protokolle (mitarbeiter, rechnername, datum, erstellt_von, notiz, erstellt_zeitpunkt) VALUES (?, ?, ?, ?, ?, ?)""", (mitarbeiter, rechnername, installiert_am, installiert_von, notiz, datetime.now().isoformat(timespec="seconds")), ) ziel_protokoll_id = cur.lastrowid anzahl = 0 for sw_id in software_ids: sw = db.execute("SELECT version FROM software WHERE id = ?", (sw_id,)).fetchone() if sw is None: continue db.execute( """INSERT INTO zuweisungen (mitarbeiter, rechnername, software_id, installierte_version, installiert_von, installiert_am, notiz, protokoll_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?)""", (mitarbeiter, rechnername, sw_id, sw["version"], installiert_von, installiert_am, notiz, ziel_protokoll_id), ) anzahl += 1 db.commit() flash(f"{anzahl} Tool(s) im Protokoll fuer {mitarbeiter} gespeichert.", "erfolg") return redirect(url_for("protokoll_detail", protokoll_id=ziel_protokoll_id)) bereits_zugewiesen = set() if bestehendes_protokoll: rows = db.execute( "SELECT software_id FROM zuweisungen WHERE protokoll_id = ?", (protokoll_id,) ).fetchall() bereits_zugewiesen = {r["software_id"] for r in rows} software_roh = db.execute( "SELECT id, name, version, kategorie FROM software ORDER BY kategorie COLLATE NOCASE, name COLLATE NOCASE" ).fetchall() software_liste = [ {"id": r["id"], "name": r["name"], "version": r["version"], "kategorie": r["kategorie"] or "Ohne Kategorie"} for r in software_roh if r["id"] not in bereits_zugewiesen ] vorausgewaehlt = request.args.get("software_id", type=int) return render_template( "assignment_form.html", software_liste=software_liste, heute=heute_str(), vorausgewaehlt=vorausgewaehlt, bestehendes_protokoll=bestehendes_protokoll, ) @app.route("/zuweisungen//loeschen", methods=["POST"]) def zuweisung_loeschen(zuweisung_id): db = get_db() eintrag = db.execute("SELECT protokoll_id FROM zuweisungen WHERE id = ?", (zuweisung_id,)).fetchone() db.execute("DELETE FROM zuweisungen WHERE id = ?", (zuweisung_id,)) db.commit() flash("Eintrag geloescht.", "erfolg") if eintrag and eintrag["protokoll_id"]: return redirect(url_for("protokoll_detail", protokoll_id=eintrag["protokoll_id"])) return redirect(url_for("protokolle")) # --------------------------------------------------------------------------- # Route: Installer-Download # --------------------------------------------------------------------------- @app.route("/software//download") def software_download(software_id): db = get_db() sw = db.execute("SELECT * FROM software WHERE id = ?", (software_id,)).fetchone() if sw is None: flash("Dieses Tool wurde nicht gefunden.", "fehler") return redirect(url_for("dashboard")) pfad = sw["installer_pfad"] if not pfad: flash("Fuer dieses Tool ist kein Installer-Pfad hinterlegt.", "fehler") return redirect(url_for("software_detail", software_id=software_id)) if pfad.lower().startswith(("http://", "https://")): return redirect(pfad) # Immer zuerst im Tool-eigenen Unterordner nach der verwalteten Datei # suchen - das gilt jetzt auch fuer aeltere Eintraege mit komplettem # Dateipfad, die frueher noch nicht automatisch verschachtelt wurden. gefunden = updater.verwaltete_datei_pfad(pfad, sw["name"]) if gefunden: return send_file(gefunden, as_attachment=True, download_name=os.path.basename(gefunden)) if sw["auto_update"]: flash("Noch keine Datei vorhanden - zuerst „Jetzt automatisch pruefen“ ausfuehren.", "fehler") return redirect(url_for("software_detail", software_id=software_id)) # Kein Auto-Update und keine einzelne verwaltete Datei gefunden -> ganzen # Tool-Ordner als ZIP anbieten (Sicherungs-Kopien ausgeschlossen). # Praktisch fuer bereits entpackte/portable Tools (z. B. Blender). ordner = updater.effektiver_ablageordner(pfad, sw["name"]) if not os.path.isdir(ordner): flash(f"Ordner nicht gefunden unter: {ordner}", "fehler") return redirect(url_for("software_detail", software_id=software_id)) zip_bytes = _zippe_ordner_in_bytes(ordner) zip_name = os.path.basename(ordner.rstrip("\\/")) + ".zip" return send_file( io.BytesIO(zip_bytes), as_attachment=True, download_name=zip_name, mimetype="application/zip" ) if __name__ == "__main__": init_db() print("=" * 60) print(" Server-Bibliothek laeuft.") print(" Im Browser oeffnen: http://localhost:5000") print(" Beenden mit STRG+C in diesem Fenster.") print("=" * 60) app.run(host="0.0.0.0", port=5000, debug=False)