from pathlib import Path from decimal import Decimal from django import forms from django.core.exceptions import ValidationError from apps.attendance.models import AttendanceRecord from apps.attendance.services import employee_in_service_on from apps.leave.models import LeaveRequest, LeaveType REQUEST_TYPES = ( ("LEAVE", "Leave Application"), ("RESIGNATION", "Resignation Application"), ("ATTENDANCE_CORRECTION", "Attendance Correction Request"), ("SALARY_ADVANCE", "Salary Advance Request"), ("LOAN", "Loan Request"), ("CERTIFICATE", "Salary / Employment Certificate"), ("PROFILE_CORRECTION", "Profile Information Correction"), ("SALARY_DISPUTE", "Salary Dispute / Payroll Correction"), ("GENERAL_HR", "General HR Request"), ) class EmployeeRequestForm(forms.Form): request_type = forms.ChoiceField(choices=REQUEST_TYPES) subject = forms.CharField(max_length=200, required=False) details = forms.CharField(widget=forms.Textarea(attrs={"rows": 4}), required=False) leave_type = forms.ModelChoiceField(queryset=LeaveType.objects.none(), required=False) leave_start_date = forms.DateField(required=False, widget=forms.DateInput(attrs={"type": "date"})) leave_end_date = forms.DateField(required=False, widget=forms.DateInput(attrs={"type": "date"})) leave_half_day = forms.BooleanField(required=False) attendance_date = forms.DateField(required=False, widget=forms.DateInput(attrs={"type": "date"})) corrected_check_in = forms.DateTimeField(required=False, widget=forms.DateTimeInput(attrs={"type": "datetime-local"}), input_formats=["%Y-%m-%dT%H:%M", "%Y-%m-%d %H:%M:%S"]) corrected_check_out = forms.DateTimeField(required=False, widget=forms.DateTimeInput(attrs={"type": "datetime-local"}), input_formats=["%Y-%m-%dT%H:%M", "%Y-%m-%d %H:%M:%S"]) corrected_status = forms.ChoiceField(choices=AttendanceRecord.Status.choices, required=False) requested_amount = forms.DecimalField(max_digits=14, decimal_places=2, required=False, min_value=0) installment_count = forms.IntegerField(required=False, min_value=1, max_value=120) required_date = forms.DateField(required=False, widget=forms.DateInput(attrs={"type": "date"})) proposed_last_working_day = forms.DateField(required=False, widget=forms.DateInput(attrs={"type": "date"})) certificate_type = forms.CharField(max_length=100, required=False) disputed_month = forms.DateField(required=False, widget=forms.DateInput(attrs={"type": "month"}), input_formats=["%Y-%m"]) attachment = forms.FileField(required=False) ALLOWED_ATTACHMENT_EXTENSIONS = {".pdf", ".jpg", ".jpeg", ".png", ".doc", ".docx", ".xls", ".xlsx", ".csv"} MAX_ATTACHMENT_BYTES = 5 * 1024 * 1024 def clean_attachment(self): upload = self.cleaned_data.get("attachment") if not upload: return upload suffix = Path(upload.name or "").suffix.lower() if suffix not in self.ALLOWED_ATTACHMENT_EXTENSIONS: raise ValidationError("Unsupported attachment type. Allowed: PDF, image, Word, Excel and CSV.") if getattr(upload, "size", 0) > self.MAX_ATTACHMENT_BYTES: raise ValidationError("Attachment size cannot exceed 5 MB.") return upload def __init__(self, *args, employee=None, **kwargs): super().__init__(*args, **kwargs) self.employee = employee if employee: self.fields["leave_type"].queryset = LeaveType.objects.filter(company=employee.company, is_active=True).order_by("name") def _clean_leave(self, cleaned): leave_type = cleaned.get("leave_type") start = cleaned.get("leave_start_date") end = cleaned.get("leave_end_date") half_day = cleaned.get("leave_half_day") if not leave_type: self.add_error("leave_type", "Leave type is required.") if not start: self.add_error("leave_start_date", "Start date is required.") if not end: self.add_error("leave_end_date", "End date is required.") if not leave_type or not start or not end: return if end < start: self.add_error("leave_end_date", "End date cannot be earlier than start date.") return if start.year != end.year: self.add_error("leave_end_date", "Cross-year leave must be submitted as separate applications.") return if not employee_in_service_on(self.employee, start) or not employee_in_service_on(self.employee, end): self.add_error("leave_start_date", "Leave dates must fall within your employment service period.") return if half_day: if start != end: self.add_error("leave_half_day", "Half-day leave is allowed only for a single date.") return if not leave_type.half_day_allowed: self.add_error("leave_half_day", "The selected leave type does not allow half-day leave.") return from apps.leave.services import count_leave_days days = count_leave_days(self.employee, start, end, half_day=bool(half_day)) if days <= 0: self.add_error("leave_end_date", "The selected range has no working day according to the current holiday setup.") return cleaned["calculated_leave_days"] = days if LeaveRequest.objects.filter( employee=self.employee, status__in=["PENDING", "UNDER_REVIEW", "APPROVED"], start_date__lte=end, end_date__gte=start, ).exists(): self.add_error("leave_start_date", "Another pending/approved leave request overlaps this date range.") def _clean_attendance(self, cleaned): work_date = cleaned.get("attendance_date") if not work_date: self.add_error("attendance_date", "Attendance date is required.") return if not employee_in_service_on(self.employee, work_date): self.add_error("attendance_date", "Attendance correction date must fall within your employment service period.") return record = AttendanceRecord.objects.filter(employee=self.employee, work_date=work_date).first() check_in = cleaned.get("corrected_check_in") check_out = cleaned.get("corrected_check_out") if check_in and check_out and check_out <= check_in: self.add_error("corrected_check_out", "Corrected check-out must be later than corrected check-in.") status = cleaned.get("corrected_status") or (record.status if record else AttendanceRecord.Status.PRESENT) cleaned["corrected_status"] = status no_punch_statuses = { AttendanceRecord.Status.ABSENT, AttendanceRecord.Status.LEAVE, AttendanceRecord.Status.HOLIDAY, AttendanceRecord.Status.WEEKLY_OFF, } if status in no_punch_statuses and (check_in or check_out): self.add_error("corrected_status", "Absent/Leave/Holiday/Weekly Off corrections cannot contain punch timestamps.") if check_out and not check_in: self.add_error("corrected_check_in", "Corrected check-in is required when corrected check-out is entered.") if status in {AttendanceRecord.Status.PRESENT, AttendanceRecord.Status.LATE} and not check_in: self.add_error("corrected_check_in", "Present/Late correction requires a check-in timestamp.") cleaned["attendance_record"] = record def clean(self): cleaned = super().clean() if not self.employee: raise ValidationError("A linked employee profile is required.") request_type = cleaned.get("request_type") if request_type == "LEAVE": self._clean_leave(cleaned) elif request_type == "ATTENDANCE_CORRECTION": self._clean_attendance(cleaned) if not (cleaned.get("details") or "").strip(): self.add_error("details", "Correction reason/details are required.") elif request_type in {"LOAN", "SALARY_ADVANCE"}: requested_amount = cleaned.get("requested_amount") if requested_amount is None or requested_amount <= 0: self.add_error("requested_amount", "Requested amount must be greater than zero.") if request_type == "LOAN" and not cleaned.get("installment_count"): self.add_error("installment_count", "Preferred installment count is required for a loan request.") if request_type == "RESIGNATION": proposed = cleaned.get("proposed_last_working_day") if not proposed: self.add_error("proposed_last_working_day", "Proposed last working day is required.") elif proposed < self.employee.joining_date: self.add_error("proposed_last_working_day", "Proposed last working day cannot be earlier than the employee joining date.") if request_type == "CERTIFICATE" and not cleaned.get("certificate_type"): self.add_error("certificate_type", "Certificate type is required.") if request_type == "SALARY_DISPUTE" and not cleaned.get("disputed_month"): self.add_error("disputed_month", "Payroll month is required.") if request_type not in {"LEAVE"} and not (cleaned.get("details") or "").strip(): self.add_error("details", "Details are required.") return cleaned RETURNABLE_TASK_SOURCE_TYPES = {"employee_request", "leave_request", "attendance_adjustment", "final_settlement"} class TaskDecisionForm(forms.Form): decision = forms.ChoiceField(choices=(("APPROVED", "Approve"), ("REJECTED", "Reject"), ("RETURNED", "Return for Correction"))) note = forms.CharField(required=False, widget=forms.Textarea(attrs={"rows": 3})) sanctioned_amount = forms.DecimalField(max_digits=14, decimal_places=2, required=False, min_value=Decimal("0.01")) approved_installment_count = forms.IntegerField(required=False, min_value=1, max_value=120) recovery_start_month = forms.DateField(required=False, widget=forms.DateInput(attrs={"type": "month"}), input_formats=["%Y-%m"]) disbursement_date = forms.DateField(required=False, widget=forms.DateInput(attrs={"type": "date"})) payment_reference = forms.CharField(max_length=120, required=False) def __init__(self, *args, task=None, **kwargs): super().__init__(*args, **kwargs) self.task = task source_type = (getattr(task, "source_type", "") or "").lower() if task and source_type not in RETURNABLE_TASK_SOURCE_TYPES: self.fields["decision"].choices = (("APPROVED", "Approve"), ("REJECTED", "Reject")) if task and source_type == "employee_request": from .models import EmployeeRequest req = EmployeeRequest.objects.filter(pk=task.source_id).first() if req and req.request_type in {"LOAN", "SALARY_ADVANCE"}: payload = req.payload or {} self.fields["sanctioned_amount"].initial = payload.get("requested_amount") self.fields["approved_installment_count"].initial = payload.get("installment_count") or 1 def clean(self): cleaned = super().clean() decision = cleaned.get("decision") if decision in {"REJECTED", "RETURNED"} and not (cleaned.get("note") or "").strip(): raise ValidationError("Reason/comment is required for Reject or Return for Correction.") source_type = (getattr(self.task, "source_type", "") or "").lower() if decision == "APPROVED" and source_type == "employee_request": from .models import EmployeeRequest req = EmployeeRequest.objects.filter(pk=self.task.source_id).first() if self.task else None if req and req.request_type in {"LOAN", "SALARY_ADVANCE"}: if not cleaned.get("sanctioned_amount"): self.add_error("sanctioned_amount", "Sanctioned amount is required.") if not cleaned.get("approved_installment_count"): self.add_error("approved_installment_count", "Approved installment count is required.") if not cleaned.get("recovery_start_month"): self.add_error("recovery_start_month", "Recovery start month is required.") if decision == "APPROVED" and source_type == "loan_disbursement": if not cleaned.get("disbursement_date"): self.add_error("disbursement_date", "Disbursement date is required.") if decision == "APPROVED" and source_type == "payroll_period": requested_action = str((getattr(self.task, "payload", {}) or {}).get("payroll_action", "")).upper() if requested_action == "PAY" and not (cleaned.get("payment_reference") or "").strip(): self.add_error("payment_reference", "Bank/payment reference is required to post payroll payment.") return cleaned class ManagementNoticeForm(forms.Form): AUDIENCE_CHOICES = (("ALL", "All active employees"), ("SELECTED", "Selected employees")) company = forms.ModelChoiceField(queryset=None) title = forms.CharField(max_length=200) message = forms.CharField(widget=forms.Textarea(attrs={"rows": 5})) publish_from = forms.DateTimeField(widget=forms.DateTimeInput(attrs={"type": "datetime-local"}), input_formats=["%Y-%m-%dT%H:%M"]) publish_until = forms.DateTimeField(required=False, widget=forms.DateTimeInput(attrs={"type": "datetime-local"}), input_formats=["%Y-%m-%dT%H:%M"]) audience = forms.ChoiceField(choices=AUDIENCE_CHOICES, initial="ALL") recipients = forms.ModelMultipleChoiceField(queryset=None, required=False) def __init__(self, *args, company_queryset=None, **kwargs): super().__init__(*args, **kwargs) from apps.organization.models import Company from apps.employees.models import Employee companies = company_queryset if company_queryset is not None else Company.objects.none() self.fields["company"].queryset = companies company_id = self.data.get("company") if self.is_bound else None self.fields["recipients"].queryset = Employee.objects.filter(company_id=company_id, status__in=["ACTIVE", "PROBATION", "CONFIRMED"]).order_by("employee_code") if company_id else Employee.objects.none() def clean(self): cleaned = super().clean() if cleaned.get("publish_until") and cleaned.get("publish_from") and cleaned["publish_until"] <= cleaned["publish_from"]: self.add_error("publish_until", "Publish Until must be later than Publish From.") if cleaned.get("audience") == "SELECTED" and not cleaned.get("recipients"): self.add_error("recipients", "Select at least one employee for a selected-audience notice.") return cleaned