0e7efe4b74
new file: .gitignore new file: Dockerfile new file: README.md new file: app.py new file: auth.py new file: config.py new file: db.py new file: db_helpers.py new file: docker-compose.yml new file: requirements.txt new file: routes/__init__.py new file: routes/admin_routes.py new file: routes/auth_routes.py new file: routes/journal_routes.py new file: static/css/style.css new file: static/js/main.js new file: templates/admin/dashboard.html new file: templates/admin/users.html new file: templates/base.html new file: templates/day.html new file: templates/future.html new file: templates/index.html new file: templates/login.html new file: templates/week.html
113 lines
4.3 KiB
Python
113 lines
4.3 KiB
Python
"""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/<int:user_id>/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/<int:user_id>/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"))
|