22104b54cd
- Neue Config-Felder: passthrough_params (true/false) pro URL-Item - Edit-Modal in admin.html + priority.html mit Checkbox - Player checkt item.passthrough === true vor appendParams() - Default: false (keine Parameter-Übergabe ohne Checkbox) - Link-Icon (ti-link / ti-link-off) in Playlist-Zeile als Status-Indikator - Doku: help.html, README.md, AGENTS.md
1989 lines
74 KiB
Python
Executable File
1989 lines
74 KiB
Python
Executable File
#!/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 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
|
||
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.1.0"
|
||
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)
|
||
|
||
@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 = get_site_list(cfg)[0] if get_site_list(cfg) else ""
|
||
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 playlist_item_enabled(item):
|
||
"""Prüft, ob ein Playlist-Eintrag aktiviert ist.
|
||
Strings sind immer aktiv, Dicts haben ein optionales "enabled"-Feld (default: True)."""
|
||
if isinstance(item, (str,)):
|
||
return True
|
||
if isinstance(item, dict):
|
||
return item.get("enabled", True)
|
||
return True
|
||
|
||
|
||
# -------------------------------------------------
|
||
# Priority-Playlist-Helfer
|
||
# -------------------------------------------------
|
||
def load_priority_files():
|
||
"""Lädt die Priority-Playlist 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 = os.path.join(MEDIA_DIR, "priority")
|
||
os.makedirs(prio_dir, exist_ok=True)
|
||
|
||
priority_files = []
|
||
prio_playlist = cfg.get("priority", {}).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": playlist_item_enabled(item),
|
||
"passthrough": item.get("passthrough_params", False) if isinstance(item, dict) else False
|
||
})
|
||
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()
|
||
ftype = "video" if ext == ".mp4" else "image"
|
||
size = os.path.getsize(file_path) // 1024
|
||
|
||
priority_files.append({
|
||
"name": item_name,
|
||
"type": ftype,
|
||
"size": size,
|
||
"enabled": playlist_item_enabled(item)
|
||
})
|
||
|
||
# 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
|
||
})
|
||
|
||
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/<site>/background/<path:filename>")
|
||
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)
|
||
|
||
|
||
# -------------------------------------------------
|
||
# 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()
|
||
# Abgelaufene Einträge bereinigen
|
||
active = [d for d in trusted
|
||
if d.get("token") != trust_token or
|
||
datetime.fromisoformat(d["expires"]) > now]
|
||
changed = len(active) != len(trusted)
|
||
for d in trusted:
|
||
if d.get("token") == trust_token:
|
||
try:
|
||
expires = datetime.fromisoformat(d["expires"])
|
||
if expires > now:
|
||
# 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")
|
||
else:
|
||
pass
|
||
except (ValueError, KeyError):
|
||
pass
|
||
if changed:
|
||
users_data[email]["trusted_devices"] = active
|
||
save_users(users_data)
|
||
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=False)
|
||
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=site_list[0] if site_list else "",
|
||
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=site_list[0] if site_list else "",
|
||
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=site_list[0] if site_list else "",
|
||
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 der Lobby-Playlist vorne platziert
|
||
"""
|
||
error = None
|
||
success = None
|
||
logo_urls = [None, None, None]
|
||
customer_names = ["", "", ""]
|
||
contacts = ["", "", ""]
|
||
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)
|
||
|
||
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:
|
||
found_logos = []
|
||
for name in filled_names:
|
||
logo = generate_welcome_page.search_customer_logo(name)
|
||
found_logos.append(logo)
|
||
|
||
valid = [(n, l) for n, l in zip(filled_names, found_logos) if l]
|
||
|
||
if not valid:
|
||
error = "Kein Logo konnte gefunden werden"
|
||
else:
|
||
valid_names = [n for n, l in valid]
|
||
valid_urls = [l for n, l in valid]
|
||
valid_contacts = []
|
||
for name in valid_names:
|
||
for i, cn in enumerate(customer_names):
|
||
if cn == name:
|
||
valid_contacts.append(contacts[i] if i < len(contacts) else "")
|
||
break
|
||
else:
|
||
valid_contacts.append("")
|
||
for i, name in enumerate(customer_names):
|
||
for j, (vn, vu) in enumerate(zip(valid_names, valid_urls)):
|
||
if vn == name:
|
||
logo_urls[i] = vu
|
||
break
|
||
|
||
bg_url = get_background_url(site)
|
||
html_filename = generate_welcome_page.save_welcome_page(valid_names, valid_urls, contacts=valid_contacts, site=site, background_url=bg_url)
|
||
|
||
if html_filename:
|
||
if generate_welcome_page.add_customer_to_lobby_playlist(html_filename, site=site):
|
||
# Letzte Kundendaten für Background-Update speichern
|
||
config = load_config()
|
||
site_cfg = config.setdefault("sites", {}).setdefault(site, {})
|
||
site_cfg["welcome_data"] = {"names": valid_names, "logo_urls": valid_urls, "contacts": valid_contacts}
|
||
save_config(config)
|
||
add_history_entry("customer_added", f"Kunden ({', '.join(valid_names)}) zur Willkommensseite an Standort '{site}' hinzugefügt", current_user.email)
|
||
success = f"✅ Willkommensseite mit {len(valid_names)} Kunden erstellt!"
|
||
else:
|
||
error = "Fehler beim Hinzufügen zur Playliste"
|
||
else:
|
||
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,
|
||
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()
|
||
|
||
html_filename = generate_welcome_page.save_welcome_page([customer_name], [logo_url], contacts=[contact], site=site)
|
||
|
||
if not html_filename:
|
||
return jsonify({"error": "Failed to create welcome page"}), 500
|
||
|
||
if generate_welcome_page.add_customer_to_lobby_playlist(html_filename, site=site):
|
||
return jsonify({
|
||
"success": True,
|
||
"message": f"Customer '{customer_name}' added successfully",
|
||
"welcome_page": html_filename,
|
||
"logo_url": logo_url,
|
||
"site": site
|
||
}), 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/<path:filename>")
|
||
def media_priority(filename):
|
||
"""Liefert globale Priority-Medien aus media/priority/ aus."""
|
||
path = os.path.join(MEDIA_DIR, "priority")
|
||
return send_from_directory(path, filename)
|
||
|
||
|
||
@app.route("/media/<site>/<screen>/<path:filename>")
|
||
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/<site>/<screen>")
|
||
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 playlist_item_enabled(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": True, "passthrough": item.get("passthrough_params", False) if isinstance(item, dict) else False})
|
||
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 = config.get("priority", {})
|
||
|
||
if priority_cfg.get("enabled", False):
|
||
prio_files = []
|
||
for item in priority_cfg.get("playlist", []):
|
||
if not playlist_item_enabled(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": True, "passthrough": item.get("passthrough_params", False) if isinstance(item, dict) else False})
|
||
continue
|
||
|
||
file_path = os.path.join(MEDIA_DIR, "priority", 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)],
|
||
prio_files=[item for item in prio_files if item.get("enabled", 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)
|
||
)
|
||
|
||
|
||
# -------------------------------------------------
|
||
# Hash für automatisches Player-Reload
|
||
# ─────────────────────────────────────────────────
|
||
# Der Player pollt alle 5s diese Route.
|
||
# Wenn sich der Hash ändert, wird neu geladen.
|
||
# -------------------------------------------------
|
||
@app.route("/playlist/<site>/<screen>/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()
|
||
|
||
relevant = {
|
||
"sites": {
|
||
site: config.get("sites", {}).get(site, {})
|
||
},
|
||
"priority": config.get("priority", {})
|
||
}
|
||
|
||
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 "<h1>Kein Standort konfiguriert</h1><p>Bitte lege einen Standort im media-Verzeichnis an.</p>", 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/<site>/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/<site>", 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).
|
||
"""
|
||
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
|
||
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("._"):
|
||
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)
|
||
|
||
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": playlist_item_enabled(item),
|
||
"passthrough": item.get("passthrough_params", False) if isinstance(item, dict) else False
|
||
})
|
||
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"
|
||
if ext in (".jpg", ".jpeg", ".png"):
|
||
ftype = "image"
|
||
if ext == ".html":
|
||
ftype = "html"
|
||
|
||
size = os.path.getsize(file_path) // 1024
|
||
|
||
files.append({
|
||
"name": item_name,
|
||
"type": ftype,
|
||
"size": size,
|
||
"enabled": playlist_item_enabled(item)
|
||
})
|
||
|
||
# 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
|
||
})
|
||
|
||
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)
|
||
|
||
return render_template(
|
||
"admin.html",
|
||
site_list=site_list,
|
||
current_site=current_site,
|
||
screens=screens,
|
||
media_files=media_files,
|
||
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/<site>/priority")
|
||
@site_access_required
|
||
def admin_priority(site):
|
||
"""Zeigt die Priority-Playlist-Seite mit Tabs (Playlist + Medien)."""
|
||
cfg = load_config()
|
||
site_list = get_accessible_sites(cfg, current_user)
|
||
priority_files = load_priority_files()
|
||
|
||
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/<site>/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/<site>/background.<ext>."""
|
||
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/<site>/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/<site>/update/<screen>", 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
|
||
|
||
save_config(config)
|
||
return redirect(f"/admin/{site}")
|
||
|
||
|
||
@app.route("/admin/<site>/update-actions/<screen>", 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/<site>/update-voice/<screen>", 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/<site>/delete-screen/<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/<site>/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/<site>/upload/<screen>", 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":
|
||
file.save(os.path.join(MEDIA_DIR, "priority", 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/<site>/add-url/<screen>", methods=["POST"])
|
||
@site_access_required
|
||
def add_url(site, screen):
|
||
"""Fügt eine URL (mit Zoom-Faktor) 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
|
||
|
||
url_data = {"url": url, "zoom": zoom_factor}
|
||
|
||
config = load_config()
|
||
if screen == "priority":
|
||
config.setdefault("priority", {}).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/<site>/delete/<screen>/<path:filename>", 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(MEDIA_DIR, "priority", filename)
|
||
playlist = cfg.get("priority", {}).get("playlist", [])
|
||
playlist[:] = [item for item in playlist if not (
|
||
(isinstance(item, str) and item == filename) or
|
||
(isinstance(item, dict) and item.get("url") == filename)
|
||
)]
|
||
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", [])
|
||
playlist[:] = [item for item in playlist if not (
|
||
(isinstance(item, str) and item == filename) or
|
||
(isinstance(item, dict) and item.get("url") == 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/<site>/playlist/<screen>", 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 = cfg.get("priority", {}).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":
|
||
cfg.setdefault("priority", {})["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/<site>/update-playlist-item/<screen>", methods=["POST"])
|
||
@site_access_required
|
||
def update_playlist_item(site, screen):
|
||
"""Aktualisiert URL und Zoom eines einzelnen Playlist-Items."""
|
||
data = request.get_json()
|
||
old_name = data.get("old_name")
|
||
new_url = data.get("new_url", "").strip()
|
||
zoom = data.get("zoom", 1.0)
|
||
|
||
if 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 = config.setdefault("priority", {}).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:
|
||
updated = {"url": new_url, "zoom": zoom}
|
||
passthrough = data.get("passthrough", None)
|
||
if passthrough is not None:
|
||
updated["passthrough_params"] = bool(passthrough)
|
||
else:
|
||
# Bestehenden Wert übernehmen falls vorhanden
|
||
existing = item.get("passthrough_params") if isinstance(item, dict) else None
|
||
if existing is not None:
|
||
updated["passthrough_params"] = existing
|
||
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=get_site_list(cfg)[0] if get_site_list(cfg) else "",
|
||
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=get_site_list(cfg)[0] if get_site_list(cfg) else "",
|
||
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=get_site_list(cfg)[0] if get_site_list(cfg) else "",
|
||
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/<path:email>", 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/<path:email>", 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/<path:email>", 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)
|