Corp/backend/finance/payroll/services.py

"""Transactional orchestration around the pure payroll calculation core."""

from __future__ import annotations

import hashlib
import json
import uuid
from collections import defaultdict
from decimal import Decimal

from django.contrib.auth import get_user_model
from django.core.exceptions import ValidationError
from django.db import IntegrityError, transaction
from django.db.models import F, Max, OuterRef, Q, Subquery, Sum
from django.utils import timezone

from finance.enums import (
    ApprovalStatus,
    InputSource,
    PayrollPeriodStatus,
    PayrollRunStatus,
    VacationPayInstructionStatus,
)
from finance.models import (
    EmployeePayRate,
    PayrollAuditEvent,
    PayrollDailyWorkEntry,
    PayrollInputLine,
    PayrollPeriod,
    PayrollRun,
    PayrollStatement,
    PayrollStatementAcknowledgement,
    PayrollStatementLine,
    PayrollWorkRecord,
    PayrollWorkSettings,
    VacationPayInstruction,
)
from payroll_core import (
    DeterministicPayrollCalculator,
    PayrollValidationError,
)

from .access import (
    SELF_APPROVAL_OVERRIDE_PERMISSION,
    can_access_payroll_management,
    can_self_approve_payroll,
    has_payroll_permission,
    has_simple_admin_access,
)
from .adapter import (
    build_core_request,
    employee_snapshot,
    point_base_accruals_for_employees,
)
from .automatic_input_copy import (
    copy_inputs_after_period_close,
    lock_automatic_input_copy_policy,
)
from .balances import opening_balances
from .components import (
    build_payroll_component_snapshot,
    get_payroll_component,
    get_payroll_components,
)
from .config import (
    base_rate_code,
    build_rules,
    ruleset_not_effective_message,
    ruleset_period_details,
)
from .exceptions import PayrollOperationError, PayrollPermissionDenied
from .scheduled_changes import (
    SCHEDULED_CHANGE_SOURCE_PREFIX,
    employee_is_eligible_for_scheduled_change,
    is_generated_scheduled_change,
    materialize_scheduled_changes,
    next_payroll_period,
    scheduled_component_for_period,
    target_cell_scheduled_delta,
)
from .work_norm import calculate_period_target_points

Employee = get_user_model()


def _require_permission(actor, permission: str) -> None:
    if (
        actor is None
        or not getattr(actor, "is_authenticated", False)
        or not has_payroll_permission(actor, permission)
    ):
        raise PayrollPermissionDenied(permission)


def _approval_audit_metadata(actor, *, self_approval: bool) -> dict[str, object]:
    """Keep legacy separation metadata only outside the temporary admin mode."""

    if has_simple_admin_access(actor):
        return {}
    if not self_approval:
        return {
            "self_approval_overridden": False,
            "approval_mode": "maker_checker",
        }
    return {
        "self_approval_overridden": True,
        "approval_mode": "self_override",
        "override_permission": SELF_APPROVAL_OVERRIDE_PERMISSION,
    }


def _hash_items(items: list[str]) -> str:
    payload = json.dumps(sorted(items), separators=(",", ":")).encode("utf-8")
    return hashlib.sha256(payload).hexdigest()


def _period_input_hash(period, results):
    hashes = [result.input_hash for _, result in results]
    if period.balance_rules_version >= 2:
        # Balance-only inputs do not produce payslips, but must still invalidate
        # an approved calculation when their amounts/settings change.
        components = {
            component.code: component for component in get_payroll_components(period)
        }
        outside_lines = (
            PayrollInputLine.objects.filter(
                period=period,
                status=ApprovalStatus.APPROVED,
            )
            .exclude(employee_id__in=[rate.employee_id for rate, _ in results])
            .order_by("pk")
        )
        for line in outside_lines:
            component = components[line.component_code]
            payload = {
                "id": line.pk,
                "employee_id": line.employee_id,
                "component_code": line.component_code,
                "amount": str(line.amount),
                "include_in_carryover": line.include_in_carryover
                and component.include_in_carryover,
                "include_in_salary": component.include_in_salary,
                "kind": component.kind,
            }
            hashes.append(
                hashlib.sha256(
                    json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()
                ).hexdigest()
            )
    return _hash_items(hashes)


def _audit(
    *,
    actor,
    action: str,
    instance,
    period: PayrollPeriod | None,
    before_hash: str = "",
    after_hash: str = "",
    metadata: dict[str, object] | None = None,
) -> None:
    PayrollAuditEvent.objects.create(
        actor=actor,
        action=action,
        object_type=instance._meta.label_lower,
        object_id=str(instance.pk),
        period=period,
        before_hash=before_hash,
        after_hash=after_hash,
        metadata=metadata or {},
    )


def _operation_error(code: str, message: str, **details):
    raise PayrollOperationError(code, message, details=details)


def _mark_linked_vacation_instruction_posted(line: PayrollInputLine) -> None:
    instruction = (
        VacationPayInstruction.objects.select_for_update()
        .select_related("vacation_terms")
        .filter(payroll_input_line=line)
        .first()
    )
    if instruction is None:
        return
    instruction.status = VacationPayInstructionStatus.POSTED
    instruction.save(update_fields=["status", "updated_at"])
    terms = instruction.vacation_terms
    terms.finance_status = "posted"
    terms.save(update_fields=["finance_status", "updated_at"])


def _sync_generated_scheduled_change_approval(line: PayrollInputLine) -> None:
    component = scheduled_component_for_period(line.component_code, line.period)
    if component is None:
        return
    PayrollInputLine.objects.filter(
        source=InputSource.API,
        source_ref=f"{SCHEDULED_CHANGE_SOURCE_PREFIX}{line.pk}",
        status=ApprovalStatus.DRAFT,
    ).update(
        status=ApprovalStatus.APPROVED,
        approved_by=line.approved_by,
        approved_at=line.approved_at,
        self_approval_overridden=line.self_approval_overridden,
    )


def _resolve_rates(
    period: PayrollPeriod,
    employee_ids: list[int],
) -> list[EmployeePayRate]:
    rates = list(
        EmployeePayRate.objects.filter(
            employee_id__in=employee_ids,
            rate_code=base_rate_code(),
            status=ApprovalStatus.APPROVED,
            effective_from__lte=period.date_to,
        )
        .select_related("employee", "employee__position")
        .order_by("employee_id", "effective_from", "revision")
    )
    grouped = defaultdict(list)
    for rate in rates:
        grouped[rate.employee_id].append(rate)

    selected = []
    for employee_id in employee_ids:
        employee_rates = grouped.get(employee_id, [])
        period_start_rates = [
            rate for rate in employee_rates if rate.effective_from <= period.date_from
        ]
        if not period_start_rates:
            _operation_error(
                "MISSING_RATE_AT_PERIOD_START",
                "Нет утверждённой ставки на начало периода.",
                employee_id=employee_id,
            )
        mid_period_rates = [
            rate
            for rate in employee_rates
            if period.date_from < rate.effective_from <= period.date_to
        ]
        if mid_period_rates:
            _operation_error(
                "MID_PERIOD_RATE_CHANGE_UNSUPPORTED",
                "Изменение ставки внутри периода требует правила пропорционирования.",
                employee_id=employee_id,
                rate_ids=[rate.pk for rate in mid_period_rates],
            )
        selected.append(
            max(
                period_start_rates,
                key=lambda rate: (
                    rate.effective_from,
                    rate.revision,
                    rate.pk,
                ),
            )
        )

    if not selected:
        _operation_error(
            "NO_APPROVED_RATES",
            "Для периода нет ни одной утверждённой базовой ставки.",
        )
    return sorted(selected, key=lambda rate: rate.employee_id)


def pay_rates_relevant_to_period(
    period: PayrollPeriod,
    *,
    queryset=None,
):
    """Return rate revisions shown as inputs for a payroll period."""

    queryset = queryset if queryset is not None else EmployeePayRate.objects.all()
    latest_effective_date = (
        EmployeePayRate.objects.filter(
            employee_id=OuterRef("employee_id"),
            rate_code=OuterRef("rate_code"),
            effective_from__lte=period.date_from,
        )
        .order_by("-effective_from")
        .values("effective_from")[:1]
    )
    return queryset.annotate(
        _latest_effective_date=Subquery(latest_effective_date)
    ).filter(
        Q(effective_from=F("_latest_effective_date"))
        | Q(
            effective_from__gt=period.date_from,
            effective_from__lte=period.date_to,
        )
    )


def _current_work_records(period):
    return {
        record.employee_id: record
        for record in PayrollWorkRecord.objects.filter(
            period=period,
            status=ApprovalStatus.APPROVED,
        ).select_related("employee")
    }


def _input_lines_by_employee(period, employee_ids):
    grouped = defaultdict(list)
    records = (
        PayrollInputLine.objects.filter(
            period=period,
            employee_id__in=employee_ids,
            status=ApprovalStatus.APPROVED,
        )
        .exclude(source=InputSource.API, source_ref__startswith="payroll-carryover:")
        .select_related("relates_to_period")
        .order_by("employee_id", "component_code", "id")
    )
    for record in records:
        grouped[record.employee_id].append(record)
    return grouped


