292 lines
10 KiB
Python
292 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
from flask import Blueprint, flash, redirect, render_template, request, url_for
|
|
from flask_login import login_required
|
|
from sqlalchemy import select
|
|
|
|
from app.extensions import db
|
|
from app.models import (
|
|
Account,
|
|
Category,
|
|
CostParticipant,
|
|
Entry,
|
|
EntryShareRule,
|
|
MonthlyEntryValue,
|
|
Month,
|
|
NotificationPreference,
|
|
User,
|
|
)
|
|
from app.seed import slugify
|
|
from app.utils.uploads import save_avatar_upload
|
|
from app.utils.decorators import admin_required
|
|
|
|
admin_bp = Blueprint("admin", __name__, url_prefix="/admin")
|
|
|
|
|
|
def _resolve_avatar_url(existing: str | None = None) -> str | None:
|
|
upload = request.files.get("avatar_file")
|
|
if upload and upload.filename:
|
|
try:
|
|
return save_avatar_upload(upload)
|
|
except ValueError as exc:
|
|
flash(str(exc), "danger")
|
|
return existing
|
|
avatar_url = request.form.get("avatar_url")
|
|
if avatar_url is not None:
|
|
avatar_url = avatar_url.strip()
|
|
return avatar_url or existing
|
|
return existing
|
|
|
|
|
|
@admin_bp.route("/")
|
|
@login_required
|
|
@admin_required
|
|
def index():
|
|
users = User.query.order_by(User.display_name.asc(), User.username.asc()).all()
|
|
participants = CostParticipant.query.order_by(CostParticipant.name.asc()).all()
|
|
accounts = db.session.scalars(select(Account).order_by(Account.sort_order.asc(), Account.name.asc())).all()
|
|
categories = db.session.scalars(
|
|
select(Category).order_by(Category.account_id.asc(), Category.sort_order.asc(), Category.name.asc())
|
|
).all()
|
|
entries = db.session.scalars(
|
|
select(Entry).order_by(Entry.category_id.asc(), Entry.sort_order.asc(), Entry.name.asc())
|
|
).all()
|
|
return render_template(
|
|
"admin/index.html",
|
|
users=users,
|
|
participants=participants,
|
|
accounts=accounts,
|
|
categories=categories,
|
|
entries=entries,
|
|
)
|
|
|
|
|
|
@admin_bp.route("/users", methods=["POST"])
|
|
@login_required
|
|
@admin_required
|
|
def create_user():
|
|
user = User(
|
|
username=request.form["username"].strip(),
|
|
display_name=request.form["display_name"].strip(),
|
|
email=request.form["email"].strip(),
|
|
avatar_url=_resolve_avatar_url(),
|
|
role=request.form.get("role", "editor"),
|
|
is_active=True,
|
|
)
|
|
user.set_password(request.form["password"])
|
|
db.session.add(user)
|
|
db.session.flush()
|
|
db.session.add(NotificationPreference(user_id=user.id))
|
|
db.session.commit()
|
|
flash("Benutzer angelegt.", "success")
|
|
return redirect(url_for("admin.index"))
|
|
|
|
|
|
@admin_bp.route("/users/<int:user_id>", methods=["POST"])
|
|
@login_required
|
|
@admin_required
|
|
def update_user(user_id: int):
|
|
user = User.query.get_or_404(user_id)
|
|
user.display_name = request.form["display_name"].strip()
|
|
user.email = request.form["email"].strip()
|
|
user.avatar_url = _resolve_avatar_url(user.avatar_url)
|
|
user.role = request.form.get("role", user.role)
|
|
user.is_active = request.form.get("is_active") == "on"
|
|
db.session.commit()
|
|
flash("Benutzer aktualisiert.", "success")
|
|
return redirect(url_for("admin.index"))
|
|
|
|
|
|
@admin_bp.route("/users/<int:user_id>/toggle", methods=["POST"])
|
|
@login_required
|
|
@admin_required
|
|
def toggle_user(user_id: int):
|
|
user = User.query.get_or_404(user_id)
|
|
user.is_active = not user.is_active
|
|
db.session.commit()
|
|
flash("Benutzerstatus aktualisiert.", "info")
|
|
return redirect(url_for("admin.index"))
|
|
|
|
|
|
@admin_bp.route("/participants", methods=["POST"])
|
|
@login_required
|
|
@admin_required
|
|
def create_participant():
|
|
participant = CostParticipant(
|
|
name=request.form["name"].strip(),
|
|
avatar_url=_resolve_avatar_url(),
|
|
is_external=request.form.get("is_external") == "on",
|
|
is_app_user=bool(request.form.get("linked_user_id")),
|
|
linked_user_id=int(request.form["linked_user_id"]) if request.form.get("linked_user_id") else None,
|
|
is_active=True,
|
|
)
|
|
db.session.add(participant)
|
|
db.session.commit()
|
|
flash("Beteiligte Person angelegt.", "success")
|
|
return redirect(url_for("admin.index"))
|
|
|
|
|
|
@admin_bp.route("/participants/<int:participant_id>", methods=["POST"])
|
|
@login_required
|
|
@admin_required
|
|
def update_participant(participant_id: int):
|
|
participant = CostParticipant.query.get_or_404(participant_id)
|
|
participant.name = request.form["name"].strip()
|
|
participant.avatar_url = _resolve_avatar_url(participant.avatar_url)
|
|
participant.is_external = request.form.get("is_external") == "on"
|
|
participant.linked_user_id = (
|
|
int(request.form["linked_user_id"]) if request.form.get("linked_user_id") else None
|
|
)
|
|
participant.is_app_user = participant.linked_user_id is not None
|
|
participant.is_active = request.form.get("is_active") == "on"
|
|
db.session.commit()
|
|
flash("Beteiligte Person aktualisiert.", "success")
|
|
return redirect(url_for("admin.index"))
|
|
|
|
|
|
@admin_bp.route("/accounts", methods=["POST"])
|
|
@login_required
|
|
@admin_required
|
|
def create_account():
|
|
name = request.form["name"].strip()
|
|
account = Account(
|
|
name=name,
|
|
slug=slugify(request.form.get("slug", "") or name),
|
|
description=request.form.get("description", "").strip() or None,
|
|
sort_order=int(request.form.get("sort_order") or 0),
|
|
is_active=request.form.get("is_active") == "on",
|
|
)
|
|
db.session.add(account)
|
|
db.session.commit()
|
|
flash("Konto angelegt.", "success")
|
|
return redirect(url_for("admin.index"))
|
|
|
|
|
|
@admin_bp.route("/accounts/<int:account_id>", methods=["POST"])
|
|
@login_required
|
|
@admin_required
|
|
def update_account(account_id: int):
|
|
account = Account.query.get_or_404(account_id)
|
|
name = request.form["name"].strip()
|
|
account.name = name
|
|
account.slug = slugify(request.form.get("slug", "") or name)
|
|
account.description = request.form.get("description", "").strip() or None
|
|
account.sort_order = int(request.form.get("sort_order") or 0)
|
|
account.is_active = request.form.get("is_active") == "on"
|
|
db.session.commit()
|
|
flash("Konto aktualisiert.", "success")
|
|
return redirect(url_for("admin.index"))
|
|
|
|
|
|
@admin_bp.route("/categories", methods=["POST"])
|
|
@login_required
|
|
@admin_required
|
|
def create_category():
|
|
name = request.form["name"].strip()
|
|
category = Category(
|
|
account_id=int(request.form["account_id"]),
|
|
name=name,
|
|
slug=slugify(request.form.get("slug", "") or name),
|
|
description=request.form.get("description", "").strip() or None,
|
|
sort_order=int(request.form.get("sort_order") or 0),
|
|
is_active=request.form.get("is_active") == "on",
|
|
)
|
|
db.session.add(category)
|
|
db.session.commit()
|
|
flash("Kategorie angelegt.", "success")
|
|
return redirect(url_for("admin.index"))
|
|
|
|
|
|
@admin_bp.route("/categories/<int:category_id>", methods=["POST"])
|
|
@login_required
|
|
@admin_required
|
|
def update_category(category_id: int):
|
|
category = Category.query.get_or_404(category_id)
|
|
name = request.form["name"].strip()
|
|
category.account_id = int(request.form["account_id"])
|
|
category.name = name
|
|
category.slug = slugify(request.form.get("slug", "") or name)
|
|
category.description = request.form.get("description", "").strip() or None
|
|
category.sort_order = int(request.form.get("sort_order") or 0)
|
|
category.is_active = request.form.get("is_active") == "on"
|
|
db.session.commit()
|
|
flash("Kategorie aktualisiert.", "success")
|
|
return redirect(url_for("admin.index"))
|
|
|
|
|
|
@admin_bp.route("/entries", methods=["POST"])
|
|
@login_required
|
|
@admin_required
|
|
def create_entry():
|
|
name = request.form["name"].strip()
|
|
entry = Entry(
|
|
category_id=int(request.form["category_id"]),
|
|
name=name,
|
|
slug=slugify(request.form.get("slug", "") or name),
|
|
description=request.form.get("description", "").strip() or None,
|
|
default_amount=request.form.get("default_amount", "0"),
|
|
amount_type=request.form.get("amount_type", "fixed"),
|
|
sort_order=int(request.form.get("sort_order") or 0),
|
|
is_active=request.form.get("is_active") == "on",
|
|
)
|
|
db.session.add(entry)
|
|
db.session.flush()
|
|
for month in Month.query.order_by(Month.year.asc(), Month.month.asc()).all():
|
|
db.session.add(
|
|
MonthlyEntryValue(
|
|
month_id=month.id,
|
|
entry_id=entry.id,
|
|
planned_amount=entry.default_amount,
|
|
)
|
|
)
|
|
db.session.commit()
|
|
flash("Eintrag angelegt und in vorhandene Monate übernommen.", "success")
|
|
return redirect(url_for("admin.index"))
|
|
|
|
|
|
@admin_bp.route("/entries/<int:entry_id>", methods=["POST"])
|
|
@login_required
|
|
@admin_required
|
|
def update_entry(entry_id: int):
|
|
entry = Entry.query.get_or_404(entry_id)
|
|
name = request.form["name"].strip()
|
|
entry.category_id = int(request.form["category_id"])
|
|
entry.name = name
|
|
entry.slug = slugify(request.form.get("slug", "") or name)
|
|
entry.description = request.form.get("description", "").strip() or None
|
|
entry.default_amount = request.form.get("default_amount", "0")
|
|
entry.amount_type = request.form.get("amount_type", "fixed")
|
|
entry.sort_order = int(request.form.get("sort_order") or 0)
|
|
entry.is_active = request.form.get("is_active") == "on"
|
|
db.session.commit()
|
|
flash("Eintrag aktualisiert.", "success")
|
|
return redirect(url_for("admin.index"))
|
|
|
|
|
|
@admin_bp.route("/entries/<int:entry_id>/share-rules", methods=["POST"])
|
|
@login_required
|
|
@admin_required
|
|
def create_share_rule(entry_id: int):
|
|
participant_id = int(request.form["participant_id"])
|
|
rule = EntryShareRule.query.filter_by(entry_id=entry_id, participant_id=participant_id).first()
|
|
if rule is None:
|
|
rule = EntryShareRule(entry_id=entry_id, participant_id=participant_id)
|
|
db.session.add(rule)
|
|
rule.share_type = request.form.get("share_type", "equal")
|
|
share_value = request.form.get("share_value", "").strip()
|
|
rule.share_value = share_value or None
|
|
db.session.commit()
|
|
flash("Beteiligungsregel gespeichert.", "success")
|
|
return redirect(url_for("admin.index"))
|
|
|
|
|
|
@admin_bp.route("/share-rules/<int:rule_id>/delete", methods=["POST"])
|
|
@login_required
|
|
@admin_required
|
|
def delete_share_rule(rule_id: int):
|
|
rule = EntryShareRule.query.get_or_404(rule_id)
|
|
db.session.delete(rule)
|
|
db.session.commit()
|
|
flash("Beteiligungsregel entfernt.", "info")
|
|
return redirect(url_for("admin.index"))
|