"""Time Tracking API views. Endpoints --------- GET /time-entries/ list entries (filterable) POST /time-entries/ create entry GET /time-entries/my/weekly/ current user's week entries GET /time-entries/{public_id}/ retrieve entry PATCH /time-entries/{public_id}/ update entry DELETE /time-entries/{public_id}/ delete entry POST /time-entries/bulk-approve/ bulk approve entries GET /projects/{id}/time-report/ project time report (via project router) """ from __future__ import annotations from datetime import date, timedelta from decimal import Decimal, InvalidOperation from django.shortcuts import get_object_or_404 from rest_framework import status from rest_framework.decorators import api_view, permission_classes from rest_framework.permissions import IsAuthenticated from rest_framework.request import Request from rest_framework.response import Response from simorgh.apps.time_tracking import selectors as sel from simorgh.apps.time_tracking import services from simorgh.apps.time_tracking.models import TimeEntry # --------------------------------------------------------------------------- # Tenant helpers # --------------------------------------------------------------------------- def _get_tenant_id(request: Request) -> int | None: tenant = getattr(request, "tenant", None) return tenant.pk if tenant else None def _get_org_node_id(request: Request) -> int | None: tenant = getattr(request, "tenant", None) if tenant is None: return None try: from simorgh.apps.organizations.models import OrganizationNode node = OrganizationNode.objects.filter(tenant=tenant).order_by("lft").first() return node.pk if node else None except Exception: return None # --------------------------------------------------------------------------- # Serializer helpers (dict-based, no DRF serializer class) # --------------------------------------------------------------------------- def _serialize_entry(entry: TimeEntry) -> dict: return { "id": entry.pk, "public_id": str(entry.public_id), "user_id": entry.user_id, "user_email": getattr(entry, "_user_email", None) or (entry.user.email if entry.user_id else None), "project_id": entry.project_id, "project_name": entry.project.name if hasattr(entry, "_project_cache") or entry.project_id else None, "task_id": entry.task_id, "task_title": entry.task.title if entry.task_id and hasattr(entry, "task") and entry.task else None, "date": str(entry.date), "hours": str(entry.hours), "description": entry.description, "is_billable": entry.is_billable, "is_approved": entry.is_approved, "approved_by_id": entry.approved_by_id, "approved_at": entry.approved_at.isoformat() if entry.approved_at else None, "created_at": entry.created_at.isoformat(), "updated_at": entry.updated_at.isoformat(), } def _serialize_entry_list(entry: TimeEntry) -> dict: """Lighter version for list views using select_related data.""" return { "id": entry.pk, "public_id": str(entry.public_id), "user_id": entry.user_id, "project_id": entry.project_id, "task_id": entry.task_id, "date": str(entry.date), "hours": str(entry.hours), "description": entry.description, "is_billable": entry.is_billable, "is_approved": entry.is_approved, } # --------------------------------------------------------------------------- # Entry list / create # --------------------------------------------------------------------------- @api_view(["GET", "POST"]) @permission_classes([IsAuthenticated]) def entry_list_create(request: Request) -> Response: tenant_id = _get_tenant_id(request) if not tenant_id: return Response({"detail": "Tenant header required."}, status=status.HTTP_400_BAD_REQUEST) if request.method == "GET": params = request.query_params entries = sel.filter_entries( tenant_id, user_id=int(params["user"]) if params.get("user") else None, project_id=int(params["project"]) if params.get("project") else None, task_id=int(params["task"]) if params.get("task") else None, date_from=date.fromisoformat(params["date_from"]) if params.get("date_from") else None, date_to=date.fromisoformat(params["date_to"]) if params.get("date_to") else None, is_billable=params["is_billable"].lower() == "true" if params.get("is_billable") else None, is_approved=params["is_approved"].lower() == "true" if params.get("is_approved") else None, ) return Response([_serialize_entry_list(e) for e in entries]) # POST data = request.data try: entry_date = date.fromisoformat(data["date"]) hours = Decimal(str(data["hours"])) except (KeyError, ValueError, InvalidOperation) as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) org_node_id = _get_org_node_id(request) or 1 try: entry = services.create_time_entry( tenant_id=tenant_id, organization_node_id=org_node_id, user_id=request.user.pk, project_id=int(data["project_id"]), date=entry_date, hours=hours, description=data.get("description", ""), is_billable=bool(data.get("is_billable", True)), task_id=int(data["task_id"]) if data.get("task_id") else None, ) except (ValueError, KeyError) as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response(_serialize_entry(entry), status=status.HTTP_201_CREATED) # --------------------------------------------------------------------------- # Entry detail # --------------------------------------------------------------------------- @api_view(["GET", "PATCH", "DELETE"]) @permission_classes([IsAuthenticated]) def entry_detail(request: Request, public_id: str) -> Response: tenant_id = _get_tenant_id(request) if not tenant_id: return Response({"detail": "Tenant header required."}, status=status.HTTP_400_BAD_REQUEST) entry = get_object_or_404(TimeEntry, tenant_id=tenant_id, public_id=public_id) if request.method == "GET": return Response(_serialize_entry(entry)) if request.method == "PATCH": data = request.data kwargs: dict = {} if "date" in data: try: kwargs["date"] = date.fromisoformat(data["date"]) except ValueError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) if "hours" in data: try: kwargs["hours"] = Decimal(str(data["hours"])) except InvalidOperation as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) if "description" in data: kwargs["description"] = data["description"] if "is_billable" in data: kwargs["is_billable"] = bool(data["is_billable"]) if "task_id" in data: kwargs["task_id"] = int(data["task_id"]) if data["task_id"] else None try: entry = services.update_time_entry(entry, **kwargs) except ValueError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response(_serialize_entry(entry)) # DELETE try: services.delete_time_entry(entry) except ValueError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response(status=status.HTTP_204_NO_CONTENT) # --------------------------------------------------------------------------- # My weekly entries # --------------------------------------------------------------------------- @api_view(["GET"]) @permission_classes([IsAuthenticated]) def my_weekly(request: Request) -> Response: tenant_id = _get_tenant_id(request) if not tenant_id: return Response({"detail": "Tenant header required."}, status=status.HTTP_400_BAD_REQUEST) # Default to current week (Mon–Sun) today = date.today() week_start_str = request.query_params.get("week_start") if week_start_str: try: week_start = date.fromisoformat(week_start_str) except ValueError: return Response({"detail": "Invalid week_start date."}, status=status.HTTP_400_BAD_REQUEST) else: week_start = today - timedelta(days=today.weekday()) week_end = week_start + timedelta(days=6) entries = sel.my_weekly_entries( tenant_id, request.user.pk, week_start, week_end ) # Build a summary: { project_id: { day: hours } } days = [(week_start + timedelta(days=i)).isoformat() for i in range(7)] by_project: dict[int, dict] = {} for e in entries: pid = e.project_id if pid not in by_project: by_project[pid] = { "project_id": pid, "project_name": e.project.name, "days": {d: "0.00" for d in days}, "entries": [], } by_project[pid]["days"][str(e.date)] = str( Decimal(by_project[pid]["days"].get(str(e.date), "0")) + e.hours ) by_project[pid]["entries"].append(_serialize_entry_list(e)) return Response({ "week_start": str(week_start), "week_end": str(week_end), "days": days, "rows": list(by_project.values()), }) # --------------------------------------------------------------------------- # Bulk approve # --------------------------------------------------------------------------- @api_view(["POST"]) @permission_classes([IsAuthenticated]) def bulk_approve(request: Request) -> Response: tenant_id = _get_tenant_id(request) if not tenant_id: return Response({"detail": "Tenant header required."}, status=status.HTTP_400_BAD_REQUEST) ids = request.data.get("ids", []) if not isinstance(ids, list): return Response({"detail": "ids must be a list."}, status=status.HTTP_400_BAD_REQUEST) count = services.approve_time_entries(ids, approved_by_id=request.user.pk, tenant_id=tenant_id) return Response({"approved": count}) # --------------------------------------------------------------------------- # Project time report (nested under /projects/{id}/time-report/) # --------------------------------------------------------------------------- @api_view(["GET"]) @permission_classes([IsAuthenticated]) def project_time_report(request: Request, project_id: int) -> Response: tenant_id = _get_tenant_id(request) if not tenant_id: return Response({"detail": "Tenant header required."}, status=status.HTTP_400_BAD_REQUEST) params = request.query_params date_from = date.fromisoformat(params["date_from"]) if params.get("date_from") else None date_to = date.fromisoformat(params["date_to"]) if params.get("date_to") else None rows = sel.project_time_report(tenant_id, project_id, date_from, date_to) result = [ { "user_id": r["user_id"], "user_email": r["user__email"], "task_id": r["task_id"], "task_title": r["task__title"], "total_hours": str(r["total_hours"]), } for r in rows ] return Response(result)