def _calculate_current_period_inputs(
    period: PayrollPeriod,
    *,
    calculator=None,
    rules=None,
):
    """Resolve and calculate the currently approved source-of-truth inputs."""

    active_rules = rules or build_rules()
    active_calculator = calculator or DeterministicPayrollCalculator()
    work_records = _current_work_records(period)
    if not work_records:
        _operation_error(
            "NO_APPROVED_WORK_RECORDS",
            "Не задан утверждённый состав сотрудников за период.",
        )
    employee_ids = sorted(work_records)
    rates = _resolve_rates(period, employee_ids)
    component_definitions = {
        component.code: component for component in get_payroll_components(period)
    }
    excluded_salary_codes = [
        code
        for code, component in component_definitions.items()
        if not component.include_in_salary
    ]
    outside_roster = list(
        PayrollInputLine.objects.filter(
            period=period,
            status=ApprovalStatus.APPROVED,
        )
        .exclude(employee_id__in=employee_ids)
        .exclude(component_code__in=excluded_salary_codes)
        .values_list("employee_id", flat=True)
        .distinct()[:20]
    )
    if outside_roster:
        _operation_error(
            "INPUT_EMPLOYEE_NOT_IN_ROSTER",
            "Утверждённые строки найдены для сотрудников вне состава периода.",
            employee_ids=outside_roster,
        )
    input_lines = _input_lines_by_employee(period, employee_ids)
    balances = opening_balances(period, employee_ids)
    requests = []
    results = []
    for rate in rates:
        if rate.currency != period.currency:
            _operation_error(
                "RATE_CURRENCY_MISMATCH",
                "Валюта ставки не совпадает с валютой периода.",
                employee_id=rate.employee_id,
                rate_id=rate.pk,
            )
        work_record = work_records.get(rate.employee_id)
        if work_record is None:
            _operation_error(
                "MISSING_APPROVED_WORK_RECORD",
                "Нет утверждённых показателей за период.",
                employee_id=rate.employee_id,
            )
        request = build_core_request(
            period=period,
            rate=rate,
            work_record=work_record,
            input_lines=input_lines[rate.employee_id],
            opening_balance=balances.get(rate.employee_id, Decimal("0")),
            rounding=active_rules.rounding.value,
            component_definitions=component_definitions,
        )
        try:
            result = active_calculator.calculate(request, active_rules)
        except PayrollValidationError as exc:
            issues = [issue.to_dict() for issue in exc.issues]
            if any(issue.code == "RULESET_NOT_EFFECTIVE" for issue in exc.issues):
                raise PayrollOperationError(
                    "RULESET_NOT_EFFECTIVE",
                    ruleset_not_effective_message(active_rules),
                    details={
                        "employee_id": rate.employee_id,
                        "issues": issues,
                        **ruleset_period_details(
                            active_rules,
                            period_from=period.date_from,
                            period_to=period.date_to,
                        ),
                    },
                ) from exc
            raise PayrollOperationError(
                "CALCULATION_VALIDATION_FAILED",
                "Ядро отклонило входные данные сотрудника.",
                details={
                    "employee_id": rate.employee_id,
                    "issues": issues,
                },
            ) from exc
        requests.append(request)
        results.append((rate, result))

    return active_rules, requests, results


def _ensure_run_is_fresh(run: PayrollRun, period: PayrollPeriod) -> None:
    """Fail closed if approved inputs, rules or persisted snapshots changed."""

    _, _, results = _calculate_current_period_inputs(period)
    current_input_hash = _period_input_hash(period, results)
    current_result_hash = _hash_items([result.result_hash for _, result in results])
    first_result = results[0][1]
    statements = {
        statement.employee_id: statement for statement in run.statements.all()
    }
    current_employee_ids = {rate.employee_id for rate, _ in results}
    snapshots_match = set(statements) == current_employee_ids
    if snapshots_match:
        for rate, result in results:
            statement = statements[rate.employee_id]
            stored_warnings = statement.result_snapshot.get("warnings")
            current_warnings = [warning.to_dict() for warning in result.warnings]
            if (
                statement.input_hash != result.input_hash
                or statement.result_hash != result.result_hash
                or statement.result_snapshot.get("result_hash") != result.result_hash
                or statement.result_snapshot.get("reconciliation_hash")
                != result.reconciliation_hash
                or stored_warnings != current_warnings
            ):
                snapshots_match = False
                break
    if (
        run.employee_count != len(results)
        or run.input_hash != current_input_hash
        or run.result_hash != current_result_hash
        or run.ruleset_id != first_result.ruleset_id
        or run.ruleset_version != first_result.ruleset_version
        or run.ruleset_hash != first_result.ruleset_hash
        or run.calculator_version != first_result.calculator_version
        or not snapshots_match
    ):
        _operation_error(
            "RUN_STALE_RECALCULATION_REQUIRED",
            "Утверждённые данные или правила изменились; выполните новый расчёт.",
            run_id=run.pk,
        )


@transaction.atomic
def calculate_period(
    period_id: int,
    *,
    actor,
    idempotency_key: uuid.UUID | None = None,
    recalculation_reason: str = "",
    calculator=None,
    rules=None,
) -> PayrollRun:
    """Calculate a complete period and persist a new immutable revision."""

    _require_permission(actor, "finance.calculate_payroll")
    period = PayrollPeriod.objects.select_for_update().get(pk=period_id)
    previous_current = None
    if period.current_run_id is not None:
        previous_current = PayrollRun.objects.select_for_update().get(
            pk=period.current_run_id
        )
    if idempotency_key is not None:
        existing = PayrollRun.objects.filter(idempotency_key=idempotency_key).first()
        if existing is not None:
            if existing.period_id != period_id:
                _operation_error(
                    "IDEMPOTENCY_KEY_CONFLICT",
                    "Ключ уже использован для другого периода.",
                )
            return existing
    if period.status == PayrollPeriodStatus.CLOSED:
        _operation_error(
            "PERIOD_CLOSED",
            "Закрытый период нельзя пересчитать.",
            period_id=period.pk,
        )
    if period.status in {
        PayrollPeriodStatus.REVIEW,
        PayrollPeriodStatus.APPROVED,
    }:
        _operation_error(
            "PERIOD_UNDER_APPROVAL",
            "Сначала верните текущую ревизию на исправление.",
            period_id=period.pk,
        )
    if previous_current is not None and not recalculation_reason.strip():
        _operation_error(
            "RECALCULATION_REASON_REQUIRED",
            "Для новой ревизии расчёта требуется основание.",
            period_id=period.pk,
        )

    active_rules, requests, results = _calculate_current_period_inputs(
        period,
        calculator=calculator,
        rules=rules,
    )
    employee_ids = [rate.employee_id for rate, _ in results]

    next_revision = (
        PayrollRun.objects.filter(period=period).aggregate(Max("revision"))[
            "revision__max"
        ]
        or 0
    ) + 1
    aggregate_input_hash = _period_input_hash(period, results)
    aggregate_result_hash = _hash_items([result.result_hash for _, result in results])
    gross_total = sum(
        (result.totals.gross_after_adjustments for _, result in results),
        Decimal("0"),
    )
    deduction_total = sum(
        (result.totals.deduction_total for _, result in results),
        Decimal("0"),
    )
    payable_total = sum(
        (result.totals.payable for _, result in results),
        Decimal("0"),
    )
    first_result = results[0][1]
    run = PayrollRun(
        period=period,
        revision=next_revision,
        status=PayrollRunStatus.CALCULATED,
        supersedes=previous_current,
        recalculation_reason=recalculation_reason.strip(),
        idempotency_key=idempotency_key or uuid.uuid4(),
        ruleset_id=first_result.ruleset_id,
        ruleset_version=first_result.ruleset_version,
        ruleset_hash=first_result.ruleset_hash,
        calculator_version=first_result.calculator_version,
        input_hash=aggregate_input_hash,
        result_hash=aggregate_result_hash,
        employee_count=len(results),
        gross_total=gross_total,
        deduction_total=deduction_total,
        payable_total=payable_total,
        requested_by=actor,
    )
    run.clean()
    try:
        with transaction.atomic():
            run.save(force_insert=True)
    except IntegrityError as exc:
        existing = None
        if idempotency_key is not None:
            existing = PayrollRun.objects.filter(
                idempotency_key=idempotency_key
            ).first()
        if existing is not None and existing.period_id == period_id:
            return existing
        if existing is not None:
            raise PayrollOperationError(
                "IDEMPOTENCY_KEY_CONFLICT",
                "Ключ уже использован для другого периода.",
            ) from exc
        raise PayrollOperationError(
            "CONCURRENT_CALCULATION_CONFLICT",
            "Параллельный расчёт создал конфликт ревизий; повторите операцию.",
        ) from exc

    prior_statements = {}
    if previous_current is not None:
        prior_statements = {
            statement.employee_id: statement
            for statement in PayrollStatement.objects.filter(
                run=previous_current,
                employee_id__in=employee_ids,
            )
        }
    employee_line_visibility = {
        component.code: component.visible_to_employee
        for component in get_payroll_components(period)
    }

    for request, (rate, result) in zip(requests, results, strict=True):
        totals = result.totals
        statement = PayrollStatement(
            run=run,
            employee=rate.employee,
            supersedes=prior_statements.get(rate.employee_id),
            employee_snapshot=employee_snapshot(rate.employee),
            currency=result.currency,
            point_delta=result.point_delta,
            gross_before_adjustments=totals.gross_before_adjustments,
            adjustment_total=totals.adjustment_total,
            gross_total=totals.gross_after_adjustments,
            deduction_total=totals.deduction_total,
            net_pay=totals.net_pay,
            payment_total=totals.payment_total,
            opening_balance=request.opening_balance,
            closing_balance=totals.closing_balance,
            payable=totals.payable,
            input_hash=result.input_hash,
            result_hash=result.result_hash,
            input_snapshot=request.to_dict(),
            result_snapshot=result.to_dict(),
        )
        statement.clean()
        statement.save(force_insert=True)
        PayrollStatementLine.objects.bulk_create(
            [
                PayrollStatementLine(
                    statement=statement,
                    position=position,
                    line_id=line.line_id,
                    code=line.code,
                    label=line.label,
                    kind=line.kind.value,
                    amount=line.amount,
                    source_ref=line.source_ref,
                    reason=line.reason,
                    visible_to_employee=employee_line_visibility.get(
                        line.code,
                        True,
                    ),
                    source_period_from=(
                        line.source_period.start if line.source_period else None
                    ),
                    source_period_to=(
                        line.source_period.end if line.source_period else None
                    ),
                    is_retro=line.is_retro,
                    calculated=line.calculated,
                )
                for position, line in enumerate(result.lines, start=1)
            ]
        )

    if previous_current and previous_current.status != PayrollRunStatus.PUBLISHED:
        previous_status = previous_current.status
        previous_current.status = PayrollRunStatus.SUPERSEDED
        previous_current.save(update_fields=["status"])
        _audit(
            actor=actor,
            action="payroll.run_superseded",
            instance=previous_current,
            period=period,
            before_hash=previous_current.result_hash,
            after_hash=run.result_hash,
            metadata={"previous_status": previous_status},
        )

    period.current_run = run
    period.status = PayrollPeriodStatus.CALCULATED
    period.lock_version += 1
    period.save(update_fields=["current_run", "status", "lock_version", "updated_at"])
    _audit(
        actor=actor,
        action="payroll.calculated",
        instance=run,
        period=period,
        after_hash=run.result_hash,
        metadata={
            "revision": run.revision,
            "employee_count": run.employee_count,
            "is_recalculation": previous_current is not None,
        },
    )
    return run


