from datetime import date, datetime, timedelta, timezone from django import template from django.conf import settings from django.core.mail import EmailMessage from django.db.models import Case, CharField, F, Value, When from django.http import QueryDict from django.shortcuts import get_object_or_404 from django.utils.decorators import method_decorator from rest_framework.pagination import PageNumberPagination from rest_framework.permissions import IsAuthenticated from rest_framework.response import Response from rest_framework.views import APIView from attendance.models import Attendance, AttendanceActivity, EmployeeShiftDay from attendance.views.clock_in_out import * from attendance.views.clock_in_out import clock_out from attendance.views.dashboard import ( find_expected_attendances, find_late_come, find_on_time, ) from attendance.views.views import * from base.backends import ConfiguredEmailBackend from base.methods import generate_pdf, is_reportingmanager from employee.filters import EmployeeFilter from recruitment.models import RecruitmentMailTemplate from ...api_decorators.base.decorators import ( manager_permission_required, permission_required, ) from ...api_methods.base.methods import groupby_queryset, permission_based_queryset from ...api_serializers.attendance.serializers import ( AttendanceActivitySerializer, AttendanceLateComeEarlyOutSerializer, AttendanceOverTimeSerializer, AttendanceRequestSerializer, AttendanceSerializer, MailTemplateSerializer, ) # Create your views here. def query_dict(data): query_dict = QueryDict("", mutable=True) for key, value in data.items(): if isinstance(value, list): for item in value: query_dict.appendlist(key, item) else: query_dict.update({key: value}) return query_dict class ClockInAPIView(APIView): """ Allows authenticated employees to clock in, determining the correct shift and attendance date, including handling night shifts. Methods: post(request): Processes and records the clock-in time. """ permission_classes = [IsAuthenticated] def post(self, request): employee, work_info = employee_exists(request) datetime_now = datetime.now() if request.__dict__.get("datetime"): datetime_now = request.datetime if employee and work_info is not None: shift = work_info.shift_id date_today = date.today() if request.__dict__.get("date"): date_today = request.date attendance_date = date_today day = date_today.strftime("%A").lower() day = EmployeeShiftDay.objects.get(day=day) now = datetime.now().strftime("%H:%M") if request.__dict__.get("time"): now = request.time.strftime("%H:%M") now_sec = strtime_seconds(now) mid_day_sec = strtime_seconds("12:00") minimum_hour, start_time_sec, end_time_sec = shift_schedule_today( day=day, shift=shift ) if start_time_sec > end_time_sec: # night shift # ------------------ # Night shift in Horilla consider a 24 hours from noon to next day noon, # the shift day taken today if the attendance clocked in after 12 O clock. if mid_day_sec > now_sec: # Here you need to create attendance for yesterday date_yesterday = date_today - timedelta(days=1) day_yesterday = date_yesterday.strftime("%A").lower() day_yesterday = EmployeeShiftDay.objects.get(day=day_yesterday) minimum_hour, start_time_sec, end_time_sec = shift_schedule_today( day=day_yesterday, shift=shift ) attendance_date = date_yesterday day = day_yesterday clock_in_attendance_and_activity( employee=employee, date_today=date_today, attendance_date=attendance_date, day=day, now=now, shift=shift, minimum_hour=minimum_hour, start_time=start_time_sec, end_time=end_time_sec, in_datetime=datetime_now, ) return Response({"message": "Clocked-In"}, status=200) return Response( { "error": "You Don't have work information filled or your employee detail neither entered " } ) class ClockOutAPIView(APIView): """ Allows authenticated employees to clock out, updating the latest attendance record and handling early outs. Methods: post(request): Records the clock-out time. """ permission_classes = [IsAuthenticated] def post(self, request): current_date = date.today() current_time = datetime.now().time() current_datetime = datetime.now() try: clock_out( Request( user=request.user, date=current_date, time=current_time, datetime=current_datetime, ) ) return Response({"message": "Clocked-Out"}, status=200) except Exception as error: logger.error("Got an error in clock_out", error) return Response({"message": "Clocked-Out"}, status=200) class AttendanceView(APIView): """ Handles CRUD operations for attendance records. Methods: get_queryset(request, type): Returns filtered attendance records. get(request, pk=None, type=None): Retrieves a specific record or a list of records. post(request): Creates a new attendance record. put(request, pk): Updates an existing attendance record. delete(request, pk): Deletes an attendance record and adjusts related overtime if needed. """ permission_classes = [IsAuthenticated] filterset_class = AttendanceFilters def get_queryset(self, request, type): if type == "ot": condition = AttendanceValidationCondition.objects.first() minot = strtime_seconds("00:30") if condition is not None: minot = strtime_seconds(condition.minimum_overtime_to_approve) queryset = Attendance.objects.filter( overtime_second__gte=minot, attendance_validated=True, ) elif type == "validated": queryset = Attendance.objects.filter(attendance_validated=True) elif type == "non-validated": queryset = Attendance.objects.filter(attendance_validated=False) else: queryset = Attendance.objects.all() user = request.user # checking user level permissions perm = "attendance.view_attendance" queryset = permission_based_queryset(user, perm, queryset, user_obj=True) return queryset def get(self, request, pk=None, type=None): # individual object workflow if pk: attendance = get_object_or_404(Attendance, pk=pk) serializer = AttendanceSerializer(instance=attendance) return Response(serializer.data, status=200) # permission based querysete attendances = self.get_queryset(request, type) # filtering queryset attendances_filter_queryset = self.filterset_class( request.GET, queryset=attendances ).qs field_name = request.GET.get("groupby_field", None) if field_name: url = request.build_absolute_uri() return groupby_queryset( request, url, field_name, attendances_filter_queryset ) # pagination workflow paginater = PageNumberPagination() page = paginater.paginate_queryset(attendances_filter_queryset, request) serializer = AttendanceSerializer(page, many=True) return paginater.get_paginated_response(serializer.data) @manager_permission_required("attendance.add_attendance") def post(self, request): serializer = AttendanceSerializer(data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=200) employee_id = request.data.get("employee_id") attendance_date = request.data.get("attendance_date", date.today()) if Attendance.objects.filter( employee_id=employee_id, attendance_date=attendance_date ).exists(): return Response( { "error": [ "Attendance for this employee on the current date already exists." ] }, status=400, ) return Response(serializer.errors, status=400) @method_decorator(permission_required("attendance.change_attendance")) def put(self, request, pk): try: attendance = Attendance.objects.get(id=pk) except Attendance.DoesNotExist: return Response({"detail": "Attendance record not found."}, status=404) serializer = AttendanceSerializer(instance=attendance, data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=200) # Customize error message for unique constraint serializer_errors = serializer.errors if "non_field_errors" in serializer.errors: unique_error_msg = ( "The fields employee_id, attendance_date must make a unique set." ) if unique_error_msg in serializer.errors["non_field_errors"]: serializer_errors = { "non_field_errors": [ "The employee already has attendance on this date." ] } return Response(serializer_errors, status=400) @method_decorator(permission_required("attendance.delete_attendance")) def delete(self, request, pk): attendance = Attendance.objects.get(id=pk) month = attendance.attendance_date month = month.strftime("%B").lower() overtime = attendance.employee_id.employee_overtime.filter(month=month).last() if overtime is not None: if attendance.attendance_overtime_approve: # Subtract overtime of this attendance total_overtime = strtime_seconds(overtime.overtime) attendance_overtime_seconds = strtime_seconds( attendance.attendance_overtime ) if total_overtime > attendance_overtime_seconds: total_overtime = total_overtime - attendance_overtime_seconds else: total_overtime = attendance_overtime_seconds - total_overtime overtime.overtime = format_time(total_overtime) overtime.save() try: attendance.delete() return Response({"status", "deleted"}, status=200) except Exception as error: return Response({"error:", f"{error}"}, status=400) else: try: attendance.delete() return Response({"status", "deleted"}, status=200) except Exception as error: return Response({"error:", f"{error}"}, status=400) class ValidateAttendanceView(APIView): """ Validates an attendance record and sends a notification to the employee. Method: put(request, pk): Marks the attendance as validated and notifies the employee. """ def put(self, request, pk): attendance = Attendance.objects.filter(id=pk).update(attendance_validated=True) attendance = Attendance.objects.filter(id=pk).first() try: notify.send( request.user.employee_get, recipient=attendance.employee_id.employee_user_id, verb=f"Your attendance for the date {attendance.attendance_date} is validated", verb_ar=f"تم تحقيق حضورك في تاريخ {attendance.attendance_date}", verb_de=f"Deine Anwesenheit für das Datum {attendance.attendance_date} ist bestätigt.", verb_es=f"Se valida tu asistencia para la fecha {attendance.attendance_date}.", verb_fr=f"Votre présence pour la date {attendance.attendance_date} est validée.", redirect="/attendance/view-my-attendance", icon="checkmark", api_redirect=f"/api/attendance/attendance?employee_id{attendance.employee_id}", ) except: pass return Response(status=200) class OvertimeApproveView(APIView): """ Approves overtime for an attendance record and sends a notification to the employee. Method: put(request, pk): Marks the overtime as approved and notifies the employee. """ def put(self, request, pk): try: attendance = Attendance.objects.filter(id=pk).update( attendance_overtime_approve=True ) except Exception as E: return Response({"error": str(E)}, status=400) attendance = Attendance.objects.filter(id=pk).first() try: notify.send( request.user.employee_get, recipient=attendance.employee_id.employee_user_id, verb=f"Your {attendance.attendance_date}'s attendance overtime approved.", verb_ar=f"تمت الموافقة على إضافة ساعات العمل الإضافية لتاريخ {attendance.attendance_date}.", verb_de=f"Die Überstunden für den {attendance.attendance_date} wurden genehmigt.", verb_es=f"Se ha aprobado el tiempo extra de asistencia para el {attendance.attendance_date}.", verb_fr=f"Les heures supplémentaires pour la date {attendance.attendance_date} ont été approuvées.", redirect="/attendance/attendance-overtime-view", icon="checkmark", api_redirect="/api/attendance/attendance-hour-account/", ) except: pass return Response(status=200) class AttendanceRequestView(APIView): """ Handles requests for creating, updating, and viewing attendance records. Methods: get(request, pk=None): Retrieves a specific attendance request by `pk` or a filtered list of requests. post(request): Creates a new attendance request. put(request, pk): Updates an existing attendance request. """ serializer_class = AttendanceRequestSerializer permission_classes = [IsAuthenticated] def get(self, request, pk=None): if pk: attendance = Attendance.objects.get(id=pk) serializer = AttendanceRequestSerializer(instance=attendance) return Response(serializer.data, status=200) requests = Attendance.objects.filter( is_validate_request=True, ) requests = filtersubordinates( request=request, perm="attendance.view_attendance", queryset=requests, ) requests = requests | Attendance.objects.filter( employee_id__employee_user_id=request.user, is_validate_request=True, ) request_filtered_queryset = AttendanceFilters(request.GET, requests).qs field_name = request.GET.get("groupby_field", None) if field_name: # groupby workflow url = request.build_absolute_uri() return groupby_queryset(request, url, field_name, request_filtered_queryset) pagenation = PageNumberPagination() page = pagenation.paginate_queryset(request_filtered_queryset, request) serializer = self.serializer_class(page, many=True) return pagenation.get_paginated_response(serializer.data) def post(self, request): serializer = AttendanceRequestSerializer(data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=200) employee_id = request.data.get("employee_id") attendance_date = request.data.get("attendance_date", date.today()) if Attendance.objects.filter( employee_id=employee_id, attendance_date=attendance_date ).exists(): return Response( { "error": [ "Attendance for this employee on the current date already exists." ] }, status=400, ) return Response(serializer.errors, status=404) @manager_permission_required("attendance.update_attendance") def put(self, request, pk): attendance = Attendance.objects.get(id=pk) serializer = AttendanceRequestSerializer(instance=attendance, data=request.data) if serializer.is_valid(): instance = serializer.save() instance.employee_id = attendance.employee_id instance.id = attendance.id if attendance.request_type != "create_request": attendance.requested_data = json.dumps(instance.serialize()) attendance.request_description = instance.request_description # set the user level validation here attendance.is_validate_request = True attendance.save() else: instance.is_validate_request_approved = False instance.is_validate_request = True instance.save() return Response(serializer.data, status=200) return Response(serializer.errors, status=404) class AttendanceRequestApproveView(APIView): """ Approves and updates an attendance request. Method: put(request, pk): Approves the attendance request, updates attendance records, and handles related activities. """ @manager_permission_required("attendance.change_attendance") def put(self, request, pk): try: attendance = Attendance.objects.get(id=pk) prev_attendance_date = attendance.attendance_date prev_attendance_clock_in_date = attendance.attendance_clock_in_date prev_attendance_clock_in = attendance.attendance_clock_in attendance.attendance_validated = True attendance.is_validate_request_approved = True attendance.is_validate_request = False attendance.request_description = None attendance.save() if attendance.requested_data is not None: requested_data = json.loads(attendance.requested_data) requested_data["attendance_clock_out"] = ( None if requested_data["attendance_clock_out"] == "None" else requested_data["attendance_clock_out"] ) requested_data["attendance_clock_out_date"] = ( None if requested_data["attendance_clock_out_date"] == "None" else requested_data["attendance_clock_out_date"] ) Attendance.objects.filter(id=pk).update(**requested_data) # DUE TO AFFECT THE OVERTIME CALCULATION ON SAVE METHOD, SAVE THE INSTANCE ONCE MORE attendance = Attendance.objects.get(id=pk) attendance.save() if ( attendance.attendance_clock_out is None or attendance.attendance_clock_out_date is None ): attendance.attendance_validated = True activity = AttendanceActivity.objects.filter( employee_id=attendance.employee_id, attendance_date=prev_attendance_date, clock_in_date=prev_attendance_clock_in_date, clock_in=prev_attendance_clock_in, ) if activity: activity.update( employee_id=attendance.employee_id, attendance_date=attendance.attendance_date, clock_in_date=attendance.attendance_clock_in_date, clock_in=attendance.attendance_clock_in, ) else: AttendanceActivity.objects.create( employee_id=attendance.employee_id, attendance_date=attendance.attendance_date, clock_in_date=attendance.attendance_clock_in_date, clock_in=attendance.attendance_clock_in, ) except Exception as E: return Response({"error": str(E)}, status=400) return Response({"status": "approved"}, status=200) class AttendanceRequestCancelView(APIView): """ Cancels an attendance request. Method: put(request, pk): Cancels the attendance request, resetting its status and data, and deletes the request if it was a create request. """ def put(self, request, pk): try: attendance = Attendance.objects.get(id=pk) if ( attendance.employee_id.employee_user_id == request.user or is_reportingmanager(request) or request.user.has_perm("attendance.change_attendance") ): attendance.is_validate_request_approved = False attendance.is_validate_request = False attendance.request_description = None attendance.requested_data = None attendance.request_type = None attendance.save() if attendance.request_type == "create_request": attendance.delete() except Exception as E: return Response({"error": str(E)}, status=400) return Response({"status": "success"}, status=200) class AttendanceOverTimeView(APIView): """ Manages CRUD operations for attendance overtime records. Methods: get(request, pk=None): Retrieves a specific overtime record by `pk` or a list of records with filtering and pagination. post(request): Creates a new overtime record. put(request, pk): Updates an existing overtime record. delete(request, pk): Deletes an overtime record. """ permission_classes = [IsAuthenticated] def get(self, request, pk=None): if pk: attendance_ot = get_object_or_404(AttendanceOverTime, pk=pk) serializer = AttendanceOverTimeSerializer(attendance_ot) return Response(serializer.data, status=200) filterset_class = AttendanceOverTimeFilter(request.GET) queryset = filterset_class.qs self_account = queryset.filter(employee_id__employee_user_id=request.user) permission_based_queryset = filtersubordinates( request, queryset, "attendance.view_attendanceovertime" ) queryset = permission_based_queryset | self_account field_name = request.GET.get("groupby_field", None) if field_name: # groupby workflow url = request.build_absolute_uri() return groupby_queryset(request, url, field_name, queryset) pagenation = PageNumberPagination() page = pagenation.paginate_queryset(queryset, request) serializer = AttendanceOverTimeSerializer(page, many=True) return pagenation.get_paginated_response(serializer.data) @manager_permission_required("attendance.add_attendanceovertime") def post(self, request): serializer = AttendanceOverTimeSerializer(data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=200) return Response(serializer.errors, status=400) @manager_permission_required("attendance.change_attendanceovertime") def put(self, request, pk): attendance_ot = get_object_or_404(AttendanceOverTime, pk=pk) serializer = AttendanceOverTimeSerializer( instance=attendance_ot, data=request.data ) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=200) return Response(serializer.errors, status=400) @method_decorator(permission_required("attendance.delete_attendanceovertime")) def delete(self, request, pk): attendance = get_object_or_404(AttendanceOverTime, pk=pk) attendance.delete() return Response({"message": "Overtime deleted successfully"}, status=204) class LateComeEarlyOutView(APIView): """ Handles retrieval and deletion of late come and early out records. Methods: get(request, pk=None): Retrieves a list of late come and early out records with filtering. delete(request, pk=None): Deletes a specific late come or early out record by `pk`. """ permission_classes = [IsAuthenticated] def get(self, request, pk=None): data = LateComeEarlyOutFilter(request.GET) serializer = AttendanceLateComeEarlyOutSerializer(data.qs, many=True) return Response(serializer.data, status=200) def delete(self, request, pk=None): attendance = get_object_or_404(AttendanceLateComeEarlyOut, pk=pk) attendance.delete() return Response({"message": "Attendance deleted successfully"}, status=204) class AttendanceActivityView(APIView): """ Retrieves attendance activity records. Method: get(request, pk=None): Retrieves a list of all attendance activity records. """ def get(self, request, pk=None): data = AttendanceActivity.objects.all() serializer = AttendanceActivitySerializer(data, many=True) return Response(serializer.data, status=200) class TodayAttendance(APIView): """ Provides the ratio of marked attendances to expected attendances for the current day. Method: get(request): Calculates and returns the attendance ratio for today. """ def get(self, request): today = datetime.today() week_day = today.strftime("%A").lower() on_time = find_on_time(request, today=today, week_day=week_day) late_come = find_late_come(start_date=today) late_come_obj = len(late_come) marked_attendances = late_come_obj + on_time expected_attendances = find_expected_attendances(week_day=week_day) marked_attendances_ratio = 0 if expected_attendances != 0: marked_attendances_ratio = ( f"{(marked_attendances / expected_attendances) * 100:.2f}" ) return Response( {"marked_attendances_ratio": marked_attendances_ratio}, status=200 ) class OfflineEmployeesCountView(APIView): """ Retrieves the count of active employees who have not clocked in today. Method: get(request): Returns the number of active employees who are not yet clocked in. """ def get(self, request): count = ( EmployeeFilter({"not_in_yet": date.today()}) .qs.exclude(employee_work_info__isnull=True) .filter(is_active=True) .count() ) return Response({"count": count}, status=200) class OfflineEmployeesListView(APIView): """ Lists active employees who have not clocked in today, including their leave status. Method: get(request): Retrieves and paginates a list of employees not clocked in today with their leave status. """ def get(self, request): queryset = ( EmployeeFilter({"not_in_yet": date.today()}) .qs.exclude(employee_work_info__isnull=True) .filter(is_active=True) ) leave_status = self.get_leave_status(queryset) pagenation = PageNumberPagination() page = pagenation.paginate_queryset(leave_status, request) return pagenation.get_paginated_response(page) def get_leave_status(self, queryset): today = date.today() queryset = queryset.distinct() # Annotate each employee with their leave status employees_with_leave_status = queryset.annotate( leave_status=Case( # Define different cases based on leave requests and attendance When( leaverequest__start_date__lte=today, leaverequest__end_date__gte=today, leaverequest__status="approved", then=Value("On Leave"), ), When( leaverequest__start_date__lte=today, leaverequest__end_date__gte=today, leaverequest__status="requested", then=Value("Waiting Approval"), ), When( leaverequest__start_date__lte=today, leaverequest__end_date__gte=today, then=Value("Canceled / Rejected"), ), When( employee_attendances__attendance_date=today, then=Value("Working") ), default=Value("Expected working"), # Default status output_field=CharField(), ), job_position_id=F("employee_work_info__job_position_id"), ).values( "employee_first_name", "employee_last_name", "leave_status", "employee_profile", "id", "job_position_id", ) for employee in employees_with_leave_status: if employee["employee_profile"]: employee["employee_profile"] = ( settings.MEDIA_URL + employee["employee_profile"] ) return employees_with_leave_status class CheckingStatus(APIView): """ Checks and provides the current attendance status for the authenticated user. Method: get(request): Returns the attendance status, duration at work, and clock-in time if available. """ permission_classes = [IsAuthenticated] @classmethod def _format_seconds(cls, seconds): hours = seconds // 3600 minutes = (seconds % 3600) // 60 seconds = seconds % 60 return f"{hours:02}:{minutes:02}:{seconds:02}" def get(self, request): attendance_activity = ( AttendanceActivity.objects.filter(employee_id=request.user.employee_get) .order_by("-id") .first() ) duration = None work_seconds = request.user.employee_get.get_forecasted_at_work()[ "forecasted_at_work_seconds" ] duration = CheckingStatus._format_seconds(int(work_seconds)) status = False clock_in_time = None today = datetime.now() attendance_activity_first = ( AttendanceActivity.objects.filter( employee_id=request.user.employee_get, clock_in_date=today ) .order_by("in_datetime") .first() ) if attendance_activity: clock_in_time = attendance_activity_first.clock_in.strftime("%I:%M %p") if attendance_activity.clock_out_date: status = False else: status = True return Response( {"status": status, "duration": duration, "clock_in": clock_in_time}, status=200, ) return Response( {"status": status, "duration": duration, "clock_in_time": clock_in_time}, status=200, ) class MailTemplateView(APIView): """ Retrieves a list of recruitment mail templates. Method: get(request): Returns all recruitment mail templates. """ permission_classes = [IsAuthenticated] def get(self, request): instances = RecruitmentMailTemplate.objects.all() serializer = MailTemplateSerializer(instances, many=True) return Response(serializer.data, status=200) class ConvertedMailTemplateConvert(APIView): """ Renders a recruitment mail template with data from a specified employee. Method: put(request): Renders the mail template body with employee and user data and returns the result. """ permission_classes = [IsAuthenticated] def put(self, request): template_id = request.data.get("template_id", None) employee_id = request.data.get("employee_id", None) employee = Employee.objects.filter(id=employee_id).first() bdy = RecruitmentMailTemplate.objects.filter(id=template_id).first() template_bdy = template.Template(bdy.body) context = template.Context( {"instance": employee, "self": request.user.employee_get} ) render_bdy = template_bdy.render(context) return Response(render_bdy) class OfflineEmployeeMailsend(APIView): """ Sends an email with attachments and rendered templates to a specified employee. Method: post(request): Renders email templates with employee and user data, attaches files, and sends the email. """ permission_classes = [IsAuthenticated] def post(self, request): employee_id = request.POST.get("employee_id") subject = request.POST.get("subject", "") bdy = request.POST.get("body", "") other_attachments = request.FILES.getlist("other_attachments") attachments = [ (file.name, file.read(), file.content_type) for file in other_attachments ] email_backend = ConfiguredEmailBackend() host = email_backend.dynamic_username employee = Employee.objects.get(id=employee_id) template_attachment_ids = request.POST.getlist("template_attachments") bodys = list( RecruitmentMailTemplate.objects.filter( id__in=template_attachment_ids ).values_list("body", flat=True) ) for html in bodys: # due to not having solid template we first need to pass the context template_bdy = template.Template(html) context = template.Context( {"instance": employee, "self": request.user.employee_get} ) render_bdy = template_bdy.render(context) attachments.append( ( "Document", generate_pdf(render_bdy, {}, path=False, title="Document").content, "application/pdf", ) ) template_bdy = template.Template(bdy) context = template.Context( {"instance": employee, "self": request.user.employee_get} ) render_bdy = template_bdy.render(context) email = EmailMessage( subject, render_bdy, host, [employee.employee_work_info.email], ) email.content_subtype = "html" email.attachments = attachments try: email.send() if employee.employee_work_info.email: return Response(f"Mail sent to {employee.get_full_name()}") else: return Response(f"Email not set for {employee.get_full_name()}") except Exception as e: return Response("Something went wrong")