Kenik
80 lines
3.7 KiB
Python
80 lines
3.7 KiB
Python
"""Расчёт дат для напоминалок админов: парсинг ввода и следующее срабатывание.
|
|
|
|
Не зависит от 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))
|