def _lock_current_run(run_id: int) -> tuple[PayrollRun, PayrollPeriod]:
    run_reference = PayrollRun.objects.only("period_id").get(pk=run_id)
    period = PayrollPeriod.objects.select_for_update().get(pk=run_reference.period_id)
    run = PayrollRun.objects.select_for_update().get(pk=run_id)
    if period.current_run_id != run.pk:
        _operation_error(
            "RUN_NOT_CURRENT",
            "Операция разрешена только для текущей ревизии.",
            run_id=run.pk,
        )
    return run, period


@transaction.atomic
def submit_run_for_review(run_id: int, *, actor) -> PayrollRun:
    _require_permission(actor, "finance.calculate_payroll")
    run, period = _lock_current_run(run_id)
    if run.status != PayrollRunStatus.CALCULATED:
        _operation_error("INVALID_RUN_STATE", "Расчёт уже передан дальше.")
    _ensure_run_is_fresh(run, period)
    reconciliation_errors = []
    for statement in run.statements.only(
        "employee_id",
        "result_snapshot",
    ):
        warning_codes = {
            warning.get("code")
            for warning in statement.result_snapshot.get("warnings", [])
        }
        if "LEGACY_EXPECTED_TOTAL_MISMATCH" in warning_codes:
            reconciliation_errors.append(statement.employee_id)
    if reconciliation_errors:
        _operation_error(
            "RECONCILIATION_MISMATCH",
            "Контрольные итоги Excel не совпадают с расчётом.",
            employee_ids=reconciliation_errors[:20],
        )
    run.status = PayrollRunStatus.REVIEW
    run.save(update_fields=["status"])
    period.status = PayrollPeriodStatus.REVIEW
    period.save(update_fields=["status", "updated_at"])
    _audit(actor=actor, action="payroll.review_requested", instance=run, period=period)
    return run


@transaction.atomic
def return_run_for_correction(
    run_id: int,
    *,
    actor,
    reason: str,
) -> PayrollRun:
    """Return a reviewed run so the operator can create a new revision."""

    if not reason.strip():
        _operation_error(
            "RETURN_REASON_REQUIRED",
            "Укажите причину возврата расчёта.",
        )
    run, period = _lock_current_run(run_id)
    if run.status == PayrollRunStatus.CALCULATED:
        _require_permission(actor, "finance.calculate_payroll")
    else:
        _require_permission(actor, "finance.approve_payroll")
    if run.status not in {
        PayrollRunStatus.CALCULATED,
        PayrollRunStatus.REVIEW,
        PayrollRunStatus.APPROVED,
    }:
        _operation_error(
            "INVALID_RUN_STATE",
            "Вернуть можно расчёт на проверке или после утверждения.",
        )
    previous_status = run.status
    run.status = PayrollRunStatus.RETURNED
    run.save(update_fields=["status"])
    period.status = PayrollPeriodStatus.CALCULATED
    period.save(update_fields=["status", "updated_at"])
    _audit(
        actor=actor,
        action="payroll.returned_for_correction",
        instance=run,
        period=period,
        metadata={
            "previous_status": previous_status,
            "reason": reason.strip(),
        },
    )
    return run


@transaction.atomic
def approve_run(run_id: int, *, actor) -> PayrollRun:
    _require_permission(actor, "finance.approve_payroll")
    run, period = _lock_current_run(run_id)
    if run.status != PayrollRunStatus.REVIEW:
        _operation_error("INVALID_RUN_STATE", "Расчёт не находится на проверке.")
    _ensure_run_is_fresh(run, period)
    self_approval_overridden = run.requested_by_id == actor.pk
    if self_approval_overridden and not can_self_approve_payroll(actor):
        _operation_error(
            "SELF_APPROVAL_FORBIDDEN",
            "Автор расчёта не может сам его утвердить.",
        )
    run.status = PayrollRunStatus.APPROVED
    run.approved_by = actor
    run.approved_at = timezone.now()
    run.self_approval_overridden = self_approval_overridden
    run.save(
        update_fields=[
            "status",
            "approved_by",
            "approved_at",
            "self_approval_overridden",
        ]
    )
    period.status = PayrollPeriodStatus.APPROVED
    period.save(update_fields=["status", "updated_at"])
    _audit(
        actor=actor,
        action="payroll.approved",
        instance=run,
        period=period,
        metadata=_approval_audit_metadata(
            actor,
            self_approval=self_approval_overridden,
        ),
    )
    return run


@transaction.atomic
def publish_run(run_id: int, *, actor) -> PayrollRun:
    _require_permission(actor, "finance.publish_payroll")
    run, period = _lock_current_run(run_id)
    if run.status != PayrollRunStatus.APPROVED:
        _operation_error("INVALID_RUN_STATE", "Расчёт ещё не утверждён.")
    _ensure_run_is_fresh(run, period)

    prior_published_runs = list(
        PayrollRun.objects.select_for_update()
        .filter(
            period=period,
            status=PayrollRunStatus.PUBLISHED,
        )
        .exclude(pk=run.pk)
    )
    for prior_run in prior_published_runs:
        prior_run.status = PayrollRunStatus.SUPERSEDED
        prior_run.save(update_fields=["status"])
        _audit(
            actor=actor,
            action="payroll.run_superseded_on_publish",
            instance=prior_run,
            period=period,
            before_hash=prior_run.result_hash,
            after_hash=run.result_hash,
            metadata={"replacement_run_id": run.pk},
        )
    now = timezone.now()
    run.status = PayrollRunStatus.PUBLISHED
    run.published_by = actor
    run.published_at = now
    run.save(update_fields=["status", "published_by", "published_at"])
    period.status = PayrollPeriodStatus.PUBLISHED
    period.save(update_fields=["status", "updated_at"])
    _audit(
        actor=actor,
        action="payroll.published",
        instance=run,
        period=period,
        after_hash=run.result_hash,
        metadata={"revision": run.revision, "employee_count": run.employee_count},
    )
    return run


@transaction.atomic
def close_period(period_id: int, *, actor) -> PayrollPeriod:
    _require_permission(actor, "finance.publish_payroll")
    lock_automatic_input_copy_policy()
    period = PayrollPeriod.objects.select_for_update().get(pk=period_id)
    if period.status == PayrollPeriodStatus.CLOSED:
        return period
    if period.status != PayrollPeriodStatus.PUBLISHED:
        _operation_error(
            "INVALID_PERIOD_STATE",
            "Закрыть можно только опубликованный период.",
        )
    period.component_snapshot = build_payroll_component_snapshot(period)
    period.status = PayrollPeriodStatus.CLOSED
    period.save(update_fields=["component_snapshot", "status", "updated_at"])
    _audit(actor=actor, action="payroll.period_closed", instance=period, period=period)
    period.automatic_input_copy = copy_inputs_after_period_close(
        source_period=period, actor=actor
    )
    return period


@transaction.atomic
def complete_period_workflow(
    period_id: int,
    *,
    actor,
    expected_lock_version: int,
    idempotency_key: uuid.UUID | None = None,
    recalculation_reason: str = "",
) -> dict[str, object]:
    """Run every remaining payroll workflow stage for a portal administrator."""

    if not has_simple_admin_access(actor):
        raise PayrollPermissionDenied("finance.full_payroll_admin")

    lock_automatic_input_copy_policy()
    period = PayrollPeriod.objects.select_for_update().get(pk=period_id)
    run = (
        PayrollRun.objects.select_for_update().get(pk=period.current_run_id)
        if period.current_run_id is not None
        else None
    )
    if period.status == PayrollPeriodStatus.CLOSED:
        if run is None:
            _operation_error(
                "CLOSED_PERIOD_RUN_MISSING",
                "У закрытого периода отсутствует опубликованный расчёт.",
                period_id=period.pk,
            )
        return {"period": period, "run": run, "completed_steps": []}
    if period.lock_version != expected_lock_version:
        _operation_error(
            "STALE_PERIOD",
            "Период уже изменён; обновите данные перед завершением расчёта.",
            expected_lock_version=expected_lock_version,
            actual_lock_version=period.lock_version,
        )

    completed_steps = []
    if run is None or run.status == PayrollRunStatus.RETURNED:
        run = calculate_period(
            period.pk,
            actor=actor,
            idempotency_key=idempotency_key,
            recalculation_reason=recalculation_reason,
        )
        completed_steps.append("calculated")
    if run.status == PayrollRunStatus.CALCULATED:
        run = submit_run_for_review(run.pk, actor=actor)
        completed_steps.append("review_requested")
    if run.status == PayrollRunStatus.REVIEW:
        run = approve_run(run.pk, actor=actor)
        completed_steps.append("approved")
    if run.status == PayrollRunStatus.APPROVED:
        run = publish_run(run.pk, actor=actor)
        completed_steps.append("published")
    if run.status == PayrollRunStatus.PUBLISHED:
        period = close_period(period.pk, actor=actor)
        completed_steps.append("period_closed")

    if period.status != PayrollPeriodStatus.CLOSED:
        _operation_error(
            "ADMIN_WORKFLOW_INCOMPLETE",
            "Не удалось завершить все этапы расчёта.",
            period_id=period.pk,
            run_status=run.status,
        )

    _audit(
        actor=actor,
        action="payroll.admin_workflow_completed",
        instance=period,
        period=period,
        metadata={"steps": completed_steps, "run_id": run.pk},
    )
    return {"period": period, "run": run, "completed_steps": completed_steps}


