"""Admin-Routen: Dashboard, Benutzerverwaltung (erstellen, löschen, Passwort zurücksetzen).""" from flask import Blueprint, render_template, request, redirect, url_for, flash, session from auth import admin_required, hash_password from db import get_connection from db_helpers import get_user_by_id admin_bp = Blueprint("admin", __name__, url_prefix="/admin") @admin_bp.route("/") @admin_required def dashboard(): with get_connection() as conn: with conn.cursor() as cur: cur.execute("SELECT COUNT(*) AS total FROM users") total_users = cur.fetchone()["total"] cur.execute("SELECT COUNT(*) AS total FROM notes") total_notes = cur.fetchone()["total"] cur.execute("SELECT COUNT(*) AS c FROM users WHERE is_admin = 1") total_admins = cur.fetchone()["c"] cur.execute( "SELECT username, email, is_admin, created_at FROM users " "ORDER BY created_at DESC LIMIT 5" ) recent = cur.fetchall() return render_template( "admin/dashboard.html", total_users=total_users, total_notes=total_notes, total_admins=total_admins, recent=recent, ) @admin_bp.route("/users", methods=["GET", "POST"]) @admin_required def users(): if request.method == "POST": username = request.form.get("username", "").strip() password = request.form.get("password", "") email = request.form.get("email", "").strip() is_admin = 1 if request.form.get("is_admin") else 0 if not username or not password: flash("Benutzername und Passwort sind Pflichtfelder.", "danger") else: try: salt, password_hash = hash_password(password) with get_connection() as conn: with conn.cursor() as cur: cur.execute( "INSERT INTO users (username, email, salt, " "password_hash, is_admin) VALUES (%s, %s, %s, %s, %s)", (username, email, salt, password_hash, is_admin), ) flash(f"Benutzer '{username}' angelegt.", "success") except Exception as exc: # Duplicate Key etc. flash(f"Benutzer konnte nicht angelegt werden: {exc}", "danger") return redirect(url_for("admin.users")) with get_connection() as conn: with conn.cursor() as cur: cur.execute( "SELECT u.*, (SELECT COUNT(*) FROM notes n WHERE n.user_id = u.id) " "AS note_count FROM users u ORDER BY u.username ASC" ) user_list = cur.fetchall() return render_template("admin/users.html", users=user_list) @admin_bp.route("/users//delete", methods=["POST"]) @admin_required def delete_user(user_id: int): # Admin darf sich nicht selbst löschen target = get_user_by_id(user_id) if not target: flash("Benutzer nicht gefunden.", "danger") return redirect(url_for("admin.users")) if target["id"] == session.get("user_id"): flash("Du kannst dich nicht selbst löschen.", "danger") return redirect(url_for("admin.users")) with get_connection() as conn: with conn.cursor() as cur: cur.execute("DELETE FROM users WHERE id = %s", (user_id,)) flash(f"Benutzer '{target['username']}' gelöscht.", "success") return redirect(url_for("admin.users")) @admin_bp.route("/users//reset-password", methods=["POST"]) @admin_required def reset_password(user_id: int): new_password = request.form.get("new_password", "") if not new_password: flash("Neues Passwort fehlt.", "danger") return redirect(url_for("admin.users")) target = get_user_by_id(user_id) if not target: flash("Benutzer nicht gefunden.", "danger") return redirect(url_for("admin.users")) salt, password_hash = hash_password(new_password) with get_connection() as conn: with conn.cursor() as cur: cur.execute( "UPDATE users SET salt = %s, password_hash = %s WHERE id = %s", (salt, password_hash, user_id), ) flash(f"Passwort von '{target['username']}' zurückgesetzt.", "success") return redirect(url_for("admin.users"))