Corp/backend/requests_app/application/statistics.py

import calendar
from datetime import date

from django.utils import timezone
from rest_framework.exceptions import ValidationError

from requests_app.enums import DAY_OFF_REQUEST_TYPES, RequestStatus, RequestType
from requests_app.models import Request as EmployeeRequest


UNPAID_VACATION_MARKERS = (
    "за свой счет",
    "за свой счёт",
    "без сохранения заработной платы",
    "без сохранения зарплаты",
    "без сохранения зп",
    "неоплачиваем",
)


def can_view_request_statistics(user) -> bool:
    return bool(
        getattr(user, "is_staff", False)
        or getattr(user, "is_superuser", False)
        or user.has_perm("requests_app.view_statistics")
        or user.has_perm("requests_app.can_view_all_requests")
    )


def resolve_stats_window_from_query(period, query_params):
    if period == "month":
        today = timezone.localdate()
        _, last_day = calendar.monthrange(today.year, today.month)
        return today.replace(day=1), today.replace(day=last_day)
    if period == "year":
        today = timezone.localdate()
        return today.replace(month=1, day=1), today.replace(month=12, day=31)
    if period != "custom":
        return None, None

    date_from_raw = (query_params.get("date_from") or "").strip()
    date_to_raw = (query_params.get("date_to") or "").strip()
    if not date_from_raw or not date_to_raw:
        raise ValidationError(
            {
                "date_from": ["Обязательное поле для custom периода."],
                "date_to": ["Обязательное поле для custom периода."],
            }
        )
    start_date = _parse_iso_date(date_from_raw, "date_from")
    end_date = _parse_iso_date(date_to_raw, "date_to")
    if start_date > end_date:
        raise ValidationError(
            {"date_to": ["Дата окончания не может быть раньше даты начала."]}
        )
    return start_date, end_date


def _parse_iso_date(raw_value, field_name):
    try:
        return date.fromisoformat(raw_value)
    except ValueError as exc:
        raise ValidationError(
            {field_name: ["Ожидается дата в формате YYYY-MM-DD."]}
        ) from exc


def _duration_days_in_window(request_obj, start_date, end_date):
    if start_date is None or end_date is None:
        if request_obj.date_from and request_obj.date_to:
            return max((request_obj.date_to - request_obj.date_from).days + 1, 0)
        return 1 if request_obj.date_from else 0
    if request_obj.date_from and request_obj.date_to:
        overlap_start = max(request_obj.date_from, start_date)
        overlap_end = min(request_obj.date_to, end_date)
        return 0 if overlap_start > overlap_end else (overlap_end - overlap_start).days + 1
    return int(
        bool(
            request_obj.date_from
            and start_date <= request_obj.date_from <= end_date
        )
    )


def _is_unpaid_vacation(request_obj):
    text = " ".join(
        [
            str(getattr(request_obj, "title", "") or ""),
            str(getattr(request_obj, "comment", "") or ""),
        ]
    ).lower()
    return any(marker in text for marker in UNPAID_VACATION_MARKERS)


def build_employee_statistics(
    *,
    employee_id,
    employee_name,
    period,
    start_date=None,
    end_date=None,
):
    base_queryset = EmployeeRequest.objects.filter(employee_id=employee_id).exclude(
        status=RequestStatus.DRAFT
    )
    if start_date and end_date:
        base_queryset = base_queryset.filter(
            created_at__date__gte=start_date,
            created_at__date__lte=end_date,
        )

    totals = {
        "sick_leave_days": 0,
        "day_off_days": 0,
        "short_day_off_minutes": 0,
        "maternity_days": 0,
        "paid_vacation_days": 0,
        "unpaid_vacation_days": 0,
    }
    approved_requests = EmployeeRequest.objects.filter(
        employee_id=employee_id,
        status=RequestStatus.APPROVED,
    ).only("type", "date_from", "date_to", "time_from", "time_to", "title", "comment")
    for request_obj in approved_requests:
        duration = _duration_days_in_window(request_obj, start_date, end_date)
        if request_obj.type == RequestType.SICK_LEAVE:
            totals["sick_leave_days"] += duration
        elif request_obj.type == RequestType.SHORT_DAY_OFF:
            if request_obj.date_from and (
                start_date is None or start_date <= request_obj.date_from
            ) and (end_date is None or request_obj.date_from <= end_date):
                totals["short_day_off_minutes"] += request_obj.duration_minutes or 0
        elif request_obj.type == RequestType.DAY_OFF:
            totals["day_off_days"] += duration
        elif request_obj.type == RequestType.MATERNITY:
            totals["maternity_days"] += duration
        elif request_obj.type == RequestType.VACATION:
            key = (
                "unpaid_vacation_days"
                if _is_unpaid_vacation(request_obj)
                else "paid_vacation_days"
            )
            totals[key] += duration

    return {
        "employee_id": employee_id,
        "employee_name": employee_name,
        "period": period,
        "date_from": start_date.isoformat() if start_date else None,
        "date_to": end_date.isoformat() if end_date else None,
        "total_submitted_requests": base_queryset.count(),
        "sick_leave_requests_count": base_queryset.filter(type=RequestType.SICK_LEAVE).count(),
        "day_off_requests_count": base_queryset.filter(type__in=DAY_OFF_REQUEST_TYPES).count(),
        "short_day_off_requests_count": base_queryset.filter(type=RequestType.SHORT_DAY_OFF).count(),
        "maternity_requests_count": base_queryset.filter(type=RequestType.MATERNITY).count(),
        **totals,
    }