@transaction.atomic
def bulk_set_point_rate(
    period_id: int,
    *,
    actor,
    employee_ids: list[int],
    mode: str = "fixed",
    point_rate: Decimal | None = None,
    reason: str,
) -> dict[str, object]:
    """Set a fixed excess price or clear it to use the dynamic in-norm price."""

    _require_permission(actor, "finance.manage_payroll_inputs")
    period = PayrollPeriod.objects.select_for_update().get(pk=period_id)
    _ensure_period_accepts_approvals(period)

    selected_employee_ids = sorted(set(employee_ids))
    rates = list(
        EmployeePayRate.objects.select_for_update(of=("self",))
        .filter(
            employee_id__in=selected_employee_ids,
            rate_code=base_rate_code(),
            effective_from__lte=period.date_to,
        )
        .exclude(status=ApprovalStatus.VOIDED)
        .select_related("employee", "replaces")
        .order_by("employee_id", "effective_from", "revision", "pk")
    )
    rates_by_employee: dict[int, list[EmployeePayRate]] = defaultdict(list)
    for rate in rates:
        rates_by_employee[rate.employee_id].append(rate)

    summary = {
        "selected_employees": len(selected_employee_ids),
        "updated_drafts": 0,
        "created_revisions": 0,
        "unchanged": 0,
        "skipped": 0,
    }
    changed_records: list[EmployeePayRate] = []

    for employee_id in selected_employee_ids:
        employee_rates = rates_by_employee.get(employee_id, [])
        period_start_dates = [
            rate.effective_from
            for rate in employee_rates
            if rate.effective_from <= period.date_from
        ]
        relevant_dates = set()
        if period_start_dates:
            relevant_dates.add(max(period_start_dates))
        relevant_dates.update(
            rate.effective_from
            for rate in employee_rates
            if period.date_from < rate.effective_from <= period.date_to
        )

        if not relevant_dates:
            summary["skipped"] += 1
            continue

        for effective_from in sorted(relevant_dates):
            current = max(
                (
                    rate
                    for rate in employee_rates
                    if rate.effective_from == effective_from
                ),
                key=lambda rate: (rate.revision, rate.pk),
            )
            if mode == "in_norm":
                resolved_point_rate = None
            else:
                resolved_point_rate = point_rate
            if mode == "fixed" and resolved_point_rate is None:
                _operation_error(
                    "POINT_RATE_REQUIRED",
                    "Не задана фиксированная цена балла сверх нормы.",
                )
            if current.point_rate == resolved_point_rate:
                summary["unchanged"] += 1
                continue

            if current.status == ApprovalStatus.DRAFT:
                if (
                    not has_simple_admin_access(actor)
                    and current.created_by_id != actor.pk
                ):
                    summary["skipped"] += 1
                    continue
                previous_point_rate = current.point_rate
                current.point_rate = resolved_point_rate
                current._expected_lock_version = current.lock_version
                try:
                    current.full_clean()
                    current.save()
                except ValidationError as exc:
                    _operation_error(
                        "INPUT_VALIDATION_FAILED",
                        "Цена балла не прошла проверку.",
                        errors=(
                            exc.message_dict
                            if hasattr(exc, "message_dict")
                            else exc.messages
                        ),
                    )
                _audit(
                    actor=actor,
                    action="payroll.rate_bulk_point_rate_updated",
                    instance=current,
                    period=period,
                    metadata={
                        "previous_point_rate": (
                            str(previous_point_rate)
                            if previous_point_rate is not None
                            else None
                        ),
                        "point_rate": (
                            str(resolved_point_rate)
                            if resolved_point_rate is not None
                            else None
                        ),
                        "point_rate_mode": mode,
                        "reason": reason,
                    },
                )
                summary["updated_drafts"] += 1
                changed_records.append(current)
                continue

            if current.status != ApprovalStatus.APPROVED:
                summary["skipped"] += 1
                continue

            revision = EmployeePayRate(
                employee=current.employee,
                rate_code=current.rate_code,
                amount=current.amount,
                point_rate=resolved_point_rate,
                currency=current.currency,
                effective_from=current.effective_from,
                revision=current.revision + 1,
                replaces=current,
                reason=reason,
                status=ApprovalStatus.DRAFT,
                source=InputSource.MANUAL,
                created_by=actor,
            )
            try:
                revision.full_clean()
                revision.save(force_insert=True)
            except ValidationError as exc:
                _operation_error(
                    "INPUT_VALIDATION_FAILED",
                    "Новая версия ставки не прошла проверку.",
                    errors=(
                        exc.message_dict
                        if hasattr(exc, "message_dict")
                        else exc.messages
                    ),
                )
            except IntegrityError:
                _operation_error(
                    "RATE_REVISION_CONFLICT",
                    ("Для одной из ставок уже создана новая версия; обновите данные."),
                    rate_id=current.pk,
                )
            _audit(
                actor=actor,
                action="payroll.rate_bulk_point_rate_revision_created",
                instance=revision,
                period=period,
                metadata={
                    "replaces_id": current.pk,
                    "previous_point_rate": (
                        str(current.point_rate)
                        if current.point_rate is not None
                        else None
                    ),
                    "point_rate": (
                        str(resolved_point_rate)
                        if resolved_point_rate is not None
                        else None
                    ),
                    "point_rate_mode": mode,
                    "reason": reason,
                },
            )
            summary["created_revisions"] += 1
            changed_records.append(revision)

    return {
        "mode": mode,
        "point_rate": point_rate,
        "summary": summary,
        "records": changed_records,
    }


@transaction.atomic
def bulk_set_pay_rate(
    period_id: int,
    *,
    actor,
    employee_ids: list[int],
    amount: Decimal,
    effective_from,
    reason: str,
) -> dict[str, object]:
    """Create or revise one base rate for each selected employee."""

    _require_permission(actor, "finance.manage_payroll_inputs")
    period = PayrollPeriod.objects.select_for_update().get(pk=period_id)
    _ensure_period_accepts_approvals(period)
    if not period.date_from <= effective_from <= period.date_to:
        _operation_error(
            "RATE_DATE_OUTSIDE_PERIOD",
            "Дата начала ставки должна входить в выбранный расчётный период.",
            period_date_from=str(period.date_from),
            period_date_to=str(period.date_to),
        )

    selected_employee_ids = sorted(set(employee_ids))
    employees = {
        employee.pk: employee
        for employee in Employee.objects.filter(
            pk__in=selected_employee_ids,
            is_active=True,
        ).order_by("pk")
    }
    rates = list(
        EmployeePayRate.objects.select_for_update(of=("self",))
        .filter(
            employee_id__in=employees,
            rate_code=base_rate_code(),
            effective_from__lte=effective_from,
        )
        .select_related("employee", "replaces")
        .order_by("employee_id", "effective_from", "revision", "pk")
    )
    rates_by_employee: dict[int, list[EmployeePayRate]] = defaultdict(list)
    for rate in rates:
        rates_by_employee[rate.employee_id].append(rate)

    summary = {
        "selected_employees": len(selected_employee_ids),
        "created_drafts": 0,
        "updated_drafts": 0,
        "created_revisions": 0,
        "unchanged": 0,
        "skipped": 0,
    }
    changed_records: list[EmployeePayRate] = []

    for employee_id in selected_employee_ids:
        employee = employees.get(employee_id)
        if employee is None:
            summary["skipped"] += 1
            continue

        employee_rates = rates_by_employee.get(employee_id, [])
        exact_rates = [
            rate for rate in employee_rates if rate.effective_from == effective_from
        ]
        current_rates = [
            rate for rate in exact_rates if rate.status != ApprovalStatus.VOIDED
        ]
        current = (
            max(current_rates, key=lambda rate: (rate.revision, rate.pk))
            if current_rates
            else None
        )

        if current is not None and current.amount == amount:
            summary["unchanged"] += 1
            continue

        if current is not None and current.status == ApprovalStatus.DRAFT:
            if not has_simple_admin_access(actor) and current.created_by_id != actor.pk:
                summary["skipped"] += 1
                continue
            previous_amount = current.amount
            current.amount = amount
            current._expected_lock_version = current.lock_version
            try:
                current.full_clean()
                current.save()
            except ValidationError as exc:
                _operation_error(
                    "INPUT_VALIDATION_FAILED",
                    "Ставка не прошла проверку.",
                    errors=(
                        exc.message_dict
                        if hasattr(exc, "message_dict")
                        else exc.messages
                    ),
                )
            _audit(
                actor=actor,
                action="payroll.rate_bulk_amount_updated",
                instance=current,
                period=period,
                metadata={
                    "previous_amount": str(previous_amount),
                    "amount": str(amount),
                    "reason": reason,
                },
            )
            summary["updated_drafts"] += 1
            changed_records.append(current)
            continue

        if current is not None and current.status == ApprovalStatus.APPROVED:
            rate = EmployeePayRate(
                employee=current.employee,
                rate_code=current.rate_code,
                amount=amount,
                point_rate=current.point_rate,
                currency=current.currency,
                effective_from=current.effective_from,
                revision=current.revision + 1,
                replaces=current,
                reason=reason,
                status=ApprovalStatus.DRAFT,
                source=InputSource.MANUAL,
                created_by=actor,
            )
            action = "payroll.rate_bulk_amount_revision_created"
            metadata = {
                "replaces_id": current.pk,
                "previous_amount": str(current.amount),
                "amount": str(amount),
                "reason": reason,
            }
            summary_key = "created_revisions"
        else:
            previous_rates = [
                rate
                for rate in employee_rates
                if rate.status != ApprovalStatus.VOIDED
                and rate.effective_from < effective_from
            ]
            previous = (
                max(
                    previous_rates,
                    key=lambda rate: (
                        rate.effective_from,
                        rate.revision,
                        rate.pk,
                    ),
                )
                if previous_rates
                else None
            )
            rate = EmployeePayRate(
                employee=employee,
                rate_code=base_rate_code(),
                amount=amount,
                point_rate=(previous.point_rate if previous else None),
                currency=period.currency,
                effective_from=effective_from,
                revision=max((item.revision for item in exact_rates), default=0) + 1,
                reason=reason,
                status=ApprovalStatus.DRAFT,
                source=InputSource.MANUAL,
                created_by=actor,
            )
            action = "payroll.rate_bulk_created"
            metadata = {
                "amount": str(amount),
                "point_rate": (
                    str(rate.point_rate) if rate.point_rate is not None else None
                ),
                "reason": reason,
            }
            summary_key = "created_drafts"

        try:
            rate.full_clean()
            rate.save(force_insert=True)
        except ValidationError as exc:
            _operation_error(
                "INPUT_VALIDATION_FAILED",
                "Новая ставка не прошла проверку.",
                errors=(
                    exc.message_dict if hasattr(exc, "message_dict") else exc.messages
                ),
            )
        except IntegrityError:
            _operation_error(
                "RATE_BULK_CONFLICT",
                "Одна из ставок уже изменена; обновите экран и повторите.",
                employee_id=employee_id,
            )
        _audit(
            actor=actor,
            action=action,
            instance=rate,
            period=period,
            metadata=metadata,
        )
        summary[summary_key] += 1
        changed_records.append(rate)

    return {
        "amount": amount,
        "effective_from": effective_from,
        "summary": summary,
        "records": changed_records,
    }


