"""Генерация .ics-приглашений на конференцию (VEVENT, `METHOD:REQUEST`). Разовая (плановая) конференция — `DTSTART`/`DTEND` в переданной таймзоне отображения (`display_timezone` настройки инстанса); закреплённая с повторением — `DTSTART` берётся из первого вхождения `expand_occurrences` (см. `services/recurrence.py`) в таймзоне самого правила повторения (`rule.timezone`), а рецидив описывается `RRULE`. `UID` стабилен (`{conference.id}@vidconf`) — календарные клиенты обновляют уже принятое приглашение по нему же, ориентируясь на растущий `SEQUENCE` (`conference.ics_sequence`, инкрементируется при правке расписания — `services/conferences.py::ConferenceService.update`). """ from datetime import UTC, datetime, timedelta from typing import cast from zoneinfo import ZoneInfo from icalendar import Calendar, Event from models.conference import Conference from services.recurrence import RecurrenceRule, expand_occurrences # Порядок ровно как в `date.weekday()`/`RecurrenceRule.weekdays` (0 — понедельник). _WEEKDAY_CODES = ("MO", "TU", "WE", "TH", "FR", "SA", "SU") # Длительность разового вхождения по умолчанию, если у конференции не указан # `duration_minutes` (совпадает с `services.conferences.DEFAULT_OCCURRENCE_DURATION_MINUTES`, # не импортируется напрямую — `ics.py` намеренно не зависит от `conferences.py`). _DEFAULT_ONE_OFF_DURATION_MINUTES = 60 # Горизонт поиска первого вхождения повторяющейся серии от `anchor_date` # (совпадает с `services.conferences.NEXT_OCCURRENCE_HORIZON`) — с запасом # покрывает самый частый шаг повторения (weekly/monthly). _OCCURRENCE_SEARCH_HORIZON = timedelta(days=400) _PRODID = "-//VidConf//vidconf.example//RU" class ConferenceHasNoScheduleError(ValueError): """У конференции нет ни `scheduled_at`, ни `recurrence` — приглашение строить не из чего.""" def build_invite( conference: Conference, *, organizer_email: str | None, join_url: str, display_timezone: str, ) -> bytes: """Собрать .ics-приглашение (`METHOD:REQUEST`) на конференцию и вернуть его байты. `display_timezone` — таймзона отображения разовых конференций (настройка инстанса); для повторяющихся используется таймзона самого правила (`RecurrenceRule.timezone`) — она обязательна в правиле и корректнее отражает намерение организатора серии. """ if conference.recurrence is not None: rule = RecurrenceRule.model_validate(conference.recurrence) tz = ZoneInfo(rule.timezone) dtstart = _first_occurrence_utc(rule).astimezone(tz) dtend = dtstart + timedelta(minutes=rule.duration_minutes) rrule_value: str | None = _build_rrule(rule) elif conference.scheduled_at is not None: tz = ZoneInfo(display_timezone) dtstart = conference.scheduled_at.astimezone(tz) duration = conference.duration_minutes or _DEFAULT_ONE_OFF_DURATION_MINUTES dtend = dtstart + timedelta(minutes=duration) rrule_value = None else: raise ConferenceHasNoScheduleError( f"у конференции {conference.id} нет ни scheduled_at, ни recurrence" ) # `Component.__init__` (общий предок `Event`/`Calendar`) не типизирован в # `icalendar` — `no-untyped-call` здесь неизбежен без переписывания # конструктора сторонней библиотеки. event = Event() # type: ignore[no-untyped-call] event.add("uid", f"{conference.id}@vidconf") event.add("sequence", conference.ics_sequence) event.add("dtstamp", datetime.now(UTC)) event.add("summary", conference.title or f"Конференция {conference.number}") event.add( "description", f"Ссылка для входа: {join_url}\nНомер конференции: {conference.number}", ) event.add("dtstart", dtstart) event.add("dtend", dtend) if organizer_email: event.add("organizer", f"mailto:{organizer_email}") if rrule_value is not None: event.add("rrule", rrule_value) calendar = Calendar() # type: ignore[no-untyped-call] calendar.add("prodid", _PRODID) calendar.add("version", "2.0") calendar.add("method", "REQUEST") calendar.add_component(event) calendar.add_missing_timezones() return cast(bytes, calendar.to_ical()) def _first_occurrence_utc(rule: RecurrenceRule) -> datetime: """Найти первое вхождение серии от `rule.anchor_date` (UTC, aware).""" horizon = _OCCURRENCE_SEARCH_HORIZON if rule.type == "every_n_days" and rule.interval_days is not None: horizon = max(horizon, timedelta(days=rule.interval_days + 2)) t_from = rule.local_datetime(rule.anchor_date).astimezone(UTC) occurrences = expand_occurrences(rule, t_from, t_from + horizon) if not occurrences: raise ConferenceHasNoScheduleError( "правило повторения не даёт ни одного вхождения в горизонте поиска" ) return occurrences[0] def _build_rrule(rule: RecurrenceRule) -> str: """Собрать значение `RRULE` из `RecurrenceRule` («.ics»).""" if rule.type == "weekly": return f"FREQ=WEEKLY;BYDAY={_byday(rule.weekdays)}" if rule.type == "biweekly": return f"FREQ=WEEKLY;INTERVAL=2;BYDAY={_byday(rule.weekdays)};WKST=MO" if rule.type == "monthly": assert rule.day_of_month is not None # гарантировано валидацией RecurrenceRule return f"FREQ=MONTHLY;BYMONTHDAY={_bymonthday(rule.day_of_month)}" if rule.type == "every_n_days": assert rule.interval_days is not None # гарантировано валидацией RecurrenceRule return f"FREQ=DAILY;INTERVAL={rule.interval_days}" raise AssertionError(f"неизвестный тип повторения: {rule.type}") def _byday(weekdays: list[int]) -> str: """Список дней недели в порядок `BYDAY` (`MO,TU,...`).""" return ",".join(_WEEKDAY_CODES[weekday] for weekday in sorted(weekdays)) def _bymonthday(day_of_month: int) -> int: """Смаппить `day_of_month` правила в `BYMONTHDAY` .ics. `31` не встречается ни в одном месяце короче — маппится в `-1` (последний день месяца), что СОВПАДАЕТ с клэмпом `expand_occurrences`. `29`/`30` НЕ клэмпаются здесь: известное задокументированное расхождение с поведением `expand_occurrences` — календарный клиент в короткие месяцы (например, февраль) вхождение пропустит, тогда как `expand_occurrences` использует клэмп к последнему дню месяца. """ return -1 if day_of_month == 31 else day_of_month