166 lines
7.9 KiB
Python
166 lines
7.9 KiB
Python
"""Recurrence-ядро для закреплённых конференций (ADR-001, п.3).
|
||
|
||
Собственная (не RRULE) модель повторения: ровно 4 типа, отображающихся на
|
||
форму UI 1:1. Правило хранит локальное время суток и IANA-таймзону —
|
||
чтобы корректно учитывать смещение UTC (переход на летнее/зимнее время в тех
|
||
зонах, где он есть); развёртка `expand_occurrences` всегда возвращает
|
||
timezone-aware метки в UTC (в БД — только UTC).
|
||
"""
|
||
|
||
import calendar
|
||
from datetime import UTC, date, datetime, time, timedelta
|
||
from typing import Literal, Self
|
||
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
|
||
|
||
from pydantic import BaseModel, Field, field_validator, model_validator
|
||
|
||
RecurrenceType = Literal["weekly", "biweekly", "monthly", "every_n_days"]
|
||
|
||
|
||
class RecurrenceRule(BaseModel):
|
||
"""Правило повторения закреплённой конференции.
|
||
|
||
Поля, специфичные для типа (`weekdays` для weekly/biweekly,
|
||
`day_of_month` для monthly, `interval_days` для every_n_days),
|
||
валидируются в `_validate_type_specific_fields` — обязательность
|
||
зависит от `type`.
|
||
"""
|
||
|
||
type: RecurrenceType
|
||
weekdays: list[int] = Field(default_factory=list)
|
||
day_of_month: int | None = None
|
||
interval_days: int | None = None
|
||
anchor_date: date
|
||
time_local: str
|
||
timezone: str
|
||
duration_minutes: int
|
||
|
||
@field_validator("weekdays")
|
||
@classmethod
|
||
def _validate_weekdays(cls, value: list[int]) -> list[int]:
|
||
"""Дни недели — 0 (понедельник) .. 6 (воскресенье), как в `date.weekday()`."""
|
||
for weekday in value:
|
||
if not 0 <= weekday <= 6:
|
||
raise ValueError("weekday должен быть в диапазоне 0..6")
|
||
return value
|
||
|
||
@field_validator("day_of_month")
|
||
@classmethod
|
||
def _validate_day_of_month(cls, value: int | None) -> int | None:
|
||
if value is not None and not 1 <= value <= 31:
|
||
raise ValueError("day_of_month должен быть в диапазоне 1..31")
|
||
return value
|
||
|
||
@field_validator("interval_days")
|
||
@classmethod
|
||
def _validate_interval_days(cls, value: int | None) -> int | None:
|
||
if value is not None and value < 1:
|
||
raise ValueError("interval_days должен быть >= 1")
|
||
return value
|
||
|
||
@field_validator("time_local")
|
||
@classmethod
|
||
def _validate_time_local(cls, value: str) -> str:
|
||
try:
|
||
datetime.strptime(value, "%H:%M")
|
||
except ValueError as exc:
|
||
raise ValueError("time_local должен быть в формате HH:MM") from exc
|
||
return value
|
||
|
||
@field_validator("timezone")
|
||
@classmethod
|
||
def _validate_timezone(cls, value: str) -> str:
|
||
try:
|
||
ZoneInfo(value)
|
||
except (ZoneInfoNotFoundError, ValueError) as exc:
|
||
raise ValueError(f"неизвестная IANA-таймзона: {value}") from exc
|
||
return value
|
||
|
||
@model_validator(mode="after")
|
||
def _validate_type_specific_fields(self) -> Self:
|
||
"""Проверить, что поля, обязательные для конкретного `type`, заполнены."""
|
||
if self.type in ("weekly", "biweekly"):
|
||
if not self.weekdays:
|
||
raise ValueError(f"{self.type} требует непустой список weekdays")
|
||
elif self.type == "monthly":
|
||
if self.day_of_month is None:
|
||
raise ValueError("monthly требует day_of_month")
|
||
elif self.type == "every_n_days":
|
||
if self.interval_days is None:
|
||
raise ValueError("every_n_days требует interval_days")
|
||
return self
|
||
|
||
def time_of_day(self) -> time:
|
||
"""Разобрать `time_local` в объект `time`."""
|
||
hours, minutes = self.time_local.split(":")
|
||
return time(int(hours), int(minutes))
|
||
|
||
def local_datetime(self, on_date: date) -> datetime:
|
||
"""Собрать локальный aware datetime для конкретной календарной даты."""
|
||
return datetime.combine(on_date, self.time_of_day(), tzinfo=ZoneInfo(self.timezone))
|
||
|
||
|
||
def _clamped_day_of_month(year: int, month: int, day_of_month: int) -> int:
|
||
"""Клэмпнуть `day_of_month` к последнему дню месяца, если тот короче (например, 31 в апреле)."""
|
||
last_day = calendar.monthrange(year, month)[1]
|
||
return min(day_of_month, last_day)
|
||
|
||
|
||
def _week_start(on_date: date) -> date:
|
||
"""Начало недели (понедельник), содержащей `on_date`."""
|
||
return on_date - timedelta(days=on_date.weekday())
|
||
|
||
|
||
def _matches(rule: RecurrenceRule, on_date: date) -> bool:
|
||
"""Проверить, попадает ли календарная дата `on_date` в правило повторения."""
|
||
if rule.type == "weekly":
|
||
return on_date.weekday() in rule.weekdays
|
||
if rule.type == "biweekly":
|
||
if on_date.weekday() not in rule.weekdays:
|
||
return False
|
||
weeks_diff = (_week_start(on_date) - _week_start(rule.anchor_date)).days // 7
|
||
return weeks_diff % 2 == 0
|
||
if rule.type == "monthly":
|
||
assert rule.day_of_month is not None # гарантировано валидацией модели
|
||
return on_date.day == _clamped_day_of_month(on_date.year, on_date.month, rule.day_of_month)
|
||
if rule.type == "every_n_days":
|
||
assert rule.interval_days is not None # гарантировано валидацией модели
|
||
# anchor_date — первое вхождение серии: даты раньше него не считаются
|
||
# (иначе `%` на отрицательной разнице даст даты «до начала» серии).
|
||
if on_date < rule.anchor_date:
|
||
return False
|
||
return (on_date - rule.anchor_date).days % rule.interval_days == 0
|
||
raise AssertionError(f"неизвестный тип повторения: {rule.type}")
|
||
|
||
|
||
def expand_occurrences(rule: RecurrenceRule, t_from: datetime, t_to: datetime) -> list[datetime]:
|
||
"""Развернуть правило повторения в список моментов начала вхождений (UTC, aware).
|
||
|
||
Диапазон `[t_from, t_to]` включителен с обеих сторон. `t_from`/`t_to`
|
||
обязаны быть timezone-aware. При `t_from > t_to` возвращается пустой
|
||
список. Перебор идёт по календарным датам в локальной таймзоне правила
|
||
с суточным запасом с каждой стороны — компенсирует случаи, когда
|
||
смещение локальной зоны отличается от зоны границ диапазона настолько,
|
||
что вхождение попадает в диапазон, а его календарная дата — нет.
|
||
"""
|
||
if t_from.tzinfo is None or t_to.tzinfo is None:
|
||
raise ValueError("t_from и t_to должны быть timezone-aware")
|
||
if t_from > t_to:
|
||
return []
|
||
|
||
tz = ZoneInfo(rule.timezone)
|
||
start_date = (t_from.astimezone(tz) - timedelta(days=1)).date()
|
||
end_date = (t_to.astimezone(tz) + timedelta(days=1)).date()
|
||
|
||
occurrences: list[datetime] = []
|
||
current = start_date
|
||
while current <= end_date:
|
||
if _matches(rule, current):
|
||
candidate = rule.local_datetime(current).astimezone(UTC)
|
||
if t_from <= candidate <= t_to:
|
||
occurrences.append(candidate)
|
||
current += timedelta(days=1)
|
||
|
||
occurrences.sort()
|
||
return occurrences
|