import datetime from datetime import timedelta from django.db.models import Sum, Q, QuerySet, Subquery, OuterRef, DurationField, Value, DecimalField from django.db.models.functions import Coalesce from django.http import JsonResponse from django.shortcuts import render, get_object_or_404, redirect from django.views import generic from django.views.decorators.csrf import csrf_exempt from django.views.decorators.http import require_POST from rest_framework.status import HTTP_400_BAD_REQUEST from dispatcher.core import import_tasks, convert_timedelta, str_to_timedelta from dispatcher.forms import ProjectForm, ImportForm, ParentForm from dispatcher.models import Project, Parent, Clocking, Task, RealSageXHours from dispatcher.serializers import ClockingSerializer, RealSageXHoursSerializer def dashboard_view(request): context = { "projects": Project.objects.all(), "parents": Parent.objects.all() } return render(request, "dashboard.html", context) def projects_view(request): context = { "class_name": "parent", "projects": Project.objects.all(), "parents": Parent.objects.all() } return render(request, "projects.html", context) def parent_view(request, id): parent = get_object_or_404(Parent, id=id) if request.method == "DELETE": parent.delete() return JsonResponse({"status": "success"}) context = { "class": "parent", "id": id } form = ParentForm(request.POST or None, request.FILES or None, instance=parent) if form.is_valid(): form.save() return redirect("parents") context["form"] = ParentForm(instance=parent) return render(request, "edit.html", context) def parent_on_delete_view(request, id): parent = get_object_or_404(Parent, id=id) projects = parent.project_set.count() tasks = Task.objects.filter(project__parent=parent).count() return JsonResponse({ "status": "success", "projects": projects, "tasks": tasks }) def project_on_delete_view(request, id): project = get_object_or_404(Project, id=id) tasks = project.task_set.count() return JsonResponse({ "status": "success", "tasks": tasks }) def new_parent_view(request): context = { "class": "parent" } form = ParentForm(request.POST or None, request.FILES or None) if form.is_valid(): form.save() return redirect("parents") context["form"] = form return render(request, "add.html", context) def project_view(request, id): project = get_object_or_404(Project, id=id) if request.method == "DELETE": project.delete() return JsonResponse({"status": "success"}) context = { "class": "project", "id": id } form = ProjectForm(request.POST or None, request.FILES or None, instance=project) if form.is_valid(): form.save() context["form"] = ProjectForm(instance=project) return render(request, "edit.html", context) def new_project_view(request): context = { "class": "project" } form = ProjectForm(request.POST or None, request.FILES or None) if form.is_valid(): form.save() return redirect("projects") context["form"] = form return render(request, "add.html", context) def table_view(request): return render(request, "table.html") @require_POST def set_parent(request, id): project = get_object_or_404(Project, id=id) parent_id = request.POST.get("parent_id") or None try: parent = Parent.objects.get(id=parent_id) except Parent.DoesNotExist: parent = None project.parent = parent project.save() return JsonResponse({"status": "success"}) def import_view(request): if request.method == "POST": form = ImportForm(request.POST, request.FILES) if form.is_valid(): print(request.FILES) import_tasks(request.FILES["file"].read().decode("utf-8")) return redirect("dashboard") return render(request, "import.html") def get_stats_by_month(request, year: int, month: int): if month < 1 or month > 12: return JsonResponse({"status": "error", "error": f"Invalid month {month}"}) sagex_subquery = RealSageXHours.objects.filter( parent=OuterRef("pk"), date__year=year, date__month=month ).values("hours")[:1] parents = Parent.objects.annotate( total_duration=Sum( "project__task__duration", filter=Q( project__task__date__year=year, project__task__date__month=month ) ) ).annotate( real_sagex_hours=Coalesce( Subquery(sagex_subquery, output_field=DurationField()), timedelta(0) ) ) projects = Project.objects.annotate( total_duration=Sum( "task__duration", filter=Q( task__date__year=year, task__date__month=month ) ) ) clockings = Clocking.objects.filter( date__year=year, date__month=month ) return JsonResponse({"status": "success", "data": format_stats(parents, projects, clockings)}) def format_stats(parents: QuerySet[Parent], projects: QuerySet[Project], clockings: QuerySet[Clocking]): data = { "parents": [ { "id": parent.id, "name": parent.name, "project_num": parent.project_num, "is_productive": parent.is_productive, "duration": parent.total_duration, "real_sagex_hours": convert_timedelta(parent.real_sagex_hours) } for parent in parents ], "projects": [ { "id": project.id, "name": project.name, "duration": project.total_duration } for project in projects ], "clockings": ClockingSerializer(clockings, many=True).data } Clocking.objects.filter() return data class ParentsView(generic.ListView): model = Parent template_name = "parents.html" context_object_name = "elements" class ProjectsView(generic.ListView): model = Project template_name = "projects.html" context_object_name = "elements" ordering = ["parent"] def get_context_data(self, **kwargs): context = super().get_context_data(**kwargs) context["parents"] = Parent.objects.all() return context def get_table_data(request, start_date: datetime.date, end_date: datetime.date): end_date = end_date + timedelta(days=1) clockings = Clocking.objects.filter( date__gte=start_date, date__lte=end_date ) parents = Parent.objects.all().order_by("id") data = { "parents": [ { "id": parent.id, "name": parent.name, "project_num": parent.project_num, "is_productive": parent.is_productive, "projects": [ { "id": project.id, "name": project.name, "tasks": [ { "date": task["date"], "duration": task["duration"] } for task in project.task_set.filter( date__gte=start_date, date__lt=end_date ).values("date").order_by("date").annotate(duration=Sum("duration")) ] } for project in parent.project_set.order_by("id") ] } for parent in parents ], "clockings": ClockingSerializer(clockings, many=True).data } return JsonResponse({"status": "success", "data": data}) @require_POST @csrf_exempt def set_clocking(request, date: datetime.date): clocking, created = Clocking.objects.get_or_create(date=date) clocking.in_am = request.POST.get("in_am", clocking.in_am) or None clocking.out_am = request.POST.get("out_am", clocking.out_am) or None clocking.in_pm = request.POST.get("in_pm", clocking.in_pm) or None clocking.out_pm = request.POST.get("out_pm", clocking.out_pm) or None remote = request.POST.get("remote", clocking.remote) or None if remote is not None: remote = str_to_timedelta(remote) else: remote = timedelta() clocking.remote = remote clocking.save() clocking.refresh_from_db() return JsonResponse({ "status": "success", "clocking": ClockingSerializer(clocking).data }) @require_POST def set_real_sagex(request, id, month: datetime.date): parent = get_object_or_404(Parent, id=id) minutes = request.POST.get("minutes") if minutes is None: return JsonResponse({ "status": "error", "error": "Missing minutes field" }, status=HTTP_400_BAD_REQUEST) try: minutes = int(minutes) except ValueError: return JsonResponse({ "status": "error", "error": "Invalid value for minutes, must be an int" }, status=HTTP_400_BAD_REQUEST) hours, minutes = divmod(minutes, 60) entry, created = RealSageXHours.objects.get_or_create(parent=parent, date=month) entry.hours = timedelta(hours=int(hours), minutes=int(minutes)) entry.save() entry.refresh_from_db() return JsonResponse({ "status": "success", "sagex": RealSageXHoursSerializer(entry).data })