import ast import calendar import hashlib import json from collections import defaultdict from datetime import date, timedelta from decimal import Decimal from django.core.exceptions import ValidationError from django.db import transaction from django.db.models.deletion import ProtectedError from django.db.models import Q, Sum from django.utils import timezone from apps.accounts.permissions import ROLE_ACCOUNTS, ROLE_HR_ADMIN, ROLE_PAYROLL_OFFICER, ROLE_SUPER_ADMIN, has_role from apps.attendance.models import AttendanceDayLedger, OvertimeEntry from apps.attendance.services import assert_attendance_period_finalized, attendance_period_fingerprint, employee_final_working_day from apps.audit.services import log_event from apps.organization.services import get_effective_policy from .calculations import hourly_rate, money, net_payable, overtime_amount from .models import ( Bonus, Deduction, EmployeeSalaryAssignment, PayrollLineItem, PayrollPeriod, PayrollRecord, SalaryComponent, ) def ensure_period_editable(period): if period.status != PayrollPeriod.Status.DRAFT: raise ValidationError("Payroll can only be calculated/recalculated while the period is in Draft status.") def month_bounds(year, month): start = date(year, month, 1) end = date(year, month, calendar.monthrange(year, month)[1]) return start, end def active_salary_assignment(employee, on_date): return ( EmployeeSalaryAssignment.objects.filter(employee=employee, effective_from__lte=on_date) .filter(Q(effective_to__isnull=True) | Q(effective_to__gte=on_date)) .select_related("structure") .order_by("-effective_from", "-pk") .first() ) def _safe_formula(expression, *, basic, gross): """Evaluate simple arithmetic only; no names other than BASIC/GROSS.""" expression = (expression or "").strip().upper() if not expression: return Decimal("0") tree = ast.parse(expression, mode="eval") allowed_ops = (ast.Add, ast.Sub, ast.Mult, ast.Div) def walk(node): if isinstance(node, ast.Expression): return walk(node.body) if isinstance(node, ast.Name): if node.id == "BASIC": return Decimal(basic) if node.id == "GROSS": return Decimal(gross) raise ValidationError(f"Unsupported formula variable: {node.id}") if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)): return Decimal(str(node.value)) if isinstance(node, ast.UnaryOp) and isinstance(node.op, (ast.UAdd, ast.USub)): value = walk(node.operand) return value if isinstance(node.op, ast.UAdd) else -value if isinstance(node, ast.BinOp) and isinstance(node.op, allowed_ops): left, right = walk(node.left), walk(node.right) if isinstance(node.op, ast.Add): return left + right if isinstance(node.op, ast.Sub): return left - right if isinstance(node.op, ast.Mult): return left * right if right == 0: raise ValidationError("Salary formula division by zero.") return left / right raise ValidationError("Unsupported salary formula syntax.") return money(walk(tree)) def structure_component_monthly_values(assignment): lines = list(assignment.structure.lines.select_related("component").filter(component__is_active=True).order_by("pk")) values = [] for line in lines: component = line.component if component.calculation_type == SalaryComponent.CalcType.FIXED: amount = Decimal(line.amount or 0) if component.code.upper() == "BASIC" and amount == 0: amount = Decimal(assignment.basic_salary) elif component.calculation_type == SalaryComponent.CalcType.PERCENTAGE: amount = Decimal(assignment.basic_salary) * Decimal(line.percentage or 0) / Decimal("100") else: amount = _safe_formula(line.formula, basic=assignment.basic_salary, gross=assignment.gross_salary) values.append({ "code": component.code, "name": component.name, "kind": component.kind, "monthly_amount": money(amount), "source": f"STRUCTURE:{assignment.structure.code}", }) if not any(v["kind"] == SalaryComponent.Kind.EARNING for v in values): values.append({ "code": "GROSS", "name": "Gross Salary", "kind": SalaryComponent.Kind.EARNING, "monthly_amount": money(assignment.gross_salary), "source": f"ASSIGNMENT:{assignment.pk}", }) return values def _assignment_monthly_gross(assignment): return money(sum((v["monthly_amount"] for v in structure_component_monthly_values(assignment) if v["kind"] == SalaryComponent.Kind.EARNING), Decimal("0"))) def _assignment_monthly_basic(assignment): values = structure_component_monthly_values(assignment) basic = sum((v["monthly_amount"] for v in values if v["kind"] == SalaryComponent.Kind.EARNING and v["code"].upper() == "BASIC"), Decimal("0")) return money(basic if basic > 0 else assignment.basic_salary) def service_bounds(employee, start, end): service_start = max(start, employee.joining_date) final_day = employee_final_working_day(employee) service_end = min(end, final_day) if final_day else end if service_end < service_start: return None, None return service_start, service_end def salary_segments(employee, start, end): service_start, service_end = service_bounds(employee, start, end) if not service_start: return [] segments = [] current_start = service_start current_assignment = None current = service_start while current <= service_end: assignment = active_salary_assignment(employee, current) if not assignment: raise ValidationError(f"No salary assignment for {employee} on {current}.") if current_assignment is None: current_assignment = assignment current_start = current elif assignment.pk != current_assignment.pk: segments.append((current_start, current - timedelta(days=1), current_assignment)) current_assignment = assignment current_start = current current += timedelta(days=1) if current_assignment: segments.append((current_start, service_end, current_assignment)) return segments def _segment_factor(policy, month_start, month_end, seg_start, seg_end): # A full calendar month always equals one full monthly salary. The fixed # divisor is used only when a segment is partial (join/exit/revision). if seg_start == month_start and seg_end == month_end: return Decimal("1") days = Decimal((seg_end - seg_start).days + 1) if policy.salary_proration_method == "FIXED_DIVISOR": divisor = Decimal(policy.salary_day_divisor) else: divisor = Decimal((month_end - month_start).days + 1) if divisor <= 0: raise ValidationError("Invalid salary proration divisor.") return days / divisor def salary_base_for_period(employee, start, end, policy): segments = salary_segments(employee, start, end) line_totals = defaultdict(lambda: {"name": "", "kind": "", "amount": Decimal("0"), "source": ""}) gross = Decimal("0") basic = Decimal("0") recurring_structure_deductions = Decimal("0") snapshot_segments = [] for seg_start, seg_end, assignment in segments: month_start = date(seg_start.year, seg_start.month, 1) month_end = date(seg_start.year, seg_start.month, calendar.monthrange(seg_start.year, seg_start.month)[1]) factor = _segment_factor(policy, month_start, month_end, seg_start, seg_end) component_values = structure_component_monthly_values(assignment) seg_gross = Decimal("0") seg_basic = Decimal("0") seg_deductions = Decimal("0") for value in component_values: amount = money(value["monthly_amount"] * factor) bucket = line_totals[value["code"]] bucket["name"] = value["name"] bucket["kind"] = value["kind"] bucket["source"] = value["source"] bucket["amount"] += amount if value["kind"] == SalaryComponent.Kind.EARNING: seg_gross += amount if value["code"].upper() == "BASIC": seg_basic += amount else: seg_deductions += amount if seg_basic == 0: seg_basic = money(Decimal(assignment.basic_salary) * factor) gross += seg_gross basic += seg_basic recurring_structure_deductions += seg_deductions snapshot_segments.append({ "assignment_id": assignment.pk, "structure": assignment.structure.code, "from": str(seg_start), "to": str(seg_end), "factor": str(factor), "gross": str(money(seg_gross)), "basic": str(money(seg_basic)), }) lines = [ {"code": code, "name": data["name"], "kind": data["kind"], "amount": money(data["amount"]), "source": data["source"]} for code, data in line_totals.items() ] return { "gross": money(gross), "basic": money(basic), "structure_deductions": money(recurring_structure_deductions), "line_items": lines, "segments": snapshot_segments, } def attendance_deduction_for_period(employee, period, policy): start, end = month_bounds(period.year, period.month) service_start, service_end = service_bounds(employee, start, end) if not service_start: return money(0), [] ledgers = AttendanceDayLedger.objects.filter( employee=employee, work_date__range=(service_start, service_end), finalized=True, ).order_by("work_date") expected = (service_end - service_start).days + 1 if ledgers.count() != expected: raise ValidationError(f"Attendance day ledger is incomplete for {employee}. Finalize attendance again before payroll.") total = Decimal("0") rows = [] month_days = Decimal(calendar.monthrange(period.year, period.month)[1]) for ledger in ledgers: fraction = Decimal(ledger.deduction_fraction or 0) if fraction <= 0: continue assignment = active_salary_assignment(employee, ledger.work_date) if not assignment: raise ValidationError(f"No salary assignment for deduction date {ledger.work_date}.") monthly_gross = _assignment_monthly_gross(assignment) divisor = Decimal(policy.salary_day_divisor) if policy.salary_proration_method == "FIXED_DIVISOR" else month_days day_amount = Decimal(monthly_gross) / divisor deduction = money(day_amount * fraction) total += deduction rows.append({"date": str(ledger.work_date), "status": ledger.day_status, "fraction": str(fraction), "amount": str(deduction), "ledger_id": ledger.pk}) return money(total), rows def period_input_totals(employee, period, policy): start, end = month_bounds(period.year, period.month) base = salary_base_for_period(employee, start, end, policy) attendance_deduction, attendance_rows = attendance_deduction_for_period(employee, period, policy) service_start, service_end = service_bounds(employee, start, end) ot_total = Decimal("0") ot_rows = [] entries = OvertimeEntry.objects.none() if service_start: entries = OvertimeEntry.objects.filter( employee=employee, work_date__range=(service_start, service_end), status="APPROVED", approved_minutes__gt=0 ).order_by("work_date") for item in entries: from apps.attendance.services import approved_leave_for_day leave = approved_leave_for_day(employee, item.work_date) if leave and not (leave.start_date == leave.end_date and Decimal(leave.days) == Decimal("0.50")): raise ValidationError( f"Approved OT conflicts with full-day approved leave for {employee} on {item.work_date}. " "Cancel/correct the leave or OT approval before payroll calculation." ) assignment = active_salary_assignment(employee, item.work_date) if not assignment or not assignment.ot_eligible or not employee.ot_eligible: continue approved_minutes = int(item.approved_minutes or 0) if approved_minutes < int(policy.ot_minimum_minutes or 0): continue rounding = max(1, int(policy.ot_rounding_minutes or 1)) approved_minutes = (approved_minutes // rounding) * rounding if approved_minutes <= 0: continue base_hourly = hourly_rate(_assignment_monthly_basic(assignment), policy.monthly_working_hours) holiday_status = None from apps.attendance.services import _holiday_status holiday_status = _holiday_status(employee, item.work_date) if Decimal(item.rate_multiplier or 0) != Decimal("1.00"): multiplier = Decimal(item.rate_multiplier) elif holiday_status == "HOLIDAY": multiplier = Decimal(policy.ot_holiday_multiplier) elif holiday_status == "WEEKLY_OFF": multiplier = Decimal(policy.ot_weekly_off_multiplier) else: multiplier = Decimal(policy.ot_default_rate_multiplier or assignment.ot_rate_multiplier) hours = Decimal(approved_minutes) / Decimal("60") amount = overtime_amount(hours, base_hourly, multiplier) ot_total += amount ot_rows.append({"id": item.pk, "date": str(item.work_date), "minutes": approved_minutes, "multiplier": str(multiplier), "amount": str(amount)}) ot_total = money(ot_total) bonus_qs = Bonus.objects.filter(employee=employee, payroll_month__year=period.year, payroll_month__month=period.month, status="APPROVED").order_by("pk") bonus_total = bonus_qs.aggregate(total=Sum("amount"))["total"] or Decimal("0.00") bonus_rows = [ {"id": item.pk, "type": item.bonus_type, "amount": str(money(item.amount)), "reason": item.reason or ""} for item in bonus_qs ] deduction_qs = Deduction.objects.filter(employee=employee, payroll_month__year=period.year, payroll_month__month=period.month, status="APPROVED").order_by("pk") variable_deduction = deduction_qs.aggregate(total=Sum("amount"))["total"] or Decimal("0.00") deduction_rows = [ {"id": item.pk, "type": item.deduction_type, "amount": str(money(item.amount)), "reason": item.reason or ""} for item in deduction_qs ] from apps.loans.services import scheduled_recovery_for_period loan_recovery = scheduled_recovery_for_period(employee, year=period.year, month=period.month) return { "base": base, "attendance_deduction": money(attendance_deduction), "attendance_rows": attendance_rows, "overtime": money(ot_total), "ot_rows": ot_rows, "bonus": money(bonus_total), "bonus_ids": [row["id"] for row in bonus_rows], "bonus_rows": bonus_rows, "deductions": money(Decimal(variable_deduction) + Decimal(base["structure_deductions"])), "variable_deduction": money(variable_deduction), "deduction_ids": [row["id"] for row in deduction_rows], "deduction_rows": deduction_rows, "loan_recovery": money(loan_recovery), "other_earnings": Decimal("0.00"), "adjustments": Decimal("0.00"), } def _payroll_adjustment_snapshot(payroll): rows = [] earnings = Decimal("0") deductions = Decimal("0") for adjustment in payroll.adjustments.all().order_by("pk"): kind = (adjustment.adjustment_type or "").strip().upper() amount = abs(Decimal(adjustment.amount or 0)) if kind in {"EARNING", "ADD", "ARREAR", "CREDIT", "INCREASE"}: earnings += amount normalized_kind = "EARNING" else: deductions += amount normalized_kind = "DEDUCTION" rows.append({ "id": adjustment.pk, "type": kind, "kind": normalized_kind, "amount": str(money(amount)), "reference_no": adjustment.reference_no, }) return rows, earnings, deductions def _payroll_input_signature(*, payroll, inputs, policy, adjustment_rows): payload = { "attendance_period_fingerprint": attendance_period_fingerprint( payroll.period.company, payroll.period.year, payroll.period.month ), "policy": { "id": policy.pk, "effective_from": str(policy.effective_from), "effective_to": str(policy.effective_to or ""), "salary_day_divisor": str(policy.salary_day_divisor), "monthly_working_hours": str(policy.monthly_working_hours), "salary_proration_method": policy.salary_proration_method, "missing_punch_action": policy.missing_punch_action, "ot_minimum_minutes": policy.ot_minimum_minutes, "ot_rounding_minutes": policy.ot_rounding_minutes, "ot_default_rate_multiplier": str(policy.ot_default_rate_multiplier), "ot_weekly_off_multiplier": str(policy.ot_weekly_off_multiplier), "ot_holiday_multiplier": str(policy.ot_holiday_multiplier), }, "salary_segments": inputs["base"]["segments"], "salary_line_items": [ {**row, "amount": str(row["amount"])} for row in inputs["base"]["line_items"] ], "attendance": inputs["attendance_rows"], "overtime": inputs["ot_rows"], "bonus_rows": inputs.get("bonus_rows", []), "deduction_rows": inputs.get("deduction_rows", []), "loan_recovery": str(money(inputs["loan_recovery"])), "payroll_adjustments": adjustment_rows, } serialized = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str) return hashlib.sha256(serialized.encode("utf-8")).hexdigest() @transaction.atomic def recalculate_payroll(payroll, *, inputs, policy, user=None): payroll = PayrollRecord.objects.select_for_update().select_related("period", "employee").get(pk=payroll.pk) ensure_period_editable(payroll.period) old_net = payroll.net_payable base = inputs["base"] # Approved/manual payroll adjustments are attached to the payroll record and # are consumed only while the period is Draft. Positive earning/arrear types # add to pay; deduction/recovery types reduce pay. adjustment_rows, adjustment_earnings, adjustment_deductions = _payroll_adjustment_snapshot(payroll) payroll.gross_salary = base["gross"] payroll.basic_salary = base["basic"] payroll.attendance_deduction = inputs["attendance_deduction"] payroll.overtime_amount = inputs["overtime"] payroll.bonus_amount = inputs["bonus"] payroll.other_earnings = money(Decimal(inputs["other_earnings"]) + adjustment_earnings) payroll.deduction_amount = money(Decimal(inputs["deductions"]) + adjustment_deductions) payroll.loan_recovery = inputs["loan_recovery"] payroll.net_payable = net_payable( gross_salary=payroll.gross_salary, overtime=payroll.overtime_amount, bonus=payroll.bonus_amount, other_earnings=payroll.other_earnings, attendance_deduction=payroll.attendance_deduction, other_deductions=payroll.deduction_amount, loan_recovery=payroll.loan_recovery, adjustments=Decimal("0.00"), ) payroll.calculation_snapshot = { "engine_version": "0.39.3-prefinal-input-signature", "input_signature": _payroll_input_signature( payroll=payroll, inputs=inputs, policy=policy, adjustment_rows=adjustment_rows ), "attendance_period_fingerprint": attendance_period_fingerprint( payroll.period.company, payroll.period.year, payroll.period.month ), "policy": { "id": policy.pk, "effective_from": str(policy.effective_from), "salary_day_divisor": str(policy.salary_day_divisor), "monthly_working_hours": str(policy.monthly_working_hours), "salary_proration_method": policy.salary_proration_method, "missing_punch_action": policy.missing_punch_action, "ot_minimum_minutes": policy.ot_minimum_minutes, "ot_rounding_minutes": policy.ot_rounding_minutes, "ot_default_rate_multiplier": str(policy.ot_default_rate_multiplier), }, "salary_segments": base["segments"], "salary_line_items": [{**x, "amount": str(x["amount"])} for x in base["line_items"]], "attendance": inputs["attendance_rows"], "overtime": inputs["ot_rows"], "bonus_ids": inputs["bonus_ids"], "bonus_rows": inputs.get("bonus_rows", []), "deduction_ids": inputs["deduction_ids"], "deduction_rows": inputs.get("deduction_rows", []), "payroll_adjustments": adjustment_rows, "gross_salary": str(payroll.gross_salary), "basic_salary": str(payroll.basic_salary), "attendance_deduction": str(payroll.attendance_deduction), "overtime_amount": str(payroll.overtime_amount), "bonus_amount": str(payroll.bonus_amount), "deduction_amount": str(payroll.deduction_amount), "loan_recovery": str(payroll.loan_recovery), "net_payable": str(payroll.net_payable), } payroll.save() # Immutable-looking component snapshot for payslip/report output; rebuilt only in Draft. payroll.line_items.all().delete() line_objects = [] for item in base["line_items"]: line_objects.append(PayrollLineItem(payroll=payroll, component_code=item["code"], component_name=item["name"], kind=item["kind"], amount=item["amount"], source=item["source"])) if payroll.overtime_amount: line_objects.append(PayrollLineItem(payroll=payroll, component_code="OT", component_name="Overtime", kind="EARNING", amount=payroll.overtime_amount, source="APPROVED_OT")) for row in inputs.get("bonus_rows", []): line_objects.append(PayrollLineItem( payroll=payroll, component_code=f"BONUS-{row['id']}", component_name=row["type"] or "Approved Bonus", kind="EARNING", amount=Decimal(row["amount"]), source=f"APPROVED_BONUS:{row['id']}" )) if payroll.attendance_deduction: line_objects.append(PayrollLineItem(payroll=payroll, component_code="ATT_DED", component_name="Attendance / Unpaid Leave Deduction", kind="DEDUCTION", amount=payroll.attendance_deduction, source="FINALIZED_DAY_LEDGER")) for row in inputs.get("deduction_rows", []): line_objects.append(PayrollLineItem( payroll=payroll, component_code=f"DED-{row['id']}", component_name=row["type"] or "Approved Deduction", kind="DEDUCTION", amount=Decimal(row["amount"]), source=f"APPROVED_DEDUCTION:{row['id']}" )) if payroll.loan_recovery: line_objects.append(PayrollLineItem(payroll=payroll, component_code="LOAN", component_name="Loan / Advance Recovery", kind="DEDUCTION", amount=payroll.loan_recovery, source="LOAN_SCHEDULE")) for row in adjustment_rows: line_objects.append(PayrollLineItem( payroll=payroll, component_code=f"ADJ-{row['id']}", component_name=f"Payroll Adjustment: {row['type'].title()}", kind=row["kind"], amount=Decimal(row["amount"]), source="PAYROLL_ADJUSTMENT" )) PayrollLineItem.objects.bulk_create(line_objects) log_event(user=user, action="RECALCULATE", module="payroll", object_id=payroll.pk, object_repr=f"{payroll.employee} / {payroll.period.year}-{payroll.period.month:02d}", old_values={"net_payable": str(old_net)}, new_values={"net_payable": str(payroll.net_payable), "policy_id": policy.pk}) return payroll @transaction.atomic def build_payroll_period(period, *, user=None): period = PayrollPeriod.objects.select_for_update().select_related("company").get(pk=period.pk) ensure_period_editable(period) assert_attendance_period_finalized(period.company, period.year, period.month) _assert_no_pending_payroll_inputs(period) start, end = month_bounds(period.year, period.month) policy = get_effective_policy(period.company, end) created = updated = skipped = 0 from apps.employees.models import Employee employees = list(Employee.objects.filter(company=period.company, joining_date__lte=end).order_by("employee_code")) eligible = [] missing_salary = [] for employee in employees: service_start, service_end = service_bounds(employee, start, end) if not service_start: skipped += 1 continue assignment = active_salary_assignment(employee, service_end) if not assignment: missing_salary.append(employee.employee_code) continue eligible.append((employee, assignment)) if missing_salary: sample = ", ".join(missing_salary[:10]) extra = f" (+{len(missing_salary) - 10} more)" if len(missing_salary) > 10 else "" raise ValidationError( f"Payroll build blocked: active/in-service employee(s) have no salary assignment for this payroll month: {sample}{extra}." ) eligible_ids = [employee.pk for employee, _ in eligible] stale_qs = period.records.exclude(employee_id__in=eligible_ids) stale_count = stale_qs.count() if stale_count: try: stale_qs.delete() except ProtectedError as exc: raise ValidationError( "Payroll build found stale employee record(s) that are already referenced by controlled financial/exit data. " "Review those records before rebuilding the period." ) from exc for employee, assignment in eligible: payroll, was_created = PayrollRecord.objects.get_or_create(period=period, employee=employee, defaults={"salary_assignment": assignment}) payroll.salary_assignment = assignment payroll.save(update_fields=["salary_assignment", "updated_at"]) inputs = period_input_totals(employee, period, policy) recalculate_payroll(payroll, inputs=inputs, policy=policy, user=user) created += int(was_created) updated += int(not was_created) if user is not None: period.prepared_by = user # Rebuilding a Draft invalidates previous downstream actor stamps. period.reviewed_by = None period.approved_by = None period.locked_by = None period.save(update_fields=["prepared_by", "reviewed_by", "approved_by", "locked_by", "updated_at"]) log_event(user=user, company=period.company, action="BUILD_PERIOD", module="payroll_period", object_id=period.pk, object_repr=f"{period.year}-{period.month:02d}", new_values={"created": created, "updated": updated, "skipped_outside_service": skipped, "stale_records_removed": stale_count, "policy_id": policy.pk, "prepared_by": getattr(user, "pk", None)}) return {"created": created, "updated": updated, "skipped": skipped, "stale_removed": stale_count} def _paid_salary_component_for_exit_month(employee, final_working_day): """Return salary-component amount already paid for the exit month. This deliberately excludes OT/bonus/loan movements so final settlement can reconcile the salary portion without treating financial side-items as base salary. A payroll is considered paid only after the locked period has an explicit payment posting/reference. """ record = PayrollRecord.objects.filter( employee=employee, period__year=final_working_day.year, period__month=final_working_day.month, period__status=PayrollPeriod.Status.LOCKED, period__payment_status=PayrollPeriod.PaymentStatus.PAID, ).select_related("period").first() if not record: return Decimal("0.00"), None structure_deduction = sum( (line.amount for line in record.line_items.filter(kind=SalaryComponent.Kind.DEDUCTION, source__startswith="STRUCTURE:")), Decimal("0.00"), ) paid_salary = money(Decimal(record.gross_salary) - Decimal(record.attendance_deduction) - Decimal(structure_deduction)) return paid_salary, record def calculate_exit_period_due(employee, final_working_day): """Calculate current exit-month salary through the authoritative final working day.""" start = final_working_day.replace(day=1) policy = get_effective_policy(employee.company, final_working_day) base = salary_base_for_period(employee, start, final_working_day, policy) # Use reconciled attendance ledgers whether or not the full month is finalized. total_deduction = Decimal("0") rows = AttendanceDayLedger.objects.filter(employee=employee, work_date__range=(start, final_working_day)).order_by("work_date") month_days = Decimal(calendar.monthrange(final_working_day.year, final_working_day.month)[1]) for ledger in rows: fraction = Decimal(ledger.deduction_fraction or 0) if fraction <= 0: continue assignment = active_salary_assignment(employee, ledger.work_date) if not assignment: continue divisor = Decimal(policy.salary_day_divisor) if policy.salary_proration_method == "FIXED_DIVISOR" else month_days total_deduction += Decimal(_assignment_monthly_gross(assignment)) / divisor * fraction actual_salary_payable = money(Decimal(base["gross"]) - total_deduction - Decimal(base["structure_deductions"])) already_paid_salary, paid_record = _paid_salary_component_for_exit_month(employee, final_working_day) payable = money(actual_salary_payable - already_paid_salary) return payable, { "base": base, "attendance_deduction": str(money(total_deduction)), "policy_id": policy.pk, "actual_exit_period_salary": str(actual_salary_payable), "already_paid_salary": str(money(already_paid_salary)), "paid_payroll_record_id": paid_record.pk if paid_record else None, "salary_balance_due": str(payable), } def _assert_no_pending_payroll_inputs(period): """Fail closed while payroll-impacting approvals for the target month are still pending.""" start, end = month_bounds(period.year, period.month) pending_ot = OvertimeEntry.objects.filter( employee__company=period.company, work_date__range=(start, end), status="PENDING" ).count() pending_bonus = Bonus.objects.filter( employee__company=period.company, payroll_month__year=period.year, payroll_month__month=period.month, status="PENDING", ).count() pending_deduction = Deduction.objects.filter( employee__company=period.company, payroll_month__year=period.year, payroll_month__month=period.month, status="PENDING", ).count() if pending_ot or pending_bonus or pending_deduction: raise ValidationError( "Payroll workflow blocked by unresolved pending input approval(s): " f"OT={pending_ot}, Bonus={pending_bonus}, Deduction={pending_deduction}. " "Approve/reject them before continuing payroll." ) def _assert_payroll_period_inputs_current(period, policy): """Ensure the stored Draft calculation still matches all authoritative inputs.""" assert_attendance_period_finalized(period.company, period.year, period.month) start, end = month_bounds(period.year, period.month) from apps.employees.models import Employee expected_ids = set() missing_salary = [] for employee in Employee.objects.filter(company=period.company, joining_date__lte=end).order_by("employee_code"): service_start, service_end = service_bounds(employee, start, end) if not service_start: continue if not active_salary_assignment(employee, service_end): missing_salary.append(employee.employee_code) continue expected_ids.add(employee.pk) if missing_salary: sample = ", ".join(missing_salary[:10]) extra = f" (+{len(missing_salary) - 10} more)" if len(missing_salary) > 10 else "" raise ValidationError(f"Payroll gate blocked: in-service employee(s) have no salary assignment: {sample}{extra}.") records = list(period.records.select_related("employee").all()) if not records: raise ValidationError("Build payroll records before continuing the payroll workflow.") actual_ids = {record.employee_id for record in records} if actual_ids != expected_ids: raise ValidationError("Payroll employee coverage changed after the last build. Return to Draft and rebuild payroll.") if any(Decimal(record.net_payable or 0) < 0 for record in records): raise ValidationError( "Payroll workflow blocked because one or more employees have a negative net payable. " "Return to Draft and correct the deductions/adjustments." ) current_fingerprint = attendance_period_fingerprint(period.company, period.year, period.month) stale_inputs = [] stale_attendance = [] for record in records: snapshot = record.calculation_snapshot or {} if snapshot.get("attendance_period_fingerprint") != current_fingerprint: stale_attendance.append(record.pk) current_inputs = period_input_totals(record.employee, period, policy) current_adjustments, _, _ = _payroll_adjustment_snapshot(record) current_signature = _payroll_input_signature( payroll=record, inputs=current_inputs, policy=policy, adjustment_rows=current_adjustments ) if snapshot.get("input_signature") != current_signature: stale_inputs.append(record.pk) if stale_attendance: raise ValidationError( "Payroll was calculated against an older attendance close. Return to Draft and rebuild after the latest attendance finalization." ) if stale_inputs: raise ValidationError( "Payroll inputs changed after the last calculation (salary/OT/bonus/deduction/loan/adjustment/policy). " "Return to Draft and rebuild payroll before continuing." ) return records @transaction.atomic def return_payroll_period_to_draft(period, *, user=None, reason=""): period = PayrollPeriod.objects.select_for_update().get(pk=period.pk) if user is None or not has_role(user, ROLE_SUPER_ADMIN, ROLE_HR_ADMIN): raise ValidationError("Only HR Admin / Super Admin may return reviewed/approved payroll to Draft.") if period.status not in {PayrollPeriod.Status.REVIEWED, PayrollPeriod.Status.APPROVED}: raise ValidationError("Only Reviewed or Approved payroll can be returned to Draft through this correction path.") if period.payment_status == PayrollPeriod.PaymentStatus.PAID: raise ValidationError("Paid payroll cannot be returned to Draft.") if not (reason or "").strip(): raise ValidationError("Return-to-Draft reason is required.") old_status = period.status period.status = PayrollPeriod.Status.DRAFT period.reviewed_by = None period.approved_by = None period.locked_by = None period.locked_at = None period.unlock_reason = f"Returned to Draft: {(reason or '').strip()}" period.save(update_fields=[ "status", "reviewed_by", "approved_by", "locked_by", "locked_at", "unlock_reason", "updated_at" ]) log_event( user=user, action="RETURN_DRAFT", module="payroll_period", object_id=period.pk, object_repr=f"{period.year}-{period.month:02d}", old_values={"status": old_status}, new_values={"status": period.status}, reason=(reason or "").strip(), ) return period @transaction.atomic def transition_payroll_period(period, *, action, user=None, reason=""): period = PayrollPeriod.objects.select_for_update().get(pk=period.pk) action = action.upper().strip() current = period.status if user is None: raise ValidationError("An authenticated decision maker is required for payroll state transitions.") if action == "REVIEW" and not has_role(user, ROLE_SUPER_ADMIN, ROLE_PAYROLL_OFFICER): raise ValidationError("Payroll review must be performed by Payroll Officer / Super Admin.") if action in {"APPROVE", "LOCK", "UNLOCK"} and not has_role(user, ROLE_SUPER_ADMIN, ROLE_HR_ADMIN): raise ValidationError("Payroll approve/lock/unlock must be performed by HR Admin / Super Admin.") if action == "MARK_PAID" and not has_role(user, ROLE_SUPER_ADMIN, ROLE_ACCOUNTS): raise ValidationError("Payroll payment must be posted by Accounts / Super Admin.") policy = get_effective_policy(period.company, date(period.year, period.month, calendar.monthrange(period.year, period.month)[1])) if policy.payroll_segregation_required and not has_role(user, ROLE_SUPER_ADMIN): if action in {"REVIEW", "APPROVE", "LOCK"} and period.prepared_by_id == user.id: raise ValidationError("Segregation of duties: the payroll preparer cannot review/approve/lock the same payroll period.") if action == "APPROVE" and period.reviewed_by_id == user.id: raise ValidationError("Segregation of duties: the payroll reviewer cannot approve the same payroll period.") if action == "MARK_PAID": if current != PayrollPeriod.Status.LOCKED: raise ValidationError("Payroll must be Locked before payment can be posted.") _assert_no_pending_payroll_inputs(period) if period.payment_status == PayrollPeriod.PaymentStatus.PAID: raise ValidationError("Payroll payment has already been posted.") if not reason.strip(): raise ValidationError("Payment/bank reference is required.") period.payment_status = PayrollPeriod.PaymentStatus.PAID period.paid_at = timezone.now() period.payment_reference = reason.strip() period.paid_by = user period.save(update_fields=["payment_status", "paid_at", "payment_reference", "paid_by", "updated_at"]) log_event(user=user, action="MARK_PAID", module="payroll_period", object_id=period.pk, object_repr=f"{period.year}-{period.month:02d}", new_values={"payment_status": period.payment_status, "payment_reference": period.payment_reference}) return period transitions = { "REVIEW": (PayrollPeriod.Status.DRAFT, PayrollPeriod.Status.REVIEWED), "APPROVE": (PayrollPeriod.Status.REVIEWED, PayrollPeriod.Status.APPROVED), "LOCK": (PayrollPeriod.Status.APPROVED, PayrollPeriod.Status.LOCKED), "UNLOCK": (PayrollPeriod.Status.LOCKED, PayrollPeriod.Status.APPROVED), } if action not in transitions: raise ValidationError("Unsupported payroll action.") expected, new_status = transitions[action] if current != expected: raise ValidationError(f"Payroll must be {expected} before {action.lower()}.") if action in {"REVIEW", "APPROVE", "LOCK"}: _assert_payroll_period_inputs_current(period, policy) if action == "UNLOCK" and period.payment_status == PayrollPeriod.PaymentStatus.PAID: raise ValidationError("A paid payroll cannot be unlocked directly. Reverse the payment posting through the controlled finance process first.") if action == "UNLOCK" and not reason.strip(): raise ValidationError("Unlock reason is required.") if action == "UNLOCK": from apps.loans.services import reverse_payroll_recoveries reverse_payroll_recoveries(period, user=user) period.status = new_status actor_fields = [] if action == "REVIEW": period.reviewed_by = user actor_fields.append("reviewed_by") elif action == "APPROVE": period.approved_by = user actor_fields.append("approved_by") if action == "LOCK": period.locked_at = timezone.now() period.locked_by = user period.unlock_reason = "" actor_fields.append("locked_by") elif action == "UNLOCK": period.locked_at = None period.locked_by = None period.unlock_reason = reason.strip() actor_fields.append("locked_by") period.save(update_fields=["status", "locked_at", "unlock_reason", *actor_fields, "updated_at"]) if action == "LOCK": from apps.loans.services import post_payroll_recoveries post_payroll_recoveries(period, user=user) log_event(user=user, action=action, module="payroll_period", object_id=period.pk, object_repr=f"{period.year}-{period.month:02d}", old_values={"status": current}, new_values={"status": new_status}, reason=reason.strip()) return period