#!/usr/bin/env python3 """ CANCOM Simple Signage – Flask-App für Digital-Signage-Anzeigen. ────────────────────────────────────────────────────────────── Diese App verwaltet mehrere Standorte (sites) mit jeweils mehreren Screens. Jeder Screen hat eine eigene Playlist aus Bildern, Videos, HTML-Seiten und URLs. Optional kann eine globale Priority-Playlist zugeschaltet werden, die auf allen Playern abwechselnd eingeblendet wird. """ import os import json import copy import hashlib import re import time import io import base64 import secrets import generate_welcome_page # Logo-Suche und Willkommensseite-Generierung import pyotp import qrcode from functools import wraps from datetime import datetime, timedelta, date from pathlib import Path from werkzeug.middleware.proxy_fix import ProxyFix from flask import ( Flask, jsonify, render_template, send_from_directory, redirect, request, abort, flash, session, url_for ) from flask_login import ( LoginManager, login_user, current_user, login_required, logout_user, UserMixin ) from werkzeug.utils import secure_filename from werkzeug.security import generate_password_hash, check_password_hash # ------------------------------------------------- # Grundkonfiguration # ------------------------------------------------- BASE_DIR = os.path.dirname(os.path.abspath(__file__)) MEDIA_DIR = os.path.join(BASE_DIR, "media") # Medien-Dateien je Standort/Screen CONFIG_FILE = os.path.join(BASE_DIR, "config.json") # Persistente Konfiguration USERS_FILE = os.path.join(BASE_DIR, "users.json") # User-Datenbank HISTORY_FILE = os.path.join(BASE_DIR, "history.json") # Aktivitätsverlauf APP_VERSION = "6.2.7" UPLOAD_EXTENSIONS = {".jpg", ".jpeg", ".png", ".mp4"} app = Flask(__name__) app.secret_key = "CHANGE_THIS_SECRET!!!" app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1) def remember_site(site): """Merkt den zuletzt besuchten Standort in der Session.""" if site: session["last_site"] = site def current_site_for(cfg, user=None): """Standort für Seiten ohne eigenen Standortbezug (Dashboard, Hilfe, User-Verwaltung, MFA). Nimmt den zuletzt besuchten Standort, sonst den ersten *zugänglichen*. Ohne das springt die Standortanzeige im Header auf den global ersten Standort - und der "Übersicht"-Link führt anschliessend dorthin statt zurück zu dem Standort, von dem man kam.""" sites = get_accessible_sites(cfg, user) if (user is not None and user.is_authenticated) else get_site_list(cfg) letzter = session.get("last_site") if letzter in sites: return letzter return sites[0] if sites else "" @app.context_processor def inject_globals(): """Stellt globale Template-Variablen für Header/Footer bereit.""" import os as _os hostname = _os.uname().nodename cfg = load_config() from flask_login import current_user if current_user.is_authenticated: site_list = get_accessible_sites(cfg, current_user) current_site_val = current_site_for(cfg, current_user) else: site_list = get_site_list(cfg) current_site_val = site_list[0] if site_list else "" return dict( site_list=site_list, current_site=current_site_val, brand_subtitle="", site_dropdown_url_prefix="/admin/", version=APP_VERSION, year=datetime.now().year, hostname=hostname ) login_manager = LoginManager(app) login_manager.login_view = "login" # Nicht eingeloggte User werden zum Login umgeleitet # ------------------------------------------------- # Config-Helfer # ------------------------------------------------- def load_config(): """Liest die config.json und gibt das Dict zurück. Legt ein leeres an, falls nicht vorhanden oder fehlerhaft.""" if not os.path.exists(CONFIG_FILE): return {"admin": {}, "sites": {}} try: with open(CONFIG_FILE) as f: return json.load(f) except (json.JSONDecodeError, OSError): return {"admin": {}, "sites": {}} def save_config(cfg): """Schreibt das Config-Dict zurück in die config.json.""" with open(CONFIG_FILE, "w") as f: json.dump(cfg, f, indent=2) HEARTBEAT_DIR = Path("/tmp/signage-heartbeat") HEARTBEAT_TIMEOUT = 60 # Sekunden def record_heartbeat(site, screen, ip="", ua="", resolution=""): """Aktualisiert den Heartbeat-Timestamp für einen Screen und speichert Client-Info.""" path = HEARTBEAT_DIR / site / f"{screen}.json" path.parent.mkdir(parents=True, exist_ok=True) data = { "ip": ip, "ua": ua[:200] if ua else "", "resolution": resolution, "last_seen": datetime.now().strftime("%d.%m.%Y %H:%M:%S") } path.write_text(json.dumps(data)) def screen_is_active(site, screen): """Prüft, ob ein Screen zuletzt innerhalb des Timeouts registriert wurde.""" path = HEARTBEAT_DIR / site / f"{screen}.json" if not path.exists(): return False return (time.time() - path.stat().st_mtime) < HEARTBEAT_TIMEOUT def get_client_info(site, screen): """Gibt die gespeicherten Client-Informationen eines Screens zurück.""" path = HEARTBEAT_DIR / site / f"{screen}.json" if not path.exists(): return {} try: return json.loads(path.read_text()) except (json.JSONDecodeError, OSError): return {} def get_site_list(cfg): """Gibt eine sortierte Liste aller konfigurierten Standort-Namen zurück.""" return sorted(cfg.get("sites", {}).keys()) def add_history_entry(action, detail, user_email): """Hängt einen Eintrag in history.json an (Aktivitätsverlauf für das Admin-Dashboard).""" entry = { "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "action": action, "detail": detail, "user": user_email } history = [] if os.path.exists(HISTORY_FILE): try: with open(HISTORY_FILE) as f: history = json.load(f) except (json.JSONDecodeError, OSError): history = [] history.insert(0, entry) # Neueste zuerst # Maximal 200 Einträge behalten history = history[:200] with open(HISTORY_FILE, "w") as f: json.dump(history, f, indent=2) def get_history(limit=50): """Gibt die letzten history-Einträge zurück.""" if not os.path.exists(HISTORY_FILE): return [] try: with open(HISTORY_FILE) as f: history = json.load(f) return history[:limit] except (json.JSONDecodeError, OSError): return [] def get_screen_config(cfg, site, screen): """Holt die Konfiguration für einen bestimmten Screen. Erzeugt automatisch fehlende Dict-Ebenen (sites → site → screens → screen).""" return ( cfg .setdefault("sites", {}) .setdefault(site, {}) .setdefault("screens", {}) .setdefault(screen, {}) ) def is_url(item): """Prüft, ob ein Playlist-Eintrag eine URL ist (String mit http/https) oder ein Dict mit "url"-Key.""" if isinstance(item, str): return item.startswith("http://") or item.startswith("https://") return isinstance(item, dict) and "url" in item def normalize_url(item): """Normalisiert einen URL-Eintrag zu einem Dict mit "url" und "zoom". Zoom wird aus Dict übernommen oder defaultmäßig auf 1.0 gesetzt.""" if isinstance(item, str): return {"url": item, "zoom": 1.0} if isinstance(item, dict): return {"url": item["url"], "zoom": item.get("zoom", 1.0)} return None def playlist_item_name(item): """Ermittelt den Anzeigenamen eines Playlist-Eintrags (String → direkt, Dict → "url" oder "name").""" if isinstance(item, str): return item if isinstance(item, dict): return item.get("url") or item.get("name", "") return "" def _window_start_date(now, start_time, end_time): """Ermittelt, an welchem Kalendertag das gerade laufende Tageszeitfenster begonnen hat. Gibt None zurück, wenn die aktuelle Uhrzeit ausserhalb des Fensters liegt. Fenster über Mitternacht (start > end) gehören zum Tag ihres Starts: um 01:00 läuft noch das Fenster, das gestern um 22:00 begonnen hat.""" current_time = now.time() if start_time <= end_time: if start_time <= current_time <= end_time: return now.date() return None # Fenster über Mitternacht, z. B. 22:00–02:00 if current_time >= start_time: return now.date() if current_time <= end_time: return (now - timedelta(days=1)).date() return None def _schedule_time(value): """Liest die Uhrzeit aus einem Schedule-Feld. Akzeptiert "HH:MM" (aktuelles Format für daily/weekly) und vollständige ISO-Zeitstempel aus Altbeständen, bei denen der Datumsanteil ohnehin ignoriert wurde. Gibt None zurück, wenn nichts Lesbares drinsteht.""" if not value: return None text = str(value).strip().replace(" ", "T") if "T" in text: try: return datetime.fromisoformat(text).time() except (ValueError, TypeError): return None for fmt in ("%H:%M", "%H:%M:%S"): try: return datetime.strptime(text, fmt).time() except ValueError: continue return None def _in_validity_range(schedule, day): """Prüft den optionalen Gültigkeitszeitraum (valid_from / valid_until) eines Schedules gegen den Tag, an dem das Zeitfenster begonnen hat. Beide Grenzen sind optional und inklusiv; unlesbare Werte werden ignoriert.""" for key, is_lower_bound in (("valid_from", True), ("valid_until", False)): value = schedule.get(key) if not value: continue try: bound = date.fromisoformat(str(value)[:10]) except (ValueError, TypeError): continue if is_lower_bound and day < bound: return False if not is_lower_bound and day > bound: return False return True @app.template_filter("schedule_text") def schedule_text(schedule): """Kurzbeschreibung einer Zeitsteuerung für Tooltips in der Playlist.""" if not schedule: return "" start = str(schedule.get("start", "")) end = str(schedule.get("end", "")) mode = schedule.get("mode", "once") def clock(value): parsed = _schedule_time(value) return parsed.strftime("%H:%M") if parsed else value if mode == "once": text = f"Einmalig: {start.replace('T', ' ')} bis {end.replace('T', ' ')}" elif mode == "daily": text = f"Täglich {clock(start)}–{clock(end)}" elif mode == "weekly": names = ["Mo", "Di", "Mi", "Do", "Fr", "Sa", "So"] days = [] for d in schedule.get("days", []): try: days.append(names[int(d)]) except (ValueError, TypeError, IndexError): continue day_text = ", ".join(days) if days else "kein Tag ausgewählt" text = f"{day_text}: {clock(start)}–{clock(end)}" else: return "" valid_from = schedule.get("valid_from") valid_until = schedule.get("valid_until") if valid_from and valid_until: text += f" (vom {valid_from} bis {valid_until})" elif valid_from: text += f" (ab {valid_from})" elif valid_until: text += f" (bis {valid_until})" return text def is_item_active(item): """Prüft, ob ein Playlist-Eintrag aktiviert und zeitlich gültig ist. Berücksichtigt enabled-Status und optionale Zeitsteuerung. Zeitfenster dürfen über Mitternacht laufen (start > end).""" now = datetime.now() # Strings haben keinen enabled-Status und keine Zeitsteuerung if not isinstance(item, dict): return True if not item.get("enabled", True): return False schedule = item.get("schedule") if not schedule: return True start_str = schedule.get("start") end_str = schedule.get("end") if not start_str or not end_str: return True mode = schedule.get("mode", "once") # "Einmalig" braucht vollständige Zeitstempel, daily/weekly nur die Uhrzeit. # Unlesbare Angaben blenden den Eintrag nicht aus - ein Konfigfehler soll # nicht dazu führen, dass stillschweigend nichts mehr läuft. if mode == "once": try: return datetime.fromisoformat(start_str) <= now <= datetime.fromisoformat(end_str) except (ValueError, TypeError): return True if mode in ("daily", "weekly"): start_time = _schedule_time(start_str) end_time = _schedule_time(end_str) if start_time is None or end_time is None: return True day = _window_start_date(now, start_time, end_time) if day is None: return False if not _in_validity_range(schedule, day): return False if mode == "daily": # Bei "täglich" zählt jeder Wochentag - days wird nicht ausgewertet return True days = schedule.get("days", []) return str(day.weekday()) in {str(d) for d in days} # Unbekannter Modus: wie ohne Zeitsteuerung behandeln return True # ------------------------------------------------- # Priority-Playlist-Helfer # ------------------------------------------------- PRIORITY_DIRNAME = "priority" def priority_dir(site): """Verzeichnis der Priority-Medien eines Standorts: media//priority/.""" return os.path.join(MEDIA_DIR, site, PRIORITY_DIRNAME) def get_priority_config(cfg, site): """Priority-Konfiguration eines Standorts (erzeugt fehlende Ebenen). Die Priority-Playlist gilt ausschliesslich fuer den eigenen Standort. Aeltere Konfigurationen hatten sie global unter cfg["priority"]; ist beim Standort noch nichts hinterlegt, wird der Altbestand einmalig uebernommen.""" site_cfg = cfg.setdefault("sites", {}).setdefault(site, {}) if "priority" not in site_cfg: alt = cfg.get("priority") site_cfg["priority"] = copy.deepcopy(alt) if isinstance(alt, dict) else {} return site_cfg["priority"] def load_priority_files(site): """Lädt die Priority-Playlist eines Standorts aus config.json plus die dazugehörigen Dateien aus media//priority/. Gibt eine Liste von Dicts mit name/type/size/enabled zurück. Neue Dateien im Ordner, die noch nicht in der Playlist sind, werden angehängt.""" cfg = load_config() prio_dir = priority_dir(site) os.makedirs(prio_dir, exist_ok=True) priority_files = [] prio_playlist = get_priority_config(cfg, site).get("playlist", []) # 1. Einträge aus der config.json-Playlist for item in prio_playlist: item_name = playlist_item_name(item) if is_url(item): url_data = normalize_url(item) if url_data: priority_files.append({ "name": url_data["url"], "type": "url", "size": "URL", "zoom": url_data.get("zoom", 1.0), "enabled": item.get("enabled", True) if isinstance(item, dict) else True, "active": is_item_active(item), "passthrough": item.get("passthrough_params", False) if isinstance(item, dict) else False, "schedule": item.get("schedule") if isinstance(item, dict) else None }) continue file_path = os.path.join(prio_dir, item_name) if not os.path.exists(file_path): continue ext = os.path.splitext(item_name)[1].lower() if ext == ".mp4": ftype = "video" elif ext == ".html": ftype = "html" else: ftype = "image" size = os.path.getsize(file_path) // 1024 priority_files.append({ "name": item_name, "type": ftype, "size": size, "enabled": item.get("enabled", True) if isinstance(item, dict) else True, "active": is_item_active(item), "schedule": item.get("schedule") if isinstance(item, dict) else None }) # 2. Neue Dateien aus dem Ordner, die noch nicht in der Playlist sind playlist_names = {playlist_item_name(item) for item in prio_playlist} for f in sorted(os.listdir(prio_dir)): if f in playlist_names or f.startswith("._"): continue file_path = os.path.join(prio_dir, f) ext = os.path.splitext(f)[1].lower() ftype = "video" if ext == ".mp4" else "image" size = os.path.getsize(file_path) // 1024 priority_files.append({ "name": f, "type": ftype, "size": size, "schedule": None }) return priority_files def prio_redirect(site): """Gibt den Redirect-Pfad zurück, der nach Priority-Operationen aufgerufen wird.""" return f"/admin/{site}/priority" def get_background_url(site): """Prüft, ob für einen Standort ein benutzerdefiniertes Hintergrundbild existiert. Gibt die URL zurück oder None (dann wird das Standard-Wallpaper verwendet).""" site_dir = os.path.join(MEDIA_DIR, site) if not os.path.isdir(site_dir): return None for f in sorted(os.listdir(site_dir)): if f.startswith("background."): ext = os.path.splitext(f)[1].lower() if ext in (".jpg", ".jpeg", ".png"): return f"/media/{site}/background/{f}" return None def _regenerate_welcome_for_site(site): """Regeneriert die Willkommensseite mit aktuellem Hintergrund, falls zuvor Kundendaten gespeichert wurden.""" config = load_config() welcome_data = config.get("sites", {}).get(site, {}).get("welcome_data") if welcome_data: names = welcome_data.get("names", []) urls = welcome_data.get("logo_urls", []) if names: bg_url = get_background_url(site) generate_welcome_page.save_welcome_page(names, urls, site=site, background_url=bg_url) print(f"♻️ Willkommensseite für {site} mit aktualisiertem Hintergrund neu generiert") # ------------------------------------------------- # Medien ausliefern (statische Dateien) # ------------------------------------------------- @app.route("/media//background/") def media_background(site, filename): """Liefert das Hintergrundbild für die Willkommensseite aus.""" path = os.path.join(MEDIA_DIR, site) return send_from_directory(path, filename) @app.route("/media//logos/") def media_logo(site, filename): """Liefert ein manuell hochgeladenes Kundenlogo aus. Ohne Login erreichbar wie die übrigen Medien - der Player ist nicht angemeldet. Die Dateien liegen direkt im Standort-Verzeichnis, damit kein Unterordner entsteht, den admin() als Screen interpretieren wuerde.""" path = os.path.join(MEDIA_DIR, site) return send_from_directory(path, filename) # ------------------------------------------------- # User-Verwaltung (Flask-Login mit users.json) # ------------------------------------------------- class User(UserMixin): """User mit E-Mail (id), Rolle, Standort-Berechtigungen und Profilfeldern.""" def __init__(self, email, password_hash="", role="user", sites=None, must_change_password=False, first_name="", last_name="", department="", notes="", mfa_enabled=False, mfa_secret="", mfa_recovery_codes=None, trusted_devices=None): self.email = email self.id = email self.password_hash = password_hash self.role = role self.sites = sites or [] self.must_change_password = must_change_password self.first_name = first_name self.last_name = last_name self.department = department self.notes = notes self.mfa_enabled = mfa_enabled self.mfa_secret = mfa_secret self.mfa_recovery_codes = mfa_recovery_codes or [] self.trusted_devices = trusted_devices or [] @property def is_admin(self): return self.role == "admin" @property def is_superuser(self): return self.role == "superuser" def can_access_site(self, site): return self.is_admin or self.is_superuser or site in self.sites def load_users(): """Liest die users.json. Gibt leeres Dict zurück, wenn nicht vorhanden oder fehlerhaft.""" if not os.path.exists(USERS_FILE): return {} try: with open(USERS_FILE) as f: return json.load(f) except (json.JSONDecodeError, OSError): return {} def save_users(users): """Schreibt das User-Dict zurück in die users.json.""" with open(USERS_FILE, "w") as f: json.dump(users, f, indent=2) def get_user(email): """Lädt einen einzelnen User aus users.json. Gibt None zurück, wenn nicht gefunden.""" users = load_users() data = users.get(email) if data: return User(email, **data) return None def init_user_db(): """Legt users.json mit einem Admin-Konto aus config.json.admin an, falls sie nicht existiert oder leer ist.""" if os.path.exists(USERS_FILE): try: with open(USERS_FILE) as f: content = f.read().strip() if content: return except OSError: pass cfg = load_config() admin = cfg.get("admin", {}) username = admin.get("username", "admin@signage.local") password = admin.get("password", "admin") users = { username: { "password_hash": generate_password_hash(password), "role": "admin", "sites": [], "must_change_password": False, "first_name": "", "last_name": "", "department": "", "notes": "" } } save_users(users) print(f"✅ users.json mit Admin-Konto ({username}) angelegt (Passwort aus config.json)") @login_manager.user_loader def load_user(user_id): """Lädt einen User aus users.json anhand der E-Mail (Flask-Login-Callback).""" return get_user(user_id) # ------------------------------------------------- # Dekoratoren: Admin / Standort-Zugriff # ------------------------------------------------- def admin_required(f): """Erfordert einen Admin-User (role=admin).""" @wraps(f) @login_required def decorated_function(*args, **kwargs): if not current_user.is_admin: abort(403) return f(*args, **kwargs) return decorated_function def site_access_required(f): """Erfordert Zugriff auf den Standort (admin →alle, user →nur freigegebene).""" @wraps(f) @login_required def decorated_function(*args, **kwargs): site = kwargs.get("site") if site and not current_user.can_access_site(site): abort(403) return f(*args, **kwargs) return decorated_function def get_accessible_sites(cfg, user): """Gibt die Liste der Standorte zurück, auf die ein User Zugriff hat.""" all_sites = sorted(cfg.get("sites", {}).keys()) if user.is_admin or user.is_superuser: return all_sites return [s for s in all_sites if s in user.sites] # ------------------------------------------------- # Login / Logout # ------------------------------------------------- @app.route("/login", methods=["GET", "POST"]) def login(): """Login-Seite. Prüft Zugangsdaten gegen users.json (gehashte Passwörter).""" error = None next_url = None if request.method == "POST": email = request.form.get("username", "").strip().lower() password = request.form.get("password", "") mfa_code = request.form.get("mfa_code", "") user = get_user(email) if user: if user.password_hash: try: valid = check_password_hash(user.password_hash, password) except (ValueError, TypeError): valid = False else: valid = False user.must_change_password = True if valid: if user.mfa_enabled: if mfa_code: totp = pyotp.TOTP(user.mfa_secret) if totp.verify(mfa_code): login_user(user) add_history_entry("user_login", f"User '{email}' angemeldet (mit MFA)", email) if user.must_change_password: return redirect("/change-password") return redirect("/admin") # Recovery-Code versuchen if mfa_code in user.mfa_recovery_codes: codes = user.mfa_recovery_codes[:] codes.remove(mfa_code) users_data = load_users() users_data[email]["mfa_recovery_codes"] = codes save_users(users_data) login_user(user) add_history_entry("user_login", f"User '{email}' angemeldet (mit Recovery-Code)", email) return redirect("/mfa/setup?recovered=1") error = "Ungültiger Sicherheitscode" else: # Trusted Device prüfen (Cookie überspringt MFA) trust_token = request.cookies.get("mfa_trust") if trust_token: now = datetime.now() treffer = None aktiv = [] for d in user.trusted_devices: try: expires = datetime.fromisoformat(d["expires"]) except (ValueError, KeyError, TypeError): continue # kaputter Eintrag fliegt raus if expires <= now: continue # abgelaufener Eintrag fliegt raus aktiv.append(d) if d.get("token") == trust_token: treffer = d if len(aktiv) != len(user.trusted_devices): users_data = load_users() users_data.setdefault(email, {})["trusted_devices"] = aktiv save_users(users_data) if treffer: # Trusted Device gültig → MFA überspringen login_user(user) add_history_entry("user_login", f"User '{email}' angemeldet (Trusted Device)", email) if user.must_change_password: return redirect("/change-password") return redirect("/admin") session["mfa_user"] = email session["mfa_password_ok"] = True return redirect("/mfa/verify") else: login_user(user) add_history_entry("user_login", f"User '{email}' angemeldet", email) if user.must_change_password: return redirect("/change-password") return redirect("/admin") elif not user.password_hash: login_user(user) add_history_entry("user_login", f"User '{email}' angemeldet (Passwort-Änderung erforderlich)", email) return redirect("/change-password") error = "Ungültige Zugangsdaten" return render_template("login.html", error=error, next=next_url) @app.route("/logout") def logout(): """Loggt den Admin aus und leitet zur Login-Seite weiter.""" email = current_user.email if current_user.is_authenticated else "unknown" logout_user() add_history_entry("user_logout", f"User '{email}' abgemeldet", email) return redirect("/login") # ------------------------------------------------- # MFA (Multi-Faktor-Authentifizierung per TOTP) # ------------------------------------------------- @app.route("/mfa/verify", methods=["GET", "POST"]) def mfa_verify(): """TOTP-Code nach erfolgreichem Login (Passwort) eingeben.""" email = session.get("mfa_user") if not email: return redirect("/login") if request.method == "POST": code = request.form.get("code", "").strip() action = request.form.get("action", "") user = get_user(email) if not user: session.pop("mfa_user", None) return redirect("/login") if action == "recovery": return render_template("login.html", error=None, show_mfa=True, mfa_user=email, recovery_mode=True) totp = pyotp.TOTP(user.mfa_secret) if code and totp.verify(code): login_user(user) session.pop("mfa_user", None) # Trusted Device speichern wenn gewünscht remember_days = request.form.get("remember") if remember_days: try: days = int(remember_days) if 1 <= days <= 365: token = secrets.token_urlsafe(32) exp_date = datetime.now() + timedelta(days=days) expires = exp_date.isoformat() users_data = load_users() trusted = users_data.setdefault(email, {}).setdefault("trusted_devices", []) trusted.append({"token": token, "expires": expires}) save_users(users_data) add_history_entry("user_login", f"User '{email}' angemeldet (mit MFA, Gerät für {days} Tage gespeichert)", email) target = "/change-password" if user.must_change_password else "/admin" resp = redirect(target) resp.set_cookie("mfa_trust", token, max_age=days * 86400, path="/", httponly=True, samesite="Lax", secure=request.is_secure) return resp except (ValueError, TypeError): pass add_history_entry("user_login", f"User '{email}' angemeldet (mit MFA)", email) if user.must_change_password: return redirect("/change-password") return redirect("/admin") # Recovery-Code prüfen if code in user.mfa_recovery_codes: codes = user.mfa_recovery_codes[:] codes.remove(code) users_data = load_users() users_data[email]["mfa_recovery_codes"] = codes save_users(users_data) login_user(user) session.pop("mfa_user", None) add_history_entry("user_login", f"User '{email}' angemeldet (mit Recovery-Code)", email) return redirect("/mfa/setup?recovered=1") return render_template("mfa_verify.html", error="Ungültiger Code", email=email) return render_template("mfa_verify.html", error=None, email=email) @app.route("/mfa/setup", methods=["GET", "POST"]) @login_required def mfa_setup(): """MFA einrichten: QR-Code anzeigen, ersten Code verifizieren.""" cfg = load_config() site_list = get_accessible_sites(cfg, current_user) email = current_user.email users_data = load_users() user_data = users_data.get(email, {}) recovered = request.args.get("recovered") disabled = request.args.get("disabled") error_param = request.args.get("error") secret = user_data.get("mfa_secret") or pyotp.random_base32() # Secret sofort persistieren, damit GET (QR-Code) und POST (Verifikation) dasselbe Secret verwenden if not user_data.get("mfa_secret"): users_data.setdefault(email, {})["mfa_secret"] = secret save_users(users_data) totp = pyotp.TOTP(secret) if request.method == "POST": code = request.form.get("code", "").strip() if totp.verify(code): recovery = [secrets.token_hex(3).upper()[:7] for _ in range(8)] users_data[email]["mfa_enabled"] = True users_data[email]["mfa_recovery_codes"] = recovery save_users(users_data) add_history_entry("mfa_enabled", f"User '{email}' hat MFA aktiviert", email) return render_template("mfa_recovery.html", codes=recovery, email=email, site_list=site_list, current_site=current_site_for(cfg, current_user), brand_subtitle="MFA", site_dropdown_url_prefix="/admin/") return render_template("mfa_setup.html", error="Code ungültig. Bitte versuche es erneut.", email=email, secret=secret, qr_data=None, site_list=site_list, current_site=current_site_for(cfg, current_user), brand_subtitle="MFA", site_dropdown_url_prefix="/admin/") uri = totp.provisioning_uri(name=email, issuer_name="CANCOM Signage") img = qrcode.make(uri) buf = io.BytesIO() img.save(buf, format="PNG") qr_b64 = base64.b64encode(buf.getvalue()).decode() return render_template("mfa_setup.html", error=None if not error_param else "Passwort ist falsch", email=email, secret=secret, qr_data=qr_b64, recovered=recovered, disabled=disabled, site_list=site_list, current_site=current_site_for(cfg, current_user), brand_subtitle="MFA", site_dropdown_url_prefix="/admin/") @app.route("/mfa/disable", methods=["POST"]) @login_required def mfa_disable(): """MFA deaktivieren (mit Passwort-Bestätigung).""" email = current_user.email password = request.form.get("password", "") user = get_user(email) if user and user.password_hash: try: valid = check_password_hash(user.password_hash, password) except (ValueError, TypeError): valid = False else: valid = False if not valid: return redirect("/mfa/setup?error=1") users_data = load_users() users_data[email]["mfa_enabled"] = False users_data[email]["mfa_secret"] = "" users_data[email]["mfa_recovery_codes"] = [] users_data[email]["trusted_devices"] = [] save_users(users_data) add_history_entry("mfa_disabled", f"User '{email}' hat MFA deaktiviert", email) resp = redirect("/mfa/setup?disabled=1") resp.delete_cookie("mfa_trust") return resp # ------------------------------------------------- # Willkommensseite (Kunden-Logos + Generierung) # ------------------------------------------------- @app.route("/willkommen", methods=["GET", "POST"]) @app.route("/customer", methods=["GET", "POST"]) @login_required def add_customer(): """ Formular für die Willkommensseite: - Bis zu 3 Kundennamen eingeben - Logos per OpenAI → Brandfetch suchen - welcome.html wird generiert und in den Playlists aller Zielscreens vorne platziert (Screens mit "welcome_enabled") """ error = None success = None logo_urls = [None, None, None] customer_names = ["", "", ""] contacts = ["", "", ""] valid_until = (request.form.get("valid_until") or "").strip() logo_hinweise = [] site = request.args.get("site") or request.form.get("site") or "" # Fallback auf ersten für den User zugänglichen Standort if not site: config = load_config() sites = get_accessible_sites(config, current_user) site = sites[0] if sites else "" # Prüfen, ob der User Zugriff auf den Standort hat if site and not current_user.can_access_site(site): abort(403) remember_site(site) if request.method == "POST": customer_names = [ request.form.get("customer_name_1", "").strip(), request.form.get("customer_name_2", "").strip(), request.form.get("customer_name_3", "").strip(), ] contacts = [ request.form.get("contact_1", "").strip(), request.form.get("contact_2", "").strip(), request.form.get("contact_3", "").strip(), ] filled_names = [n for n in customer_names if n] print(f"🔍 APP.PY - Suche nach Kunden: {filled_names} (site: {site})") if not filled_names: error = "Mindestens ein Kundenname erforderlich" elif not site: error = "Kein Standort ausgewählt" else: try: # Logo je Kunde bestimmen: ein hochgeladenes Logo hat Vorrang vor # der automatischen Suche - der Upload ist eine bewusste Entscheidung. # Zuletzt verwendete Logos, um einen früheren Upload nicht bei jedem # erneuten Generieren zu verlieren frueher = load_config().get("sites", {}).get(site, {}).get("welcome_data", {}) frueher_namen = frueher.get("names", []) frueher_urls = frueher.get("logo_urls", []) logo_prefix = f"/media/{site}/{generate_welcome_page.LOGO_URL_PREFIX}/" found_logos = [None, None, None] for i, name in enumerate(customer_names): if not name: continue datei = request.files.get(f"logo_file_{i + 1}") # Eindeutig protokollieren, ob ueberhaupt eine Datei ankam - # ein fehlendes enctype im Formular (etwa aus dem Browser-Cache) # sieht sonst genauso aus wie "kein Logo hochgeladen". print(f"🔍 APP.PY - Slot {i + 1} ({name}): Upload = " f"{datei.filename if datei and datei.filename else 'keiner'}") if datei and datei.filename: url, hinweis = generate_welcome_page.save_uploaded_logo( datei, site, i + 1, freistellen=(f"logo_keep_bg_{i + 1}" not in request.form)) found_logos[i] = url if hinweis: logo_hinweise.append(f"{name}: {hinweis}") elif not url: logo_hinweise.append(f"{name}: Logo konnte nicht übernommen werden") continue # Kein neuer Upload: ein zuvor hochgeladenes Logo weiterverwenden, # solange der Kundenname in diesem Feld derselbe geblieben ist alt_url = frueher_urls[i] if i < len(frueher_urls) else None alt_name = frueher_namen[i] if i < len(frueher_namen) else None if alt_url and alt_name == name and str(alt_url).startswith(logo_prefix): found_logos[i] = alt_url continue found_logos[i] = generate_welcome_page.search_customer_logo(name) logo_urls = list(found_logos) valid = [(customer_names[i], found_logos[i], contacts[i]) for i in range(len(customer_names)) if customer_names[i] and found_logos[i]] if not valid: error = "Kein Logo gefunden – bitte für die betroffenen Kunden ein Logo hochladen" else: valid_names = [n for n, u, k in valid] valid_urls = [u for n, u, k in valid] valid_contacts = [k for n, u, k in valid] bg_url = get_background_url(site) targets = generate_welcome_page.welcome_target_screens(load_config(), site) if not targets: error = ("Kein Screen für die Willkommensseite ausgewählt – " "im Tab 'Einstellungen' eines Screens aktivieren") html_filename = None else: html_filename = generate_welcome_page.save_welcome_page( valid_names, valid_urls, contacts=valid_contacts, site=site, background_url=bg_url, screens=targets) if html_filename: if generate_welcome_page.add_customer_to_playlists( html_filename, site=site, screens=targets, valid_until=valid_until): # Letzte Kundendaten für Background-Update speichern config = load_config() site_cfg = config.setdefault("sites", {}).setdefault(site, {}) # Feldweise speichern (Slot 1-3), nicht nur die erfolgreichen - # sonst rutscht Kunde 2 beim naechsten Aufruf in Feld 1 und die # Zuordnung von Logo und Ansprechpartner verschiebt sich mit. site_cfg["welcome_data"] = { "names": list(customer_names), "logo_urls": list(found_logos), "contacts": list(contacts) } save_config(config) add_history_entry("customer_added", f"Kunden ({', '.join(valid_names)}) zur Willkommensseite an Standort '{site}' hinzugefügt (Screens: {', '.join(targets)})", current_user.email) success = (f"✅ Willkommensseite mit {len(valid_names)} Kunden erstellt " f"– angezeigt auf: {', '.join(targets)}") if valid_until: success += f" (bis einschliesslich {valid_until})" else: error = "Fehler beim Hinzufügen zur Playliste" elif targets: # Ohne Zielscreens steht die Meldung dazu bereits in error error = "Fehler beim Speichern der Willkommensseite" except ValueError as e: error = f"Konfigurationsfehler: {str(e)}" except Exception as e: error = f"Fehler: {str(e)}" config = load_config() site_list = get_site_list(config) background_url = get_background_url(site) if request.args.get("reset") and site: site_cfg = config.setdefault("sites", {}).setdefault(site, {}) site_cfg.pop("welcome_data", None) save_config(config) return redirect(url_for("add_customer", site=site)) welcome_data = config.get("sites", {}).get(site, {}).get("welcome_data", {}) saved_contacts = welcome_data.get("contacts", ["", "", ""]) if request.method != "POST": contacts = saved_contacts customer_names = welcome_data.get("names", ["", "", ""]) logo_urls = welcome_data.get("logo_urls", [None, None, None]) return render_template( "customer.html", error=error, success=success, logo_urls=logo_urls, customer_names=customer_names, contacts=contacts, welcome_screens=generate_welcome_page.welcome_target_screens(config, site) if site else [], logo_hinweise=logo_hinweise, site=site, site_list=site_list, current_site=site, version=APP_VERSION, year=datetime.now().year, hostname=os.uname().nodename, background_url=background_url ) @app.route("/api/customer", methods=["POST"]) @login_required def api_add_customer(): """ API-Endpunkt für die Willkommensseite (JSON). Erwartet: {"customer_name": "...", "site": "..."} Gibt die Logo-URL und den Status zurück. """ try: data = request.get_json() customer_name = data.get("customer_name", "").strip() site = data.get("site", "").strip() if not customer_name: return jsonify({"error": "Customer name required"}), 400 # Default auf ersten Standort if not site: config = load_config() sites = get_site_list(config) site = sites[0] if sites else "" if not site: return jsonify({"error": "No site configured"}), 400 if not current_user.can_access_site(site): return jsonify({"error": "Access denied"}), 403 logo_url = generate_welcome_page.search_customer_logo(customer_name) if not logo_url: return jsonify({"error": "Logo not found"}), 400 contact = data.get("contact", "").strip() valid_until = (data.get("valid_until") or "").strip() targets = generate_welcome_page.welcome_target_screens(load_config(), site) if not targets: return jsonify({"error": "No welcome screen configured for this site"}), 400 html_filename = generate_welcome_page.save_welcome_page( [customer_name], [logo_url], contacts=[contact], site=site, screens=targets) if not html_filename: return jsonify({"error": "Failed to create welcome page"}), 500 if generate_welcome_page.add_customer_to_playlists( html_filename, site=site, screens=targets, valid_until=valid_until): return jsonify({ "success": True, "message": f"Customer '{customer_name}' added successfully", "welcome_page": html_filename, "logo_url": logo_url, "site": site, "screens": targets }), 201 else: return jsonify({"error": "Failed to add to playlist"}), 500 except ValueError as e: return jsonify({"error": str(e)}), 400 except Exception as e: return jsonify({"error": str(e)}), 500 # ------------------------------------------------- # Medien ausliefern (statische Dateien) # ------------------------------------------------- @app.route("/media//priority/") def media_priority(site, filename): """Liefert die Priority-Medien eines Standorts aus media//priority/ aus.""" return send_from_directory(priority_dir(site), filename) @app.route("/media///") def media(site, screen, filename): """Liefert Medien-Dateien für einen bestimmten Screen aus.""" path = os.path.join(MEDIA_DIR, site, screen) return send_from_directory(path, filename) # ------------------------------------------------- # Player # ------------------------------------------------- @app.route("/player//") def player(site, screen): """ Player-Ansicht für einen Screen. Baut die Playlist (normal + optional priority) zusammen und rendert player.html. Der Player zeigt abwechselnd Bilder, Videos, HTML-Seiten und URLs an. """ config = load_config() sites_cfg = config.setdefault("sites", {}) site_cfg = sites_cfg.get(site) if not site_cfg: abort(404) screens_cfg = site_cfg.setdefault("screens", {}) screen_cfg = screens_cfg.get(screen) if not screen_cfg: abort(404) folder = os.path.join(MEDIA_DIR, site, screen) playlist = screen_cfg.get("playlist", []) show_images = screen_cfg.get("show_images", True) show_videos = screen_cfg.get("show_videos", True) def allowed_file(name): """Prüft, ob eine Datei angezeigt werden darf (abhängig von show_images / show_videos).""" if name.startswith("._"): return False ext = name.lower() if ext.endswith((".jpg", ".jpeg", ".png")): return show_images if ext.endswith(".mp4"): return show_videos if ext.endswith((".html", ".htm")): return True return False # ─── Normale Screen-Playlist ─── normal_files = [] # 1. Playlist-Reihenfolge aus config.json for item in playlist: if not is_item_active(item): continue item_name = playlist_item_name(item) if is_url(item): url_data = normalize_url(item) if url_data: normal_files.append({"kind": "url", "url": url_data["url"], "zoom": url_data.get("zoom", 1.0), "enabled": item.get("enabled", True) if isinstance(item, dict) else True, "active": is_item_active(item), "passthrough": item.get("passthrough_params", False) if isinstance(item, dict) else False, "schedule": item.get("schedule") if isinstance(item, dict) else None}) continue if item_name and os.path.exists(os.path.join(folder, item_name)) and allowed_file(item_name): normal_files.append({"kind": "file", "name": item_name, "enabled": True}) existing_file_names = { item["name"] for item in normal_files if item["kind"] == "file" } # 2. Neue Dateien aus dem Medien-Ordner (ohne Playlist-Eintrag) hinten anhängen if not playlist: for f in sorted(os.listdir(folder)): if f in existing_file_names or f.startswith("._"): continue if allowed_file(f): normal_files.append({"kind": "file", "name": f}) # ─── Priority-Playlist (global) ─── priority_cfg = get_priority_config(config, site) if priority_cfg.get("enabled", False): prio_files = [] for item in priority_cfg.get("playlist", []): if not is_item_active(item): continue item_name = playlist_item_name(item) if is_url(item): url_data = normalize_url(item) if url_data: prio_files.append({"kind": "url", "url": url_data["url"], "zoom": url_data.get("zoom", 1.0), "enabled": item.get("enabled", True) if isinstance(item, dict) else True, "active": is_item_active(item), "passthrough": item.get("passthrough_params", False) if isinstance(item, dict) else False, "schedule": item.get("schedule") if isinstance(item, dict) else None}) continue file_path = os.path.join(priority_dir(site), item_name) if os.path.exists(file_path): prio_files.append({"kind": "file", "name": item_name, "enabled": True}) else: prio_files = [] return render_template( "player.html", site=site, screen=screen, normal_files=[item for item in normal_files if item.get("enabled", True) and item.get("active", True)], prio_files=[item for item in prio_files if item.get("enabled", True) and item.get("active", True)], interval=screen_cfg.get("interval", 10), newsticker_text=screen_cfg.get("newsticker_text", ""), newsticker_enabled=screen_cfg.get("newsticker_enabled", False), custom_url=screen_cfg.get("custom_url", ""), custom_url_label=screen_cfg.get("custom_url_label", ""), custom_url_enabled=screen_cfg.get("custom_url_enabled", False), custom_url_target=screen_cfg.get("custom_url_target", "overlay"), custom_url_position=screen_cfg.get("custom_url_position", "top-left"), stay_on_first=screen_cfg.get("stay_on_first", False), voice_agent_enabled=screen_cfg.get("voice_agent_enabled", False), voice_agent_show_image=screen_cfg.get("voice_agent_show_image") if screen_cfg.get("voice_agent_show_image") is not None else False, voice_agent_url=config.get("voice_agent_url", ""), voice_agent_label=screen_cfg.get("voice_agent_label", "Digitaler Assistent"), voice_agent_target=screen_cfg.get("voice_agent_target", "overlay"), voice_agent_position=screen_cfg.get("voice_agent_position", "top-left"), player_params=dict(request.args), now=datetime.now().isoformat() ) # ------------------------------------------------- # Hash für automatisches Player-Reload # ───────────────────────────────────────────────── # Der Player pollt alle 5s diese Route. # Wenn sich der Hash ändert, wird neu geladen. # ------------------------------------------------- @app.route("/playlist///hash") def playlist_hash(site, screen): """Berechnet einen MD5-Hash über die aktuelle Konfiguration (Screen + Priority). Gibt JSON mit Hash und aktuellem stay_on_first-Wert zurück. Zeichnet gleichzeitig den Heartbeat auf (Player aktiv). Nimmt optionale Query-Parameter w (Breite) und h (Höhe) für Client-Info.""" ip = request.remote_addr or "" ua = request.headers.get("User-Agent", "") resolution = f"{request.args.get('w', '?')}x{request.args.get('h', '?')}" record_heartbeat(site, screen, ip=ip, ua=ua, resolution=resolution) config = load_config() screen_playlist = ( config.get("sites", {}).get(site, {}).get("screens", {}).get(screen, {}).get("playlist", []) ) prio_cfg = get_priority_config(config, site) prio_playlist = prio_cfg.get("playlist", []) if prio_cfg.get("enabled", False) else [] relevant = { "sites": { site: config.get("sites", {}).get(site, {}) }, "priority": prio_cfg, # Aktueller Zeitsteuerungs-Zustand: sonst würde der Player ein # ablaufendes Zeitfenster erst beim nächsten Reload bemerken. "active": [is_item_active(item) for item in screen_playlist] + [is_item_active(item) for item in prio_playlist] } blob = json.dumps(relevant, sort_keys=True).encode() h = hashlib.md5(blob).hexdigest() screen_cfg = config.get("sites", {}).get(site, {}).get("screens", {}).get(screen, {}) return {"hash": h, "stay_on_first": screen_cfg.get("stay_on_first", False)} # ------------------------------------------------- # Admin # ------------------------------------------------- @app.route("/", strict_slashes=False) def root_redirect(): """Root weiterleiten zum Login.""" return redirect("/login") @app.route("/admin", strict_slashes=False) @login_required def admin_redirect(): """Leitet zum ersten für den User zugänglichen Standort weiter.""" cfg = load_config() sites = get_accessible_sites(cfg, current_user) if sites: return redirect(f"/admin/{sites[0]}") return "

