"""Payroll rules. net = base salary + overtime pay + additions - lateness deduction - deductions hourly rate = base salary / monthly_standard_hours; overtime pay = overtime hours x rate x multiplier; lateness deduction = late hours x rate (if enabled). Worked/late/overtime minutes come from the attendance report. Absences are not deducted automatically (weekends differ per company); use the manual `deductions` field. Rates are frozen on the run when it is created. Scope of payroll.view: `self` = my payslip of approved runs; `organization` = everything. """ from __future__ import annotations import datetime from decimal import ROUND_HALF_UP, Decimal from django.db import transaction from django.utils import timezone from rest_framework.exceptions import NotFound, ValidationError from application.use_cases.reports import BuildMonthlyReport from apps.audit import services as audit from apps.organizations.models import OrganizationMembership from apps.payroll.models import PayrollItem, PayrollRun, PayrollSettings from apps.personnel.models import Contract from domain.access_policy import scope_rank from domain.exceptions import DomainError from domain.value_objects import jalali_month_to_gregorian_range from interfaces.api.support import repositories_for class PayrollStateError(DomainError): pass def _sees_all(ctx) -> bool: scope = ctx.effective.scope_of("payroll.view") return scope is not None and scope_rank(scope) >= scope_rank("organization") def visible_runs(ctx): qs = PayrollRun.objects.filter(organization=ctx.organization) return qs if _sees_all(ctx) else qs.filter(status="approved") def get_run(ctx, pk: int, *, for_update: bool = False) -> PayrollRun: qs = PayrollRun.objects.filter(organization=ctx.organization) if for_update else visible_runs(ctx) run = qs.filter(pk=pk).first() if run is None: raise NotFound() return run def visible_items(ctx, run: PayrollRun): qs = run.items.select_related("membership__user") return qs if _sees_all(ctx) else qs.filter(membership__user=ctx.user) def get_settings(ctx) -> PayrollSettings: return PayrollSettings.objects.get_or_create(organization=ctx.organization)[0] def update_settings(ctx, *, data: dict, request=None) -> PayrollSettings: settings = get_settings(ctx) if "monthly_standard_hours" in data: if not 1 <= data["monthly_standard_hours"] <= 744: raise ValidationError({"monthly_standard_hours": "Must be between 1 and 744."}) settings.monthly_standard_hours = data["monthly_standard_hours"] if "overtime_multiplier" in data: if not Decimal("1") <= data["overtime_multiplier"] <= Decimal("5"): raise ValidationError({"overtime_multiplier": "Must be between 1 and 5."}) settings.overtime_multiplier = data["overtime_multiplier"] if "deduct_lateness" in data: settings.deduct_lateness = data["deduct_lateness"] settings.save() audit.record( actor=ctx.user, action="payroll.settings", organization=ctx.organization, entity_type="payroll_settings", entity_id=settings.pk, after={k: str(v) for k, v in data.items()}, request=request, ) return settings def create_run(ctx, *, year: int, month: int, request=None) -> PayrollRun: if not (1 <= month <= 12 and 1300 <= year <= 1500): raise ValidationError({"month": "Invalid Jalali year/month."}) if PayrollRun.objects.filter(organization=ctx.organization, year=year, month=month).exists(): raise ValidationError({"month": "A payroll run already exists for this month."}) settings = get_settings(ctx) run = PayrollRun.objects.create( organization=ctx.organization, year=year, month=month, created_by=ctx.user, hourly_divisor=settings.monthly_standard_hours, overtime_multiplier=settings.overtime_multiplier, deduct_lateness=settings.deduct_lateness, ) audit.record( actor=ctx.user, action="payroll.run_create", organization=ctx.organization, entity_type="payroll_run", entity_id=run.pk, after={"year": year, "month": month}, request=request, ) return run def _hm_to_minutes(value: str | None) -> int: if not value: return 0 hours, minutes = value.split(":") return int(hours) * 60 + int(minutes) def _money(value: Decimal) -> int: return int(value.quantize(Decimal("1"), rounding=ROUND_HALF_UP)) def _active_contract(membership, start: datetime.date, end: datetime.date): return ( Contract.objects.filter(membership=membership, status="active", start_date__lte=end) .exclude(end_date__lt=start) .order_by("-start_date", "-id") .first() ) def _recompute_net(run: PayrollRun, item: PayrollItem) -> None: item.net = item.base_salary + item.overtime_pay + item.additions - item.late_deduction - item.deductions def _standard_daily_hours(repos) -> Decimal: """Length of the organization's default working day (8 hours when no schedule is set).""" schedule = repos.schedule.get_default() if schedule is None: return Decimal(8) start = datetime.datetime.combine(datetime.date.today(), schedule.standard_check_in) end = datetime.datetime.combine(datetime.date.today(), schedule.standard_check_out) minutes = (end - start).total_seconds() / 60 return Decimal(minutes) / 60 if minutes >= 60 else Decimal(8) def compute_pay(pay_type: str, amount: int, stats: dict, run: PayrollRun, daily_hours: Decimal) -> dict: """Earnings for one person from the contract's pay type and the attendance figures. monthly: base = amount; hourly rate = amount / monthly standard hours. daily: base = amount x paid days (worked + paid leave); hourly rate = amount / working-day hours. hourly: base = amount x regular hours (worked minus overtime); no lateness deduction because unworked time is simply not paid. Overtime = overtime hours x hourly rate x multiplier; lateness = late hours x hourly rate (monthly/daily). """ amount_d = Decimal(amount) overtime_h = Decimal(stats["overtime"]) / 60 late_h = Decimal(stats["late"]) / 60 paid_days = 0 if pay_type == "hourly": regular_h = Decimal(max(stats["worked"] - stats["overtime"], 0)) / 60 base, rate, late_pay = amount_d * regular_h, amount_d, Decimal(0) elif pay_type == "daily": paid_days = stats["present_days"] + stats["paid_leave_days"] base, rate = amount_d * paid_days, amount_d / daily_hours late_pay = rate * late_h if run.deduct_lateness else Decimal(0) else: base, rate = amount_d, amount_d / Decimal(run.hourly_divisor) late_pay = rate * late_h if run.deduct_lateness else Decimal(0) return { "base": _money(base), "paid_days": paid_days, "overtime_pay": _money(rate * overtime_h * run.overtime_multiplier), "late_deduction": _money(late_pay), } @transaction.atomic def calculate_run(ctx, run: PayrollRun, *, request=None) -> dict: """(Re)build items from contracts and attendance. Manual additions/deductions are kept.""" if run.status != "draft": raise PayrollStateError("An approved payroll run cannot be recalculated.") start, end = jalali_month_to_gregorian_range(run.year, run.month) repos = repositories_for(ctx.organization) report = BuildMonthlyReport(repos.attendance, repos.holiday, repos.leave, repos.schedule) daily_hours = _standard_daily_hours(repos) existing = {i.membership_id: i for i in run.items.all()} seen, skipped = set(), [] memberships = OrganizationMembership.objects.filter(organization=ctx.organization).exclude(status="left").select_related("user") for membership in memberships: contract = _active_contract(membership, start, end) if contract is None or not contract.pay_amount: skipped.append(membership.pk) continue rows = report.execute(membership.user_id, run.year, run.month).rows stats = { "worked": sum(_hm_to_minutes(r.worked) for r in rows), "late": sum(_hm_to_minutes(r.late) for r in rows), "overtime": sum(_hm_to_minutes(r.overtime) for r in rows), "present_days": sum(1 for r in rows if r.check_in), "paid_leave_days": sum( 1 for r in rows if not r.check_in and any(lv.type == "daily" and lv.category != "unpaid" for lv in r.leaves) ), } pay = compute_pay(contract.pay_type, contract.pay_amount, stats, run, daily_hours) item = existing.get(membership.pk) or PayrollItem(run=run, membership=membership) item.pay_type, item.unit_rate = contract.pay_type, contract.pay_amount item.paid_days = pay["paid_days"] item.base_salary = pay["base"] item.worked_minutes, item.late_minutes, item.overtime_minutes = stats["worked"], stats["late"], stats["overtime"] item.overtime_pay, item.late_deduction = pay["overtime_pay"], pay["late_deduction"] _recompute_net(run, item) item.save() seen.add(membership.pk) stale = [pk for pk in existing if pk not in seen] if stale: # no longer has an active contract: drop the line run.items.filter(membership_id__in=stale).delete() audit.record( actor=ctx.user, action="payroll.calculate", organization=ctx.organization, entity_type="payroll_run", entity_id=run.pk, after={"items": len(seen), "skipped": len(skipped)}, request=request, ) return {"items": len(seen), "skipped": len(skipped)} def update_item(ctx, item: PayrollItem, *, data: dict, request=None) -> PayrollItem: if item.run.status != "draft": raise PayrollStateError("Items of an approved payroll run cannot be changed.") for field in ("additions", "deductions"): if field in data: if data[field] < 0: raise ValidationError({field: "Cannot be negative."}) setattr(item, field, data[field]) if "notes" in data: item.notes = data["notes"] _recompute_net(item.run, item) item.save() audit.record( actor=ctx.user, action="payroll.item_update", organization=ctx.organization, entity_type="payroll_item", entity_id=item.pk, after={k: str(v) for k, v in data.items() if k != "notes"}, request=request, ) return item def get_item(ctx, pk: int) -> PayrollItem: item = PayrollItem.objects.filter(pk=pk, run__organization=ctx.organization).select_related("run", "membership__user").first() if item is None: raise NotFound() return item @transaction.atomic def approve_run(ctx, run: PayrollRun, *, request=None) -> PayrollRun: if run.status != "draft": raise PayrollStateError("The payroll run is already approved.") if not run.items.exists(): raise PayrollStateError("Calculate the payroll run before approving it.") run.status = "approved" run.approved_by = ctx.user run.approved_at = timezone.now() run.save() audit.record(actor=ctx.user, action="payroll.approve", organization=ctx.organization, entity_type="payroll_run", entity_id=run.pk, request=request) return run def delete_run(ctx, run: PayrollRun, *, request=None) -> None: if run.status != "draft": raise PayrollStateError("An approved payroll run cannot be deleted.") pk = run.pk run.delete() audit.record(actor=ctx.user, action="payroll.run_delete", organization=ctx.organization, entity_type="payroll_run", entity_id=pk, request=request)