"""Правила напоминаний (ТЗ 3.21, подраздел «Напоминания»).
Здесь только правила — **чистые функции**: ни сети, ни часов, ни базы. «Сегодня»
приходит аргументом, поэтому правила проверяются тестами без ожидания и без
подмены системного времени, а планировщик (`services/scheduler.py`) остаётся
тонким: собрать кандидатов, отсеять отправленное, отдать в push.
Что напоминаем:
- **строгий дедлайн разовой задачи** — накануне, в день срока и один раз о
просрочке. Нестрогий период («в течение недели/месяца/года») не напоминает
вовсе: в модели у него нет дня отсчёта, и выдумывать якорь владелец не захотел.
Регулярных эта ветка не касается (0.91): у них своя дата и свой повод — ритм;
- **ритм регулярной задачи** — «пора: зарядка». День берётся тем же правилом, что
рождает следующий экземпляр (`services/recurrence.py`), но напоминание не ждёт
закрытия предыдущего — иначе о задаче вспоминаешь только тогда, когда её уже
сделал;
- **сводка на сегодня** — отдельная функция-окно, содержимое считает планировщик.
Молчим о `done` и `cancelled` (работа сделана или отменена) и о `deferred`:
«сейчас не в приоритете» — это осознанное решение владельца, и напоминание о сроке
противоречило бы ему.
"""
from collections.abc import Iterable
from dataclasses import dataclass, field
from datetime import date, datetime, time, timedelta
from app.models import Task
from app.services import recurrence
# Статусы, о которых напоминаем. `done`/`cancelled`/`deferred` — молчание
ACTIVE_STATUSES = ("to_do", "in_progress")
# Виды напоминаний: они же — ключи текстов в services/push_texts.py и первая
# половина ключа дедупа в push_deliveries
KIND_DUE_TOMORROW = "task.due_tomorrow"
KIND_DUE_TODAY = "task.due_today"
KIND_OVERDUE = "task.overdue"
KIND_RHYTHM = "task.rhythm"
KIND_SUMMARY = "summary"
# Окно сводки по локальному времени: утром её ещё читают, к вечеру она про
# прошедший день и смысла не имеет
SUMMARY_START_HOUR = 9
SUMMARY_END_HOUR = 14
@dataclass(frozen=True)
class Reminder:
"""Напоминание до отправки: вид, ключ дедупа и поля для подстановки в текст."""
kind: str
# Уникален в пределах (пользователь, вид): что уже отправлено, решает
# push_deliveries. Перенос срока или новый день ритма дают новый ключ
ref_key: str
task_id: int | None
fields: dict[str, str] = field(default_factory=dict)
def is_remindable(task: Task) -> bool:
"""Задача ещё живая: не сделана, не отменена, не отложена «на потом»."""
return task.status in ACTIVE_STATUSES
def _days_left(task: Task, today: date) -> int | None:
if task.deadline_date is None:
return None
return (task.deadline_date - today).days
def deadline_reminders(task: Task, today: date) -> list[Reminder]:
"""Напоминания о строгом дедлайне: за день, в день и один раз о просрочке.
У регулярной задачи дата — это дата следующего экземпляра (3.5), а не срок
работы, и напоминает о ней ритм. Без этой оговорки клон получал бы два
уведомления об одном и том же в один день: «Срок сегодня» и «Пора».
"""
if not is_remindable(task) or task.task_type == "recurring":
return []
deadline = task.deadline_date
if deadline is None:
return []
days = (deadline - today).days
if days == 1:
kind = KIND_DUE_TOMORROW
elif days == 0:
kind = KIND_DUE_TODAY
elif days < 0:
kind = KIND_OVERDUE
else:
return []
# Дата в ключе: срок перенесли — напоминание о новом сроке законное, а о
# прежнем уже отправлено и повторяться не должно. Просрочка тоже привязана к
# дате срока, поэтому звучит один раз, а не каждый день
ref_key = f"{task.id}:{deadline.isoformat()}"
return [Reminder(kind, ref_key, task.id, {"title": task.title})]
def rhythm_reminders(task: Task, today: date) -> list[Reminder]:
"""«Пора» для регулярной задачи: сегодня по её же правилу повторения.
День сверяем тем же `recurrence.next_date`, что создаёт следующий экземпляр:
отдельного «календаря напоминаний» не заводим, иначе правила разъедутся.
"""
if not is_remindable(task) or task.task_type != "recurring" or task.recur_kind is None:
return []
# next_date ищет строго после переданной даты, поэтому «сегодня по правилу»
# проверяется как «следующая дата после вчера равна сегодня»
if recurrence.next_date(task, today - timedelta(days=1)) != today:
return []
# Дата в ключе — как у дедлайна (id задачи + день): без id один пользователь
# получал за день ровно одно «Пора», сколько бы регулярных ни выпало на
# сегодня, — остальные молча съедала уникальность push_deliveries
return [Reminder(KIND_RHYTHM, f"{task.id}:{today.isoformat()}", task.id, {"title": task.title})]
def task_reminders(task: Task, today: date) -> list[Reminder]:
"""Все напоминания по одной задаче (может не быть ни одного)."""
return deadline_reminders(task, today) + rhythm_reminders(task, today)
# --- Тихие часы ---------------------------------------------------------------
def parse_quiet_hours(value: str | None) -> tuple[time, time] | None:
"""Разобрать «HH:MM-HH:MM»; пусто или битое значение → тихих часов нет.
Битое значение молча выключает тихие часы намеренно: это пользовательская
настройка, а не инвариант, и падать из-за опечатки в ней планировщик не должен.
"""
if not value:
return None
parts = value.split("-")
if len(parts) != 2:
return None
parsed: list[time] = []
for part in parts:
try:
hour, minute = part.strip().split(":")
parsed.append(time(int(hour), int(minute)))
except ValueError:
return None
return parsed[0], parsed[1]
def in_quiet_hours(value: str | None, now: datetime) -> bool:
"""Попадает ли момент в тихое окно (оно может пересекать полночь)."""
window = parse_quiet_hours(value)
if window is None:
return False
start, end = window
if start == end:
# Окно нулевой длины — не «весь день», а «выключено»
return False
moment = now.time()
if start < end:
return start <= moment < end
return moment >= start or moment < end
# --- Сводка -------------------------------------------------------------------
def summary_due(now: datetime) -> bool:
"""Сводка уместна в утренне-дневном окне (по локальному времени)."""
return SUMMARY_START_HOUR <= now.hour < SUMMARY_END_HOUR
def summary_counts(tasks: Iterable[Task], today: date) -> dict[str, int]:
"""Что сказать в сводке: просрочено, срок сегодня и ждёт приёмки.
Ждущие приёмки — закрытые агентом задачи (3.20): их статус `done`, но работа
не принята, и это единственное, что о них напоминает.
"""
counts = {"overdue": 0, "today": 0, "review": 0}
for task in tasks:
if task.accept_state == "pending":
counts["review"] += 1
# Срок регулярной — это дата следующего экземпляра (3.5): в сводке он
# повторял бы ритм, который об этой же задаче и так скажет
if not is_remindable(task) or task.task_type == "recurring":
continue
days = _days_left(task, today)
if days is None:
continue
if days < 0:
counts["overdue"] += 1
elif days == 0:
counts["today"] += 1
return counts