from __future__ import annotations from pathlib import Path from uuid import uuid4 from flask import Blueprint, current_app, flash, jsonify, redirect, render_template, request, url_for from flask_login import current_user, login_required from werkzeug.utils import secure_filename from ..extensions import csrf, db from ..forms import AdminUserForm, QuickTaskConfigForm, SettingsProfileForm from ..models import BadgeDefinition, MonthlyScoreSnapshot, NotificationLog, PushSubscription, TaskInstance, TaskTemplate, User from ..services.app_settings import get_quick_task_config, set_setting_int, set_setting_str from ..services.badges import earned_badges_for_user from ..services.notifications import push_enabled bp = Blueprint("settings", __name__, url_prefix="/settings") def _require_admin(): if not current_user.is_admin: flash("Dieser Bereich ist nur für Admins verfügbar.", "error") return False return True def _settings_tabs(): tabs = [("settings.index", "Profil & Team", "gear")] if current_user.is_admin: tabs.append(("settings.badges", "Badges", "award")) return tabs def _save_avatar(file_storage) -> str: filename = secure_filename(file_storage.filename or "") ext = Path(filename).suffix.lower() or ".png" relative_path = Path("avatars") / f"{uuid4().hex}{ext}" absolute_path = Path(current_app.config["UPLOAD_FOLDER"]) / relative_path absolute_path.parent.mkdir(parents=True, exist_ok=True) file_storage.save(absolute_path) return relative_path.as_posix() @bp.route("", methods=["GET", "POST"]) @login_required def index(): form = SettingsProfileForm(original_email=current_user.email, obj=current_user) admin_form = AdminUserForm(prefix="admin") quick_task_config_form = QuickTaskConfigForm(prefix="quickconfig") quick_task_config = get_quick_task_config() if request.method == "GET": quick_task_config_form.fast_label.data = quick_task_config["fast"]["label"] quick_task_config_form.fast_points.data = quick_task_config["fast"]["points"] quick_task_config_form.normal_label.data = quick_task_config["normal"]["label"] quick_task_config_form.normal_points.data = quick_task_config["normal"]["points"] quick_task_config_form.medium_label.data = quick_task_config["medium"]["label"] quick_task_config_form.medium_points.data = quick_task_config["medium"]["points"] quick_task_config_form.heavy_label.data = quick_task_config["heavy"]["label"] quick_task_config_form.heavy_points.data = quick_task_config["heavy"]["points"] if form.validate_on_submit(): current_user.name = form.name.data.strip() current_user.email = form.email.data.lower().strip() current_user.notification_task_due_enabled = form.notification_task_due_enabled.data current_user.notification_monthly_winner_enabled = form.notification_monthly_winner_enabled.data if form.password.data: current_user.set_password(form.password.data) if form.avatar.data: current_user.avatar_path = _save_avatar(form.avatar.data) db.session.commit() flash("Deine Einstellungen wurden gespeichert.", "success") return redirect(url_for("settings.index")) subscriptions = PushSubscription.query.filter_by(user_id=current_user.id).all() return render_template( "settings/index.html", form=form, admin_form=admin_form, quick_task_config_form=quick_task_config_form, quick_task_config=quick_task_config, users=User.query.order_by(User.is_admin.desc(), User.name.asc()).all(), earned_badges=earned_badges_for_user(current_user.id), push_ready=push_enabled(), has_subscription=bool(subscriptions), settings_tabs=_settings_tabs(), active_settings_tab="settings.index", ) @bp.route("/badges") @login_required def badges(): if not _require_admin(): return redirect(url_for("settings.index")) badges = BadgeDefinition.query.order_by(BadgeDefinition.name.asc()).all() return render_template( "settings/badges.html", badges=badges, settings_tabs=_settings_tabs(), active_settings_tab="settings.badges", ) @bp.route("/badges/", methods=["POST"]) @login_required def update_badge(badge_id: int): if not _require_admin(): return redirect(url_for("settings.index")) badge = BadgeDefinition.query.get_or_404(badge_id) badge.threshold = max(1, request.form.get("threshold", type=int, default=badge.threshold)) badge.bonus_points = max(0, request.form.get("bonus_points", type=int, default=badge.bonus_points)) badge.active = request.form.get("active") == "on" db.session.commit() flash(f"Badge „{badge.name}“ wurde aktualisiert.", "success") return redirect(url_for("settings.badges")) @bp.route("/users", methods=["POST"]) @login_required def create_user(): if not _require_admin(): return redirect(url_for("settings.index")) form = AdminUserForm(prefix="admin") if not form.validate_on_submit(): for field_errors in form.errors.values(): for error in field_errors: flash(error, "error") return redirect(url_for("settings.index")) user = User( name=form.name.data.strip(), email=form.email.data.lower().strip(), is_admin=form.is_admin.data, ) user.set_password(form.password.data) db.session.add(user) db.session.commit() flash(f"Nutzer „{user.name}“ wurde angelegt.", "success") return redirect(url_for("settings.index")) @bp.route("/quick-task-config", methods=["POST"]) @login_required def update_quick_task_config(): if not _require_admin(): return redirect(url_for("settings.index")) form = QuickTaskConfigForm(prefix="quickconfig") if not form.validate_on_submit(): for field_errors in form.errors.values(): for error in field_errors: flash(error, "error") return redirect(url_for("settings.index")) set_setting_str("quick_task_label_fast", form.fast_label.data) set_setting_int("quick_task_points_fast", form.fast_points.data) set_setting_str("quick_task_label_normal", form.normal_label.data) set_setting_int("quick_task_points_normal", form.normal_points.data) set_setting_str("quick_task_label_medium", form.medium_label.data) set_setting_int("quick_task_points_medium", form.medium_points.data) set_setting_str("quick_task_label_heavy", form.heavy_label.data) set_setting_int("quick_task_points_heavy", form.heavy_points.data) db.session.commit() flash("Schnellaufgaben-Aufwand und Punkte wurden aktualisiert.", "success") return redirect(url_for("settings.index")) @bp.route("/users//toggle-admin", methods=["POST"]) @login_required def toggle_admin(user_id: int): if not _require_admin(): return redirect(url_for("settings.index")) user = User.query.get_or_404(user_id) make_admin = request.form.get("make_admin") == "1" if user.id == current_user.id and not make_admin: flash("Du kannst dir die Admin-Rechte nicht selbst entziehen.", "error") return redirect(url_for("settings.index")) if not make_admin and User.query.filter_by(is_admin=True).count() <= 1: flash("Mindestens ein Admin muss erhalten bleiben.", "error") return redirect(url_for("settings.index")) user.is_admin = make_admin db.session.commit() flash(f"Admin-Status für „{user.name}“ wurde aktualisiert.", "success") return redirect(url_for("settings.index")) @bp.route("/users//delete", methods=["POST"]) @login_required def delete_user(user_id: int): if not _require_admin(): return redirect(url_for("settings.index")) user = User.query.get_or_404(user_id) if user.id == current_user.id: flash("Du kannst deinen aktuell eingeloggten Account nicht löschen.", "error") return redirect(url_for("settings.index")) if user.is_admin and User.query.filter_by(is_admin=True).count() <= 1: flash("Der letzte Admin kann nicht gelöscht werden.", "error") return redirect(url_for("settings.index")) TaskTemplate.query.filter_by(default_assigned_user_id=user.id).update({"default_assigned_user_id": None}) TaskInstance.query.filter_by(assigned_user_id=user.id).update({"assigned_user_id": None}) TaskInstance.query.filter_by(completed_by_user_id=user.id).update({"completed_by_user_id": None}) MonthlyScoreSnapshot.query.filter_by(user_id=user.id).delete() NotificationLog.query.filter_by(user_id=user.id).delete() PushSubscription.query.filter_by(user_id=user.id).delete() db.session.delete(user) db.session.commit() flash("Nutzer wurde entfernt.", "success") return redirect(url_for("settings.index")) @bp.route("/push/subscribe", methods=["POST"]) @login_required @csrf.exempt def push_subscribe(): if not push_enabled(): return jsonify({"ok": False, "message": "VAPID ist nicht konfiguriert."}), 400 data = request.get_json(silent=True) or {} endpoint = data.get("endpoint") keys = data.get("keys", {}) if not endpoint or not keys.get("p256dh") or not keys.get("auth"): return jsonify({"ok": False, "message": "Subscription unvollständig."}), 400 subscription = PushSubscription.query.filter_by(endpoint=endpoint).first() if not subscription: subscription = PushSubscription(user_id=current_user.id, endpoint=endpoint, p256dh=keys["p256dh"], auth=keys["auth"]) db.session.add(subscription) else: subscription.user_id = current_user.id subscription.p256dh = keys["p256dh"] subscription.auth = keys["auth"] db.session.commit() return jsonify({"ok": True}) @bp.route("/push/unsubscribe", methods=["POST"]) @login_required @csrf.exempt def push_unsubscribe(): data = request.get_json(silent=True) or {} endpoint = data.get("endpoint") if endpoint: subscription = PushSubscription.query.filter_by(endpoint=endpoint, user_id=current_user.id).first() if subscription: db.session.delete(subscription) db.session.commit() return jsonify({"ok": True})