248 lines
8.5 KiB
Python
248 lines
8.5 KiB
Python
"""Helpers for signed, one-period payroll component changes."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from decimal import Decimal
|
|
|
|
from django.db.models import Q
|
|
|
|
from employees.constants import ACTION_DISMISSED
|
|
from employees.services.personnel_state import resolve_employee_personnel_state
|
|
from finance.enums import ApprovalStatus, InputSource, PayrollComponentKind
|
|
from finance.models import PayrollCustomComponent, PayrollInputLine, PayrollPeriod
|
|
from requests_app.enums import RequestStatus
|
|
|
|
SCHEDULED_CHANGE_SOURCE_PREFIX = "payroll-scheduled-change:"
|
|
|
|
|
|
def next_payroll_period(source_period: PayrollPeriod, *, for_update: bool = False):
|
|
"""Return the nearest later period in the same payroll timeline."""
|
|
|
|
periods = PayrollPeriod.objects.filter(
|
|
currency=source_period.currency,
|
|
date_from__gt=source_period.date_to,
|
|
).order_by("date_from", "pk")
|
|
if for_update:
|
|
periods = periods.select_for_update()
|
|
return periods.first()
|
|
|
|
|
|
def materialize_scheduled_changes(
|
|
*,
|
|
source_period: PayrollPeriod,
|
|
target_period: PayrollPeriod,
|
|
) -> list[PayrollInputLine]:
|
|
"""Apply source markers once when their immediately next period exists."""
|
|
|
|
next_period = next_payroll_period(source_period, for_update=True)
|
|
if next_period is None or next_period.pk != target_period.pk:
|
|
return []
|
|
|
|
from finance.payroll.components import get_payroll_component, get_payroll_components
|
|
|
|
scheduled_components = {
|
|
component.code: component
|
|
for component in get_payroll_components(source_period)
|
|
if component.kind == PayrollComponentKind.SCHEDULED_CHANGE
|
|
}
|
|
if not scheduled_components:
|
|
return []
|
|
|
|
target_components = {
|
|
component.code: component for component in get_payroll_components(target_period)
|
|
}
|
|
source_lines = list(
|
|
PayrollInputLine.objects.select_for_update(of=("self",))
|
|
.select_related("created_by", "approved_by")
|
|
.filter(
|
|
period=source_period,
|
|
component_code__in=scheduled_components,
|
|
status__in=(ApprovalStatus.DRAFT, ApprovalStatus.APPROVED),
|
|
)
|
|
.order_by("pk")
|
|
)
|
|
existing_refs = set(
|
|
PayrollInputLine.objects.filter(
|
|
source=InputSource.API,
|
|
source_ref__in=[
|
|
f"{SCHEDULED_CHANGE_SOURCE_PREFIX}{line.pk}" for line in source_lines
|
|
],
|
|
).values_list("source_ref", flat=True)
|
|
)
|
|
generated = []
|
|
for source_line in source_lines:
|
|
source_ref = f"{SCHEDULED_CHANGE_SOURCE_PREFIX}{source_line.pk}"
|
|
if source_ref in existing_refs:
|
|
continue
|
|
scheduled_component = scheduled_components[source_line.component_code]
|
|
target_code = scheduled_component.scheduled_target_component_code
|
|
target = target_components.get(target_code)
|
|
if target is None:
|
|
target = get_payroll_component(target_code, period=target_period)
|
|
if target.kind == PayrollComponentKind.SCHEDULED_CHANGE:
|
|
raise ValueError(
|
|
"Запланированное изменение не может изменять другой маркер."
|
|
)
|
|
line = PayrollInputLine(
|
|
period=target_period,
|
|
employee_id=source_line.employee_id,
|
|
component_code=target_code,
|
|
amount=source_line.amount,
|
|
value_sign=source_line.value_sign,
|
|
reason=(
|
|
f"Применено из колонки «{scheduled_component.name}» "
|
|
f"периода {source_period.code}"
|
|
),
|
|
status=source_line.status,
|
|
source=InputSource.API,
|
|
source_ref=source_ref,
|
|
include_in_carryover=False,
|
|
created_by=source_line.created_by,
|
|
approved_by=source_line.approved_by,
|
|
approved_at=source_line.approved_at,
|
|
self_approval_overridden=source_line.self_approval_overridden,
|
|
)
|
|
line.full_clean()
|
|
line.save(force_insert=True)
|
|
generated.append(line)
|
|
return generated
|
|
|
|
|
|
def is_generated_scheduled_change(line: PayrollInputLine) -> bool:
|
|
return line.source_ref.startswith(SCHEDULED_CHANGE_SOURCE_PREFIX)
|
|
|
|
|
|
def generated_source_line_id(line: PayrollInputLine) -> int | None:
|
|
if not is_generated_scheduled_change(line):
|
|
return None
|
|
raw_id = line.source_ref.removeprefix(SCHEDULED_CHANGE_SOURCE_PREFIX)
|
|
try:
|
|
return int(raw_id)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def scheduled_component_for_period(code: str, period: PayrollPeriod):
|
|
from finance.payroll.components import get_payroll_component
|
|
|
|
try:
|
|
component = get_payroll_component(code, period=period)
|
|
except ValueError:
|
|
return None
|
|
if component.kind != PayrollComponentKind.SCHEDULED_CHANGE:
|
|
return None
|
|
return component
|
|
|
|
|
|
def scheduled_components_targeting(code: str):
|
|
from finance.payroll.components import get_payroll_components
|
|
|
|
return [
|
|
component
|
|
for component in get_payroll_components()
|
|
if component.kind == PayrollComponentKind.SCHEDULED_CHANGE
|
|
and component.scheduled_target_component_code == code
|
|
]
|
|
|
|
|
|
def employee_is_eligible_for_scheduled_change(employee, period: PayrollPeriod) -> bool:
|
|
if not employee.is_active:
|
|
return False
|
|
actions = getattr(employee, "payroll_personnel_actions", None)
|
|
if actions is None:
|
|
actions = list(
|
|
employee.actions.filter(
|
|
Q(source_request__isnull=True)
|
|
| Q(source_request__status=RequestStatus.APPROVED)
|
|
).select_related("source_request")
|
|
)
|
|
state = resolve_employee_personnel_state(employee, period.date_to, actions=actions)
|
|
return state.status != ACTION_DISMISSED
|
|
|
|
|
|
def effective_period_lines(
|
|
period: PayrollPeriod,
|
|
employee,
|
|
lines: list[PayrollInputLine],
|
|
*,
|
|
scheduled_codes: set[str] | None = None,
|
|
) -> list[PayrollInputLine]:
|
|
"""Remove marker inputs and inapplicable generated effects from calculation."""
|
|
|
|
eligible = employee_is_eligible_for_scheduled_change(employee, period)
|
|
if scheduled_codes is None:
|
|
scheduled_codes = set(
|
|
PayrollCustomComponent.objects.filter(
|
|
kind=PayrollComponentKind.SCHEDULED_CHANGE,
|
|
).values_list("code", flat=True)
|
|
)
|
|
return [
|
|
line
|
|
for line in lines
|
|
if line.component_code not in scheduled_codes
|
|
and (eligible or not is_generated_scheduled_change(line))
|
|
]
|
|
|
|
|
|
def target_cell_scheduled_delta(
|
|
*,
|
|
period: PayrollPeriod,
|
|
employee_id: int,
|
|
component_code: str,
|
|
) -> Decimal:
|
|
lines = PayrollInputLine.objects.filter(
|
|
period=period,
|
|
employee_id=employee_id,
|
|
component_code=component_code,
|
|
status__in=(ApprovalStatus.DRAFT, ApprovalStatus.APPROVED),
|
|
source_ref__startswith=SCHEDULED_CHANGE_SOURCE_PREFIX,
|
|
)
|
|
return sum((line.signed_amount for line in lines), Decimal("0"))
|
|
|
|
|
|
def applied_change_payloads(lines: list[PayrollInputLine]) -> list[dict[str, object]]:
|
|
from finance.payroll.components import get_payroll_components
|
|
|
|
generated = [line for line in lines if is_generated_scheduled_change(line)]
|
|
source_ids = {
|
|
source_id
|
|
for line in generated
|
|
if (source_id := generated_source_line_id(line)) is not None
|
|
}
|
|
sources = {
|
|
line.pk: line
|
|
for line in PayrollInputLine.objects.filter(pk__in=source_ids).select_related(
|
|
"period"
|
|
)
|
|
}
|
|
components_by_period = {
|
|
period_id: {
|
|
component.code: component
|
|
for component in get_payroll_components(source.period)
|
|
}
|
|
for period_id, source in {
|
|
source.period_id: source for source in sources.values()
|
|
}.items()
|
|
}
|
|
payloads = []
|
|
for generated_line in generated:
|
|
source = sources.get(generated_source_line_id(generated_line))
|
|
if source is None:
|
|
continue
|
|
component = components_by_period.get(source.period_id, {}).get(
|
|
source.component_code
|
|
)
|
|
if component is None:
|
|
continue
|
|
payloads.append(
|
|
{
|
|
"target_component_code": generated_line.component_code,
|
|
"amount": str(generated_line.signed_amount),
|
|
"source_period_id": source.period_id,
|
|
"source_period_code": source.period.code,
|
|
"source_component_code": source.component_code,
|
|
"source_component_name": component.name,
|
|
}
|
|
)
|
|
return payloads
|