"""Производственный календарь (D30): загрузка года из файла xmlcalendar.ru и предыдущий рабочий день для даты отчёта.""" from __future__ import annotations import logging from dataclasses import dataclass from datetime import date, datetime from app.db.base import SessionLocal from app.imports.xmlcalendar import ParsedCalendar, weekend_workdays from app.repositories.calendar import CalendarRepository from app.services.dates import BusinessCalendar, previous_business_day logger = logging.getLogger(__name__) @dataclass(frozen=True) class CalendarYearRow: year: int non_working: int weekday_holidays: int working_weekends: int updated_by: str updated_at: datetime async def load(first_year: int, last_year: int) -> BusinessCalendar: async with SessionLocal() as session: rows = await CalendarRepository(session).list_between( date(first_year, 1, 1), date(last_year, 12, 31) ) days = frozenset(day for day, _, _ in rows) return BusinessCalendar(frozenset(day.year for day in days), days) async def report_date(today: date | None = None) -> date: """Предыдущий рабочий день по календарю. Берём и прошлый год: в начале января предыдущий рабочий день — в декабре.""" today = today or date.today() calendar = await load(today.year - 1, today.year) if today.year not in calendar.years: logger.warning( "Производственный календарь на %d не загружен, рабочими считаются будни", today.year, ) return previous_business_day(today, calendar) async def save_year(parsed: ParsedCalendar, user: str) -> None: """Год заменяется целиком: повторная загрузка исправляет календарь.""" async with SessionLocal() as session: await CalendarRepository(session).replace_year( parsed.year, parsed.non_working, user ) await session.commit() async def list_years() -> list[CalendarYearRow]: """Загруженные годы, новые сверху.""" async with SessionLocal() as session: rows = await CalendarRepository(session).list_between(date.min, date.max) by_year: dict[int, list[tuple[date, str, datetime]]] = {} for row in rows: by_year.setdefault(row[0].year, []).append(row) result = [] for year, items in sorted(by_year.items(), reverse=True): days = {day for day, _, _ in items} _, updated_by, updated_at = max(items, key=lambda item: item[2]) result.append( CalendarYearRow( year=year, non_working=len(days), weekday_holidays=sum(1 for day in days if day.weekday() < 5), working_weekends=len(weekend_workdays(year, days)), updated_by=updated_by, updated_at=updated_at, ) ) return result