"""Payroll business rules. Salary formula (per employee, per calendar month) ------------------------------------------------- gross = basic + house rent + medical + transport + other allowance per_day = gross / days_in_month deduction_days = absent + unpaid leave + 1/2 x half days + days before joining / after leaving + late_days // late_days_per_absence (if enabled) + unmarked working days (only if "unmarked = absent") absence_deduction = per_day x deduction_days overtime_amount = overtime_hours x (basic / (days x work_hours)) x multiplier net = gross - absence_deduction + overtime + bonus - advance_deduction - other_deduction Weekly-off days and holidays are paid. Paid leave is paid. Days with no attendance record count as present unless the company turns on "unmarked = absent" in Payroll Settings. """ import calendar from datetime import date from decimal import ROUND_HALF_UP, Decimal from django.db import transaction from django.db.models import Q, Sum from accounts.branch_utils import scope from .models import ( Attendance, Employee, Holiday, PayrollSetting, Payslip, SalaryAdvance, ) ZERO = Decimal("0") CENT = Decimal("0.01") def q2(value): return Decimal(value).quantize(CENT, rounding=ROUND_HALF_UP) # -------------------------------------------------------------------------- # Small helpers # -------------------------------------------------------------------------- def month_bounds(year, month): days = calendar.monthrange(year, month)[1] return date(year, month, 1), date(year, month, days), days def parse_month(value, today=None): """'2026-09' -> (2026, 9). Falls back to the current month on bad input.""" today = today or date.today() try: y, m = str(value).split("-") y, m = int(y), int(m) if 1 <= m <= 12 and 2000 <= y <= 2100: return y, m except (ValueError, AttributeError): pass return today.year, today.month def shift_month(year, month, delta): idx = year * 12 + (month - 1) + delta return idx // 12, idx % 12 + 1 def get_setting(company): setting, _ = PayrollSetting.objects.get_or_create(company=company) return setting def employees_in_scope(company, branch, *, active_only=False): qs = scope(Employee.objects.filter(company=company), branch) # Employees always have a branch (never NULL), so the legacy-NULL rule in # scope() simply never matches here. if active_only: qs = qs.filter(status=Employee.STATUS_ACTIVE) return qs def next_employee_code(company): highest = 0 for code in Employee.objects.filter(company=company).values_list("employee_code", flat=True): digits = "".join(ch for ch in code if ch.isdigit()) if digits: highest = max(highest, int(digits)) return f"EMP-{highest + 1:04d}" def holiday_dates(company, branch, start, end): """Set of holiday dates in [start, end] that apply to ``branch``. A holiday with no branch applies to every branch. """ qs = Holiday.objects.filter(company=company, date__range=[start, end]) if branch is not None: qs = qs.filter(Q(branch=branch.pk) | Q(branch__isnull=True)) return set(qs.values_list("date", flat=True)) def employee_holidays(company, employee, start, end): return set( Holiday.objects.filter(company=company, date__range=[start, end]) .filter(Q(branch=employee.branch_id) | Q(branch__isnull=True)) .values_list("date", flat=True) ) # -------------------------------------------------------------------------- # Salary calculation # -------------------------------------------------------------------------- def attendance_breakdown(employee, year, month, setting, att_by_date, holidays): """Classify every day of the month for one employee. Returns a dict of counts.""" start, end, days = month_bounds(year, month) off_days = setting.weekly_off_set c = { "present": 0, "late": 0, "half": 0, "absent": 0, "paid_leave": 0, "unpaid_leave": 0, "weekly_off": 0, "holiday": 0, "unmarked": 0, "not_employed": 0, "overtime_hours": ZERO, "days": days, } for day in range(1, days + 1): d = date(year, month, day) if d < employee.join_date or (employee.left_date and d > employee.left_date): c["not_employed"] += 1 continue rec = att_by_date.get(d) is_off = d.weekday() in off_days is_holiday = d in holidays if rec: c["overtime_hours"] += rec.overtime_hours or ZERO if is_off or is_holiday: # Paid day off. If the person worked, it just counts as attended. if rec and rec.status in (Attendance.PRESENT, Attendance.LATE, Attendance.HALF_DAY): c["present"] += 1 else: c["weekly_off" if is_off else "holiday"] += 1 continue if rec is None: c["unmarked"] += 1 continue if rec.status == Attendance.PRESENT: c["present"] += 1 elif rec.status == Attendance.LATE: c["late"] += 1 elif rec.status == Attendance.HALF_DAY: c["half"] += 1 elif rec.status == Attendance.ABSENT: c["absent"] += 1 elif rec.status == Attendance.PAID_LEAVE: c["paid_leave"] += 1 elif rec.status == Attendance.UNPAID_LEAVE: c["unpaid_leave"] += 1 return c def advance_due_for_month(advances, month_end): """Default advance instalment to recover in a payslip.""" total = ZERO for adv in advances: if adv.date <= month_end: total += adv.next_installment return total def compute_payslip_values(employee, year, month, setting, att_by_date, holidays, advances=(), bonus=ZERO, other_deduction=ZERO): """Pure calculation - returns a dict of Payslip field values.""" start, end, days = month_bounds(year, month) c = attendance_breakdown(employee, year, month, setting, att_by_date, holidays) gross = q2(employee.gross_salary) per_day = gross / Decimal(days) late_penalty = 0 if setting.late_days_per_absence: late_penalty = c["late"] // setting.late_days_per_absence ded_days = ( Decimal(c["absent"] + c["unpaid_leave"] + c["not_employed"] + late_penalty) + Decimal("0.5") * c["half"] ) if setting.unmarked_as_absent: ded_days += c["unmarked"] ded_days = min(ded_days, Decimal(days)) absence_deduction = q2(per_day * ded_days) hourly_base = employee.basic_salary or gross hourly = Decimal(hourly_base) / (Decimal(days) * Decimal(setting.work_hours_per_day or 8)) overtime_amount = q2(c["overtime_hours"] * hourly * Decimal(setting.overtime_multiplier or 1)) bonus = q2(bonus or ZERO) other_deduction = q2(other_deduction or ZERO) earnings = gross + overtime_amount + bonus available = max(ZERO, earnings - absence_deduction - other_deduction) advance_deduction = q2(min(advance_due_for_month(advances, end), available)) net = earnings - absence_deduction - advance_deduction - other_deduction return { "basic": employee.basic_salary, "house_rent": employee.house_rent, "medical": employee.medical_allowance, "transport": employee.transport_allowance, "other_allowance": employee.other_allowance, "gross_salary": gross, "days_in_month": days, "present_days": c["present"], "late_days": c["late"], "half_days": c["half"], "absent_days": c["absent"], "paid_leave_days": c["paid_leave"], "unpaid_leave_days": c["unpaid_leave"], "weekly_off_days": c["weekly_off"], "holiday_days": c["holiday"], "unmarked_days": c["unmarked"], "not_employed_days": c["not_employed"], "deduction_days": ded_days, "overtime_hours": c["overtime_hours"], "absence_deduction": absence_deduction, "overtime_amount": overtime_amount, "bonus": bonus, "advance_deduction": advance_deduction, "other_deduction": other_deduction, "net_salary": q2(net), } def recalc_net(payslip): """Recompute net_salary after a manual edit of bonus / deductions.""" net = ( (payslip.gross_salary or ZERO) + (payslip.overtime_amount or ZERO) + (payslip.bonus or ZERO) - (payslip.absence_deduction or ZERO) - (payslip.advance_deduction or ZERO) - (payslip.other_deduction or ZERO) ) payslip.net_salary = q2(net) return payslip def eligible_employees(company, branch, year, month): """Employees who should appear on the salary sheet of this month.""" start, end, _ = month_bounds(year, month) qs = employees_in_scope(company, branch).filter(join_date__lte=end) qs = qs.filter(Q(left_date__isnull=True) | Q(left_date__gte=start)) # Active staff, plus anyone who left *during or after* this month. return qs.filter(Q(status=Employee.STATUS_ACTIVE) | Q(left_date__gte=start)) def generate_payroll(company, branch, year, month, user=None): """Create / refresh DRAFT payslips for every eligible employee. Approved or paid payslips are never touched. Manual bonus and other deductions on existing drafts are preserved. """ start, end, _ = month_bounds(year, month) setting = get_setting(company) created = updated = skipped = 0 with transaction.atomic(using=company.db_alias): for emp in eligible_employees(company, branch, year, month): existing = Payslip.objects.filter(employee=emp, year=year, month=month).first() if existing and existing.is_locked: skipped += 1 continue att = { a.date: a for a in Attendance.objects.filter( employee=emp, date__range=[start, end] ) } holidays = employee_holidays(company, emp, start, end) advances = [ a for a in SalaryAdvance.objects.filter(company=company, employee=emp) if a.outstanding > 0 ] values = compute_payslip_values( emp, year, month, setting, att, holidays, advances, bonus=existing.bonus if existing else ZERO, other_deduction=existing.other_deduction if existing else ZERO, ) if existing: for k, v in values.items(): setattr(existing, k, v) existing.branch_id = emp.branch_id existing.save() updated += 1 else: Payslip.objects.create( company=company, branch_id=emp.branch_id, employee=emp, year=year, month=month, created_by=user, **values, ) created += 1 return {"created": created, "updated": updated, "skipped": skipped} # -------------------------------------------------------------------------- # Status changes # -------------------------------------------------------------------------- def _recover_advances(company, employee, amount): """Apply ``amount`` against the employee's open advances, oldest first.""" remaining = Decimal(amount) if remaining <= 0: return advances = ( SalaryAdvance.objects.select_for_update() .filter(company=company, employee=employee).order_by("date", "created_at") ) for adv in advances: if remaining <= 0: break take = min(adv.outstanding, remaining) if take > 0: adv.recovered += take adv.save(update_fields=["recovered"]) remaining -= take if remaining > 0: raise ValueError( f"{employee.full_name}: advance deduction is more than the outstanding advance. " "Refresh the salary sheet or edit the payslip." ) def _restore_advances(company, employee, amount): """Undo a recovery (used when a paid payslip is reverted), newest first.""" remaining = Decimal(amount) if remaining <= 0: return advances = ( SalaryAdvance.objects.select_for_update() .filter(company=company, employee=employee, recovered__gt=0).order_by("-date", "-created_at") ) for adv in advances: if remaining <= 0: break give_back = min(adv.recovered, remaining) adv.recovered -= give_back adv.save(update_fields=["recovered"]) remaining -= give_back def approve_payslip(payslip): if payslip.status != Payslip.DRAFT: raise ValueError("Only draft payslips can be approved.") payslip.status = Payslip.APPROVED payslip.save(update_fields=["status", "updated_at"]) def reopen_payslip(payslip): if payslip.status != Payslip.APPROVED: raise ValueError("Only approved payslips can be re-opened.") payslip.status = Payslip.DRAFT payslip.save(update_fields=["status", "updated_at"]) def pay_payslip(company, payslip, paid_date, method, reference=""): """Mark paid and record the advance recovery. Call inside atomic(using=alias).""" payslip = Payslip.objects.select_for_update().select_related("employee").get(pk=payslip.pk) if payslip.status == Payslip.PAID: raise ValueError("This payslip is already paid.") if payslip.net_salary < 0: raise ValueError("Net salary cannot be negative.") _recover_advances(company, payslip.employee, payslip.advance_deduction) payslip.status = Payslip.PAID payslip.paid_date = paid_date payslip.payment_method = method or payslip.employee.payment_method payslip.payment_reference = reference payslip.save() return payslip def revert_payment(company, payslip): payslip = Payslip.objects.select_for_update().select_related("employee").get(pk=payslip.pk) if payslip.status != Payslip.PAID: raise ValueError("Only paid payslips can be reverted.") _restore_advances(company, payslip.employee, payslip.advance_deduction) payslip.status = Payslip.APPROVED payslip.paid_date = None payslip.payment_reference = "" payslip.save() return payslip # -------------------------------------------------------------------------- # Numbers for dashboards / reports # -------------------------------------------------------------------------- def payroll_cost(company, start, end, branch=None): """Salary cost of PAID payslips whose payment date is in [start, end]. Cost = net salary + advance recovered (an advance is a prepayment of salary, so it is still salary cost when it is recovered). """ qs = scope( Payslip.objects.filter( company=company, status=Payslip.PAID, paid_date__range=[start, end] ), branch, ) agg = qs.aggregate(net=Sum("net_salary"), adv=Sum("advance_deduction")) return (agg["net"] or ZERO) + (agg["adv"] or ZERO) def attendance_for_day(company, branch, day): """Counts of each attendance status for active staff on one day.""" emps = employees_in_scope(company, branch, active_only=True).filter(join_date__lte=day) emp_ids = list(emps.values_list("pk", flat=True)) counts = {k: 0 for k, _ in Attendance.STATUS_CHOICES} rows = Attendance.objects.filter(company=company, date=day, employee_id__in=emp_ids) for status in rows.values_list("status", flat=True): counts[status] = counts.get(status, 0) + 1 marked = sum(counts.values()) counts["total"] = len(emp_ids) counts["unmarked"] = len(emp_ids) - marked return counts def month_locked_employee_ids(company, year, month): """Employees whose payslip for this month is approved/paid (attendance is frozen).""" return set( Payslip.objects.filter(company=company, year=year, month=month) .exclude(status=Payslip.DRAFT).values_list("employee_id", flat=True) )