@transaction.atomic
def bulk_set_target_points(
    period_id: int,
    *,
    actor,
    employee_ids: list[int],
    mode: str,
    target_points: Decimal | None,
    reason: str,
) -> dict[str, object]:
    """Set one target for selected employees or restore their automatic target."""

    _require_permission(actor, "finance.manage_payroll_inputs")
    period = PayrollPeriod.objects.select_for_update().get(pk=period_id)
    _ensure_period_accepts_approvals(period)
    selected_employee_ids = sorted(set(employee_ids))
    employees = {
        employee.pk: employee
        for employee in Employee.objects.filter(
            pk__in=selected_employee_ids,
            is_active=True,
        )
        .select_related("position")
        .order_by("pk")
    }
    records = list(
        PayrollWorkRecord.objects.select_for_update(of=("self",))
        .filter(period=period, employee_id__in=employees)
        .select_related("employee", "replaces")
        .order_by("employee_id", "revision", "pk")
    )
    records_by_employee: dict[int, list[PayrollWorkRecord]] = defaultdict(list)
    for record in records:
        records_by_employee[record.employee_id].append(record)

    summary = {
        "selected_employees": len(selected_employee_ids),
        "created_drafts": 0,
        "updated_drafts": 0,
        "created_revisions": 0,
        "unchanged": 0,
        "skipped": 0,
    }
    changed_records: list[PayrollWorkRecord] = []

    for employee_id in selected_employee_ids:
        employee = employees.get(employee_id)
        if employee is None:
            summary["skipped"] += 1
            continue
        employee_records = records_by_employee.get(employee_id, [])
        active_records = [
            record
            for record in employee_records
            if record.status != ApprovalStatus.VOIDED
        ]
        current = (
            max(active_records, key=lambda record: (record.revision, record.pk))
            if active_records
            else None
        )
        automatic = mode == "automatic"
        resolved_target = (
            calculate_period_target_points(period, employee=employee)[0]
            if automatic
            else target_points
        )
        if resolved_target is None or resolved_target <= 0:
            _operation_error(
                "TARGET_POINTS_REQUIRED",
                "Не удалось определить положительную норму баллов.",
                employee_id=employee_id,
            )
        if current is None and automatic:
            # The table already calculates this value dynamically when no work
            # record exists, so an empty automatic record would only add noise.
            summary["unchanged"] += 1
            continue
        if (
            current is not None
            and current.target_points == resolved_target
            and current.target_points_overridden is not automatic
        ):
            summary["unchanged"] += 1
            continue

        if current is not None and current.status == ApprovalStatus.DRAFT:
            if not has_simple_admin_access(actor) and current.created_by_id != actor.pk:
                summary["skipped"] += 1
                continue
            previous_target = current.target_points
            previous_automatic = not current.target_points_overridden
            current.target_points = resolved_target
            current.target_points_overridden = not automatic
            current.expected_point_amount = None
            current.expected_gross = None
            current.expected_recalculated_gross = None
            current.expected_payable = None
            current.reason = reason
            current.source = InputSource.MANUAL
            current.source_ref = ""
            current._expected_lock_version = current.lock_version
            try:
                current.full_clean()
                current.save()
            except ValidationError as exc:
                _operation_error(
                    "INPUT_VALIDATION_FAILED",
                    "Норма баллов не прошла проверку.",
                    errors=(
                        exc.message_dict
                        if hasattr(exc, "message_dict")
                        else exc.messages
                    ),
                )
            _audit(
                actor=actor,
                action="payroll.work_bulk_target_updated",
                instance=current,
                period=period,
                metadata={
                    "previous_target_points": str(previous_target),
                    "previous_automatic": previous_automatic,
                    "target_points": str(resolved_target),
                    "automatic": automatic,
                    "reason": reason,
                },
            )
            summary["updated_drafts"] += 1
            changed_records.append(current)
            continue

        if current is not None and current.status == ApprovalStatus.APPROVED:
            if PayrollWorkRecord.objects.filter(replaces=current).exists():
                summary["skipped"] += 1
                continue
            record = PayrollWorkRecord(
                period=period,
                employee=employee,
                target_points=resolved_target,
                target_points_overridden=not automatic,
                actual_points=current.actual_points,
                revision=current.revision + 1,
                replaces=current,
                reason=reason,
                status=ApprovalStatus.DRAFT,
                source=InputSource.MANUAL,
                created_by=actor,
            )
            action = "payroll.work_bulk_target_revision_created"
            metadata = {
                "replaces_id": current.pk,
                "previous_target_points": str(current.target_points),
                "target_points": str(resolved_target),
                "automatic": automatic,
                "reason": reason,
            }
            summary_key = "created_revisions"
        elif current is None:
            record = PayrollWorkRecord(
                period=period,
                employee=employee,
                target_points=resolved_target,
                target_points_overridden=not automatic,
                actual_points=Decimal("0"),
                revision=max(
                    (record.revision for record in employee_records),
                    default=0,
                )
                + 1,
                reason=reason,
                status=ApprovalStatus.DRAFT,
                source=InputSource.MANUAL,
                created_by=actor,
            )
            action = "payroll.work_bulk_target_created"
            metadata = {
                "target_points": str(resolved_target),
                "automatic": automatic,
                "reason": reason,
            }
            summary_key = "created_drafts"
        else:
            summary["skipped"] += 1
            continue

        try:
            record.full_clean()
            record.save(force_insert=True)
        except ValidationError as exc:
            _operation_error(
                "INPUT_VALIDATION_FAILED",
                "Норма баллов не прошла проверку.",
                errors=(
                    exc.message_dict if hasattr(exc, "message_dict") else exc.messages
                ),
            )
        except IntegrityError:
            _operation_error(
                "WORK_BULK_CONFLICT",
                "Одна из записей выработки уже изменена; обновите экран и повторите.",
                employee_id=employee_id,
            )
        _audit(
            actor=actor,
            action=action,
            instance=record,
            period=period,
            metadata=metadata,
        )
        summary[summary_key] += 1
        changed_records.append(record)

    return {
        "mode": mode,
        "target_points": target_points,
        "summary": summary,
        "records": changed_records,
    }


def sync_daily_work_record(
    *,
    period: PayrollPeriod,
    employee,
    actor,
    channel: str = "employee_api",
) -> PayrollWorkRecord:
    totals = PayrollDailyWorkEntry.objects.filter(
        period=period,
        employee=employee,
    ).aggregate(
        actual_points=Sum("actual_points"),
    )
    actual_points = totals["actual_points"] or Decimal("0")

    records = list(
        PayrollWorkRecord.objects.select_for_update()
        .filter(period=period, employee=employee)
        .order_by("revision", "pk")
    )
    active_records = [
        record for record in records if record.status != ApprovalStatus.VOIDED
    ]
    current = (
        max(active_records, key=lambda record: (record.revision, record.pk))
        if active_records
        else None
    )
    calculated_target, workdays_count, target_source = calculate_period_target_points(
        period, employee=employee
    )
    uses_saved_target = current is not None and current.target_points_overridden
    target_points = current.target_points if uses_saved_target else calculated_target
    if target_points <= 0:
        _operation_error(
            "DAILY_WORK_EMPTY",
            "Для периода не удалось определить положительную норму выработки.",
        )
    if (
        current is not None
        and current.target_points == target_points
        and current.actual_points == actual_points
    ):
        return current

    reason = "Агрегация ежедневной выработки сотрудника"
    if current is not None and current.status == ApprovalStatus.DRAFT:
        previous_target = current.target_points
        previous_actual = current.actual_points
        current.target_points = target_points
        current.actual_points = actual_points
        current.expected_point_amount = None
        current.expected_gross = None
        current.expected_recalculated_gross = None
        current.expected_payable = None
        if not uses_saved_target:
            current.reason = reason
            current.source = InputSource.API
            current.source_ref = (
                f"daily-work:{period.pk}:{employee.pk}:r{current.revision}"
            )
            current.target_points_overridden = False
        current._expected_lock_version = current.lock_version
        try:
            current.full_clean()
            current.save()
        except ValidationError as exc:
            _operation_error(
                "INPUT_VALIDATION_FAILED",
                "Итоговая выработка не прошла проверку.",
                errors=(
                    exc.message_dict if hasattr(exc, "message_dict") else exc.messages
                ),
            )
        _audit(
            actor=actor,
            action="payroll.daily_work_aggregate_updated",
            instance=current,
            period=period,
            metadata={
                "channel": channel,
                "previous_target_points": str(previous_target),
                "previous_actual_points": str(previous_actual),
                "target_points": str(target_points),
                "actual_points": str(actual_points),
                "target_source": (
                    "saved_record" if uses_saved_target else target_source
                ),
                "workdays_count": workdays_count,
            },
        )
        return current

    if current is not None and current.status == ApprovalStatus.APPROVED:
        if PayrollWorkRecord.objects.filter(replaces=current).exists():
            _operation_error(
                "WORK_REVISION_CONFLICT",
                "Для итоговой выработки уже создавалась новая версия.",
                record_id=current.pk,
            )
        revision = current.revision + 1
        record = PayrollWorkRecord(
            period=period,
            employee=employee,
            target_points=target_points,
            target_points_overridden=uses_saved_target,
            actual_points=actual_points,
            revision=revision,
            replaces=current,
            reason=reason,
            status=ApprovalStatus.DRAFT,
            source=(current.source if uses_saved_target else InputSource.API),
            source_ref=f"daily-work:{period.pk}:{employee.pk}:r{revision}",
            created_by=actor,
        )
        action = "payroll.daily_work_aggregate_revision_created"
        metadata = {
            "channel": channel,
            "replaces_id": current.pk,
            "target_points": str(target_points),
            "actual_points": str(actual_points),
            "target_source": ("saved_record" if uses_saved_target else target_source),
            "workdays_count": workdays_count,
        }
    elif current is None:
        revision = max((record.revision for record in records), default=0) + 1
        record = PayrollWorkRecord(
            period=period,
            employee=employee,
            target_points=target_points,
            target_points_overridden=False,
            actual_points=actual_points,
            revision=revision,
            reason=reason,
            status=ApprovalStatus.DRAFT,
            source=InputSource.API,
            source_ref=f"daily-work:{period.pk}:{employee.pk}:r{revision}",
            created_by=actor,
        )
        action = "payroll.daily_work_aggregate_created"
        metadata = {
            "channel": channel,
            "target_points": str(target_points),
            "actual_points": str(actual_points),
            "target_source": target_source,
            "workdays_count": workdays_count,
        }
    else:
        _operation_error(
            "WORK_RECORD_STATE_INVALID",
            "Итоговую запись выработки нельзя изменить.",
            record_status=current.status,
        )

    try:
        record.full_clean()
        record.save(force_insert=True)
    except ValidationError as exc:
        _operation_error(
            "INPUT_VALIDATION_FAILED",
            "Итоговая выработка не прошла проверку.",
            errors=(exc.message_dict if hasattr(exc, "message_dict") else exc.messages),
        )
    except IntegrityError:
        _operation_error(
            "WORK_RECORD_CONFLICT",
            "Итоговая выработка уже изменена; обновите данные и повторите.",
        )
    _audit(
        actor=actor,
        action=action,
        instance=record,
        period=period,
        metadata=metadata,
    )
    return record


