"""Shared payroll input-copy service, independent of HTTP adapters.""" from decimal import Decimal from django.utils import timezone from employees.constants import ACTION_DISMISSED from employees.services.personnel_state import resolve_employee_personnel_state from finance.enums import ApprovalStatus, InputSource, PayrollComponentKind from finance.models import PayrollAuditEvent, PayrollInputLine from finance.payroll.components import get_payroll_components from finance.payroll.scheduled_changes import ( employee_is_eligible_for_scheduled_change, is_generated_scheduled_change, ) def _audit_copy_mutation(*, actor, action, instance, period, metadata): PayrollAuditEvent.objects.create( actor=actor, action=action, object_type=instance._meta.label_lower, object_id=str(instance.pk), period=period, metadata=metadata, ) def _copyable_input_lines(source_period, *, target_period=None): component_codes = { component.code for component in get_payroll_components(source_period) if component.kind != PayrollComponentKind.SCHEDULED_CHANGE } if target_period is not None: component_codes &= { component.code for component in get_payroll_components(target_period) if component.kind != PayrollComponentKind.SCHEDULED_CHANGE } return ( PayrollInputLine.objects.filter( period=source_period, employee__is_hidden_in_ui=False, status__in=(ApprovalStatus.DRAFT, ApprovalStatus.APPROVED), component_code__in=component_codes, ) .exclude( source=InputSource.API, source_ref__startswith="payroll-carryover:", ) .select_related( "employee", "employee__position", "relates_to_period", ) .prefetch_related( "employee__departments_links__department", "employee__actions__source_request", ) .order_by("employee_id", "component_code", "id") ) def _employee_active_for_period(employee, period): if not employee.is_active: return False actions = employee._prefetched_objects_cache.get("actions") state = resolve_employee_personnel_state( employee, period.date_to, actions=actions, ) return state.status != ACTION_DISMISSED def _collapse_scheduled_effective_cells(lines): """Fold a marker-adjusted source cell into one ordinary copied value.""" lines = [ line for line in lines if not is_generated_scheduled_change(line) or employee_is_eligible_for_scheduled_change(line.employee, line.period) ] grouped = {} for line in lines: grouped.setdefault((line.employee_id, line.component_code), []).append(line) result = [] for (employee_id, component_code), cell_lines in grouped.items(): if not any(is_generated_scheduled_change(line) for line in cell_lines): result.extend(cell_lines) continue amount = sum((line.signed_amount for line in cell_lines), Decimal("0")) if amount <= 0: continue source_ids = ",".join(str(line.pk) for line in cell_lines) effective = PayrollInputLine( period=cell_lines[0].period, employee=cell_lines[0].employee, component_code=component_code, amount=amount, value_sign=1, relates_to_period=None, reason="Перенесено итоговое значение с применённым маркером", include_in_carryover=any(line.include_in_carryover for line in cell_lines), created_by=cell_lines[0].created_by, ) effective._copy_source_token = ( f"effective-{employee_id}-{component_code}-{source_ids}" ) result.append(effective) return result def apply_input_copy( *, target_period, source_period, actor, data, approved_only=False, channel="admin_api", ): copy_prefix = ( f"payroll-period-copy:{target_period.pk}:{source_period.pk}:" f"{data['idempotency_key']}:" ) existing = list( PayrollInputLine.objects.filter( period=target_period, source_ref__startswith=copy_prefix, ) .order_by("id") .values_list("id", flat=True) ) if existing: return { "source_period_id": source_period.pk, "target_period_id": target_period.pk, "summary": { "created": len(existing), "employees": PayrollInputLine.objects.filter(pk__in=existing) .values("employee_id") .distinct() .count(), "skipped": 0, "replaced_cells": 0, "voided": 0, }, "record_ids": existing, "idempotent_replay": True, } lines = list( _copyable_input_lines( source_period, target_period=target_period, ).filter(component_code__in=data["component_codes"]) ) if approved_only: lines = [line for line in lines if line.status == ApprovalStatus.APPROVED] lines = _collapse_scheduled_effective_cells(lines) if data["employee_scope"] == "selected": selected_ids = set(data["employee_ids"]) lines = [line for line in lines if line.employee_id in selected_ids] elif data["employee_scope"] == "active": lines = [ line for line in lines if _employee_active_for_period(line.employee, target_period) ] selected_cells = {(line.employee_id, line.component_code) for line in lines} existing_lines = [] if selected_cells: possible_existing_lines = list( PayrollInputLine.objects.select_for_update() .filter( period=target_period, employee_id__in={employee_id for employee_id, _ in selected_cells}, component_code__in={ component_code for _, component_code in selected_cells }, status__in=(ApprovalStatus.DRAFT, ApprovalStatus.APPROVED), ) .order_by("pk") ) existing_lines = [ line for line in possible_existing_lines if (line.employee_id, line.component_code) in selected_cells and not is_generated_scheduled_change(line) and not ( line.source == InputSource.API and line.source_ref.startswith("payroll-carryover:") ) ] occupied_cells = { (line.employee_id, line.component_code) for line in existing_lines } skipped = 0 replaced_cells = 0 voided = 0 if data["existing_mode"] == "skip" and occupied_cells: kept_lines = [] for line in lines: if (line.employee_id, line.component_code) in occupied_cells: skipped += 1 else: kept_lines.append(line) lines = kept_lines elif data["existing_mode"] == "replace" and existing_lines: replaced_cells = len(occupied_cells) voided = len(existing_lines) now = timezone.now() for line in existing_lines: previous_status = line.status line.status = ApprovalStatus.VOIDED line.voided_by = actor line.voided_at = now line.void_reason = ( f"Заменено переносом записей из периода {source_period.code}" ) line.save( update_fields=[ "status", "voided_by", "voided_at", "void_reason", ] ) _audit_copy_mutation( actor=actor, action="payroll.input_line_voided_from_period_copy", instance=line, period=target_period, metadata={ "channel": channel, "previous_status": previous_status, "source_period_id": source_period.pk, }, ) records = PayrollInputLine.objects.bulk_create( [ PayrollInputLine( period=target_period, employee=line.employee, component_code=line.component_code, amount=line.amount, value_sign=line.value_sign, relates_to_period=line.relates_to_period, reason=line.reason, status=ApprovalStatus.DRAFT, source=InputSource.API, source_ref=( f"{copy_prefix}{getattr(line, '_copy_source_token', line.pk)}" ), include_in_carryover=line.include_in_carryover, created_by=actor, ) for line in lines ] ) PayrollAuditEvent.objects.bulk_create( [ PayrollAuditEvent( actor=actor, action="payroll.input_line_draft_created", object_type=PayrollInputLine._meta.label_lower, object_id=str(record.pk), period=target_period, metadata={ "channel": channel, "copied_from_period_id": source_period.pk, "copied_from_input_line_id": source.pk, }, ) for record, source in zip(records, lines, strict=True) ] ) return { "source_period_id": source_period.pk, "target_period_id": target_period.pk, "summary": { "created": len(records), "employees": len({record.employee_id for record in records}), "skipped": skipped, "replaced_cells": replaced_cells, "voided": voided, }, "record_ids": [record.pk for record in records], "idempotent_replay": False, }