Files
putzliga/app/routes/tasks.py

299 lines
11 KiB
Python

from __future__ import annotations
import calendar
from collections import defaultdict
from datetime import date
from flask import Blueprint, flash, redirect, render_template, request, url_for
from flask_login import current_user, login_required
from sqlalchemy import or_
from ..forms import QuickTaskForm, TaskForm
from ..models import QuickWin, TaskInstance, User
from ..services.app_settings import get_quick_task_config
from ..services.dates import month_label, today_local
from ..services.tasks import (
complete_task,
create_quick_task,
create_task_template_and_instance,
delete_task_instance,
refresh_task_statuses,
update_template_and_instance,
)
bp = Blueprint("tasks", __name__, url_prefix="")
def _user_choices() -> list[tuple[int, str]]:
return [(user.id, user.name) for user in User.query.order_by(User.name.asc()).all()]
def _secondary_user_choices() -> list[tuple[int, str]]:
return [(0, "Keine zweite Person")] + _user_choices()
@bp.route("/my-tasks")
@login_required
def my_tasks():
tasks = (
TaskInstance.query.filter(
or_(
TaskInstance.assigned_user_id == current_user.id,
TaskInstance.assigned_user_secondary_id == current_user.id,
)
)
.order_by(TaskInstance.completed_at.is_(None).desc(), TaskInstance.due_date.asc())
.all()
)
refresh_task_statuses(tasks)
sections = {
"open": [],
"due_today": [],
"due_tomorrow": [],
"due_day_after_tomorrow": [],
"overdue": [],
"completed": [],
}
for task in tasks:
sections[task.status].append(task)
completed_count = len(sections["completed"])
active_count = (
len(sections["open"])
+ len(sections["due_today"])
+ len(sections["due_tomorrow"])
+ len(sections["due_day_after_tomorrow"])
+ len(sections["overdue"])
)
completion_ratio = 0 if completed_count + active_count == 0 else round(completed_count / (completed_count + active_count) * 100)
return render_template(
"tasks/my_tasks.html",
sections=sections,
completion_ratio=completion_ratio,
today=today_local(),
)
@bp.route("/tasks")
@login_required
def all_tasks():
query = TaskInstance.query
status = request.args.get("status", "all")
mine = request.args.get("mine")
user_filter = request.args.get("user_id", type=int)
sort = request.args.get("sort", "due")
if mine == "1":
query = query.filter(
or_(
TaskInstance.assigned_user_id == current_user.id,
TaskInstance.assigned_user_secondary_id == current_user.id,
)
)
elif user_filter:
query = query.filter(
or_(
TaskInstance.assigned_user_id == user_filter,
TaskInstance.assigned_user_secondary_id == user_filter,
)
)
if sort == "points":
query = query.order_by(TaskInstance.points_awarded.desc(), TaskInstance.due_date.asc())
elif sort == "user":
query = query.order_by(TaskInstance.assigned_user_id.asc(), TaskInstance.due_date.asc())
else:
query = query.order_by(TaskInstance.completed_at.is_(None).desc(), TaskInstance.due_date.asc())
tasks = query.all()
refresh_task_statuses(tasks)
if status != "all":
if status == "soon":
tasks = [task for task in tasks if task.status in {"due_today", "due_tomorrow", "due_day_after_tomorrow"}]
else:
status_map = {
"completed": "completed",
"overdue": "overdue",
"open": "open",
"today": "due_today",
"tomorrow": "due_tomorrow",
"day_after_tomorrow": "due_day_after_tomorrow",
}
selected = status_map.get(status)
if selected:
tasks = [task for task in tasks if task.status == selected]
return render_template(
"tasks/all_tasks.html",
tasks=tasks,
users=User.query.order_by(User.name.asc()).all(),
filters={"status": status, "mine": mine, "user_id": user_filter, "sort": sort},
)
@bp.route("/tasks/new", methods=["GET", "POST"])
@login_required
def create():
form = TaskForm()
form.assigned_user_id.choices = _user_choices()
form.assigned_user_secondary_id.choices = _secondary_user_choices()
if request.method == "GET" and not form.due_date.data:
form.due_date.data = today_local()
if form.validate_on_submit():
task = create_task_template_and_instance(form)
flash(f"Aufgabe „{task.title}“ wurde angelegt.", "success")
return redirect(url_for("tasks.my_tasks"))
return render_template("tasks/task_form.html", form=form, mode="create", task=None)
@bp.route("/tasks/quick", methods=["POST"])
@login_required
def quick_create():
config = get_quick_task_config()
created_titles: list[str] = []
selected_ids = request.form.getlist("quick_win_ids")
if selected_ids:
quick_wins = QuickWin.query.filter(QuickWin.id.in_(selected_ids), QuickWin.active.is_(True)).order_by(QuickWin.id.asc()).all()
for quick_win in quick_wins:
task = create_quick_task(quick_win.title, quick_win.effort, current_user, description="Quick-Win")
complete_task(task, current_user.id)
created_titles.append(task.title)
if request.form.get("include_custom") == "1":
form = QuickTaskForm(prefix="quick")
form.effort.choices = [(key, values["label"]) for key, values in config.items()]
custom_title = (form.title.data or "").strip()
extra_errors: list[str] = []
if not custom_title:
extra_errors.append("Bitte gib für „Sonstiges“ einen Titel ein.")
if not form.effort.data or form.effort.data not in config:
extra_errors.append("Bitte wähle für „Sonstiges“ einen Aufwand aus.")
if not form.validate_on_submit() or extra_errors:
for field_errors in form.errors.values():
for error in field_errors:
flash(error, "error")
for error in extra_errors:
flash(error, "error")
return redirect(request.referrer or url_for("tasks.my_tasks"))
task = create_quick_task(custom_title, form.effort.data, current_user, description="Quick-Win")
complete_task(task, current_user.id)
created_titles.append(task.title)
if not created_titles:
flash("Bitte wähle mindestens einen Quick-Win aus.", "error")
return redirect(request.referrer or url_for("tasks.my_tasks"))
if len(created_titles) == 1:
flash(f"Quick-Win „{created_titles[0]}“ wurde als erledigt gespeichert.", "success")
else:
flash(f"{len(created_titles)} Quick-Wins wurden als erledigt gespeichert.", "success")
return redirect(request.referrer or url_for("tasks.my_tasks"))
@bp.route("/tasks/<int:task_id>/edit", methods=["GET", "POST"])
@login_required
def edit(task_id: int):
task = TaskInstance.query.get_or_404(task_id)
form = TaskForm(obj=task.task_template)
form.assigned_user_id.choices = _user_choices()
form.assigned_user_secondary_id.choices = _secondary_user_choices()
next_url = request.args.get("next") or request.form.get("next") or request.referrer or url_for("tasks.all_tasks")
if request.method == "GET":
form.title.data = task.title
form.description.data = task.description
form.default_points.data = task.task_template.default_points
form.assigned_user_id.data = task.assigned_user_id or _user_choices()[0][0]
form.assigned_user_secondary_id.data = task.assigned_user_secondary_id or 0
form.due_date.data = task.due_date
form.recurrence_interval_value.data = task.task_template.recurrence_interval_value or 1
form.recurrence_interval_unit.data = task.task_template.recurrence_interval_unit
form.active.data = task.task_template.active
if form.validate_on_submit():
update_template_and_instance(task, form)
flash("Aufgabe und Vorlage wurden aktualisiert.", "success")
return redirect(next_url)
return render_template("tasks/task_form.html", form=form, mode="edit", task=task, next_url=next_url)
@bp.route("/tasks/<int:task_id>/delete", methods=["POST"])
@login_required
def delete(task_id: int):
task = TaskInstance.query.get_or_404(task_id)
title = task.title
next_url = request.form.get("next") or url_for("tasks.all_tasks")
delete_task_instance(task)
flash(f"Aufgabe „{title}“ wurde gelöscht.", "success")
return redirect(next_url)
@bp.route("/tasks/<int:task_id>/complete", methods=["POST"])
@login_required
def complete(task_id: int):
task = TaskInstance.query.get_or_404(task_id)
choice = request.form.get("completed_for", "me")
if task.is_completed:
flash("Diese Aufgabe ist bereits erledigt.", "info")
return redirect(request.referrer or url_for("tasks.my_tasks"))
completed_by_id = current_user.id
allowed_ids = {current_user.id}
if task.assigned_user_id:
allowed_ids.add(task.assigned_user_id)
if task.assigned_user_secondary_id:
allowed_ids.add(task.assigned_user_secondary_id)
if choice != "me":
selected_user_id = request.form.get("completed_for", type=int)
if selected_user_id in allowed_ids:
completed_by_id = selected_user_id
complete_task(task, completed_by_id)
flash("Punkte verbucht. Gute Arbeit.", "success")
return redirect(request.referrer or url_for("tasks.my_tasks"))
@bp.route("/calendar")
@login_required
def calendar_view():
today = today_local()
year = request.args.get("year", type=int) or today.year
month = request.args.get("month", type=int) or today.month
view = request.args.get("view", "calendar")
tasks = TaskInstance.query.filter(
TaskInstance.due_date >= date(year, month, 1),
TaskInstance.due_date <= date(year, month, calendar.monthrange(year, month)[1]),
).order_by(TaskInstance.due_date.asc()).all()
refresh_task_statuses(tasks)
tasks_by_day: dict[int, list[TaskInstance]] = defaultdict(list)
for task in tasks:
tasks_by_day[task.due_date.day].append(task)
mobile_day_groups = [
{"day": day, "tasks": grouped_tasks}
for day, grouped_tasks in sorted(tasks_by_day.items(), key=lambda item: item[0])
]
month_calendar = calendar.Calendar(firstweekday=0).monthdayscalendar(year, month)
return render_template(
"tasks/calendar.html",
current_year=year,
current_month=month,
current_label=month_label(year, month),
month_calendar=month_calendar,
tasks_by_day=tasks_by_day,
mobile_day_groups=mobile_day_groups,
view=view,
tasks=tasks,
)