@transaction.atomic
def save_own_daily_work_entry(
    period_id: int,
    *,
    actor,
    work_date,
    actual_points: Decimal,
    note: str = "",
    expected_lock_version: int | None = None,
) -> tuple[PayrollDailyWorkEntry, PayrollWorkRecord, str]:
    """Upsert one daily entry and refresh the employee's period aggregate."""

    if actor is None or not getattr(actor, "is_authenticated", False):
        _operation_error(
            "AUTHENTICATION_REQUIRED",
            "Для сохранения выработки необходимо войти в систему.",
        )
    period = (
        PayrollPeriod.objects.select_for_update(of=("self",))
        .select_related("current_run")
        .get(pk=period_id)
    )
    if period.status != PayrollPeriodStatus.OPEN:
        _operation_error(
            "WORK_PERIOD_LOCKED",
            "Выработка за этот период уже закрыта для изменений.",
            period_status=period.status,
        )
    if period.employee_work_entry_locked and not can_access_payroll_management(actor):
        _operation_error(
            "WORK_ENTRY_LOCKED",
            "Ввод выработки за этот период закрыт. Обратитесь к администрации.",
        )
    if not period.date_from <= work_date <= period.date_to:
        _operation_error(
            "WORK_DATE_OUTSIDE_PERIOD",
            "Дата выработки должна входить в выбранный расчётный период.",
            period_date_from=str(period.date_from),
            period_date_to=str(period.date_to),
        )
    if work_date > timezone.localdate():
        _operation_error(
            "WORK_DATE_IN_FUTURE",
            "Нельзя указать выработку за будущую дату.",
        )

    entry = (
        PayrollDailyWorkEntry.objects.select_for_update()
        .filter(period=period, employee=actor, work_date=work_date)
        .first()
    )
    note = note.strip()
    if entry is not None:
        if expected_lock_version is None:
            _operation_error(
                "DAILY_WORK_VERSION_REQUIRED",
                "Обновите данные и повторите сохранение.",
            )
        if entry.lock_version != expected_lock_version:
            _operation_error(
                "STALE_DAILY_WORK_ENTRY",
                "Запись уже изменена; обновите данные и повторите.",
                expected_lock_version=expected_lock_version,
                actual_lock_version=entry.lock_version,
            )
        if entry.actual_points == actual_points and entry.note == note:
            aggregate = sync_daily_work_record(
                period=period,
                employee=actor,
                actor=actor,
            )
            return entry, aggregate, "unchanged"

        previous_actual = entry.actual_points
        entry.actual_points = actual_points
        entry.note = note
        entry.lock_version += 1
        operation = "updated"
        action = "payroll.daily_work_entry_updated"
        metadata = {
            "channel": "employee_api",
            "work_date": str(work_date),
            "previous_actual_points": str(previous_actual),
            "target_points": str(entry.target_points),
            "actual_points": str(actual_points),
        }
    else:
        target_points = PayrollWorkSettings.get_daily_target_points()
        entry = PayrollDailyWorkEntry(
            period=period,
            employee=actor,
            work_date=work_date,
            target_points=target_points,
            actual_points=actual_points,
            note=note,
        )
        operation = "created"
        action = "payroll.daily_work_entry_created"
        metadata = {
            "channel": "employee_api",
            "work_date": str(work_date),
            "target_points": str(target_points),
            "actual_points": str(actual_points),
        }

    try:
        entry.full_clean()
        entry.save()
    except ValidationError as exc:
        _operation_error(
            "INPUT_VALIDATION_FAILED",
            "Ежедневная выработка не прошла проверку.",
            errors=(exc.message_dict if hasattr(exc, "message_dict") else exc.messages),
        )
    except IntegrityError:
        _operation_error(
            "DAILY_WORK_ENTRY_CONFLICT",
            "Запись за эту дату уже создана; обновите данные.",
        )
    _audit(
        actor=actor,
        action=action,
        instance=entry,
        period=period,
        metadata=metadata,
    )
    aggregate = sync_daily_work_record(
        period=period,
        employee=actor,
        actor=actor,
    )
    return entry, aggregate, operation


def _approve_record(
    record,
    *,
    actor,
    object_label: str,
    expected_lock_version: int,
):
    if (
        not isinstance(expected_lock_version, int)
        or isinstance(expected_lock_version, bool)
        or expected_lock_version < 0
    ):
        _operation_error(
            "APPROVAL_VERSION_REQUIRED",
            "Укажите версию черновика, которую проверил утверждающий.",
        )
    if record.lock_version != expected_lock_version:
        _operation_error(
            "APPROVAL_VERSION_CONFLICT",
            "Черновик изменился после проверки; откройте и проверьте его заново.",
            expected_lock_version=expected_lock_version,
            actual_lock_version=record.lock_version,
        )
    if record.status != ApprovalStatus.DRAFT:
        _operation_error("INVALID_INPUT_STATE", "Утвердить можно только черновик.")
    self_approval_overridden = record.created_by_id == actor.pk
    if self_approval_overridden and not can_self_approve_payroll(actor):
        _operation_error(
            "SELF_APPROVAL_FORBIDDEN",
            "Автор входных данных не может сам их утвердить.",
        )
    now = timezone.now()
    replaced = None
    replaces_id = getattr(record, "replaces_id", None)
    if replaces_id:
        replaced = record.__class__.objects.select_for_update().get(pk=replaces_id)
        record._state.fields_cache["replaces"] = replaced
    previous_status = replaced.status if replaced is not None else ""
    if replaced is not None and replaced.status != ApprovalStatus.APPROVED:
        _operation_error(
            "REPLACEMENT_SOURCE_NOT_APPROVED",
            "Заменить можно только действующую утверждённую запись.",
            replaced_id=replaced.pk,
        )
    if replaced:
        replaced.status = ApprovalStatus.VOIDED
        replaced.voided_by = actor
        replaced.voided_at = now
        replaced.save(update_fields=["status", "voided_by", "voided_at"])
    record.status = ApprovalStatus.APPROVED
    record.approved_by = actor
    record.approved_at = now
    record.self_approval_overridden = self_approval_overridden
    try:
        record.full_clean()
    except ValidationError as exc:
        raise PayrollOperationError(
            "INPUT_VALIDATION_FAILED",
            "Входные данные не прошли проверку.",
            details={
                "errors": exc.message_dict
                if hasattr(exc, "message_dict")
                else exc.messages
            },
        ) from exc
    record.save(
        update_fields=[
            "status",
            "approved_by",
            "approved_at",
            "self_approval_overridden",
        ]
    )
    period = getattr(record, "period", None)
    if replaced is not None:
        _audit(
            actor=actor,
            action=f"payroll.{object_label}_voided_on_replacement",
            instance=replaced,
            period=period,
            metadata={
                "previous_status": previous_status,
                "replacement_id": record.pk,
            },
        )
    _audit(
        actor=actor,
        action=f"payroll.{object_label}_approved",
        instance=record,
        period=period,
        metadata={
            "approved_lock_version": record.lock_version,
            **_approval_audit_metadata(
                actor,
                self_approval=self_approval_overridden,
            ),
        },
    )
    return record


def _ensure_period_accepts_approvals(period: PayrollPeriod) -> None:
    if period.status == PayrollPeriodStatus.CLOSED:
        _operation_error(
            "PERIOD_INPUTS_LOCKED",
            "Данные закрытого периода заблокированы.",
        )
    if period.current_run_id is None:
        return
    current_run = PayrollRun.objects.select_for_update().get(pk=period.current_run_id)
    if current_run.status in {
        PayrollRunStatus.CALCULATED,
        PayrollRunStatus.REVIEW,
        PayrollRunStatus.APPROVED,
    }:
        _operation_error(
            "PERIOD_INPUTS_LOCKED",
            "Сначала верните текущий расчёт на исправление.",
            run_id=current_run.pk,
        )


