"""Планировщик рассылок. Расписания читаются из БД (таблица ``bot.schedules``, редактируется в админке). Если база недоступна, используется расписание по умолчанию из кода и последний успешно прочитанный набор. Тексты берутся из ``bot.message_templates`` через ``common.texts.store`` (TTL-кеш, правки без перезапуска). """ from __future__ import annotations import asyncio import logging import time from dataclasses import dataclass from datetime import datetime from zoneinfo import ZoneInfo from aiogram import Bot from bot.services import users as users_service from common.config import get_settings from common.repositories import schedules_repo from common.texts import format_usernames, render from common.texts.store import get_text logger = logging.getLogger(__name__) #: Размер порции упоминаний в одном сообщении. BATCH_SIZE = 10 #: Как часто писать в лог, если БД с расписаниями недоступна (секунды). WARNING_INTERVAL = 300.0 @dataclass(frozen=True, slots=True) class ScheduledMessage: """Одна запланированная рассылка из БД (или из значений по умолчанию). :param weekdays: дни недели по ISO (1=Пн … 7=Вс). Пусто — ежедневно. :param timezone: таймзона расписания (IANA), время в ней. """ hour: int minute: int template_key: str weekdays: tuple[int, ...] = () timezone: str = "Europe/Moscow" def matches(self, now: datetime) -> bool: """Совпадает ли момент ``now`` (aware) с этим расписанием.""" local = now.astimezone(ZoneInfo(self.timezone)) if self.hour != local.hour or self.minute != local.minute: return False return not self.weekdays or local.isoweekday() in self.weekdays #: Расписание по умолчанию — на случай недоступной БД (совпадает с V16-сидом). DEFAULT_SCHEDULES: tuple[ScheduledMessage, ...] = ( ScheduledMessage(16, 30, "notify.blago_30"), ScheduledMessage(16, 45, "notify.blago_45"), ScheduledMessage(16, 55, "notify.blago_55"), ScheduledMessage(17, 0, "notify.blago"), ScheduledMessage(15, 0, "notify.vklad", weekdays=(5,)), # пятница ScheduledMessage(14, 0, "notify.vklad", weekdays=(7,)), # воскресенье ) #: Последний успешно прочитанный набор — чтобы пережить временный сбой БД. _last_good: tuple[ScheduledMessage, ...] | None = None _last_warning_at = 0.0 def _from_row(row: dict) -> ScheduledMessage: return ScheduledMessage( hour=int(row["hour"]), minute=int(row["minute"]), template_key=str(row["template_key"]), weekdays=tuple(int(day) for day in (row["weekdays"] or ())), timezone=str(row["timezone"] or "Europe/Moscow"), ) async def _load_schedules() -> tuple[ScheduledMessage, ...]: """Прочитать активные расписания; при сбое — последние известные или дефолт.""" global _last_good, _last_warning_at try: rows = await schedules_repo.list_active() except Exception: now = time.monotonic() if now - _last_warning_at > WARNING_INTERVAL: logger.warning( "Не удалось прочитать расписания из БД — использую последние известные", exc_info=True, ) _last_warning_at = now return _last_good if _last_good is not None else DEFAULT_SCHEDULES _last_good = tuple(_from_row(row) for row in rows) return _last_good async def _send_broadcast(bot: Bot, template: str, subscribers: list[dict]) -> None: chunks = [subscribers[i : i + BATCH_SIZE] for i in range(0, len(subscribers), BATCH_SIZE)] or [[]] for chunk in chunks: text = render(template, {"usernames": format_usernames(chunk)}) await bot.send_message(chat_id=get_settings().chat_id, text=text) async def _tick(bot: Bot, now: datetime) -> None: for schedule in await _load_schedules(): if not schedule.matches(now): continue template = await get_text(schedule.template_key) if not template: logger.warning("Шаблон %s пуст или не найден — рассылка пропущена", schedule.template_key) continue subscribers = await users_service.list_subscribers() await _send_broadcast(bot, template, subscribers) async def run_notifications(bot: Bot) -> None: """Бесконечно проверять расписание раз в минуту.""" tz = get_settings().tzinfo while True: now = datetime.now(tz).replace(second=0, microsecond=0) try: await _tick(bot, now) except asyncio.CancelledError: raise except Exception: logger.exception("Ошибка планировщика рассылок") # Спать до начала следующей минуты, чтобы обрабатывать каждую минуту один раз. await asyncio.sleep(60 - datetime.now(tz).second)