From 4403a4bb219027454ffb2f599d3ec7d14a39e709 Mon Sep 17 00:00:00 2001 From: MpakobecEx Date: Mon, 26 Oct 2026 02:58:31 +0500 Subject: [PATCH] =?UTF-8?q?=D0=9F=D0=B5=D1=80=D0=B5=D0=B7=D0=B0=D0=BF?= =?UTF-8?q?=D1=83=D1=81=D0=BA=20=D0=B1=D0=BE=D1=82=D0=B0=20=D0=B1=D0=BE?= =?UTF-8?q?=D0=BB=D1=8C=D1=88=D0=B5=20=D0=BD=D0=B5=20=D0=BF=D0=BE=D0=B2?= =?UTF-8?q?=D1=82=D0=BE=D1=80=D1=8F=D0=B5=D1=82=20=D1=80=D0=B0=D1=81=D1=81?= =?UTF-8?q?=D1=8B=D0=BB=D0=BA=D1=83?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Расписание common.check в 00:35 совпало с выкаткой: старый процесс успел отправить цепочку, новый после старта отправил её ещё раз. Теперь каждая отработанная минута атомарно отмечается в bot.schedule_runs; если отметка уже есть, рассылка пропускается. Старые отметки чистятся через 7 дней. --- README.md | 6 ++- bot/services/notifications.py | 12 +++++- common/repositories/schedules_repo.py | 21 ++++++++++ common/version.py | 2 +- sql/V25__schedule_runs.sql | 11 ++++++ tests/test_notifications_schedule.py | 56 +++++++++++++++++++++++++++ tests/test_version.py | 2 +- 7 files changed, 105 insertions(+), 5 deletions(-) create mode 100644 sql/V25__schedule_runs.sql diff --git a/README.md b/README.md index 1ead8b9..abc75f8 100644 --- a/README.md +++ b/README.md @@ -73,7 +73,9 @@ HttpOnly-cookie (`ADMIN_SESSION_SECRET`), все формы защищены CSR Раздел «Расписания» — CRUD рассылок (`bot.schedules` + история): время, дни недели, таймзона, шаблон и вкл/выкл. Бот раз в минуту читает активные расписания из БД; если -база недоступна — использует расписание из кода (совпадает с сидом `V16`). +база недоступна — использует расписание из кода (совпадает с сидом `V16`). Каждая +отправленная минута фиксируется в `bot.schedule_runs`, поэтому перезапуск бота в ту же +минуту не повторяет рассылку. Раздел «Пользователи» — список с поиском по нику/username/имени/Telegram-ID и карточка: уровень, опыт и серия, баланс (`SUM(money.amount)`), яички, сообщения, очищения, @@ -173,7 +175,7 @@ make backup E=test # на домашнем сервере: ./scripts/clickho ## Версионирование -- Единый источник версии — `common/version.py` (`__version__`), текущая — `0.6.8`. +- Единый источник версии — `common/version.py` (`__version__`), текущая — `0.6.9`. - Схема — SemVer-подобная (`MAJOR.MINOR.PATCH`); git-тег = `v` + версия (`v0.1`). - Docker-образы тегируются версией (Блок 2), события ClickHouse получают `bot_version` (Блок 3). - Выпуск: изменить `common/version.py` → коммит → аннотированный тег `vX.Y` → `git push --tags`. diff --git a/bot/services/notifications.py b/bot/services/notifications.py index ab03509..4f13e68 100644 --- a/bot/services/notifications.py +++ b/bot/services/notifications.py @@ -12,7 +12,7 @@ import asyncio import logging import time from dataclasses import dataclass -from datetime import datetime +from datetime import UTC, datetime from zoneinfo import ZoneInfo from aiogram import Bot @@ -45,6 +45,7 @@ class ScheduledMessage: template_key: str weekdays: tuple[int, ...] = () timezone: str = "Europe/Moscow" + schedule_id: int | None = None def matches(self, now: datetime) -> bool: """Совпадает ли момент ``now`` (aware) с этим расписанием.""" @@ -76,6 +77,7 @@ def _from_row(row: dict) -> ScheduledMessage: template_key=str(row["template_key"]), weekdays=tuple(int(day) for day in (row["weekdays"] or ())), timezone=str(row["timezone"] or "Europe/Moscow"), + schedule_id=int(row["id"]) if row.get("id") is not None else None, ) @@ -114,6 +116,14 @@ async def _tick(bot: Bot, now: datetime) -> None: for schedule in await _load_schedules(): if not schedule.matches(now): continue + + # Защита от повторной отправки после перезапуска в ту же минуту. + if schedule.schedule_id is not None: + run_at = now.astimezone(UTC).replace(second=0, microsecond=0) + if not await schedules_repo.claim_run(schedule.schedule_id, run_at): + logger.info("Рассылка %s за %s уже отправлена — пропускаю", schedule.schedule_id, run_at) + continue + subscribers = await users_service.list_subscribers() await _send_broadcast(bot, schedule.template_key, subscribers) diff --git a/common/repositories/schedules_repo.py b/common/repositories/schedules_repo.py index 648c754..5295403 100644 --- a/common/repositories/schedules_repo.py +++ b/common/repositories/schedules_repo.py @@ -50,6 +50,27 @@ async def list_active() -> list[dict[str, Any]]: return await cur.fetchall() +async def claim_run(schedule_id: int, run_at: Any) -> bool: + """Атомарно отметить минуту как отработанную. + + ``True`` — эту минуту ещё никто не отправлял; ``False`` — уже была отправка + (например, до перезапуска бота). + """ + async with get_pool().connection() as conn: + cur = await conn.execute( + """ + INSERT INTO bot.schedule_runs (schedule_id, run_at) + VALUES (%s, %s) + ON CONFLICT DO NOTHING + """, + (schedule_id, run_at), + ) + claimed = bool(cur.rowcount) + if claimed: + await conn.execute("DELETE FROM bot.schedule_runs WHERE run_at < now() - interval '7 days'") + return claimed + + async def get_schedule(schedule_id: int) -> dict[str, Any] | None: async with get_pool().connection() as conn: cur = await conn.execute( diff --git a/common/version.py b/common/version.py index 3a5bc8a..361bd45 100644 --- a/common/version.py +++ b/common/version.py @@ -7,4 +7,4 @@ from __future__ import annotations -__version__ = "0.6.8" +__version__ = "0.6.9" diff --git a/sql/V25__schedule_runs.sql b/sql/V25__schedule_runs.sql new file mode 100644 index 0000000..539b1b4 --- /dev/null +++ b/sql/V25__schedule_runs.sql @@ -0,0 +1,11 @@ +-- Отметки отправленных минут: защита от повторной рассылки после перезапуска бота. +CREATE TABLE IF NOT EXISTS bot.schedule_runs ( + schedule_id bigint NOT NULL, + run_at timestamptz NOT NULL, + PRIMARY KEY (schedule_id, run_at) +); + +CREATE INDEX IF NOT EXISTS idx_schedule_runs_run_at + ON bot.schedule_runs (run_at); + +COMMENT ON TABLE bot.schedule_runs IS 'Какие минуты рассылок уже отработаны (атомарная защита от дублей)'; diff --git a/tests/test_notifications_schedule.py b/tests/test_notifications_schedule.py index a0e2da7..89e5121 100644 --- a/tests/test_notifications_schedule.py +++ b/tests/test_notifications_schedule.py @@ -6,6 +6,7 @@ import re from datetime import UTC, datetime from pathlib import Path +from bot.services import notifications from bot.services.notifications import DEFAULT_SCHEDULES, ScheduledMessage, _from_row SEED_FILE = Path(__file__).resolve().parents[1] / "sql" / "V16__seed_schedules.sql" @@ -66,3 +67,58 @@ def test_default_schedules_match_seed() -> None: ) assert seeded == list(DEFAULT_SCHEDULES) + + +async def test_tick_skips_already_sent_minute(monkeypatch) -> None: + schedule = ScheduledMessage(12, 0, "notify.blago", schedule_id=1) + sent: list[str] = [] + + async def load() -> tuple[ScheduledMessage, ...]: + return (schedule,) + + async def claim(schedule_id: int, run_at: datetime) -> bool: + return False + + async def subscribers() -> list[dict]: + return [] + + async def broadcast(bot, key: str, rows: list[dict]) -> None: + sent.append(key) + + monkeypatch.setattr(notifications, "_load_schedules", load) + monkeypatch.setattr(notifications.schedules_repo, "claim_run", claim) + monkeypatch.setattr(notifications.users_service, "list_subscribers", subscribers) + monkeypatch.setattr(notifications, "_send_broadcast", broadcast) + + await notifications._tick(None, datetime(2026, 1, 1, 9, 0, tzinfo=UTC)) + + assert sent == [] + + +async def test_tick_sends_once_when_claimed(monkeypatch) -> None: + schedule = ScheduledMessage(12, 0, "notify.blago", schedule_id=1) + sent: list[str] = [] + claims: list[tuple[int, datetime]] = [] + + async def load() -> tuple[ScheduledMessage, ...]: + return (schedule,) + + async def claim(schedule_id: int, run_at: datetime) -> bool: + claims.append((schedule_id, run_at)) + return True + + async def subscribers() -> list[dict]: + return [] + + async def broadcast(bot, key: str, rows: list[dict]) -> None: + sent.append(key) + + monkeypatch.setattr(notifications, "_load_schedules", load) + monkeypatch.setattr(notifications.schedules_repo, "claim_run", claim) + monkeypatch.setattr(notifications.users_service, "list_subscribers", subscribers) + monkeypatch.setattr(notifications, "_send_broadcast", broadcast) + + await notifications._tick(None, datetime(2026, 1, 1, 9, 0, tzinfo=UTC)) + + assert sent == ["notify.blago"] + assert claims == [(1, datetime(2026, 1, 1, 9, 0, tzinfo=UTC))] diff --git a/tests/test_version.py b/tests/test_version.py index dedad10..e503e9f 100644 --- a/tests/test_version.py +++ b/tests/test_version.py @@ -7,4 +7,4 @@ from common import __version__ def test_version_is_set(): assert __version__ - assert __version__ == "0.6.8" + assert __version__ == "0.6.9"