@transaction.atomic
def clear_component_input_lines(
    period_id: int,
    *,
    employee_id: int,
    component_code: str,
    actor,
    validate_scheduled_result: bool = True,
) -> dict[str, int]:
    """Clear one spreadsheet-like component cell without losing its audit trail."""

    _require_permission(actor, "finance.manage_payroll_inputs")
    period = PayrollPeriod.objects.select_for_update().filter(pk=period_id).first()
    if period is None:
        _operation_error("PERIOD_NOT_FOUND", "Расчётный период не найден.")
    _ensure_period_accepts_approvals(period)
    employee = Employee.objects.filter(pk=employee_id).first()
    if employee is None:
        _operation_error("EMPLOYEE_NOT_FOUND", "Сотрудник не найден.")
    scheduled_component = scheduled_component_for_period(component_code, period)
    if (
        scheduled_component is None
        and validate_scheduled_result
        and employee_is_eligible_for_scheduled_change(employee, period)
    ):
        scheduled_delta = target_cell_scheduled_delta(
            period=period,
            employee_id=employee_id,
            component_code=component_code,
        )
        if scheduled_delta < 0:
            _operation_error(
                "NEGATIVE_SCHEDULED_TARGET",
                (
                    "Нельзя очистить обычное значение: применённый маркер "
                    "сделает итог колонки отрицательным."
                ),
                scheduled_delta=str(scheduled_delta),
            )
    lines = list(
        PayrollInputLine.objects.select_for_update()
        .filter(
            period=period,
            employee_id=employee_id,
            component_code=component_code,
            status__in=(ApprovalStatus.DRAFT, ApprovalStatus.APPROVED),
        )
        .order_by("pk")
    )
    if scheduled_component is None:
        lines = [line for line in lines if not is_generated_scheduled_change(line)]
    generated_lines = []
    if scheduled_component is not None and lines:
        generated_refs = [
            f"{SCHEDULED_CHANGE_SOURCE_PREFIX}{line.pk}" for line in lines
        ]
        generated_lines = list(
            PayrollInputLine.objects.select_for_update()
            .filter(
                source_ref__in=generated_refs,
                status__in=(ApprovalStatus.DRAFT, ApprovalStatus.APPROVED),
            )
            .order_by("pk")
        )
        for target_period in {generated.period for generated in generated_lines}:
            try:
                _ensure_period_accepts_approvals(target_period)
            except PayrollOperationError as exc:
                _operation_error(
                    "TARGET_PERIOD_INPUTS_LOCKED",
                    "Период применения уже заблокирован для изменений.",
                    application_period_id=target_period.pk,
                    original_code=exc.code,
                )
    now = timezone.now()
    summary = {"draft": 0, "approved": 0, "total": len(lines)}
    for line in lines:
        previous_status = line.status
        summary[previous_status] += 1
        line.status = ApprovalStatus.VOIDED
        line.voided_by = actor
        line.voided_at = now
        line.void_reason = "Значение очищено в итоговой таблице"
        line.save(
            update_fields=[
                "status",
                "voided_by",
                "voided_at",
                "void_reason",
            ]
        )
        _audit(
            actor=actor,
            action="payroll.input_line_voided_from_table",
            instance=line,
            period=period,
            metadata={"previous_status": previous_status},
        )
    for generated in generated_lines:
        generated.status = ApprovalStatus.VOIDED
        generated.voided_by = actor
        generated.voided_at = now
        generated.void_reason = "Исходное запланированное изменение очищено"
        generated.save(
            update_fields=["status", "voided_by", "voided_at", "void_reason"]
        )
    return summary


@transaction.atomic
def set_component_input_line_value(
    period_id: int,
    *,
    employee_id: int,
    component_code: str,
    amount: Decimal,
    actor,
) -> PayrollInputLine:
    """Replace the aggregate value of one editable spreadsheet component cell."""

    period = PayrollPeriod.objects.select_for_update().get(pk=period_id)
    try:
        get_payroll_component(component_code, period=period)
    except ValueError as exc:
        raise PayrollOperationError(
            "COMPONENT_NOT_AVAILABLE",
            "Колонка недоступна в выбранном расчётном периоде.",
            details={"component_code": component_code},
        ) from exc
    scheduled_component = scheduled_component_for_period(component_code, period)
    target_period = None
    if scheduled_component is not None:
        target_period = next_payroll_period(period, for_update=True)
        if target_period is not None:
            try:
                _ensure_period_accepts_approvals(target_period)
            except PayrollOperationError as exc:
                _operation_error(
                    "TARGET_PERIOD_INPUTS_LOCKED",
                    "Следующий период уже заблокирован для изменений.",
                    application_period_id=target_period.pk,
                    original_code=exc.code,
                )
    signed_amount = Decimal(amount)
    if signed_amount == 0:
        _operation_error(
            "INPUT_VALIDATION_FAILED",
            "Значение ячейки не может быть нулевым.",
        )
    stored_amount = signed_amount
    if scheduled_component is None:
        employee = Employee.objects.filter(pk=employee_id).first()
        if employee is None:
            _operation_error("EMPLOYEE_NOT_FOUND", "Сотрудник не найден.")
        if employee_is_eligible_for_scheduled_change(employee, period):
            stored_amount -= target_cell_scheduled_delta(
                period=period,
                employee_id=employee_id,
                component_code=component_code,
            )
        if stored_amount <= 0:
            _operation_error(
                "INVALID_COMPONENT_BASE",
                (
                    "Итоговое значение должно быть больше уже применённого "
                    "запланированного изменения."
                ),
                desired_amount=str(signed_amount),
                scheduled_delta=str(signed_amount - stored_amount),
            )
    clear_component_input_lines(
        period_id,
        employee_id=employee_id,
        component_code=component_code,
        actor=actor,
        validate_scheduled_result=False,
    )
    line = PayrollInputLine(
        period=period,
        employee_id=employee_id,
        component_code=component_code,
        amount=abs(stored_amount),
        value_sign=-1 if stored_amount < 0 else 1,
        reason="Перезаписано в итоговой таблице",
        status=ApprovalStatus.DRAFT,
        source=InputSource.MANUAL,
        created_by=actor,
    )
    try:
        line.full_clean()
        line.save(force_insert=True)
    except ValidationError as exc:
        raise PayrollOperationError(
            "INPUT_VALIDATION_FAILED",
            "Новое значение ячейки не прошло проверку.",
            details={
                "errors": (
                    exc.message_dict if hasattr(exc, "message_dict") else exc.messages
                )
            },
        ) from exc
    if scheduled_component is not None and target_period is not None:
        try:
            materialize_scheduled_changes(
                source_period=period,
                target_period=target_period,
            )
            generated = PayrollInputLine.objects.get(
                source=InputSource.API,
                source_ref=f"{SCHEDULED_CHANGE_SOURCE_PREFIX}{line.pk}",
            )
        except ValidationError as exc:
            raise PayrollOperationError(
                "INPUT_VALIDATION_FAILED",
                "Запланированное изменение не прошло проверку.",
                details={
                    "errors": (
                        exc.message_dict
                        if hasattr(exc, "message_dict")
                        else exc.messages
                    )
                },
            ) from exc
        target_lines = PayrollInputLine.objects.filter(
            period=target_period,
            employee_id=employee_id,
            component_code=generated.component_code,
            status__in=(ApprovalStatus.DRAFT, ApprovalStatus.APPROVED),
        )
        base_amount = sum(
            (
                target_line.signed_amount
                for target_line in target_lines
                if not is_generated_scheduled_change(target_line)
            ),
            Decimal("0"),
        )
        scheduled_delta = sum(
            (
                target_line.signed_amount
                for target_line in target_lines
                if is_generated_scheduled_change(target_line)
            ),
            Decimal("0"),
        )
        result_amount = base_amount + scheduled_delta
        if result_amount < 0:
            _operation_error(
                "NEGATIVE_SCHEDULED_TARGET",
                "Запланированное изменение делает значение колонки отрицательным.",
                application_period_id=target_period.pk,
                target_component_code=generated.component_code,
                current_amount=str(base_amount),
                scheduled_delta=str(scheduled_delta),
                result_amount=str(result_amount),
            )
        target_point_base = point_base_accruals_for_employees(
            period=target_period,
            employee_ids=[employee_id],
        )[employee_id]
        if target_point_base < 0:
            _operation_error(
                "NEGATIVE_POINT_BASE",
                "Изменение создаёт отрицательную балльную базу в периоде применения.",
                employee_id=employee_id,
                component_code=generated.component_code,
                application_period_id=target_period.pk,
                point_base=str(target_point_base),
            )
    point_base_accrual = point_base_accruals_for_employees(
        period=period,
        employee_ids=[employee_id],
    )[employee_id]
    if point_base_accrual < 0:
        _operation_error(
            "NEGATIVE_POINT_BASE",
            (
                "Значение создаёт отрицательную балльную базу. "
                "Измените сумму или настройки колонки."
            ),
            employee_id=employee_id,
            component_code=component_code,
            point_base=str(point_base_accrual),
        )
    _audit(
        actor=actor,
        action="payroll.input_line_draft_created_from_table",
        instance=line,
        period=period,
        metadata={"component_code": component_code, "amount": str(signed_amount)},
    )
    return line


