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 == "half_year": today = timezone.localdate() month_index = today.year * 12 + today.month - 1 - 5 start_year, start_month_index = divmod(month_index, 12) return date(start_year, start_month_index + 1, 1), today 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 _vacation_paid_days_in_window(request_obj, duration): terms = getattr(request_obj, "vacation_terms", None) if terms is None: return 0 if _is_unpaid_vacation(request_obj) else duration return min(terms.paid_days, duration) 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, ) .select_related("vacation_terms") .only( "type", "date_from", "date_to", "time_from", "time_to", "title", "comment", "vacation_terms__paid_days", ) ) 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: paid_days = _vacation_paid_days_in_window(request_obj, duration) totals["paid_vacation_days"] += paid_days totals["unpaid_vacation_days"] += max(duration - paid_days, 0) 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, }