new file: .env.example
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
This commit is contained in:
112
routes/admin_routes.py
Normal file
112
routes/admin_routes.py
Normal file
@@ -0,0 +1,112 @@
|
||||
"""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"))
|
||||
Reference in New Issue
Block a user