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, QuickWinForm, SettingsProfileForm from ..models import BadgeDefinition, MonthlyScoreSnapshot, NotificationLog, PushSubscription, QuickWin, 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"), ("settings.quick_wins", "Quick-Wins", "plus"), ] 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(): current_user.ensure_calendar_feed_token() form = SettingsProfileForm(original_email=current_user.email, obj=current_user) admin_form = AdminUserForm(prefix="admin") 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, users=User.query.order_by(User.is_admin.desc(), User.name.asc()).all(), earned_badges=earned_badges_for_user(current_user.id), calendar_feed_url=url_for("main.calendar_feed", token=current_user.calendar_feed_token, _external=True), push_ready=push_enabled(), has_subscription=bool(subscriptions), settings_tabs=_settings_tabs(), active_settings_tab="settings.index", ) @bp.route("/quick-wins", methods=["GET", "POST"]) @login_required def quick_wins(): quick_win_form = QuickWinForm(prefix="quickwin") quick_task_config_form = QuickTaskConfigForm(prefix="quickconfig") quick_task_config = get_quick_task_config() quick_win_form.effort.choices = [(key, values["label"]) for key, values in quick_task_config.items()] 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"] quick_task_config_form.super_heavy_label.data = quick_task_config["super_heavy"]["label"] quick_task_config_form.super_heavy_points.data = quick_task_config["super_heavy"]["points"] if quick_win_form.validate_on_submit(): existing_quick_win = QuickWin.query.filter_by(title=quick_win_form.title.data.strip(), active=True).first() if existing_quick_win: flash("Diesen Quick-Win gibt es bereits.", "error") return redirect(url_for("settings.quick_wins")) quick_win = QuickWin( title=quick_win_form.title.data.strip(), effort=quick_win_form.effort.data, active=True, created_by_user_id=current_user.id, ) db.session.add(quick_win) db.session.commit() flash(f"Quick-Win „{quick_win.title}“ wurde gespeichert.", "success") return redirect(url_for("settings.quick_wins")) return render_template( "settings/quick_wins.html", quick_win_form=quick_win_form, quick_task_config_form=quick_task_config_form, quick_task_config=quick_task_config, quick_wins=QuickWin.query.filter_by(active=True).order_by(QuickWin.id.asc()).all(), settings_tabs=_settings_tabs(), active_settings_tab="settings.quick_wins", ) @bp.route("/calendar-feed/regenerate", methods=["POST"]) @login_required def regenerate_calendar_feed(): current_user.calendar_feed_token = None current_user.ensure_calendar_feed_token() db.session.commit() flash("Dein persönlicher Kalender-Link wurde neu erzeugt.", "success") return redirect(url_for("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) set_setting_str("quick_task_label_super_heavy", form.super_heavy_label.data) set_setting_int("quick_task_points_super_heavy", form.super_heavy_points.data) db.session.commit() flash("Quick-Win-Aufwand und Punkte wurden aktualisiert.", "success") return redirect(url_for("settings.quick_wins")) @bp.route("/quick-wins//delete", methods=["POST"]) @login_required def delete_quick_win(quick_win_id: int): quick_win = QuickWin.query.get_or_404(quick_win_id) if quick_win.created_by_user_id != current_user.id and not current_user.is_admin: flash("Diesen Quick-Win kannst du nicht entfernen.", "error") return redirect(url_for("settings.quick_wins")) quick_win.active = False db.session.commit() flash("Quick-Win wurde ausgeblendet.", "success") return redirect(url_for("settings.quick_wins")) @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})