335 lines
12 KiB
Python
335 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
from flask import Blueprint, flash, g, redirect, render_template, request, url_for
|
|
from werkzeug.security import generate_password_hash
|
|
|
|
from .auth import admin_required, can_remove_last_admin, url_with_scroll_position, validate_admin_user_form, wants_to_stay_on_form
|
|
from .constants import BUILDER_DESCRIPTIONS, BUILDER_OPTIONS, DEFAULT_CATEGORIES, DEFAULT_CATEGORY_BUILDERS, ROLE_LABELS
|
|
from .db import get_db
|
|
|
|
|
|
admin_bp = Blueprint("admin", __name__, url_prefix="/admin")
|
|
|
|
|
|
def get_household_user(user_id: int):
|
|
user = get_db().execute(
|
|
"""
|
|
SELECT users.*, households.name AS household_name
|
|
FROM users
|
|
LEFT JOIN households ON households.id = users.household_id
|
|
WHERE users.id = ? AND users.household_id = ?
|
|
""",
|
|
(user_id, g.user["household_id"]),
|
|
).fetchone()
|
|
if user is None:
|
|
raise ValueError("Der Nutzer wurde nicht gefunden.")
|
|
return user
|
|
|
|
|
|
def fetch_household_categories():
|
|
return get_db().execute(
|
|
"""
|
|
SELECT *
|
|
FROM household_categories
|
|
WHERE household_id = ?
|
|
ORDER BY is_active DESC, sort_order, LOWER(name)
|
|
""",
|
|
(g.user["household_id"],),
|
|
).fetchall()
|
|
|
|
|
|
@admin_bp.get("/users")
|
|
@admin_required
|
|
def user_list():
|
|
users = get_db().execute(
|
|
"""
|
|
SELECT *
|
|
FROM users
|
|
WHERE household_id = ?
|
|
ORDER BY is_active DESC, LOWER(COALESCE(display_name, username))
|
|
""",
|
|
(g.user["household_id"],),
|
|
).fetchall()
|
|
return render_template("admin/users_list.html", users=users, role_labels=ROLE_LABELS)
|
|
|
|
|
|
@admin_bp.route("/users/new", methods=("GET", "POST"))
|
|
@admin_required
|
|
def user_create():
|
|
form_data = {
|
|
"display_name": "",
|
|
"username": "",
|
|
"email": "",
|
|
"role": "member",
|
|
"is_active": True,
|
|
}
|
|
|
|
if request.method == "POST":
|
|
database = get_db()
|
|
form_data = {
|
|
"display_name": request.form.get("display_name", "").strip(),
|
|
"username": request.form.get("username", "").strip().lower(),
|
|
"email": request.form.get("email", "").strip().lower(),
|
|
"role": request.form.get("role", "member").strip(),
|
|
"is_active": request.form.get("is_active", "1") == "1",
|
|
}
|
|
password = request.form.get("password", "")
|
|
password_repeat = request.form.get("password_repeat", "")
|
|
|
|
error = validate_admin_user_form(
|
|
database,
|
|
username=form_data["username"],
|
|
email=form_data["email"] or None,
|
|
role=form_data["role"],
|
|
is_active=form_data["is_active"],
|
|
password=password,
|
|
password_repeat=password_repeat,
|
|
)
|
|
|
|
if error is None:
|
|
database.execute(
|
|
"""
|
|
INSERT INTO users (household_id, username, email, display_name, role, is_active, password_hash)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
g.user["household_id"],
|
|
form_data["username"],
|
|
form_data["email"] or None,
|
|
form_data["display_name"],
|
|
form_data["role"],
|
|
1 if form_data["is_active"] else 0,
|
|
generate_password_hash(password),
|
|
),
|
|
)
|
|
database.commit()
|
|
flash("Der Nutzer wurde angelegt.", "success")
|
|
if wants_to_stay_on_form():
|
|
new_user = database.execute(
|
|
"""
|
|
SELECT id
|
|
FROM users
|
|
WHERE household_id = ? AND username = ?
|
|
ORDER BY id DESC
|
|
LIMIT 1
|
|
""",
|
|
(g.user["household_id"], form_data["username"]),
|
|
).fetchone()
|
|
if new_user is not None:
|
|
return redirect(url_with_scroll_position(url_for("admin.user_edit", user_id=int(new_user["id"]))))
|
|
return redirect(url_for("admin.user_list"))
|
|
|
|
flash(error, "error")
|
|
|
|
return render_template("admin/user_form.html", user=None, form_data=form_data, role_labels=ROLE_LABELS)
|
|
|
|
|
|
@admin_bp.route("/users/<int:user_id>/edit", methods=("GET", "POST"))
|
|
@admin_required
|
|
def user_edit(user_id: int):
|
|
try:
|
|
user = get_household_user(user_id)
|
|
except ValueError as exc:
|
|
flash(str(exc), "error")
|
|
return redirect(url_for("admin.user_list"))
|
|
|
|
form_data = {
|
|
"display_name": user["display_name"] or "",
|
|
"username": user["username"],
|
|
"email": user["email"] or "",
|
|
"role": user["role"],
|
|
"is_active": bool(user["is_active"]),
|
|
}
|
|
|
|
if request.method == "POST":
|
|
database = get_db()
|
|
form_data = {
|
|
"display_name": request.form.get("display_name", "").strip(),
|
|
"username": request.form.get("username", "").strip().lower(),
|
|
"email": request.form.get("email", "").strip().lower(),
|
|
"role": request.form.get("role", "member").strip(),
|
|
"is_active": request.form.get("is_active", "0") == "1",
|
|
}
|
|
password = request.form.get("password", "")
|
|
password_repeat = request.form.get("password_repeat", "")
|
|
|
|
error = validate_admin_user_form(
|
|
database,
|
|
username=form_data["username"],
|
|
email=form_data["email"] or None,
|
|
role=form_data["role"],
|
|
is_active=form_data["is_active"],
|
|
password=password,
|
|
password_repeat=password_repeat,
|
|
current_user_id=user_id,
|
|
)
|
|
if error is None and can_remove_last_admin(user_id, form_data["role"], form_data["is_active"]):
|
|
error = "Mindestens ein aktiver Admin sollte im Haushalt bleiben."
|
|
|
|
if error is None:
|
|
database.execute(
|
|
"""
|
|
UPDATE users
|
|
SET username = ?,
|
|
email = ?,
|
|
display_name = ?,
|
|
role = ?,
|
|
is_active = ?,
|
|
updated_at = CURRENT_TIMESTAMP
|
|
WHERE id = ?
|
|
""",
|
|
(
|
|
form_data["username"],
|
|
form_data["email"] or None,
|
|
form_data["display_name"],
|
|
form_data["role"],
|
|
1 if form_data["is_active"] else 0,
|
|
user_id,
|
|
),
|
|
)
|
|
if password:
|
|
database.execute(
|
|
"""
|
|
UPDATE users
|
|
SET password_hash = ?, updated_at = CURRENT_TIMESTAMP
|
|
WHERE id = ?
|
|
""",
|
|
(generate_password_hash(password), user_id),
|
|
)
|
|
database.commit()
|
|
flash("Der Nutzer wurde aktualisiert.", "success")
|
|
if wants_to_stay_on_form():
|
|
return redirect(url_with_scroll_position(url_for("admin.user_edit", user_id=user_id)))
|
|
return redirect(url_for("admin.user_list"))
|
|
|
|
flash(error, "error")
|
|
|
|
return render_template("admin/user_form.html", user=user, form_data=form_data, role_labels=ROLE_LABELS)
|
|
|
|
|
|
@admin_bp.route("/categories", methods=("GET", "POST"))
|
|
@admin_required
|
|
def category_settings():
|
|
if request.method == "POST":
|
|
name = request.form.get("name", "").strip()
|
|
builder_key = request.form.get("builder_key", "neutral").strip()
|
|
if not name:
|
|
flash("Bitte einen Kategorienamen eintragen.", "error")
|
|
else:
|
|
existing = get_db().execute(
|
|
"""
|
|
SELECT id
|
|
FROM household_categories
|
|
WHERE household_id = ? AND LOWER(name) = LOWER(?)
|
|
""",
|
|
(g.user["household_id"], name),
|
|
).fetchone()
|
|
if existing:
|
|
get_db().execute(
|
|
"UPDATE household_categories SET is_active = 1, builder_key = ? WHERE id = ?",
|
|
(builder_key, existing["id"]),
|
|
)
|
|
flash("Die Kategorie ist wieder aktiv.", "success")
|
|
else:
|
|
sort_row = get_db().execute(
|
|
"SELECT COALESCE(MAX(sort_order), 0) AS max_sort FROM household_categories WHERE household_id = ?",
|
|
(g.user["household_id"],),
|
|
).fetchone()
|
|
get_db().execute(
|
|
"""
|
|
INSERT INTO household_categories (household_id, name, builder_key, sort_order, is_active)
|
|
VALUES (?, ?, ?, ?, 1)
|
|
""",
|
|
(g.user["household_id"], name, builder_key, int(sort_row["max_sort"]) + 10),
|
|
)
|
|
flash("Die Kategorie wurde ergänzt.", "success")
|
|
get_db().commit()
|
|
return redirect(url_for("admin.category_settings"))
|
|
|
|
return render_template(
|
|
"admin/categories.html",
|
|
categories=fetch_household_categories(),
|
|
default_categories=DEFAULT_CATEGORIES,
|
|
default_category_builders=DEFAULT_CATEGORY_BUILDERS,
|
|
builder_options=BUILDER_OPTIONS,
|
|
builder_descriptions=BUILDER_DESCRIPTIONS,
|
|
)
|
|
|
|
|
|
@admin_bp.post("/categories/<int:category_id>/toggle")
|
|
@admin_required
|
|
def category_toggle(category_id: int):
|
|
category = get_db().execute(
|
|
"""
|
|
SELECT *
|
|
FROM household_categories
|
|
WHERE id = ? AND household_id = ?
|
|
""",
|
|
(category_id, g.user["household_id"]),
|
|
).fetchone()
|
|
if category is None:
|
|
flash("Die Kategorie wurde nicht gefunden.", "error")
|
|
return redirect(url_for("admin.category_settings"))
|
|
|
|
new_state = 0 if category["is_active"] else 1
|
|
get_db().execute(
|
|
"UPDATE household_categories SET is_active = ? WHERE id = ?",
|
|
(new_state, category_id),
|
|
)
|
|
get_db().commit()
|
|
flash("Die Kategorie wurde aktualisiert.", "success")
|
|
return redirect(url_for("admin.category_settings"))
|
|
|
|
|
|
@admin_bp.post("/categories/<int:category_id>/update")
|
|
@admin_required
|
|
def category_update(category_id: int):
|
|
category = get_db().execute(
|
|
"""
|
|
SELECT *
|
|
FROM household_categories
|
|
WHERE id = ? AND household_id = ?
|
|
""",
|
|
(category_id, g.user["household_id"]),
|
|
).fetchone()
|
|
if category is None:
|
|
flash("Die Kategorie wurde nicht gefunden.", "error")
|
|
return redirect(url_for("admin.category_settings"))
|
|
|
|
builder_key = request.form.get("builder_key", "neutral").strip()
|
|
get_db().execute(
|
|
"UPDATE household_categories SET builder_key = ? WHERE id = ?",
|
|
(builder_key, category_id),
|
|
)
|
|
get_db().commit()
|
|
flash("Die Zuordnung wurde aktualisiert.", "success")
|
|
return redirect(url_for("admin.category_settings"))
|
|
|
|
|
|
@admin_bp.post("/categories/<int:category_id>/delete")
|
|
@admin_required
|
|
def category_delete(category_id: int):
|
|
category = get_db().execute(
|
|
"""
|
|
SELECT *
|
|
FROM household_categories
|
|
WHERE id = ? AND household_id = ?
|
|
""",
|
|
(category_id, g.user["household_id"]),
|
|
).fetchone()
|
|
if category is None:
|
|
flash("Die Kategorie wurde nicht gefunden.", "error")
|
|
return redirect(url_for("admin.category_settings"))
|
|
|
|
if category["name"] in DEFAULT_CATEGORIES:
|
|
flash("Standardkategorien bleiben erhalten. Du kannst sie bei Bedarf pausieren.", "info")
|
|
return redirect(url_for("admin.category_settings"))
|
|
|
|
get_db().execute(
|
|
"DELETE FROM household_categories WHERE id = ? AND household_id = ?",
|
|
(category_id, g.user["household_id"]),
|
|
)
|
|
get_db().commit()
|
|
flash("Die Kategorie wurde entfernt.", "success")
|
|
return redirect(url_for("admin.category_settings"))
|