396 lines
16 KiB
Python
396 lines
16 KiB
Python
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 sqlalchemy import func
|
|
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,
|
|
sort_order=(db.session.query(func.max(QuickWin.sort_order)).scalar() or -1) + 1,
|
|
)
|
|
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.sort_order.asc(), 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/<int:badge_id>", 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/<int:quick_win_id>/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("/quick-wins/<int:quick_win_id>/update", methods=["POST"])
|
|
@login_required
|
|
def update_quick_win(quick_win_id: int):
|
|
quick_win = QuickWin.query.get_or_404(quick_win_id)
|
|
quick_task_config = get_quick_task_config()
|
|
|
|
title = (request.form.get("title") or "").strip()
|
|
effort = request.form.get("effort") or ""
|
|
|
|
if len(title) < 2:
|
|
flash("Quick-Wins brauchen einen Titel mit mindestens 2 Zeichen.", "error")
|
|
return redirect(url_for("settings.quick_wins"))
|
|
|
|
if len(title) > 160:
|
|
flash("Quick-Win-Titel dürfen maximal 160 Zeichen lang sein.", "error")
|
|
return redirect(url_for("settings.quick_wins"))
|
|
|
|
if effort not in quick_task_config:
|
|
flash("Bitte wähle einen gültigen Aufwand.", "error")
|
|
return redirect(url_for("settings.quick_wins"))
|
|
|
|
duplicate = (
|
|
QuickWin.query.filter(QuickWin.id != quick_win.id, QuickWin.title == title, QuickWin.active.is_(True))
|
|
.first()
|
|
)
|
|
if duplicate:
|
|
flash("Diesen Quick-Win gibt es bereits.", "error")
|
|
return redirect(url_for("settings.quick_wins"))
|
|
|
|
quick_win.title = title
|
|
quick_win.effort = effort
|
|
db.session.commit()
|
|
flash(f"Quick-Win „{quick_win.title}“ wurde aktualisiert.", "success")
|
|
return redirect(url_for("settings.quick_wins"))
|
|
|
|
|
|
@bp.route("/quick-wins/reorder", methods=["POST"])
|
|
@login_required
|
|
@csrf.exempt
|
|
def reorder_quick_wins():
|
|
payload = request.get_json(silent=True)
|
|
if payload is not None:
|
|
raw_ids = payload.get("ids", [])
|
|
else:
|
|
raw_ids = request.form.get("ids", "").split(",")
|
|
ordered_ids = [int(item) for item in raw_ids if str(item).isdigit()]
|
|
|
|
quick_wins = QuickWin.query.filter_by(active=True).all()
|
|
quick_wins_by_id = {quick_win.id: quick_win for quick_win in quick_wins}
|
|
|
|
for position, quick_win_id in enumerate(ordered_ids):
|
|
quick_win = quick_wins_by_id.get(quick_win_id)
|
|
if quick_win:
|
|
quick_win.sort_order = position
|
|
|
|
used_ids = set(ordered_ids)
|
|
remaining = [quick_win for quick_win in quick_wins if quick_win.id not in used_ids]
|
|
for offset, quick_win in enumerate(sorted(remaining, key=lambda item: (item.sort_order, item.id)), start=len(ordered_ids)):
|
|
quick_win.sort_order = offset
|
|
|
|
db.session.commit()
|
|
if payload is not None:
|
|
return jsonify({"ok": True})
|
|
|
|
flash("Die Quick-Win-Reihenfolge wurde gespeichert.", "success")
|
|
return redirect(url_for("settings.quick_wins"))
|
|
|
|
|
|
@bp.route("/users/<int:user_id>/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/<int:user_id>/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})
|
|
TaskTemplate.query.filter_by(default_assigned_user_secondary_id=user.id).update({"default_assigned_user_secondary_id": None})
|
|
TaskInstance.query.filter_by(assigned_user_id=user.id).update({"assigned_user_id": None})
|
|
TaskInstance.query.filter_by(assigned_user_secondary_id=user.id).update({"assigned_user_secondary_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})
|