Игорь Надеин
3da3100328
All checks were successful
Deploy production / deploy (push) Successful in 11m10s
356 lines
12 KiB
Python
356 lines
12 KiB
Python
"""Mapping between approved Django records and payroll-core DTOs."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections import defaultdict
|
|
from decimal import ROUND_HALF_UP, Context, Decimal, localcontext
|
|
|
|
from finance.enums import ApprovalStatus, InputSource, PayrollComponentKind
|
|
from finance.models import (
|
|
EmployeePayRate,
|
|
PayrollInputLine,
|
|
PayrollPeriod,
|
|
PayrollWorkRecord,
|
|
)
|
|
from payroll_core import (
|
|
ExpectedTotals,
|
|
InputLine,
|
|
LineKind,
|
|
PayrollRequest,
|
|
)
|
|
from payroll_core import (
|
|
PayrollPeriod as CorePayrollPeriod,
|
|
)
|
|
|
|
from .config import base_rate_code
|
|
from .scheduled_changes import effective_period_lines
|
|
|
|
POINT_BASE_NEGATIVE_KINDS = frozenset(
|
|
{
|
|
"adjustment_debit",
|
|
"deduction",
|
|
"payment",
|
|
}
|
|
)
|
|
|
|
|
|
def _core_period(period: PayrollPeriod) -> CorePayrollPeriod:
|
|
return CorePayrollPeriod(start=period.date_from, end=period.date_to)
|
|
|
|
|
|
def _expected_totals(work_record: PayrollWorkRecord) -> ExpectedTotals | None:
|
|
values = (
|
|
work_record.expected_point_amount,
|
|
work_record.expected_gross,
|
|
work_record.expected_recalculated_gross,
|
|
work_record.expected_payable,
|
|
)
|
|
if all(value is None for value in values):
|
|
return None
|
|
return ExpectedTotals(
|
|
point_amount=work_record.expected_point_amount,
|
|
gross=work_record.expected_gross,
|
|
recalculated_gross=work_record.expected_recalculated_gross,
|
|
payable=work_record.expected_payable,
|
|
)
|
|
|
|
|
|
def _component_map(period, component_definitions=None):
|
|
if component_definitions is not None:
|
|
return component_definitions
|
|
from .components import get_payroll_components
|
|
|
|
return {component.code: component for component in get_payroll_components(period)}
|
|
|
|
|
|
def _mapped_input_lines(
|
|
input_lines: list[PayrollInputLine],
|
|
component_definitions,
|
|
) -> tuple[InputLine, ...]:
|
|
relevant_lines = [
|
|
record
|
|
for record in input_lines
|
|
if component_definitions[record.component_code].kind
|
|
!= PayrollComponentKind.SCHEDULED_CHANGE
|
|
]
|
|
signed_codes = {
|
|
record.component_code for record in relevant_lines if record.value_sign < 0
|
|
}
|
|
mapped_lines = []
|
|
for component_code in signed_codes:
|
|
records = [
|
|
record
|
|
for record in relevant_lines
|
|
if record.component_code == component_code
|
|
]
|
|
component = component_definitions[component_code]
|
|
amount = sum((record.signed_amount for record in records), Decimal("0"))
|
|
if amount < 0:
|
|
raise ValueError(
|
|
f"Итоговое значение колонки {component.name} не может быть отрицательным."
|
|
)
|
|
if amount == 0:
|
|
continue
|
|
mapped_lines.append(
|
|
InputLine(
|
|
line_id=(
|
|
f"effective:{component_code}:"
|
|
+ ",".join(str(record.pk) for record in records)
|
|
),
|
|
code=component.code,
|
|
label=component.name,
|
|
kind=LineKind(component.kind),
|
|
amount=amount,
|
|
source_ref="scheduled-effective-value",
|
|
include_in_salary=component.include_in_salary,
|
|
reason="; ".join(
|
|
dict.fromkeys(record.reason for record in records if record.reason)
|
|
),
|
|
include_in_carryover=(
|
|
component.include_in_carryover
|
|
and any(record.include_in_carryover for record in records)
|
|
),
|
|
)
|
|
)
|
|
for record in relevant_lines:
|
|
if record.component_code in signed_codes:
|
|
continue
|
|
component = component_definitions[record.component_code]
|
|
source_period = None
|
|
if record.relates_to_period_id:
|
|
source_period = _core_period(record.relates_to_period)
|
|
mapped_lines.append(
|
|
InputLine(
|
|
line_id=f"input:{record.pk}",
|
|
code=component.code,
|
|
label=component.name,
|
|
kind=LineKind(component.kind),
|
|
amount=record.amount,
|
|
source_ref=(
|
|
record.source_ref or f"{record.source}:payroll-input:{record.pk}"
|
|
),
|
|
reason=record.reason,
|
|
source_period=source_period,
|
|
is_retro=source_period is not None,
|
|
include_in_salary=component.include_in_salary,
|
|
include_in_carryover=(
|
|
component.include_in_carryover and record.include_in_carryover
|
|
),
|
|
)
|
|
)
|
|
return tuple(mapped_lines)
|
|
|
|
|
|
def _effective_point_rate(
|
|
rate: EmployeePayRate | None,
|
|
point_base_accrual: Decimal,
|
|
target_points: Decimal | None,
|
|
*,
|
|
rounding: str = ROUND_HALF_UP,
|
|
) -> Decimal:
|
|
"""Resolve a nullable excess price without persisting the derived value."""
|
|
|
|
if rate is not None and rate.point_rate is not None:
|
|
return rate.point_rate
|
|
if target_points is None or target_points <= 0:
|
|
return Decimal("0")
|
|
with localcontext(Context(prec=80, rounding=rounding)):
|
|
return (point_base_accrual / target_points).quantize(
|
|
Decimal("0.0001"),
|
|
rounding=rounding,
|
|
)
|
|
|
|
|
|
def _point_base_accrual(
|
|
rate: EmployeePayRate | None,
|
|
input_lines: list[PayrollInputLine],
|
|
component_definitions,
|
|
) -> Decimal:
|
|
"""Return the amount whose payout is proportional to actual points.
|
|
|
|
Component settings explicitly decide which columns participate. By
|
|
default only the regular bonus is included; both correction columns remain
|
|
fixed changes of the final amount.
|
|
"""
|
|
|
|
base_amount = rate.amount if rate is not None else Decimal("0")
|
|
point_base_lines = Decimal("0")
|
|
for record in input_lines:
|
|
component = component_definitions[record.component_code]
|
|
if not component.include_in_point_base or not component.include_in_salary:
|
|
continue
|
|
direction = -1 if component.kind in POINT_BASE_NEGATIVE_KINDS else 1
|
|
if component.kind == PayrollComponentKind.SCHEDULED_CHANGE:
|
|
continue
|
|
point_base_lines += record.signed_amount * direction
|
|
return base_amount + point_base_lines
|
|
|
|
|
|
def point_base_accruals_for_employees(
|
|
*,
|
|
period: PayrollPeriod,
|
|
employee_ids,
|
|
component_definitions=None,
|
|
) -> dict[int, Decimal]:
|
|
"""Return current editable point bases using the table-preview semantics."""
|
|
|
|
normalized_ids = {int(employee_id) for employee_id in employee_ids}
|
|
if not normalized_ids:
|
|
return {}
|
|
|
|
grouped_rates = defaultdict(list)
|
|
rates = EmployeePayRate.objects.filter(
|
|
employee_id__in=normalized_ids,
|
|
rate_code=base_rate_code(),
|
|
status__in=[ApprovalStatus.DRAFT, ApprovalStatus.APPROVED],
|
|
effective_from__lte=period.date_from,
|
|
).order_by("employee_id", "effective_from", "revision", "id")
|
|
for rate in rates:
|
|
grouped_rates[rate.employee_id].append(rate)
|
|
selected_rates = {
|
|
employee_id: max(
|
|
records,
|
|
key=lambda record: (
|
|
record.effective_from,
|
|
record.revision,
|
|
record.pk,
|
|
),
|
|
)
|
|
for employee_id, records in grouped_rates.items()
|
|
}
|
|
|
|
grouped_lines = defaultdict(list)
|
|
lines = (
|
|
PayrollInputLine.objects.filter(
|
|
period=period,
|
|
employee_id__in=normalized_ids,
|
|
status__in=[ApprovalStatus.DRAFT, ApprovalStatus.APPROVED],
|
|
)
|
|
.exclude(source=InputSource.API, source_ref__startswith="payroll-carryover:")
|
|
.select_related("relates_to_period")
|
|
.order_by("employee_id", "component_code", "id")
|
|
)
|
|
for line in lines:
|
|
grouped_lines[line.employee_id].append(line)
|
|
|
|
components_by_code = _component_map(period, component_definitions)
|
|
return {
|
|
employee_id: _point_base_accrual(
|
|
selected_rates.get(employee_id),
|
|
grouped_lines.get(employee_id, []),
|
|
components_by_code,
|
|
)
|
|
for employee_id in normalized_ids
|
|
}
|
|
|
|
|
|
def build_core_request(
|
|
*,
|
|
period: PayrollPeriod,
|
|
rate: EmployeePayRate,
|
|
work_record: PayrollWorkRecord | None,
|
|
input_lines: list[PayrollInputLine],
|
|
opening_balance: Decimal = Decimal("0"),
|
|
rounding: str = ROUND_HALF_UP,
|
|
component_definitions=None,
|
|
) -> PayrollRequest:
|
|
components_by_code = _component_map(period, component_definitions)
|
|
input_lines = effective_period_lines(
|
|
period,
|
|
rate.employee,
|
|
input_lines,
|
|
scheduled_codes={
|
|
code
|
|
for code, component in components_by_code.items()
|
|
if component.kind == PayrollComponentKind.SCHEDULED_CHANGE
|
|
},
|
|
)
|
|
target_points = work_record.target_points if work_record else None
|
|
point_base_accrual = _point_base_accrual(
|
|
rate,
|
|
input_lines,
|
|
components_by_code,
|
|
)
|
|
return PayrollRequest(
|
|
employee_ref=f"employee:{rate.employee_id}",
|
|
balance_rules_version=period.balance_rules_version,
|
|
period=_core_period(period),
|
|
currency=period.currency,
|
|
base_accrual=rate.amount,
|
|
base_source_ref=f"pay-rate:{rate.pk}:revision:{rate.revision}",
|
|
opening_balance=opening_balance,
|
|
point_base_accrual=point_base_accrual,
|
|
target_points=target_points,
|
|
actual_points=(work_record.actual_points if work_record else None),
|
|
point_rate=_effective_point_rate(
|
|
rate,
|
|
point_base_accrual,
|
|
target_points,
|
|
rounding=rounding,
|
|
),
|
|
point_rate_overridden=rate.point_rate is not None,
|
|
lines=_mapped_input_lines(input_lines, components_by_code),
|
|
expected_totals=(
|
|
_expected_totals(work_record) if work_record is not None else None
|
|
),
|
|
)
|
|
|
|
|
|
def build_core_preview_request(
|
|
*,
|
|
period: PayrollPeriod,
|
|
employee_id: int,
|
|
rate: EmployeePayRate | None,
|
|
target_points: Decimal,
|
|
actual_points: Decimal,
|
|
input_lines: list[PayrollInputLine],
|
|
opening_balance: Decimal = Decimal("0"),
|
|
rounding: str = ROUND_HALF_UP,
|
|
component_definitions=None,
|
|
) -> PayrollRequest:
|
|
"""Build a non-persisted calculation request from draft or partial inputs."""
|
|
|
|
components_by_code = _component_map(period, component_definitions)
|
|
point_base_accrual = _point_base_accrual(
|
|
rate,
|
|
input_lines,
|
|
components_by_code,
|
|
)
|
|
return PayrollRequest(
|
|
employee_ref=f"employee:{employee_id}",
|
|
balance_rules_version=period.balance_rules_version,
|
|
period=_core_period(period),
|
|
currency=period.currency,
|
|
base_accrual=rate.amount if rate is not None else Decimal("0"),
|
|
base_source_ref=(
|
|
f"pay-rate:{rate.pk}:revision:{rate.revision}"
|
|
if rate is not None
|
|
else "preview:missing-pay-rate"
|
|
),
|
|
opening_balance=opening_balance,
|
|
point_base_accrual=point_base_accrual,
|
|
target_points=target_points,
|
|
actual_points=actual_points,
|
|
point_rate=_effective_point_rate(
|
|
rate,
|
|
point_base_accrual,
|
|
target_points,
|
|
rounding=rounding,
|
|
),
|
|
point_rate_overridden=bool(rate is not None and rate.point_rate is not None),
|
|
lines=_mapped_input_lines(input_lines, components_by_code),
|
|
expected_totals=None,
|
|
)
|
|
|
|
|
|
def employee_snapshot(employee) -> dict[str, object]:
|
|
"""Return the minimum identity snapshot needed for a historical payslip."""
|
|
|
|
display_name = employee.get_full_name().strip() or f"Сотрудник #{employee.pk}"
|
|
position = getattr(employee, "position", None)
|
|
return {
|
|
"employee_id": employee.pk,
|
|
"display_name": display_name,
|
|
"position": ({"id": position.pk, "name": position.name} if position else None),
|
|
}
|