Files

T

Kenik 6d4dd16caf Напоминалки: план реализации (docs)

2026-09-02 21:56:18 +03:00

44 KiB

Raw Permalink Blame History

Настраиваемые напоминалки для админов — план реализации

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Добавить в VK-бота раздел «⏰ Напоминалки»: админ создаёт напоминание (дата/время, текст, получатели, периодичность разово/неделя/месяц/год), бот сам присылает его в срок и, если периодическое, пересчитывает следующее срабатывание.

Architecture: Новая таблица SQLite reminders + CRUD в database.py. Чистая логика расчёта дат (парсинг ввода, следующее срабатывание с учётом коротких месяцев/високосного 29 февраля) — в новом файле reminders.py, без зависимостей от bot.py, что делает её тривиально юнит-тестируемой. Пошаговый мастер ввода (ADMIN_REMINDER_* состояния) и общий список для всех админов (add/delete, без edit) — в bot.py, по образцу уже существующего мастера «Быстрые ответы» (ADMIN_FAQ_*). Отправка в срок — отдельный фоновый поток (threading.Thread, как уже используется для рассылки/прогрева медиа), проверяющий БД раз в 60 секунд; при простое бота пропущенное напоминание досылается один раз при следующей проверке, дальше планируется в будущее.

Tech Stack: Python, SQLite (sqlite3 через существующий DatabaseManager), vk_api (VkKeyboard), pytest для тестов.

Важно: НЕ пушить в git до полного прогона тестов и ручной проверки в консольном эмуляторе бота (см. Task 7). Локальные коммиты после каждой задачи — можно и нужно.

Task 1: `reminders.py` — парсинг даты и расчёт следующего срабатывания

Files:

Create: reminders.py
Test: tests/test_reminders.py
Step 1: Написать падающие тесты

Создать tests/test_reminders.py:

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))

Step 2: Убедиться, что тесты падают (модуля ещё нет)

Run: python -m pytest tests/test_reminders.py -v Expected: FAIL с ModuleNotFoundError: No module named 'reminders'

Step 3: Реализовать reminders.py

Создать reminders.py:

"""Расчёт дат для напоминалок админов: парсинг ввода и следующее срабатывание.

Не зависит от 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

Step 4: Прогнать тесты, убедиться, что все проходят

Run: python -m pytest tests/test_reminders.py -v Expected: PASS (7 passed)

Step 5: Коммит

git add reminders.py tests/test_reminders.py
git commit -m "Напоминалки: чистая логика расчёта дат (reminders.py)"

Task 2: `database.py` — таблица `reminders` и CRUD

Files:

Modify: database.py
Test: tests/test_database.py
Step 1: Написать падающие тесты

Добавить в конец tests/test_database.py:

def test_reminders_crud(tmp_db):
    rid = tmp_db.add_reminder("Продлить домен", "2027-03-15 09:00:00", "yearly", [1, 2], created_by=1)
    active = tmp_db.get_active_reminders()
    assert any(r["id"] == rid for r in active)
    r = next(r for r in active if r["id"] == rid)
    assert r["recipients"] == [1, 2] and r["recurrence"] == "yearly"

    due = tmp_db.get_due_reminders("2027-03-15 09:00:00")
    assert any(r["id"] == rid for r in due)
    assert tmp_db.get_due_reminders("2027-03-14 09:00:00") == []

    tmp_db.update_reminder_next_at(rid, "2028-03-15 09:00:00")
    r2 = next(r for r in tmp_db.get_active_reminders() if r["id"] == rid)
    assert r2["next_at"] == "2028-03-15 09:00:00"

    tmp_db.deactivate_reminder(rid)
    assert not any(r["id"] == rid for r in tmp_db.get_active_reminders())


def test_reminders_all_recipients_and_delete(tmp_db):
    rid = tmp_db.add_reminder("Оплатить сервер", "2027-01-01 09:00:00", "once", "all", created_by=1)
    r = next(r for r in tmp_db.get_active_reminders() if r["id"] == rid)
    assert r["recipients"] == "all"
    assert tmp_db.delete_reminder(rid)
    assert not tmp_db.delete_reminder(rid)  # уже удалено, повторное удаление — False

Step 2: Убедиться, что тесты падают

Run: python -m pytest tests/test_database.py -v -k reminders Expected: FAIL с AttributeError: 'DatabaseManager' object has no attribute 'add_reminder'

Step 3: Добавить таблицу в _init_db

В database.py найти блок создания таблицы bot_flags (последняя таблица перед conn.commit() в _init_db):

            # Table for feature flags (показ/скрытие кнопок меню и разделов)
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS bot_flags (
                    key TEXT PRIMARY KEY,
                    enabled INTEGER NOT NULL,
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                )
            """)

            conn.commit()

