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/<int:software_id>")
|
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/<int:software_id>/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/<int:software_id>/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/<int:software_id>/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/<int:software_id>/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/<int:software_id>/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/<int:software_id>/bild/<int:bild_id>")
|
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/<int:software_id>/bild/<int:bild_id>/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/<int:protokoll_id>")
|
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/<int:protokoll_id>/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/<int:zuweisung_id>/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/<int:software_id>/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)
|
