from datetime import date, timedelta from app.services.calculations import compute_net_minutes from app.services.targets import monday_of, target_map_for_weeks, week_starts_between from app.services.vacations import expand_vacation_dates from app.services.workdays import DEFAULT_WORKING_DAYS, is_workday def compute_effective_span_totals( *, entries: list, range_start: date, range_end: date, weekly_target_minutes: int, vacation_dates: set[date] | None, non_working_dates: set[date] | None, count_as_worktime_dates: set[date] | None, overtime_adjustment_minutes_by_date: dict[date, int] | None, overtime_start_date: date | None, relevant_weekdays: set[int] | None = None, ) -> dict[str, int]: if range_end < range_start: return { "ist_minutes": 0, "soll_minutes": 0, "delta_minutes": 0, "eligible_workdays": 0, "vacation_workdays": 0, } blocked_before = overtime_start_date vacation_dates = vacation_dates or set() non_working_dates = non_working_dates or set() count_as_worktime_dates = count_as_worktime_dates or set() overtime_adjustment_minutes_by_date = overtime_adjustment_minutes_by_date or {} relevant_weekdays = relevant_weekdays or set(DEFAULT_WORKING_DAYS) workdays_per_week = max(1, len(relevant_weekdays)) net_by_date: dict[date, int] = {} for entry in entries: if entry.date < range_start or entry.date > range_end: continue net_by_date[entry.date] = compute_net_minutes( entry.start_minutes, entry.end_minutes, entry.break_minutes, ) eligible_workdays = 0 vacation_workdays = 0 ist_minutes = 0 overtime_adjustment_minutes = 0 current = range_start while current <= range_end: overtime_adjustment_minutes += int(overtime_adjustment_minutes_by_date.get(current, 0)) if blocked_before is None or current >= blocked_before: day_counts_as_worktime = current in count_as_worktime_dates and is_workday(current, relevant_weekdays) day_target_minutes = int(round(weekly_target_minutes / workdays_per_week)) if is_workday(current, relevant_weekdays) else 0 if day_counts_as_worktime: ist_minutes += day_target_minutes elif current not in non_working_dates: ist_minutes += net_by_date.get(current, 0) if is_workday(current, relevant_weekdays): if current in vacation_dates and not day_counts_as_worktime: vacation_workdays += 1 elif current in non_working_dates and not day_counts_as_worktime: pass else: eligible_workdays += 1 current += timedelta(days=1) soll_minutes = int(round((weekly_target_minutes / workdays_per_week) * eligible_workdays)) delta_minutes = ist_minutes - soll_minutes + overtime_adjustment_minutes return { "ist_minutes": ist_minutes, "soll_minutes": soll_minutes, "delta_minutes": delta_minutes, "eligible_workdays": eligible_workdays, "vacation_workdays": vacation_workdays, "overtime_adjustment_minutes": overtime_adjustment_minutes, } def compute_effective_week_totals( *, entries: list, week_start: date, weekly_target_minutes: int, vacation_dates: set[date] | None, non_working_dates: set[date] | None, count_as_worktime_dates: set[date] | None, overtime_adjustment_minutes_by_date: dict[date, int] | None, overtime_start_date: date | None, relevant_weekdays: set[int] | None = None, ) -> dict[str, int]: week_end = week_start + timedelta(days=6) totals = compute_effective_span_totals( entries=entries, range_start=week_start, range_end=week_end, weekly_target_minutes=weekly_target_minutes, vacation_dates=vacation_dates, non_working_dates=non_working_dates, count_as_worktime_dates=count_as_worktime_dates, overtime_adjustment_minutes_by_date=overtime_adjustment_minutes_by_date, overtime_start_date=overtime_start_date, relevant_weekdays=relevant_weekdays, ) return { "weekly_ist": totals["ist_minutes"], "weekly_soll": totals["soll_minutes"], "weekly_delta": totals["delta_minutes"], } def compute_cumulative_overtime_minutes( *, entries: list, rules: list, weekly_target_fallback: int, vacation_periods: list, non_working_dates: set[date] | None, count_as_worktime_dates: set[date] | None, overtime_adjustment_minutes_by_date: dict[date, int] | None, selected_week_start: date, overtime_start_date: date | None, overtime_expiry_days: int | None, expire_negative_overtime: bool, relevant_weekdays: set[int] | None = None, ) -> int: selected_week_end = selected_week_start + timedelta(days=6) return compute_cumulative_overtime_until_date( entries=entries, rules=rules, weekly_target_fallback=weekly_target_fallback, vacation_periods=vacation_periods, non_working_dates=non_working_dates, count_as_worktime_dates=count_as_worktime_dates, overtime_adjustment_minutes_by_date=overtime_adjustment_minutes_by_date, as_of_date=selected_week_end, overtime_start_date=overtime_start_date, overtime_expiry_days=overtime_expiry_days, expire_negative_overtime=expire_negative_overtime, relevant_weekdays=relevant_weekdays, ) def compute_cumulative_overtime_until_date( *, entries: list, rules: list, weekly_target_fallback: int, vacation_periods: list, non_working_dates: set[date] | None, count_as_worktime_dates: set[date] | None, overtime_adjustment_minutes_by_date: dict[date, int] | None, as_of_date: date, overtime_start_date: date | None, overtime_expiry_days: int | None, expire_negative_overtime: bool, relevant_weekdays: set[int] | None = None, ) -> int: relevant_weekdays = relevant_weekdays or set(DEFAULT_WORKING_DAYS) workdays_per_week = max(1, len(relevant_weekdays)) overtime_adjustment_minutes_by_date = overtime_adjustment_minutes_by_date or {} earliest_entry_date = min((entry.date for entry in entries), default=None) earliest_adjustment_date = min(overtime_adjustment_minutes_by_date.keys(), default=None) range_start_candidates = [candidate for candidate in [earliest_entry_date, earliest_adjustment_date] if candidate is not None] if not range_start_candidates: return 0 range_start = min(range_start_candidates) if range_start > as_of_date: return 0 first_week_start = monday_of(range_start) relevant_weeks = week_starts_between(first_week_start, monday_of(as_of_date)) base_target_map = target_map_for_weeks(rules, relevant_weeks, weekly_target_fallback) vacation_dates = expand_vacation_dates( vacation_periods, range_start, as_of_date, relevant_weekdays=relevant_weekdays, ) non_working_dates = non_working_dates or set() count_as_worktime_dates = count_as_worktime_dates or set() net_by_date: dict[date, int] = {} for entry in entries: if entry.date < range_start or entry.date > as_of_date: continue net_by_date[entry.date] = compute_net_minutes( entry.start_minutes, entry.end_minutes, entry.break_minutes, ) cutoff_date: date | None = None if overtime_expiry_days is not None and overtime_expiry_days > 0: cutoff_date = as_of_date - timedelta(days=overtime_expiry_days) total = 0.0 current = range_start while current <= as_of_date: week_start = monday_of(current) weekly_target = base_target_map.get(week_start, weekly_target_fallback) day_adjustment = float(overtime_adjustment_minutes_by_date.get(current, 0)) regular_delta_allowed = overtime_start_date is None or current >= overtime_start_date day_counts_as_worktime = current in count_as_worktime_dates and current.weekday() in relevant_weekdays if regular_delta_allowed and current.weekday() in relevant_weekdays and (current not in vacation_dates or day_counts_as_worktime): if current in non_working_dates and not day_counts_as_worktime: day_target = 0.0 else: day_target = weekly_target / workdays_per_week else: day_target = 0.0 if regular_delta_allowed: if day_counts_as_worktime: day_net = day_target else: day_net = 0.0 if current in non_working_dates else float(net_by_date.get(current, 0)) else: day_net = 0.0 delta = day_net - day_target + day_adjustment expired = cutoff_date is not None and current < cutoff_date if expired: if delta > 0: current += timedelta(days=1) continue if delta < 0 and expire_negative_overtime: current += timedelta(days=1) continue total += delta current += timedelta(days=1) return int(round(total))