909 lines
35 KiB
Python
909 lines
35 KiB
Python
from datetime import date, datetime, timedelta, timezone
|
|
|
|
from django import template
|
|
from django.conf import settings
|
|
from django.contrib.auth.decorators import permission_required
|
|
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.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
|
|
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):
|
|
datetime_now = datetime.now()
|
|
if request.__dict__.get("datetime"):
|
|
datetime_now = request.datetime
|
|
employee, work_info = employee_exists(request)
|
|
shift = work_info.shift_id
|
|
date_today = date.today()
|
|
if request.__dict__.get("date"):
|
|
date_today = request.date
|
|
day = date_today.strftime("%A").lower()
|
|
day = EmployeeShiftDay.objects.get(day=day)
|
|
attendance = (
|
|
Attendance.objects.filter(employee_id=employee)
|
|
.order_by("id", "attendance_date")
|
|
.last()
|
|
)
|
|
if attendance is not None:
|
|
day = attendance.attendance_day
|
|
now = datetime.now().strftime("%H:%M")
|
|
if request.__dict__.get("time"):
|
|
now = request.time.strftime("%H:%M")
|
|
minimum_hour, start_time_sec, end_time_sec = shift_schedule_today(
|
|
day=day, shift=shift
|
|
)
|
|
early_out_instance = attendance.late_come_early_out.filter(type="early_out")
|
|
if not early_out_instance.exists():
|
|
early_out(
|
|
attendance=attendance, start_time=start_time_sec, end_time=end_time_sec
|
|
)
|
|
|
|
clock_out_attendance_and_activity(
|
|
employee=employee, date_today=date_today, now=now, out_datetime=datetime_now
|
|
)
|
|
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():
|
|
validated_data = serializer.validated_data
|
|
instance = Attendance(**validated_data)
|
|
instance.save()
|
|
return Response(serializer.data, status=200)
|
|
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)
|
|
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")
|