Напоминалки: чистая логика расчёта дат (reminders.py)
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,65 @@
|
||||
"""Расчёт дат для напоминалок админов: парсинг ввода и следующее срабатывание.
|
||||
|
||||
Не зависит от 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
|
||||
@@ -0,0 +1,48 @@
|
||||
from datetime import datetime
|
||||
from reminders import parse_reminder_datetime, compute_next_occurrence
|
||||
|
||||
|
||||
def test_parse_reminder_datetime_valid():
|
||||
dt = parse_reminder_datetime("15.03.2027 09:00")
|
||||
assert dt == datetime(2027, 3, 15, 9, 0)
|
||||
|
||||
|
||||
def test_parse_reminder_datetime_invalid():
|
||||
assert parse_reminder_datetime("не дата") is None
|
||||
assert parse_reminder_datetime("2027-03-15 09:00") is None
|
||||
assert parse_reminder_datetime("32.13.2027 09:00") is None
|
||||
|
||||
|
||||
def test_compute_next_weekly():
|
||||
anchor = datetime(2026, 9, 2, 9, 0)
|
||||
nxt = compute_next_occurrence(anchor, "weekly", after=anchor)
|
||||
assert nxt == datetime(2026, 9, 9, 9, 0)
|
||||
|
||||
|
||||
def test_compute_next_monthly_end_of_month_clamp_no_drift():
|
||||
"""31.01 -> 28.02 (2026 не високосный), а дальше 31.03 — не 28.03 (без дрейфа)."""
|
||||
anchor = datetime(2026, 1, 31, 9, 0)
|
||||
first = compute_next_occurrence(anchor, "monthly", after=anchor)
|
||||
assert first == datetime(2026, 2, 28, 9, 0)
|
||||
second = compute_next_occurrence(anchor, "monthly", after=first)
|
||||
assert second == datetime(2026, 3, 31, 9, 0)
|
||||
|
||||
|
||||
def test_compute_next_yearly_leap_day():
|
||||
anchor = datetime(2024, 2, 29, 9, 0)
|
||||
nxt = compute_next_occurrence(anchor, "yearly", after=anchor)
|
||||
assert nxt == datetime(2025, 2, 28, 9, 0)
|
||||
|
||||
|
||||
def test_compute_next_skips_missed_periods_during_downtime():
|
||||
"""Бот стоял два месяца — пропущенные периоды пропускаются, берём ближайшее будущее."""
|
||||
anchor = datetime(2026, 1, 1, 9, 0)
|
||||
after = datetime(2026, 3, 15, 0, 0)
|
||||
nxt = compute_next_occurrence(anchor, "monthly", after=after)
|
||||
assert nxt == datetime(2026, 4, 1, 9, 0)
|
||||
|
||||
|
||||
def test_compute_next_unknown_recurrence_raises():
|
||||
import pytest
|
||||
with pytest.raises(ValueError):
|
||||
compute_next_occurrence(datetime(2026, 1, 1, 9, 0), "daily", after=datetime(2026, 1, 1, 9, 0))
|
||||
Reference in New Issue
Block a user