"""Расчёт дат для напоминалок админов: парсинг ввода и следующее срабатывание. Не зависит от bot.py/database.py — чистая логика дат, легко тестируется отдельно. """ import calendar from datetime import datetime, timedelta from typing import Optional DT_FORMAT = "%Y-%m-%d %H:%M:%S" # формат хранения в БД INPUT_FORMAT = "%d.%m.%Y %H:%M" # формат ввода админом RECURRENCES = ("once", "weekly", "monthly", "yearly") def parse_reminder_datetime(text: str) -> Optional[datetime]: """Парсит дату/время в формате ДД.ММ.ГГГГ ЧЧ:ММ. None, если формат неверный.""" text = text.strip() try: return datetime.strptime(text, INPUT_FORMAT) except ValueError: return None def _clamp_day(year: int, month: int, day: int) -> int: """День месяца, не превышающий число дней в этом месяце (31.01 -> 28/29.02, 31.03 и т.п. — без ползучего дрейфа, см. compute_next_occurrence).""" last_day = calendar.monthrange(year, month)[1] return min(day, last_day) def compute_next_occurrence(anchor: datetime, recurrence: str, after: datetime) -> datetime: """Следующее срабатывание строго после `after`, отталкиваясь от якорной даты `anchor` (а не от предыдущего срабатывания) — так периодичность не «дрейфует» из-за клампинга коротких месяцев. Если пропущено несколько периодов (бот был выключен) — промежуточные пропускаются, возвращается ближайшее будущее относительно `after`. """ if recurrence not in RECURRENCES: raise ValueError(f"Неизвестная периодичность: {recurrence}") if recurrence == "weekly": candidate = anchor while candidate <= after: candidate += timedelta(days=7) return candidate if recurrence == "monthly": year, month = anchor.year, anchor.month candidate = anchor while candidate <= after: month += 1 if month > 12: month = 1 year += 1 day = _clamp_day(year, month, anchor.day) candidate = candidate.replace(year=year, month=month, day=day) return candidate # yearly year = anchor.year candidate = anchor while candidate <= after: year += 1 day = _clamp_day(year, anchor.month, anchor.day) candidate = candidate.replace(year=year, day=day) return candidate def server_utc_offset_hours() -> float: """Собственное смещение сервера от UTC в часах. Часовой пояс сервера нигде явно не настроен (может быть любым в зависимости от хостинга), поэтому вычисляется на лету через локальную таймзону ОС.""" offset = datetime.now().astimezone().utcoffset() return offset.total_seconds() / 3600 if offset else 0.0 def convert_time(dt: datetime, from_offset: float, to_offset: float) -> datetime: """Переводит наивный datetime из одного часового пояса (смещение от UTC в часах) в другой — просто сдвигает время на разницу смещений.""" return dt + timedelta(hours=(to_offset - from_offset))