Заменить на (добавлена таблица reminders перед conn.commit()):

            # Table for feature flags (показ/скрытие кнопок меню и разделов)
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS bot_flags (
                    key TEXT PRIMARY KEY,
                    enabled INTEGER NOT NULL,
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                )
            """)

            # Table for admin reminders (настраиваемые напоминания: дата, текст, периодичность)
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS reminders (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    text TEXT NOT NULL,
                    remind_at TEXT NOT NULL,   -- якорная дата первого срабатывания, не меняется
                    next_at TEXT NOT NULL,     -- следующее срабатывание, двигается вперёд
                    recurrence TEXT NOT NULL,  -- 'once' | 'weekly' | 'monthly' | 'yearly'
                    recipients TEXT NOT NULL,  -- JSON: список vk_id, либо строка "all"
                    created_by INTEGER NOT NULL,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    active INTEGER NOT NULL DEFAULT 1
                )
            """)

            conn.commit()

Step 4: Добавить CRUD-методы

В database.py найти метод get_admins (в блоке # --- Администраторы ---):

    def get_admins(self) -> list:
        with self._get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute("SELECT user_id FROM admins ORDER BY user_id")
            return [r[0] for r in cursor.fetchall()]

Сразу после него (перед блоком # --- Быстрые ответы (custom FAQ) ---) добавить новый блок:

    # --- Напоминалки для админов (дата/время, периодичность) ---
    def add_reminder(self, text: str, remind_at: str, recurrence: str,
                      recipients, created_by: int) -> int:
        """recipients — список vk_id (list[int]) либо строка 'all'."""
        with self._get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute(
                "INSERT INTO reminders (text, remind_at, next_at, recurrence, recipients, created_by) "
                "VALUES (?, ?, ?, ?, ?, ?)",
                (text, remind_at, remind_at, recurrence, json.dumps(recipients), created_by)
            )
            conn.commit()
            return cursor.lastrowid

    def _rows_with_parsed_recipients(self, cursor) -> list:
        rows = [dict(r) for r in cursor.fetchall()]
        for r in rows:
            r["recipients"] = json.loads(r["recipients"])
        return rows

    def get_active_reminders(self) -> list:
        """Активные напоминания для отображения в админке, по next_at."""
        with self._get_connection() as conn:
            conn.row_factory = sqlite3.Row
            cursor = conn.cursor()
            cursor.execute(
                "SELECT id, text, remind_at, next_at, recurrence, recipients, created_by "
                "FROM reminders WHERE active = 1 ORDER BY next_at"
            )
            return self._rows_with_parsed_recipients(cursor)

    def get_due_reminders(self, now_str: str) -> list:
        """Активные напоминания, чьё время (next_at) уже наступило."""
        with self._get_connection() as conn:
            conn.row_factory = sqlite3.Row
            cursor = conn.cursor()
            cursor.execute(
                "SELECT id, text, remind_at, next_at, recurrence, recipients, created_by "
                "FROM reminders WHERE active = 1 AND next_at <= ?", (now_str,)
            )
            return self._rows_with_parsed_recipients(cursor)

    def update_reminder_next_at(self, reminder_id: int, next_at: str):
        with self._get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute("UPDATE reminders SET next_at = ? WHERE id = ?", (next_at, reminder_id))
            conn.commit()

    def deactivate_reminder(self, reminder_id: int):
        with self._get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute("UPDATE reminders SET active = 0 WHERE id = ?", (reminder_id,))
            conn.commit()

    def delete_reminder(self, reminder_id: int) -> bool:
        with self._get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute("DELETE FROM reminders WHERE id = ?", (reminder_id,))
            conn.commit()
            return cursor.rowcount > 0

(json уже импортирован в начале database.py — import json в строке 3, отдельно импортировать не нужно.)

Step 5: Прогнать тесты

Run: python -m pytest tests/test_database.py -v Expected: PASS (все тесты, включая старые)

Step 6: Коммит

git add database.py tests/test_database.py
git commit -m "Напоминалки: таблица reminders и CRUD в database.py"

Task 3: `keyboards.py` — кнопки

Files:

Modify: keyboards.py
Step 1: Добавить кнопку «⏰ Напоминалки» в админ-панель

В keyboards.py найти get_admin_keyboard:

def get_admin_keyboard(is_super: bool = False, muted: bool = False) -> str:
    """Клавиатура админ-панели с разделами. Супер-админу — раздел «Администраторы»."""
    keyboard = VkKeyboard(one_time=False)
    keyboard.add_button("Статистика 📊", color=VkKeyboardColor.SECONDARY)
    keyboard.add_button("Быстрые ответы 💬", color=VkKeyboardColor.SECONDARY)
    keyboard.add_line()
    keyboard.add_button("Рассылка 📣", color=VkKeyboardColor.SECONDARY)
    keyboard.add_button("Ссылки 🔗", color=VkKeyboardColor.SECONDARY)
    keyboard.add_line()
    keyboard.add_button("Кнопки меню 🧩", color=VkKeyboardColor.SECONDARY)
    keyboard.add_button("Уведомления: выкл 🔕" if muted else "Уведомления: вкл 🔔",
                        color=VkKeyboardColor.SECONDARY)
    keyboard.add_line()
    if is_super:
        keyboard.add_button("Администраторы 👑", color=VkKeyboardColor.SECONDARY)
        keyboard.add_line()
    keyboard.add_button("В главное меню ↩️", color=VkKeyboardColor.SECONDARY)
    return keyboard.get_keyboard()

Заменить на (добавлена строка с кнопкой напоминалок перед разделом супер-админа):

def get_admin_keyboard(is_super: bool = False, muted: bool = False) -> str:
    """Клавиатура админ-панели с разделами. Супер-админу — раздел «Администраторы»."""
    keyboard = VkKeyboard(one_time=False)
    keyboard.add_button("Статистика 📊", color=VkKeyboardColor.SECONDARY)
    keyboard.add_button("Быстрые ответы 💬", color=VkKeyboardColor.SECONDARY)
    keyboard.add_line()
    keyboard.add_button("Рассылка 📣", color=VkKeyboardColor.SECONDARY)
    keyboard.add_button("Ссылки 🔗", color=VkKeyboardColor.SECONDARY)
    keyboard.add_line()
    keyboard.add_button("Кнопки меню 🧩", color=VkKeyboardColor.SECONDARY)
    keyboard.add_button("Уведомления: выкл 🔕" if muted else "Уведомления: вкл 🔔",
                        color=VkKeyboardColor.SECONDARY)
    keyboard.add_line()
    keyboard.add_button("⏰ Напоминалки", color=VkKeyboardColor.SECONDARY)
    keyboard.add_line()
    if is_super:
        keyboard.add_button("Администраторы 👑", color=VkKeyboardColor.SECONDARY)
        keyboard.add_line()
    keyboard.add_button("В главное меню ↩️", color=VkKeyboardColor.SECONDARY)
    return keyboard.get_keyboard()

Step 2: Добавить клавиатуру выбора периодичности

В keyboards.py сразу после функции get_cancel_admin_keyboard:

def get_cancel_admin_keyboard() -> str:
    """Кнопка отмены для админских мастеров ввода."""
    keyboard = VkKeyboard(one_time=False)
    keyboard.add_button("Отмена ❌", color=VkKeyboardColor.NEGATIVE)
    return keyboard.get_keyboard()

Добавить новую функцию:

def get_reminder_period_keyboard() -> str:
    """Клавиатура выбора периодичности напоминания (шаг мастера ADMIN_REMINDER_PERIOD)."""
    keyboard = VkKeyboard(one_time=False)
    keyboard.add_button("Разово", color=VkKeyboardColor.SECONDARY)
    keyboard.add_line()
    keyboard.add_button("Каждую неделю", color=VkKeyboardColor.SECONDARY)
    keyboard.add_button("Каждый месяц", color=VkKeyboardColor.SECONDARY)
    keyboard.add_line()
    keyboard.add_button("Каждый год", color=VkKeyboardColor.SECONDARY)
    keyboard.add_line()
    keyboard.add_button("Отмена ❌", color=VkKeyboardColor.NEGATIVE)
    return keyboard.get_keyboard()

Step 3: Проверить, что ничего не сломалось

Run: python -m pytest tests/ -v Expected: PASS (все существующие тесты по-прежнему проходят — keyboards.py не имеет собственных тестов, но используется в test_flow.py)

Step 4: Коммит

git add keyboards.py
git commit -m "Напоминалки: кнопка в админ-панели и клавиатура выбора периода"

Task 4: `bot.py` — мастер создания и раздел списка/удаления

Files:

Modify: bot.py
Step 1: Импорт и константы

В bot.py найти блок импортов:

from database import DatabaseManager
import config
import mailer

Заменить на:

from database import DatabaseManager
from reminders import parse_reminder_datetime, compute_next_occurrence
from datetime import datetime
import config
import mailer

Найти константу:

# Сколько секунд кэшировать список рантайм-админов из БД
ADMINS_CACHE_TTL = 30

Заменить на:

# Сколько секунд кэшировать список рантайм-админов из БД
ADMINS_CACHE_TTL = 30
# Как часто фоновый поток проверяет, не пора ли отправить напоминание
REMINDER_CHECK_INTERVAL_SEC = 60

REMINDER_PERIOD_LABELS = {
    "once": "разово",
    "weekly": "каждую неделю",
    "monthly": "каждый месяц",
    "yearly": "каждый год",
}
REMINDER_PERIOD_BY_LABEL = {v: k for k, v in REMINDER_PERIOD_LABELS.items()}

Затем найти внутри run() локальный импорт (он больше не нужен — datetime теперь импортирован на уровне модуля):

        try:
            from datetime import datetime
            ts = datetime.now().strftime("%d.%m.%Y %H:%M")

Заменить на:

        try:
            ts = datetime.now().strftime("%d.%m.%Y %H:%M")

Step 2: Прогнать тесты (регрессия импортов)

Run: python -m pytest tests/ -v Expected: PASS (импорт reminders из Task 1 уже существует, ничего не сломано)

Step 3: Добавить _all_admin_ids и раздел списка/удаления напоминаний

В bot.py найти конец метода _handle_admin_remove_admin (перед _cleanup_antiflood):

        if db.remove_admin(rem_id):
            self._admins_cache = None  # инвалидируем кэш админов
            self.send_message(user_id, f"✅ Пользователь vk.com/id{rem_id} больше не администратор.")
        else:
            self.send_message(user_id, f"⚠️ vk.com/id{rem_id} нет среди добавленных из бота "
                                       "(админов из .env и супер-админов отсюда снять нельзя).")
        self._send_admins_panel(user_id)

    def _cleanup_antiflood(self):

Заменить на (добавлены новые методы между ними):

        if db.remove_admin(rem_id):
            self._admins_cache = None  # инвалидируем кэш админов
            self.send_message(user_id, f"✅ Пользователь vk.com/id{rem_id} больше не администратор.")
        else:
            self.send_message(user_id, f"⚠️ vk.com/id{rem_id} нет среди добавленных из бота "
                                       "(админов из .env и супер-админов отсюда снять нельзя).")
        self._send_admins_panel(user_id)

    def _all_admin_ids(self) -> list:
        """Полный список текущих админов: супер-админы + .env + добавленные в рантайме,
        без дублей, отсортирован (используется для «все админы» и нумерации в мастере)."""
        ids = set(getattr(config, "SUPER_ADMIN_IDS", []))
        ids |= set(getattr(config, "ADMIN_IDS", []))
        ids |= self._runtime_admins()
        return sorted(ids)

    def _send_reminders_admin(self, user_id: int):
        """Раздел «Напоминалки»: список активных + инструкции."""
        rows = db.get_active_reminders()
        listing = ""
        for r in rows:
            dt = datetime.strptime(r["next_at"], "%Y-%m-%d %H:%M:%S")
            who = "всем админам" if r["recipients"] == "all" else \
                ", ".join(f"id{uid}" for uid in r["recipients"])
            period = REMINDER_PERIOD_LABELS.get(r["recurrence"], r["recurrence"])
            short = (r["text"][:50] + "…") if len(r["text"]) > 50 else r["text"]
            listing += (f"🔹 [{r['id']}] {dt.strftime('%d.%m.%Y %H:%M')} ({period}) → {who}\n"
                        f"   «{short}»\n")
        if not listing:
            listing = "Активных напоминаний нет.\n"
        kb = VkKeyboard(inline=True)
        kb.add_button("➕ Добавить", color=VkKeyboardColor.PRIMARY)
        kb.add_line()
        kb.add_button("В админку ↩️", color=VkKeyboardColor.SECONDARY)
        self.send_message(
            user_id,
            "⏰ Напоминалки\n\n"
            f"{listing}\n"
            "Чтобы удалить: напоминалка удалить <id>\n"
            "Чтобы добавить: нажмите «➕ Добавить»",
            keyboard=kb.get_keyboard()
        )

    def _handle_reminder_delete(self, user_id: int, text: str):
        arg = text[len("напоминалка удалить"):].strip()
        if not arg.isdigit():
            self.send_message(user_id, "⚠️ Укажите номер: напоминалка удалить 3")
            return
        if db.delete_reminder(int(arg)):
            self.send_message(user_id, f"✅ Напоминание №{arg} удалено.")
        else:
            self.send_message(user_id, f"⚠️ Напоминание №{arg} не найдено.")
        self._send_reminders_admin(user_id)

    def _cleanup_antiflood(self):

Step 4: Подключить раздел в главном роутере handle_message

В bot.py найти в handle_message блок обработки быстрых ответов/ссылок:

            elif text_lower in ["ссылки 🔗", "ссылки", "ссылка"]:
                self._send_links_admin(user_id)
                return

Заменить на (добавлены три новых elif сразу после):

            elif text_lower in ["ссылки 🔗", "ссылки", "ссылка"]:
                self._send_links_admin(user_id)
                return
            elif text_lower in ["⏰ напоминалки", "напоминалки"]:
                self._send_reminders_admin(user_id)
                return
            elif text_lower == "➕ добавить":
                db.update_user_state(user_id, "ADMIN_REMINDER_DATE")
                self.send_message(
                    user_id,
                    "⏰ Новое напоминание.\nВведите дату и время: ДД.ММ.ГГГГ ЧЧ:ММ\n"
                    "Например: 15.03.2027 09:00",
                    keyboard=get_cancel_admin_keyboard()
                )
                return
            elif text_lower.startswith("напоминалка удалить"):
                self._handle_reminder_delete(user_id, text_clean)
                return

Step 5: Добавить обработку состояний мастера в _handle_admin_wizard

В bot.py найти конец обработки администраторов (перед сбросом неизвестного состояния):

        if state == "ADMIN_ADMIN_DEL_PICK":
            if not self._is_super_admin(user_id):
                db.update_user_state(user_id, "FREE_CHAT")
                self._send_admin_panel(user_id)
                return
            self._handle_admin_remove_admin(user_id, "админ удал " + pick_id(text))
            db.update_user_state(user_id, "FREE_CHAT")
            return

        # Неизвестное ADMIN_-состояние — сброс
        self._admin_ctx.pop(user_id, None)
        db.update_user_state(user_id, "FREE_CHAT")
        self._send_admin_panel(user_id)

Заменить на (новый блок # ---------- Напоминалки ---------- вставлен перед сбросом):

        if state == "ADMIN_ADMIN_DEL_PICK":
            if not self._is_super_admin(user_id):
                db.update_user_state(user_id, "FREE_CHAT")
                self._send_admin_panel(user_id)
                return
            self._handle_admin_remove_admin(user_id, "админ удал " + pick_id(text))
            db.update_user_state(user_id, "FREE_CHAT")
            return

        # ---------- Напоминалки ----------
        if state == "ADMIN_REMINDER_DATE":
            dt = parse_reminder_datetime(text)
            if not dt:
                self.send_message(user_id, "⚠️ Формат: ДД.ММ.ГГГГ ЧЧ:ММ (например, 15.03.2027 09:00)",
                                  keyboard=get_cancel_admin_keyboard())
                return
            if dt <= datetime.now():
                self.send_message(user_id, "⚠️ Дата и время должны быть в будущем.",
                                  keyboard=get_cancel_admin_keyboard())
                return
            self._admin_ctx[user_id] = {"remind_at": dt.strftime("%Y-%m-%d %H:%M:%S")}
            db.update_user_state(user_id, "ADMIN_REMINDER_TEXT")
            self.send_message(user_id, "✏️ Введите текст напоминания:", keyboard=get_cancel_admin_keyboard())
            return

        if state == "ADMIN_REMINDER_TEXT":
            if not text.strip():
                self.send_message(user_id, "⚠️ Текст не может быть пустым.", keyboard=get_cancel_admin_keyboard())
                return
            self._admin_ctx.setdefault(user_id, {})["text"] = text.strip()
            admins = self._all_admin_ids()
            self._admin_ctx[user_id]["admins_list"] = admins
            listing = "\n".join(f"{i + 1}. vk.com/id{a}" for i, a in enumerate(admins))
            db.update_user_state(user_id, "ADMIN_REMINDER_RECIPIENTS")
            self.send_message(
                user_id,
                "👥 Кому отправлять напоминание?\n"
                f"{listing}\n\n"
                "Введите номер (например, 2), несколько через запятую (1,3) или 0 — все админы.",
                keyboard=get_cancel_admin_keyboard()
            )
            return

        if state == "ADMIN_REMINDER_RECIPIENTS":
            ctx = self._admin_ctx.get(user_id, {})
            admins = ctx.get("admins_list", [])
            raw = text.strip()
            if raw == "0":
                recipients = "all"
            else:
                picks = [p.strip() for p in raw.split(",") if p.strip()]
                if not picks or not all(p.isdigit() for p in picks):
                    self.send_message(user_id, "⚠️ Введите номера через запятую или 0 для всех админов.",
                                      keyboard=get_cancel_admin_keyboard())
                    return
                indices = [int(p) for p in picks]
                if any(i < 1 or i > len(admins) for i in indices):
                    self.send_message(user_id, "⚠️ Есть номер вне списка. Проверьте и введите снова.",
                                      keyboard=get_cancel_admin_keyboard())
                    return
                recipients = sorted({admins[i - 1] for i in indices})
            ctx["recipients"] = recipients
            db.update_user_state(user_id, "ADMIN_REMINDER_PERIOD")
            self.send_message(user_id, "🔁 Периодичность напоминания:", keyboard=get_reminder_period_keyboard())
            return

        if state == "ADMIN_REMINDER_PERIOD":
            recurrence = REMINDER_PERIOD_BY_LABEL.get(text_lower)
            if not recurrence:
                self.send_message(user_id, "⚠️ Выберите период кнопкой ниже.",
                                  keyboard=get_reminder_period_keyboard())
                return
            ctx = self._admin_ctx.pop(user_id, {})
            remind_at = ctx.get("remind_at")
            rtext = ctx.get("text")
            recipients = ctx.get("recipients")
            if not (remind_at and rtext and recipients is not None):
                db.update_user_state(user_id, "FREE_CHAT")
                self.send_message(user_id, "⚠️ Что-то пошло не так, начните заново кнопкой «➕ Добавить».")
                self._send_reminders_admin(user_id)
                return
            db.add_reminder(rtext, remind_at, recurrence, recipients, user_id)
            db.update_user_state(user_id, "FREE_CHAT")
            self.send_message(user_id, "✅ Напоминание создано.")
            self._send_reminders_admin(user_id)
            return

        # Неизвестное ADMIN_-состояние — сброс
        self._admin_ctx.pop(user_id, None)
        db.update_user_state(user_id, "FREE_CHAT")
        self._send_admin_panel(user_id)

Step 6: Прогнать полный набор тестов

Run: python -m pytest tests/ -v Expected: PASS (мастер пока не покрыт тестами — добавим в Task 6; ничего не должно сломаться из существующего)

Step 7: Коммит

git add bot.py
git commit -m "Напоминалки: мастер создания, список и удаление в bot.py"

Task 5: `bot.py` — фоновый планировщик отправки

Files:

Modify: bot.py
Step 1: Добавить _check_due_reminders и _reminders_loop

В bot.py найти конец метода _run_broadcast (перед def run(self):):

        self.send_message(admin_id, f"✅ Рассылка завершена. Доставлено получателям: {sent} из {len(recipients)}.")

    def run(self):

Заменить на:

        self.send_message(admin_id, f"✅ Рассылка завершена. Доставлено получателям: {sent} из {len(recipients)}.")

    def _check_due_reminders(self):
        """Проверяет и рассылает напоминания, чьё время наступило.
        Вызывается фоновым потоком `_reminders_loop`, но выделена отдельно,
        чтобы её можно было дёргать напрямую из тестов без запуска потока."""
        now = datetime.now()
        now_str = now.strftime("%Y-%m-%d %H:%M:%S")
        for r in db.get_due_reminders(now_str):
            targets = self._all_admin_ids() if r["recipients"] == "all" else r["recipients"]
            body = f"⏰ Напоминание:\n{r['text']}"
            for uid in targets:
                try:
                    self.send_message(uid, body)
                except Exception as e:
                    logger.error(f"Не удалось отправить напоминание {r['id']} пользователю {uid}: {e}")
            if r["recurrence"] == "once":
                db.deactivate_reminder(r["id"])
            else:
                anchor = datetime.strptime(r["remind_at"], "%Y-%m-%d %H:%M:%S")
                next_dt = compute_next_occurrence(anchor, r["recurrence"], now)
                db.update_reminder_next_at(r["id"], next_dt.strftime("%Y-%m-%d %H:%M:%S"))

    def _reminders_loop(self):
        """Фоновый цикл: раз в REMINDER_CHECK_INTERVAL_SEC секунд проверяет и шлёт напоминания.
        Обрыв на одной итерации (например, сбой отправки) не должен останавливать цикл."""
        while True:
            try:
                self._check_due_reminders()
            except Exception as e:
                logger.error(f"Ошибка обработки напоминаний: {e}")
            time.sleep(REMINDER_CHECK_INTERVAL_SEC)

    def run(self):

Step 2: Запустить поток в run()

В bot.py найти начало run():

    def run(self):
        """Запуск бесконечного цикла прослушивания событий VK LongPoll."""
        if not self.longpoll:

Заменить на:

    def run(self):
        """Запуск бесконечного цикла прослушивания событий VK LongPoll."""
        threading.Thread(target=self._reminders_loop, daemon=True).start()
        if not self.longpoll:

Step 3: Прогнать полный набор тестов

Run: python -m pytest tests/ -v Expected: PASS

Step 4: Коммит

git add bot.py
git commit -m "Напоминалки: фоновый планировщик отправки в срок"

Task 6: Сквозные тесты мастера в `test_flow.py`

Files:

Modify: tests/test_flow.py
Step 1: Написать тесты полного сценария

Добавить в конец tests/test_flow.py:

def test_admin_reminder_wizard_creates_reminder(app, monkeypatch):
    admin = 80001
    monkeypatch.setattr(bot.config, "ADMIN_IDS", [admin])

    app.handle_message(admin, "Админка ⚙️")
    app.handle_message(admin, "⏰ Напоминалки")
    app.handle_message(admin, "➕ Добавить")
    assert _state(admin) == "ADMIN_REMINDER_DATE"

    app.handle_message(admin, "не дата")  # невалидный формат — остаёмся на шаге
    assert _state(admin) == "ADMIN_REMINDER_DATE"

    app.handle_message(admin, "01.01.2099 09:00")
    assert _state(admin) == "ADMIN_REMINDER_TEXT"

    app.handle_message(admin, "Продлить домен")
    assert _state(admin) == "ADMIN_REMINDER_RECIPIENTS"

    app.handle_message(admin, "0")  # все админы
    assert _state(admin) == "ADMIN_REMINDER_PERIOD"

    app.handle_message(admin, "Каждый год")
    assert _state(admin) == "FREE_CHAT"

    rows = bot.db.get_active_reminders()
    assert len(rows) == 1
    assert rows[0]["recipients"] == "all"
    assert rows[0]["recurrence"] == "yearly"
    assert rows[0]["text"] == "Продлить домен"


def test_admin_reminder_wizard_specific_recipients(app, monkeypatch):
    admin, other = 80003, 80004
    monkeypatch.setattr(bot.config, "ADMIN_IDS", [admin, other])

    app.handle_message(admin, "⏰ Напоминалки")
    app.handle_message(admin, "➕ Добавить")
    app.handle_message(admin, "01.01.2099 09:00")
    app.handle_message(admin, "Оплатить сервер")
    app.handle_message(admin, "9")  # номер вне списка (только 2 админа)
    assert _state(admin) == "ADMIN_REMINDER_RECIPIENTS"
    app.handle_message(admin, "1")  # первый по сортировке id
    assert _state(admin) == "ADMIN_REMINDER_PERIOD"
    app.handle_message(admin, "Разово")

    rows = bot.db.get_active_reminders()
    assert rows[0]["recipients"] == [min(admin, other)]
    assert rows[0]["recurrence"] == "once"


def test_admin_reminder_delete(app, monkeypatch):
    admin = 80002
    monkeypatch.setattr(bot.config, "ADMIN_IDS", [admin])
    rid = bot.db.add_reminder("Тест", "2099-01-01 09:00:00", "once", [admin], created_by=admin)

    app.handle_message(admin, f"напоминалка удалить {rid}")
    assert not bot.db.get_active_reminders()

    app.handle_message(admin, f"напоминалка удалить {rid}")  # повторно — не найдено
    last_msg = app._sent[-1]["msg"] if app._sent else ""
    assert "не найдено" in last_msg

Step 2: Прогнать новые тесты

Run: python -m pytest tests/test_flow.py -v -k reminder Expected: PASS (3 passed)

Step 3: Прогнать весь набор тестов проекта

Run: python -m pytest tests/ -v Expected: PASS (все тесты проекта, включая старые)

Step 4: Коммит

git add tests/test_flow.py
git commit -m "Напоминалки: сквозные тесты мастера создания и удаления"

Task 7: Ручная проверка в консольном эмуляторе (перед push)

Files: нет изменений кода — только проверка.

Step 1: Запустить бота в режиме эмуляции

Run: python bot.py

(Без валидного VK_TOKEN в .env бот запускается в консольном режиме — это ожидаемо и удобно для ручной проверки без реального VK.)

Step 2: Пройти мастер вручную

В консоли эмулятора (пользователь по умолчанию не админ — на реальном проекте временно добавьте свой тестовый ID в ADMIN_IDS в .env и перезапустите, либо отредактируйте config.ADMIN_IDS перед запуском для проверки):

Админка ⚙️
⏰ Напоминалки
➕ Добавить
15.03.2027 09:00
Тестовое напоминание
0
Каждый месяц

Expected: после каждого шага — соответствующий вопрос мастера; в конце — «✅ Напоминание создано.» и список с одной записью.

Step 3: Проверить срабатывание фонового потока

Создать через код (не дожидаясь реальной даты) напоминание на ближайшую минуту и подождать: временно в интерактивной python-сессии (python, в каталоге проекта):

import bot
from datetime import datetime, timedelta
soon = (datetime.now() + timedelta(seconds=5)).strftime("%Y-%m-%d %H:%M:%S")
bot.db.add_reminder("Проверка планировщика", soon, "once", "all", created_by=1)
app = bot.VKBotApplication()
import time
time.sleep(1)
app._check_due_reminders()  # без реального VK_TOKEN send_message пишет в лог, не падает
print(bot.db.get_active_reminders())  # должно быть пусто — 'once' деактивировался

Expected: [] — напоминание отправлено (проверьте bot_run.log на строку с текстом напоминания либо консольный вывод) и деактивировано.

Step 4: Убедиться, что база не испортила существующие данные

Run: python -m pytest tests/ -v Expected: PASS — полный прогон перед завершением работы.

Step 5: Сообщить пользователю результат ручной проверки

Только после успешного прохождения Step 1–4 — сообщить, что фича готова к ревью. Не выполнять git push — по договорённости с пользователем, push делается им самим (или по отдельному явному запросу) после того, как он лично протестирует бота.

44 KiB Raw Permalink Blame History Unescape Escape

Настраиваемые напоминалки для админов — план реализации

Task 1: reminders.py — парсинг даты и расчёт следующего срабатывания

Task 2: database.py — таблица reminders и CRUD

Task 3: keyboards.py — кнопки

Task 4: bot.py — мастер создания и раздел списка/удаления

Task 5: bot.py — фоновый планировщик отправки

Task 6: Сквозные тесты мастера в test_flow.py

Task 7: Ручная проверка в консольном эмуляторе (перед push)

44 KiB

Raw Permalink Blame History

Task 1: `reminders.py` — парсинг даты и расчёт следующего срабатывания

Task 2: `database.py` — таблица `reminders` и CRUD

Task 3: `keyboards.py` — кнопки

Task 4: `bot.py` — мастер создания и раздел списка/удаления

Task 5: `bot.py` — фоновый планировщик отправки

Task 6: Сквозные тесты мастера в `test_flow.py`