@transaction.atomic
def approve_period_drafts(period_id: int, *, actor) -> dict[str, int]:
    """Approve every draft shown in a period queue as one transaction."""

    _require_permission(actor, "finance.approve_payroll_inputs")

    # Keep the global lock order used by single-record rate approval: periods,
    # active runs, rates, work records, then input lines.
    periods = list(PayrollPeriod.objects.select_for_update().order_by("pk"))
    period = next((item for item in periods if item.pk == period_id), None)
    if period is None:
        _operation_error("PERIOD_NOT_FOUND", "Расчётный период не найден.")

    open_periods = [
        item for item in periods if item.status != PayrollPeriodStatus.CLOSED
    ]
    current_run_ids = [
        item.current_run_id for item in open_periods if item.current_run_id is not None
    ]
    current_runs = list(
        PayrollRun.objects.select_for_update()
        .filter(pk__in=current_run_ids)
        .order_by("pk")
    )
    _ensure_period_accepts_approvals(period)

    rates = list(
        pay_rates_relevant_to_period(
            period,
            queryset=EmployeePayRate.objects.select_for_update(),
        )
        .filter(status=ApprovalStatus.DRAFT)
        .order_by("pk")
    )
    work_records = list(
        PayrollWorkRecord.objects.select_for_update()
        .select_related("period")
        .filter(period=period, status=ApprovalStatus.DRAFT)
        .order_by("pk")
    )
    input_refs = list(
        PayrollInputLine.objects.filter(
            period=period,
            status=ApprovalStatus.DRAFT,
        )
        .exclude(source_ref__startswith=SCHEDULED_CHANGE_SOURCE_PREFIX)
        .order_by("pk")
        .values_list("pk", "component_code")
    )
    input_lines = list(
        PayrollInputLine.objects.select_for_update()
        .select_related("period")
        .filter(pk__in=[line_id for line_id, _ in input_refs])
        .order_by("pk")
    )

    active_run_statuses = {
        PayrollRunStatus.CALCULATED,
        PayrollRunStatus.REVIEW,
        PayrollRunStatus.APPROVED,
    }
    for rate in rates:
        for run in current_runs:
            if (
                run.status in active_run_statuses
                and run.statements.filter(employee_id=rate.employee_id).exists()
            ):
                _operation_error(
                    "RATE_USED_BY_ACTIVE_RUN",
                    "Ставку нельзя утвердить, пока расчёт сотрудника согласуется.",
                    run_id=run.pk,
                )

    for record in work_records:
        if period.status == PayrollPeriodStatus.PUBLISHED and not record.replaces_id:
            _operation_error(
                "PUBLISHED_WORK_REPLACEMENT_REQUIRED",
                "После публикации показатели меняются только новой ревизией.",
            )

    for line in input_lines:
        component = scheduled_component_for_period(line.component_code, period)
        if component is not None:
            target_period = next_payroll_period(period)
            if target_period is None:
                continue
            try:
                _ensure_period_accepts_approvals(target_period)
            except PayrollOperationError as exc:
                _operation_error(
                    "TARGET_PERIOD_INPUTS_LOCKED",
                    "Период применения уже заблокирован для изменений.",
                    application_period_id=target_period.pk,
                    original_code=exc.code,
                )

    for rate in rates:
        _approve_record(
            rate,
            actor=actor,
            object_label="rate",
            expected_lock_version=rate.lock_version,
        )
    for record in work_records:
        _approve_record(
            record,
            actor=actor,
            object_label="work_record",
            expected_lock_version=record.lock_version,
        )
    for line in input_lines:
        approved_line = _approve_record(
            line,
            actor=actor,
            object_label="input_line",
            expected_lock_version=line.lock_version,
        )
        _mark_linked_vacation_instruction_posted(approved_line)
        _sync_generated_scheduled_change_approval(approved_line)

    summary = {
        "rates": len(rates),
        "work_records": len(work_records),
        "input_lines": len(input_lines),
    }
    summary["total"] = sum(summary.values())
    _audit(
        actor=actor,
        action="payroll.period_drafts_approved",
        instance=period,
        period=period,
        metadata=summary,
    )
    return summary


@transaction.atomic
def approve_pay_rate(
    rate_id: int,
    *,
    actor,
    expected_lock_version: int,
) -> EmployeePayRate:
    _require_permission(actor, "finance.approve_payroll_inputs")
    # A rate can affect any open period. Lock periods before runs and the rate,
    # matching the global payroll lock order and serializing against calculate.
    periods = list(
        PayrollPeriod.objects.select_for_update()
        .exclude(status=PayrollPeriodStatus.CLOSED)
        .order_by("pk")
    )
    current_run_ids = [
        period.current_run_id for period in periods if period.current_run_id is not None
    ]
    current_runs = list(
        PayrollRun.objects.select_for_update()
        .filter(pk__in=current_run_ids)
        .order_by("pk")
    )
    record = EmployeePayRate.objects.select_for_update().get(pk=rate_id)
    for run in current_runs:
        if (
            run.status
            in {
                PayrollRunStatus.CALCULATED,
                PayrollRunStatus.REVIEW,
                PayrollRunStatus.APPROVED,
            }
            and run.statements.filter(employee_id=record.employee_id).exists()
        ):
            _operation_error(
                "RATE_USED_BY_ACTIVE_RUN",
                "Ставку нельзя утвердить, пока расчёт сотрудника согласуется.",
                run_id=run.pk,
            )
    return _approve_record(
        record,
        actor=actor,
        object_label="rate",
        expected_lock_version=expected_lock_version,
    )


@transaction.atomic
def approve_work_record(
    record_id: int,
    *,
    actor,
    expected_lock_version: int,
) -> PayrollWorkRecord:
    _require_permission(actor, "finance.approve_payroll_inputs")
    reference = PayrollWorkRecord.objects.only("period_id").get(pk=record_id)
    period = PayrollPeriod.objects.select_for_update().get(pk=reference.period_id)
    _ensure_period_accepts_approvals(period)
    record = PayrollWorkRecord.objects.select_for_update().get(pk=record_id)
    if period.status == PayrollPeriodStatus.PUBLISHED and not record.replaces_id:
        _operation_error(
            "PUBLISHED_WORK_REPLACEMENT_REQUIRED",
            "После публикации показатели меняются только новой ревизией.",
        )
    return _approve_record(
        record,
        actor=actor,
        object_label="work_record",
        expected_lock_version=expected_lock_version,
    )


@transaction.atomic
def approve_input_line(
    line_id: int,
    *,
    actor,
    expected_lock_version: int,
) -> PayrollInputLine:
    _require_permission(actor, "finance.approve_payroll_inputs")
    reference = PayrollInputLine.objects.only("period_id").get(pk=line_id)
    period = PayrollPeriod.objects.select_for_update().get(pk=reference.period_id)
    _ensure_period_accepts_approvals(period)
    record = PayrollInputLine.objects.select_for_update().get(pk=line_id)
    if is_generated_scheduled_change(record):
        _operation_error(
            "SCHEDULED_CHANGE_DERIVED",
            "Автоматическая строка утверждается вместе с исходным маркером.",
        )
    component = scheduled_component_for_period(record.component_code, period)
    if component is not None:
        target_period = next_payroll_period(period, for_update=True)
        if target_period is not None:
            try:
                _ensure_period_accepts_approvals(target_period)
            except PayrollOperationError as exc:
                _operation_error(
                    "TARGET_PERIOD_INPUTS_LOCKED",
                    "Период применения уже заблокирован для изменений.",
                    application_period_id=target_period.pk,
                    original_code=exc.code,
                )
    approved_line = _approve_record(
        record,
        actor=actor,
        object_label="input_line",
        expected_lock_version=expected_lock_version,
    )
    _mark_linked_vacation_instruction_posted(approved_line)
    _sync_generated_scheduled_change_approval(approved_line)
    return approved_line


@transaction.atomic
def acknowledge_statement(
    statement_id: int,
    *,
    actor,
    idempotency_key: uuid.UUID | None = None,
) -> PayrollStatementAcknowledgement:
    """Acknowledge receiving a current published statement, idempotently."""

    if actor is None or not getattr(actor, "is_authenticated", False):
        raise PayrollPermissionDenied("finance.acknowledge_own_statement")
    statement_reference = PayrollStatement.objects.only(
        "run_id",
        "employee_id",
    ).get(pk=statement_id)
    run_reference = PayrollRun.objects.only("period_id").get(
        pk=statement_reference.run_id
    )
    period = PayrollPeriod.objects.select_for_update().get(pk=run_reference.period_id)
    run = PayrollRun.objects.select_for_update().get(pk=statement_reference.run_id)
    statement = PayrollStatement.objects.select_for_update().get(pk=statement_id)
    if statement.employee_id != actor.pk:
        _operation_error(
            "STATEMENT_NOT_FOUND",
            "Расчётный листок не найден.",
        )
    latest_published = (
        PayrollRun.objects.filter(
            period_id=period.pk,
            status=PayrollRunStatus.PUBLISHED,
        )
        .order_by("-revision")
        .first()
    )
    if latest_published is None or latest_published.pk != run.pk:
        _operation_error(
            "STATEMENT_NOT_CURRENT",
            "Подтвердить можно только актуальный опубликованный листок.",
        )

    if idempotency_key is not None:
        keyed_acknowledgement = PayrollStatementAcknowledgement.objects.filter(
            idempotency_key=idempotency_key
        ).first()
        if keyed_acknowledgement is not None:
            if keyed_acknowledgement.statement_id != statement.pk:
                _operation_error(
                    "IDEMPOTENCY_KEY_CONFLICT",
                    "Ключ уже использован для другого листка.",
                )
            return keyed_acknowledgement

    acknowledgement = PayrollStatementAcknowledgement.objects.filter(
        statement=statement
    ).first()
    if acknowledgement and acknowledgement.acknowledged_at:
        return acknowledgement

    now = timezone.now()
    if acknowledgement is None:
        acknowledgement = PayrollStatementAcknowledgement(
            statement=statement,
            employee=actor,
            content_hash=statement.result_hash,
            idempotency_key=idempotency_key or uuid.uuid4(),
            viewed_at=now,
            acknowledged_at=now,
        )
    else:
        acknowledgement.viewed_at = acknowledgement.viewed_at or now
        acknowledgement.acknowledged_at = now
    acknowledgement.full_clean()
    try:
        with transaction.atomic():
            acknowledgement.save()
    except IntegrityError as exc:
        existing = None
        if idempotency_key is not None:
            existing = PayrollStatementAcknowledgement.objects.filter(
                idempotency_key=idempotency_key
            ).first()
        if existing is not None and existing.statement_id == statement.pk:
            return existing
        raise PayrollOperationError(
            "ACKNOWLEDGEMENT_CONFLICT",
            "Подтверждение уже обработано другой операцией.",
        ) from exc
    _audit(
        actor=actor,
        action="payroll.statement_acknowledged",
        instance=acknowledgement,
        period=period,
        after_hash=statement.result_hash,
    )
    return acknowledgement