Corp/backend/employees/services/personnel_state.py

from __future__ import annotations

from dataclasses import dataclass
from datetime import date, datetime, time
from typing import Iterable

from django.utils import timezone

from employees.constants import (
    ACTION_DISMISSED,
    ACTION_LABELS,
    ACTION_REMOTE,
    ACTION_WORKING,
    ACTIVATING_MARKER_ACTIONS,
    MARKER_ACTIONS,
    PERMANENT_ACTIONS,
    RETURN_MARKER_ACTIONS,
    TEMPORARY_ACTION_PRIORITIES,
    TEMPORARY_START_ACTIONS,
)
from employees.models import Employee, EmployeeAction

PERSONNEL_STATUS_NORMAL = "normal"
EMPLOYMENT_START_ACTIONS = ACTIVATING_MARKER_ACTIONS | {
    ACTION_WORKING,
    ACTION_REMOTE,
}


@dataclass(frozen=True)
class EmployeePersonnelState:
    status: str = PERSONNEL_STATUS_NORMAL
    label: str = ""
    action_id: int | None = None
    date_from: date | None = None
    date_to: date | None = None
    expects_attendance: bool = True
    approved_absence_intervals: tuple[tuple[time, time], ...] = ()

    @property
    def is_non_working(self) -> bool:
        return not self.expects_attendance

    @property
    def approved_absence_minutes(self) -> int:
        return sum(
            _time_to_minutes(end) - _time_to_minutes(start)
            for start, end in self.approved_absence_intervals
        )


def resolve_employee_personnel_state(
    employee: Employee,
    target_date: date,
    *,
    actions: Iterable[EmployeeAction] | None = None,
) -> EmployeePersonnelState:
    employee_actions = _sorted_actions(
        actions
        if actions is not None
        else employee.actions.select_related("source_request").all()
    )
    relevant_actions = [
        action
        for action in employee_actions
        if _action_date(action) is not None and _action_date(action) <= target_date
    ]
    if not relevant_actions:
        return EmployeePersonnelState()

    partial_day_off_intervals = _short_day_off_intervals(
        employee_actions,
        target_date,
    )
    partial_kwargs = {
        "approved_absence_intervals": partial_day_off_intervals,
    }

    permanent_action = _resolve_permanent_action(relevant_actions)
    if permanent_action and permanent_action.action == ACTION_DISMISSED:
        return EmployeePersonnelState(
            status=ACTION_DISMISSED,
            label=ACTION_LABELS[ACTION_DISMISSED],
            action_id=permanent_action.id,
            date_from=_action_date(permanent_action),
            expects_attendance=False,
            **partial_kwargs,
        )

    active_temporary = _active_temporary_actions(employee_actions, target_date)
    if active_temporary:
        action, date_to = active_temporary[0]
        return EmployeePersonnelState(
            status=action.action,
            label=action.get_action_display(),
            action_id=action.id,
            date_from=_action_date(action),
            date_to=date_to,
            expects_attendance=False,
            **partial_kwargs,
        )

    marker_action = _active_marker_action(relevant_actions, target_date)
    if marker_action:
        return EmployeePersonnelState(
            status=marker_action.action,
            label=marker_action.get_action_display(),
            action_id=marker_action.id,
            date_from=_action_date(marker_action),
            expects_attendance=True,
            **partial_kwargs,
        )

    if permanent_action:
        return EmployeePersonnelState(
            status=permanent_action.action,
            label=permanent_action.get_action_display(),
            action_id=permanent_action.id,
            date_from=_action_date(permanent_action),
            **partial_kwargs,
        )

    return EmployeePersonnelState(
        status=ACTION_WORKING,
        label=ACTION_LABELS[ACTION_WORKING],
        **partial_kwargs,
    )


def _sorted_actions(actions: Iterable[EmployeeAction]) -> list[EmployeeAction]:
    return sorted(
        actions,
        key=lambda action: (
            _action_date(action) or date.min,
            action.id or 0,
        ),
    )


def calculate_employee_tenure_days(
    employee: Employee,
    target_date: date,
    *,
    actions: Iterable[EmployeeAction] | None = None,
) -> int | None:
    """Считает фактический стаж по кадровым событиям.

    Период работы открывается приемом/восстановлением или рабочим постоянным
    статусом и закрывается увольнением. Разрывы между увольнением и повторным
    приемом в стаж не входят.
    """
    employee_actions = _sorted_actions(
        actions
        if actions is not None
        else employee.actions.select_related("source_request").all()
    )
    total_days = 0
    current_start: date | None = None
    has_employment_period = False

    for action in employee_actions:
        action_date = _action_date(action)
        if action_date is None or action_date > target_date:
            continue

        if action.action in EMPLOYMENT_START_ACTIONS:
            if current_start is None:
                current_start = action_date
                has_employment_period = True
            continue

        if action.action == ACTION_DISMISSED and current_start is not None:
            total_days += max((action_date - current_start).days, 0)
            current_start = None

    if current_start is not None:
        total_days += max((target_date - current_start).days, 0)

    return total_days if has_employment_period else None


def _action_date(action: EmployeeAction) -> date | None:
    if not action.date:
        return None
    if isinstance(action.date, datetime):
        if timezone.is_aware(action.date):
            return timezone.localtime(action.date).date()
        return action.date.date()
    return action.date