Kein Standort konfiguriert

Bitte lege einen Standort im media-Verzeichnis an.

", 200 @app.route("/add-site") @admin_required def add_site(): """Legt einen neuen Standort an (Verzeichnis + Config-Eintrag). Nur für Admins.""" name = request.args.get("name", "").strip().lower() if not name: return redirect("/admin") cfg = load_config() cfg.setdefault("sites", {}).setdefault(name, {"screens": {}}) os.makedirs(os.path.join(MEDIA_DIR, name), exist_ok=True) save_config(cfg) add_history_entry("site_created", f"Standort '{name}' angelegt", current_user.email) return redirect(f"/admin/{name}") @app.route("/admin//delete-site", methods=["POST"]) @admin_required def delete_site(site): """Löscht einen Standort inkl. aller Screens und Medien (Config + Verzeichnis). Nur für Admins.""" cfg = load_config() cfg.get("sites", {}).pop(site, None) save_config(cfg) site_dir = os.path.join(MEDIA_DIR, site) if os.path.isdir(site_dir): import shutil shutil.rmtree(site_dir) add_history_entry("site_deleted", f"Standort '{site}' gelöscht", current_user.email) sites = get_site_list(cfg) return redirect(f"/admin/{sites[0]}" if sites else "/admin") @app.route("/admin/", strict_slashes=False) @site_access_required def admin(site): """ Admin-Dashboard für einen Standort. Zeigt alle Screens als Cards mit Tabler-Tabs (Playlist, Einstellungen, Medien). """ remember_site(site) cfg = load_config() site_list = get_accessible_sites(cfg, current_user) # Prüfen, ob der Standort existiert if site not in get_site_list(cfg): if os.path.isdir(os.path.join(MEDIA_DIR, site)): cfg.setdefault("sites", {})[site] = {"screens": {}} save_config(cfg) site_list = get_site_list(cfg) else: abort(404) current_site = site screens = {} media_files = {} screen_status = {} client_info = {} site_media_base = os.path.join(MEDIA_DIR, current_site) # Screens erkennen: aus Config + Dateisystem known_screens = set() site_cfg = cfg.setdefault("sites", {}).setdefault(current_site, {"screens": {}}) config_screens = site_cfg.setdefault("screens", {}) # Ordner im Medienverzeichnis = Screens. # Reservierte Namen: "logos" (hochgeladene Kundenlogos) und "priority" # (Priority-Medien des Standorts). Solche Ordner wuerden sonst dauerhaft als # Screen in der Konfiguration landen; "priority" ist ausserdem in den Routen # bereits als Kennzeichen fuer die Priority-Playlist belegt. reserviert = {generate_welcome_page.LOGO_URL_PREFIX, PRIORITY_DIRNAME} if os.path.isdir(site_media_base): for entry in os.listdir(site_media_base): entry_path = os.path.join(site_media_base, entry) if (os.path.isdir(entry_path) and not entry.startswith("._") and entry not in reserviert): known_screens.add(entry) # Screens aus Config ergänzen (auch ohne Ordner) for s in config_screens: known_screens.add(s) for screen_name in sorted(known_screens): path = os.path.join(site_media_base, screen_name) os.makedirs(path, exist_ok=True) files = [] screen_cfg = get_screen_config(cfg, current_site, screen_name) playlist = screen_cfg.get("playlist", []) screen_cfg["newsticker_text"] = screen_cfg.get("newsticker_text", "") screen_cfg["newsticker_enabled"] = screen_cfg.get("newsticker_enabled", False) # 1. Playlist-Einträge in Config-Reihenfolge for item in playlist: item_name = playlist_item_name(item) schedule = item.get("schedule") if isinstance(item, dict) else None if is_url(item): url_data = normalize_url(item) if url_data: files.append({ "name": url_data["url"], "type": "url", "size": "URL", "zoom": url_data.get("zoom", 1.0), "enabled": item.get("enabled", True) if isinstance(item, dict) else True, "active": is_item_active(item), "passthrough": item.get("passthrough_params", False) if isinstance(item, dict) else False, "schedule": schedule }) continue file_path = os.path.join(path, item_name) if not os.path.exists(file_path): continue ext = os.path.splitext(item_name)[1].lower() if ext == ".mp4": ftype = "video" elif ext == ".html": ftype = "html" else: ftype = "image" size = os.path.getsize(file_path) // 1024 files.append({ "name": item_name, "type": ftype, "size": size, "enabled": item.get("enabled", True) if isinstance(item, dict) else True, "active": is_item_active(item), "schedule": schedule }) # 2. Neue Dateien (ohne Playlist-Eintrag) anhängen playlist_names = {playlist_item_name(item) for item in playlist} for f in sorted(os.listdir(path)): if f in playlist_names or f.startswith("._"): continue file_path = os.path.join(path, f) ext = os.path.splitext(f)[1].lower() ftype = "video" if ext == ".mp4" else "image" size = os.path.getsize(file_path) // 1024 files.append({ "name": f, "type": ftype, "size": size, "schedule": None }) screens[screen_name] = screen_cfg media_files[screen_name] = files screen_status[screen_name] = "online" if screen_is_active(current_site, screen_name) else "offline" client_info[screen_name] = get_client_info(current_site, screen_name) # Priority-Einträge des Standorts nur zur Anzeige mitgeben: sie gelten für # alle Screens, sind aber in einem eigenen Menüpunkt und werden hier leicht # übersehen. Bedienung bleibt ausschließlich auf der Priority-Seite. prio_cfg = get_priority_config(cfg, current_site) if current_site else {} priority_preview = [] if current_site and prio_cfg.get("enabled", False): priority_preview = [f for f in load_priority_files(current_site) if "enabled" in f] return render_template( "admin.html", site_list=site_list, current_site=current_site, screens=screens, media_files=media_files, priority_preview=priority_preview, screen_status=screen_status, client_info=client_info, version=APP_VERSION, year=datetime.now().year, hostname=os.uname().nodename, server_url=cfg.get("server_url", ""), voice_agent_url=cfg.get("voice_agent_url", "") ) # ------------------------------------------------- # Admin: Priority-Playlist (eigene Seite) # ------------------------------------------------- @app.route("/admin//priority") @site_access_required def admin_priority(site): """Zeigt die Priority-Playlist-Seite mit Tabs (Playlist + Medien).""" remember_site(site) cfg = load_config() site_list = get_accessible_sites(cfg, current_user) priority_files = load_priority_files(site) return render_template( "priority.html", site_list=site_list, current_site=site, priority_files=priority_files, version=APP_VERSION, year=datetime.now().year, hostname=os.uname().nodename, server_url=cfg.get("server_url", "") ) # ------------------------------------------------- # Admin: Hintergrundbild für Willkommensseite hochladen # ------------------------------------------------- @app.route("/admin//upload-background", methods=["POST"]) @site_access_required def upload_background(site): """Lädt ein Hintergrundbild für die Willkommensseite hoch (pro Standort). Gespeichert als media//background..""" file = request.files.get("background") if not file: return redirect(f"/willkommen?site={site}") ext = os.path.splitext(file.filename)[1].lower() if ext not in UPLOAD_EXTENSIONS: return "Ungültiger Dateityp", 400 site_dir = os.path.join(MEDIA_DIR, site) os.makedirs(site_dir, exist_ok=True) for f in os.listdir(site_dir): if f.startswith("background."): os.remove(os.path.join(site_dir, f)) filename = f"background{ext}" file.save(os.path.join(site_dir, filename)) _regenerate_welcome_for_site(site) add_history_entry("background_uploaded", f"Hintergrundbild für Standort '{site}' hochgeladen", current_user.email) return redirect(f"/willkommen?site={site}") @app.route("/admin//delete-background", methods=["POST"]) @site_access_required def delete_background(site): """Löscht das Hintergrundbild für die Willkommensseite.""" site_dir = os.path.join(MEDIA_DIR, site) if os.path.isdir(site_dir): for f in os.listdir(site_dir): if f.startswith("background."): os.remove(os.path.join(site_dir, f)) _regenerate_welcome_for_site(site) add_history_entry("background_deleted", f"Hintergrundbild für Standort '{site}' zurückgesetzt", current_user.email) return redirect(f"/willkommen?site={site}") # ------------------------------------------------- # Admin: Screen-Einstellungen speichern # ------------------------------------------------- @app.route("/admin//update/", methods=["POST"]) @site_access_required def update_screen(site, screen): """Speichert die allgemeinen Einstellungen für einen Screen (Intervall, Newsticker, Show-Bilder/Videos).""" config = load_config() cfg = get_screen_config(config, site, screen) cfg["interval"] = int(request.form.get("interval") or 10) cfg["show_images"] = "show_images" in request.form cfg["show_videos"] = "show_videos" in request.form cfg["newsticker_text"] = request.form.get("newsticker_text", "") cfg["newsticker_enabled"] = "newsticker_enabled" in request.form cfg["stay_on_first"] = "stay_on_first" in request.form welcome_an = "welcome_enabled" in request.form vorher = cfg.get("welcome_enabled") cfg["welcome_enabled"] = welcome_an # Beim Umlegen des Schalters die Willkommensseite sofort platzieren bzw. # entfernen - sie ist je Standort identisch, ein erneutes Generieren # waere also nur ein zusaetzlicher Handgriff ohne anderes Ergebnis. if vorher != welcome_an: generate_welcome_page.sync_welcome_for_screen(config, site, screen, welcome_an) save_config(config) return redirect(f"/admin/{site}") @app.route("/admin//update-actions/", methods=["POST"]) @site_access_required def update_screen_actions(site, screen): """Speichert die Aktionen-Einstellungen für einen Screen (Custom-URL + Position).""" config = load_config() cfg = get_screen_config(config, site, screen) cfg["custom_url"] = request.form.get("custom_url", "").strip() cfg["custom_url_label"] = request.form.get("custom_url_label", "").strip() cfg["custom_url_enabled"] = "custom_url_enabled" in request.form cfg["custom_url_target"] = request.form.get("custom_url_target", "overlay") cfg["custom_url_position"] = request.form.get("custom_url_position", "top-left") save_config(config) return redirect(f"/admin/{site}") # ------------------------------------------------- # Admin: Voice-Agent-Einstellungen speichern # ------------------------------------------------- @app.route("/admin//update-voice/", methods=["POST"]) @site_access_required def update_screen_voice(site, screen): """Speichert die Voice-Agent-Einstellungen für einen Screen (Enabled, Label, Target, Position).""" config = load_config() cfg = get_screen_config(config, site, screen) cfg["voice_agent_enabled"] = "voice_agent_enabled" in request.form cfg["voice_agent_show_image"] = "voice_agent_show_image" in request.form cfg["voice_agent_label"] = request.form.get("voice_agent_label", "").strip() cfg["voice_agent_target"] = request.form.get("voice_agent_target", "overlay") cfg["voice_agent_position"] = request.form.get("voice_agent_position", "top-left") save_config(config) return redirect(f"/admin/{site}") # ------------------------------------------------- # Admin: Screen löschen # ------------------------------------------------- @app.route("/admin//delete-screen/", methods=["POST"]) @site_access_required def delete_screen(site, screen): """Löscht einen Screen (Config-Eintrag + Medien-Verzeichnis).""" config = load_config() screens_cfg = config.get("sites", {}).get(site, {}).get("screens", {}) screens_cfg.pop(screen, None) save_config(config) screen_dir = os.path.join(MEDIA_DIR, site, screen) if os.path.isdir(screen_dir): import shutil shutil.rmtree(screen_dir) add_history_entry("screen_deleted", f"Screen '{screen}' an Standort '{site}' gelöscht", current_user.email) return redirect(f"/admin/{site}") # ------------------------------------------------- # Admin: Neuen Screen anlegen # ------------------------------------------------- @app.route("/admin//add-screen", methods=["GET"]) @site_access_required def add_screen(site): """Legt einen neuen Screen an (Verzeichnis + Config-Eintrag).""" screen_name = request.args.get("name", "").strip().lower() if not screen_name: return redirect(f"/admin/{site}") config = load_config() os.makedirs(os.path.join(MEDIA_DIR, site, screen_name), exist_ok=True) screens_cfg = config.setdefault("sites", {}).setdefault(site, {}).setdefault("screens", {}) if screen_name not in screens_cfg: screens_cfg[screen_name] = {"playlist": [], "interval": 10, "show_images": True, "show_videos": True} save_config(config) add_history_entry("screen_created", f"Screen '{screen_name}' an Standort '{site}' angelegt", current_user.email) return redirect(f"/admin/{site}#screen-{screen_name}") # ------------------------------------------------- # Admin: Medien-Upload # ------------------------------------------------- @app.route("/admin//upload/", methods=["POST"]) @site_access_required def upload(site, screen): """Lädt eine Datei hoch (Bild oder Video). Bei screen="priority" ins globale Verzeichnis.""" file = request.files.get("file") if not file: return redirect(prio_redirect(site) if screen == "priority" else f"/admin/{site}") ext = os.path.splitext(file.filename)[1].lower() if ext not in UPLOAD_EXTENSIONS: return "Ungültiger Dateityp", 400 filename = secure_filename(file.filename) if screen == "priority": os.makedirs(priority_dir(site), exist_ok=True) file.save(os.path.join(priority_dir(site), filename)) else: file.save(os.path.join(MEDIA_DIR, site, screen, filename)) return redirect(prio_redirect(site) if screen == "priority" else f"/admin/{site}") # ------------------------------------------------- # Admin: URL zur Playlist hinzufügen # ------------------------------------------------- @app.route("/admin//add-url/", methods=["POST"]) @site_access_required def add_url(site, screen): """Fügt eine URL (mit Zoom-Faktor und optionaler Zeitsteuerung) zur Playlist eines Screens oder der Priority-Playlist hinzu.""" url = request.form.get("url", "").strip() zoom = request.form.get("zoom", "1.0").strip() if not is_url(url): return redirect(prio_redirect(site) if screen == "priority" else f"/admin/{site}") try: zoom_factor = float(zoom) if zoom_factor <= 0: zoom_factor = 1.0 except (ValueError, TypeError): zoom_factor = 1.0 # Zeitsteuerung verarbeiten schedule_enabled = "schedule_enabled" in request.form schedule = None if schedule_enabled: mode = request.form.get("schedule_mode", "once") # "Einmalig" braucht Datum + Uhrzeit, daily/weekly nur die Uhrzeit if mode in ("daily", "weekly"): start = request.form.get("schedule_start_time") end = request.form.get("schedule_end_time") else: start = request.form.get("schedule_start") end = request.form.get("schedule_end") if not start or not end: flash("Start- und Endzeit müssen angegeben werden", "error") return redirect(f"/admin/{site}") schedule = { "mode": mode, "start": start, "end": end } if mode == "weekly": days = request.form.getlist("schedule_days") schedule["days"] = days # Optionaler Gültigkeitszeitraum (nur bei daily/weekly sinnvoll) if mode in ("daily", "weekly"): valid_from = (request.form.get("schedule_valid_from") or "").strip() valid_until = (request.form.get("schedule_valid_until") or "").strip() if valid_from: schedule["valid_from"] = valid_from if valid_until: schedule["valid_until"] = valid_until url_data = {"url": url, "zoom": zoom_factor} if schedule: url_data["schedule"] = schedule config = load_config() if screen == "priority": get_priority_config(config, site).setdefault("playlist", []).append(url_data) else: cfg = get_screen_config(config, site, screen) cfg.setdefault("playlist", []).append(url_data) save_config(config) return redirect(prio_redirect(site) if screen == "priority" else f"/admin/{site}") # ------------------------------------------------- # Admin: Datei aus Playlist löschen # ------------------------------------------------- @app.route("/admin//delete//", methods=["POST"]) @site_access_required def delete_file(site, screen, filename): """Löscht eine Datei aus der Playlist (Config + Dateisystem).""" cfg = load_config() if screen == "priority": path = os.path.join(priority_dir(site), filename) playlist = get_priority_config(cfg, site).get("playlist", []) else: path = os.path.join(MEDIA_DIR, site, screen, filename) screens_cfg = cfg.get("sites", {}).get(site, {}).get("screens", {}) playlist = screens_cfg.get(screen, {}).get("playlist", []) # Ueber playlist_item_name vergleichen: Datei-Eintraege stehen als # {"name": ...}, URLs als {"url": ...}, Altbestand als reiner String. playlist[:] = [item for item in playlist if playlist_item_name(item) != filename] if os.path.exists(path): os.remove(path) save_config(cfg) return redirect(prio_redirect(site) if screen == "priority" else f"/admin/{site}") # ------------------------------------------------- # Admin: Playlist-Reihenfolge speichern (Drag & Drop) # ------------------------------------------------- @app.route("/admin//playlist/", methods=["POST"]) @site_access_required def save_playlist(site, screen): """ Speichert die über SortableJS per Drag & Drop neu sortierte Playlist. Der Client schickt ein JSON mit der neuen Reihenfolge und enabled-Status. """ cfg = load_config() data = request.get_json() new_order = data["playlist"] enabled_map = data.get("enabled", {}) if screen == "priority": old_playlist = get_priority_config(cfg, site).get("playlist", []) else: screens_cfg = cfg.get("sites", {}).get(site, {}).get("screens", {}) old_playlist = screens_cfg.get(screen, {}).get("playlist", []) # Mapping: Name → Original-Item (für Zoom etc.) item_map = {} for item in old_playlist: item_map[playlist_item_name(item)] = item # Neue Playlist in der gewünschten Reihenfolge bauen new_playlist = [] for key in new_order: if key in item_map: item = dict(item_map[key]) if isinstance(item_map[key], dict) else {"name": item_map[key]} item["enabled"] = enabled_map.get(key, True if not isinstance(item_map[key], dict) else item_map[key].get("enabled", True)) new_playlist.append(item) else: new_playlist.append({"url": key, "zoom": 1.0, "enabled": enabled_map.get(key, True)} if is_url(key) else {"name": key, "enabled": enabled_map.get(key, True)}) if screen == "priority": get_priority_config(cfg, site)["playlist"] = new_playlist else: cfg.setdefault("sites", {}).setdefault(site, {}).setdefault("screens", {}).setdefault(screen, {})["playlist"] = new_playlist save_config(cfg) return "", 204 # ------------------------------------------------- # Admin: Playlist-Item bearbeiten (URL + Zoom) # ------------------------------------------------- @app.route("/admin//update-playlist-item/", methods=["POST"]) @site_access_required def update_playlist_item(site, screen): """Aktualisiert URL, Zoom und Zeitsteuerung eines einzelnen Playlist-Items.""" data = request.get_json() old_name = data.get("old_name") new_url = (data.get("new_url") or "").strip() zoom = data.get("zoom", 1.0) passthrough = data.get("passthrough") schedule = data.get("schedule") # Kann None, {} oder schedule-Objekt sein # new_url ist nur bei URL-Einträgen gesetzt; bei Dateien wird ausschliesslich # die Zeitsteuerung bearbeitet und der Dateiname bleibt unverändert. is_url_item = bool(new_url) if is_url_item and not is_url(new_url): return jsonify({"error": "Ungültige URL"}), 400 try: zoom = float(zoom) if zoom <= 0: zoom = 1.0 except (ValueError, TypeError): zoom = 1.0 config = load_config() if screen == "priority": playlist = get_priority_config(config, site).setdefault("playlist", []) else: playlist = config["sites"][site]["screens"][screen].setdefault("playlist", []) for i, item in enumerate(playlist): name = playlist_item_name(item) if name == old_name: if is_url_item: updated = {"url": new_url, "zoom": zoom} else: updated = {"name": old_name} # enabled-Status des bestehenden Eintrags nicht verlieren if isinstance(item, dict) and "enabled" in item: updated["enabled"] = item["enabled"] # Passthrough beibehalten oder aktualisieren if passthrough is not None: updated["passthrough_params"] = bool(passthrough) else: existing = item.get("passthrough_params") if isinstance(item, dict) else None if existing is not None: updated["passthrough_params"] = existing # Zeitsteuerung beibehalten, entfernen oder aktualisieren if schedule is not None: # Schedule wurde explizit gesendet (kann {} für Entfernen sein) if schedule: # Nicht leer updated["schedule"] = schedule # Wenn schedule leer ist {}, wird es nicht hinzugefügt (entfernt) else: # Nicht im Request → bestehenden Wert beibehalten if isinstance(item, dict) and "schedule" in item: updated["schedule"] = item["schedule"] playlist[i] = updated save_config(config) return jsonify({"success": True}) return jsonify({"error": "Item nicht gefunden"}), 404 # ------------------------------------------------- # Change-Password (für First-Login / Reset) # ------------------------------------------------- @app.route("/change-password", methods=["GET", "POST"]) @login_required def change_password(): """Seite zum Ändern des Passworts (bei First-Login oder nach Admin-Reset).""" cfg = load_config() site_list = get_accessible_sites(cfg, current_user) current_site = site_list[0] if site_list else "" error = None success = None if request.method == "POST": new_pw = request.form.get("new_password", "") confirm = request.form.get("confirm_password", "") if not new_pw or len(new_pw) < 6: error = "Passwort muss mindestens 6 Zeichen lang sein" elif new_pw != confirm: error = "Passwörter stimmen nicht überein" else: users = load_users() email = current_user.email if email in users: users[email]["password_hash"] = generate_password_hash(new_pw) users[email]["must_change_password"] = False save_users(users) add_history_entry("password_changed", f"Passwort für User '{email}' geändert", current_user.email) return redirect("/admin") return render_template( "change_password.html", error=error, success=success, site_list=site_list, current_site=current_site, version=APP_VERSION, year=datetime.now().year, hostname=os.uname().nodename, server_url=cfg.get("server_url", "") ) # ------------------------------------------------- # User-Verwaltung (nur für Admins) # ------------------------------------------------- @app.route("/admin/dashboard") @admin_required def admin_dashboard(): """Admin-Dashboard mit Statistiken: Sites, Screens, User, Admins, Superuser.""" cfg = load_config() users_data = load_users() sites = get_site_list(cfg) screen_count = sum(len(site_cfg.get("screens", {})) for site_cfg in cfg.get("sites", {}).values()) user_count = len(users_data) admin_count = sum(1 for u in users_data.values() if u.get("role") == "admin") superuser_count = sum(1 for u in users_data.values() if u.get("role") == "superuser") history = get_history(30) # Sparkline-Daten aus dem Verlauf berechnen def build_sparkline(entries, creates, deletes): seq = list(reversed(entries)) data = [0] for e in seq: a = e.get("action") if a in creates: data.append(data[-1] + 1) elif a in deletes: data.append(data[-1] - 1) if len(data) < 3: data = [0, 0, 0] return data site_spark = build_sparkline(history, {"site_created"}, {"site_deleted"}) screen_spark = build_sparkline(history, {"screen_created"}, {"screen_deleted"}) user_spark = build_sparkline(history, {"user_created"}, {"user_deleted"}) admin_spark = build_sparkline(history, {"admin_created"}, {"admin_deleted"}) superuser_spark = build_sparkline(history, {"superuser_created"}, {"superuser_deleted"}) def trend_val(creates, deletes): c = sum(1 for e in history if e.get("action") in creates) d = sum(1 for e in history if e.get("action") in deletes) return c - d site_trend = trend_val({"site_created"}, {"site_deleted"}) screen_trend = trend_val({"screen_created"}, {"screen_deleted"}) user_trend = trend_val({"user_created"}, {"user_deleted"}) admin_trend = trend_val({"admin_created"}, {"admin_deleted"}) superuser_trend = trend_val({"superuser_created"}, {"superuser_deleted"}) return render_template( "admin_dashboard.html", site_count=len(sites), screen_count=screen_count, user_count=user_count, admin_count=admin_count, superuser_count=superuser_count, site_spark=site_spark, screen_spark=screen_spark, user_spark=user_spark, admin_spark=admin_spark, superuser_spark=superuser_spark, site_trend=site_trend, screen_trend=screen_trend, user_trend=user_trend, admin_trend=admin_trend, superuser_trend=superuser_trend, history=history, site_list=get_accessible_sites(cfg, current_user), current_site=current_site_for(cfg, current_user), version=APP_VERSION, year=datetime.now().year, hostname=os.uname().nodename, server_url=cfg.get("server_url", "") ) @app.route("/admin/help") @login_required def admin_help(): """Hilfeseite zur Bedienung des Tools.""" cfg = load_config() return render_template( "help.html", site_list=get_accessible_sites(cfg, current_user), current_site=current_site_for(cfg, current_user), version=APP_VERSION, year=datetime.now().year, hostname=os.uname().nodename, server_url=cfg.get("server_url", "") ) @app.route("/admin/users") @admin_required def admin_users(): """Listet alle User.""" cfg = load_config() users_data = load_users() users_list = [] for email, data in users_data.items(): users_list.append({ "email": email, "role": data.get("role", "user"), "sites": data.get("sites", []), "must_change_password": data.get("must_change_password", False), "first_name": data.get("first_name", ""), "last_name": data.get("last_name", ""), "department": data.get("department", ""), "notes": data.get("notes", ""), "mfa_enabled": data.get("mfa_enabled", False) }) return render_template( "user_list.html", users=sorted(users_list, key=lambda u: (u["last_name"].lower(), u["first_name"].lower())), site_list=get_accessible_sites(cfg, current_user), current_site=current_site_for(cfg, current_user), version=APP_VERSION, year=datetime.now().year, hostname=os.uname().nodename, server_url=cfg.get("server_url", "") ) @app.route("/admin/users/create", methods=["GET"]) @admin_required def admin_users_create_page(): """Zeigt das Formular zum Anlegen eines neuen Users.""" cfg = load_config() all_sites = get_site_list(cfg) return render_template( "user_create.html", all_sites=all_sites, site_list=get_accessible_sites(cfg, current_user), current_site=all_sites[0] if all_sites else "", version=APP_VERSION, year=datetime.now().year, hostname=os.uname().nodename, server_url=cfg.get("server_url", "") ) @app.route("/admin/users/create", methods=["POST"]) @admin_required def admin_users_create(): """Legt einen neuen User an (E-Mail=Username, Passwort, Rolle, Standorte).""" email = request.form.get("email", "").strip().lower() password = request.form.get("password", "") role = request.form.get("role", "user") sites = request.form.getlist("sites") # Validierung if not email or not re.match(r"[^@]+@[^@]+\.[^@]+", email): return "Ungültige E-Mail-Adresse", 400 if not password or len(password) < 6: return "Passwort muss mindestens 6 Zeichen lang sein", 400 if role not in ("admin", "superuser", "user"): return "Ungültige Rolle", 400 users = load_users() if email in users: return "User existiert bereits", 400 users[email] = { "password_hash": generate_password_hash(password), "role": role, "sites": sites if role == "user" else [], "must_change_password": True, "first_name": request.form.get("first_name", "").strip(), "last_name": request.form.get("last_name", "").strip(), "department": request.form.get("department", "").strip(), "notes": request.form.get("notes", "").strip() } save_users(users) add_history_entry("user_created", f"User '{email}' angelegt (Rolle: {role})", current_user.email) if role == "admin": add_history_entry("admin_created", f"Admin '{email}' angelegt", current_user.email) elif role == "superuser": add_history_entry("superuser_created", f"Superuser '{email}' angelegt", current_user.email) return redirect("/admin/users") @app.route("/admin/users/edit/", methods=["GET", "POST"]) @admin_required def admin_users_edit(email): """Bearbeitet einen User (E-Mail, Rolle, Standorte).""" cfg = load_config() all_sites = get_site_list(cfg) users_data = load_users() if email not in users_data: return "User nicht gefunden", 404 error = None if request.method == "POST": new_email = request.form.get("email", "").strip().lower() new_role = request.form.get("role", "user") new_sites = request.form.getlist("sites") if not new_email or not re.match(r"[^@]+@[^@]+\.[^@]+", new_email): error = "Ungültige E-Mail-Adresse" elif new_role not in ("admin", "superuser", "user"): error = "Ungültige Rolle" else: user_data = users_data.pop(email) old_role = user_data.get("role", "user") user_data["role"] = new_role user_data["sites"] = new_sites if new_role == "user" else [] user_data["first_name"] = request.form.get("first_name", "").strip() user_data["last_name"] = request.form.get("last_name", "").strip() user_data["department"] = request.form.get("department", "").strip() user_data["notes"] = request.form.get("notes", "").strip() users_data[new_email] = user_data save_users(users_data) add_history_entry("user_updated", f"User '{email}' bearbeitet (Rolle: {old_role} → {new_role})", current_user.email) if old_role == "admin" and new_role != "admin": add_history_entry("admin_deleted", f"Admin-Rechte von '{email}' entzogen", current_user.email) elif old_role != "admin" and new_role == "admin": add_history_entry("admin_created", f"'{email}' zum Admin ernannt", current_user.email) if old_role == "superuser" and new_role != "superuser": add_history_entry("superuser_deleted", f"Superuser-Rechte von '{email}' entzogen", current_user.email) elif old_role != "superuser" and new_role == "superuser": add_history_entry("superuser_created", f"'{email}' zum Superuser ernannt", current_user.email) return redirect(f"/admin/users?updated={new_email}") user = { "email": email, "role": users_data[email].get("role", "user"), "sites": users_data[email].get("sites", []), "must_change_password": users_data[email].get("must_change_password", False), "first_name": users_data[email].get("first_name", ""), "last_name": users_data[email].get("last_name", ""), "department": users_data[email].get("department", ""), "notes": users_data[email].get("notes", "") } return render_template( "user_edit.html", user=user, all_sites=all_sites, error=error, site_list=get_accessible_sites(cfg, current_user), current_site=all_sites[0] if all_sites else "", version=APP_VERSION, year=datetime.now().year, hostname=os.uname().nodename, server_url=cfg.get("server_url", "") ) @app.route("/admin/users/delete/", methods=["POST"]) @admin_required def admin_users_delete(email): """Löscht einen User. Der eigene Account kann nicht gelöscht werden. Der letzte verbleibende Admin kann nicht gelöscht werden.""" if email == current_user.email: return "Du kannst dich nicht selbst löschen", 400 users = load_users() role = users.get(email, {}).get("role", "user") # Prüfen, ob es der letzte Admin wäre if role == "admin": admin_count = sum(1 for u in users.values() if u.get("role") == "admin") if admin_count <= 1: return "Letzter Admin kann nicht gelöscht werden", 400 users.pop(email, None) save_users(users) add_history_entry("user_deleted", f"User '{email}' gelöscht", current_user.email) if role == "admin": add_history_entry("admin_deleted", f"Admin '{email}' gelöscht", current_user.email) elif role == "superuser": add_history_entry("superuser_deleted", f"Superuser '{email}' gelöscht", current_user.email) return redirect("/admin/users") @app.route("/admin/users/reset-password/", methods=["POST"]) @admin_required def admin_users_reset_password(email): """Setzt ein temporäres Passwort und zwingt User zur Änderung beim nächsten Login.""" import secrets import string users = load_users() if email not in users: return "User nicht gefunden", 404 temp_password = secrets.choice(string.ascii_lowercase) + secrets.choice(string.ascii_uppercase) + secrets.choice(string.digits) + secrets.token_hex(4) users[email]["password_hash"] = generate_password_hash(temp_password) users[email]["must_change_password"] = True users[email]["mfa_enabled"] = False users[email]["mfa_secret"] = "" users[email]["mfa_recovery_codes"] = [] users[email]["trusted_devices"] = [] save_users(users) add_history_entry("user_password_reset", f"Passwort für User '{email}' zurückgesetzt (MFA deaktiviert)", current_user.email) return redirect(f"/admin/users?reset={email}&temp={temp_password}") # ------------------------------------------------- # Main # ------------------------------------------------- try: init_user_db() except Exception as e: print(f"⚠️ init_user_db fehlgeschlagen: {e}") if __name__ == "__main__": app.run(debug=True, host="0.0.0.0", port=5005)