396 lines
14 KiB
Python
396 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import calendar
|
|
from collections import defaultdict
|
|
from datetime import date, timedelta
|
|
|
|
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()
|
|
|
|
|
|
def _my_tasks_soon_priority(task: TaskInstance) -> int:
|
|
order = {
|
|
"due_tomorrow": 0,
|
|
"due_day_after_tomorrow": 1,
|
|
"open": 2,
|
|
}
|
|
return order.get(task.status, 99)
|
|
|
|
|
|
def _task_sort_key(task: TaskInstance, sort: str) -> tuple:
|
|
if sort == "points":
|
|
return (-task.points_awarded, task.due_date, task.title.lower())
|
|
if sort == "user":
|
|
return (task.assignee_label.lower(), task.due_date, task.title.lower())
|
|
return (task.due_date, task.title.lower())
|
|
|
|
|
|
def _group_active_tasks(tasks: list[TaskInstance], *, sort: str = "due") -> dict[str, list[TaskInstance]]:
|
|
sections = {
|
|
"overdue": [],
|
|
"today": [],
|
|
"soon": [],
|
|
"open": [],
|
|
}
|
|
|
|
for task in tasks:
|
|
if task.is_completed:
|
|
continue
|
|
if task.status == "overdue":
|
|
sections["overdue"].append(task)
|
|
elif task.status == "due_today":
|
|
sections["today"].append(task)
|
|
elif task.status in {"due_tomorrow", "due_day_after_tomorrow"}:
|
|
sections["soon"].append(task)
|
|
else:
|
|
sections["open"].append(task)
|
|
|
|
sections["overdue"].sort(key=lambda task: _task_sort_key(task, sort))
|
|
sections["today"].sort(key=lambda task: _task_sort_key(task, sort))
|
|
sections["soon"].sort(
|
|
key=lambda task: (_my_tasks_soon_priority(task),) + _task_sort_key(task, sort)
|
|
)
|
|
sections["open"].sort(key=lambda task: _task_sort_key(task, sort))
|
|
return sections
|
|
|
|
|
|
def _archive_day_priority(day: date, today: date) -> tuple[int, int]:
|
|
if day == today:
|
|
return (0, 0)
|
|
if day == today - timedelta(days=1):
|
|
return (1, 0)
|
|
if day == today - timedelta(days=2):
|
|
return (2, 0)
|
|
return (3, -day.toordinal())
|
|
|
|
|
|
def _archive_day_label(day: date, today: date) -> str:
|
|
if day == today:
|
|
return "Heute"
|
|
if day == today - timedelta(days=1):
|
|
return "Gestern"
|
|
if day == today - timedelta(days=2):
|
|
return "Vorgestern"
|
|
return day.strftime("%d.%m.%Y")
|
|
|
|
|
|
@bp.route("/my-tasks")
|
|
@login_required
|
|
def my_tasks():
|
|
all_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(all_tasks)
|
|
sections = _group_active_tasks(all_tasks)
|
|
|
|
completed_count = len([task for task in all_tasks if task.is_completed])
|
|
active_count = (
|
|
len(sections["overdue"])
|
|
+ len(sections["today"])
|
|
+ len(sections["soon"])
|
|
+ len(sections["open"])
|
|
)
|
|
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_tomorrow", "due_day_after_tomorrow"}]
|
|
else:
|
|
status_map = {
|
|
"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]
|
|
tasks = [task for task in tasks if not task.is_completed]
|
|
else:
|
|
tasks = [task for task in tasks if not task.is_completed]
|
|
|
|
sections = _group_active_tasks(tasks, sort=sort)
|
|
|
|
return render_template(
|
|
"tasks/all_tasks.html",
|
|
tasks=tasks,
|
|
sections=sections,
|
|
users=User.query.order_by(User.name.asc()).all(),
|
|
filters={"status": status, "mine": mine, "user_id": user_filter, "sort": sort},
|
|
)
|
|
|
|
|
|
@bp.route("/archive")
|
|
@login_required
|
|
def archive_view():
|
|
selected_user_id = request.args.get("user_id", type=int) or current_user.id
|
|
archive_users = User.query.order_by(User.name.asc()).all()
|
|
selected_user = next((user for user in archive_users if user.id == selected_user_id), current_user)
|
|
ordered_users = sorted(archive_users, key=lambda user: (user.id != current_user.id, user.name.lower()))
|
|
|
|
completed_tasks = (
|
|
TaskInstance.query.filter_by(completed_by_user_id=selected_user.id)
|
|
.filter(TaskInstance.completed_at.isnot(None))
|
|
.order_by(TaskInstance.completed_at.desc(), TaskInstance.updated_at.desc())
|
|
.all()
|
|
)
|
|
|
|
today = today_local()
|
|
grouped: dict[date, list[TaskInstance]] = defaultdict(list)
|
|
for task in completed_tasks:
|
|
grouped[task.completed_at.date()].append(task)
|
|
|
|
archive_sections = [
|
|
{
|
|
"label": _archive_day_label(day, today),
|
|
"day": day,
|
|
"tasks": grouped[day],
|
|
}
|
|
for day in sorted(grouped.keys(), key=lambda value: _archive_day_priority(value, today))
|
|
]
|
|
|
|
return render_template(
|
|
"tasks/archive.html",
|
|
archive_users=ordered_users,
|
|
selected_user=selected_user,
|
|
archive_sections=archive_sections,
|
|
)
|
|
|
|
|
|
@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,
|
|
)
|