def _resolve_permanent_action(
    actions: Iterable[EmployeeAction],
) -> EmployeeAction | None:
    ordered_actions = list(actions)
    permanent_actions = [
        action for action in ordered_actions if action.action in PERMANENT_ACTIONS
    ]
    if not permanent_actions:
        return None

    last_permanent = permanent_actions[-1]
    if last_permanent.action != ACTION_DISMISSED:
        return last_permanent

    dismissal_index = ordered_actions.index(last_permanent)
    activation_after_dismissal = [
        action
        for action in ordered_actions[dismissal_index + 1 :]
        if action.action in ACTIVATING_MARKER_ACTIONS
    ]
    if not activation_after_dismissal:
        return last_permanent

    restored_state = [
        action for action in permanent_actions[:-1] if action.action != ACTION_DISMISSED
    ]
    return restored_state[-1] if restored_state else None


def _active_marker_action(
    actions: list[EmployeeAction],
    target_date: date,
) -> EmployeeAction | None:
    marker_actions = [
        action
        for action in actions
        if action.action in MARKER_ACTIONS
        and action.date
        and _action_date(action) == target_date
    ]
    return marker_actions[-1] if marker_actions else None


def _active_temporary_actions(
    actions: list[EmployeeAction],
    target_date: date,
) -> list[tuple[EmployeeAction, date | None]]:
    active = []
    for action in actions:
        if action.action not in TEMPORARY_START_ACTIONS or not action.date:
            continue
        if _is_short_day_off_action(action):
            continue

        date_from = _action_date(action)
        if date_from is None:
            continue
        if date_from > target_date:
            continue

        date_to = _resolve_temporary_action_date_to(action)
        if date_to is not None and date_to < target_date:
            continue

        active.append((action, date_to))

    return sorted(
        active,
        key=lambda item: (
            TEMPORARY_ACTION_PRIORITIES.get(item[0].action, 99),
            -(item[0].date.timestamp() if item[0].date else 0),
            -(item[0].id or 0),
        ),
    )


def _resolve_temporary_action_date_to(action: EmployeeAction) -> date | None:
    if action.date_to:
        return action.date_to
    if action.date:
        return _action_date(action)
    return None


def _is_short_day_off_action(action: EmployeeAction) -> bool:
    source_request = getattr(action, "source_request", None)
    if source_request is not None:
        return getattr(source_request, "type", None) == "short_day_off"
    return bool(isinstance(action.extra, dict) and action.extra.get("partial_day"))


def _short_day_off_intervals(
    actions: list[EmployeeAction],
    target_date: date,
) -> tuple[tuple[time, time], ...]:
    intervals: list[tuple[time, time]] = []
    for action in actions:
        if not _is_short_day_off_action(action) or _action_date(action) != target_date:
            continue
        source_request = getattr(action, "source_request", None)
        if source_request is not None:
            if getattr(source_request, "status", None) != "approved":
                continue
            start = getattr(source_request, "time_from", None)
            end = getattr(source_request, "time_to", None)
        else:
            extra = action.extra if isinstance(action.extra, dict) else {}
            try:
                start = time.fromisoformat(
                    str(extra.get("starts_at", "")).split("T")[-1]
                )
                end = time.fromisoformat(str(extra.get("ends_at", "")).split("T")[-1])
            except ValueError:
                continue
        if start:
            start = start.replace(tzinfo=None)
        if end:
            end = end.replace(tzinfo=None)
        if start and end and end > start:
            intervals.append((start, end))
    return _merge_time_intervals(intervals)


def _merge_time_intervals(
    intervals: list[tuple[time, time]],
) -> tuple[tuple[time, time], ...]:
    if not intervals:
        return ()
    ordered = sorted(intervals, key=lambda item: _time_to_minutes(item[0]))
    merged: list[tuple[time, time]] = [ordered[0]]
    for start, end in ordered[1:]:
        previous_start, previous_end = merged[-1]
        if _time_to_minutes(start) <= _time_to_minutes(previous_end):
            if end > previous_end:
                merged[-1] = (previous_start, end)
        else:
            merged.append((start, end))
    return tuple(merged)


def _time_to_minutes(value: time) -> int:
    return value.hour * 60 + value.minute


def approved_absence_minutes_within_schedule(
    state: EmployeePersonnelState,
    schedule: dict | None,
) -> int:
    """Return approved partial absence clipped to the employee work schedule."""
    if not state.approved_absence_intervals:
        return 0
    try:
        schedule_start = time.fromisoformat(str((schedule or {}).get("start_time")))
        schedule_end = time.fromisoformat(str((schedule or {}).get("end_time")))
    except ValueError:
        return state.approved_absence_minutes
    start_minutes = _time_to_minutes(schedule_start)
    end_minutes = _time_to_minutes(schedule_end)
    if end_minutes <= start_minutes:
        return state.approved_absence_minutes
    return sum(
        max(
            min(_time_to_minutes(end), end_minutes)
            - max(_time_to_minutes(start), start_minutes),
            0,
        )
        for start, end in state.approved_absence_intervals
    )


def is_temporary_start_action(action: str) -> bool:
    return action in TEMPORARY_START_ACTIONS


def is_temporary_return_action(action: str) -> bool:
    return action in RETURN_MARKER_ACTIONS


def action_expects_attendance(action: str) -> bool:
    return action not in TEMPORARY_START_ACTIONS and action != ACTION_DISMISSED