Kenik
1014 lines
44 KiB
Markdown
1014 lines
44 KiB
Markdown
# Настраиваемые напоминалки для админов — план реализации
|
||
|
||
> **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`:
|
||
|
||
```python
|
||
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`:
|
||
|
||
```python
|
||
"""Расчёт дат для напоминалок админов: парсинг ввода и следующее срабатывание.
|
||
|
||
Не зависит от 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: Коммит**
|
||
|
||
```bash
|
||
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`:
|
||
|
||
```python
|
||
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`):
|
||
|
||
```python
|
||
# 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()`):
|
||
|
||
```python
|
||
# 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` (в блоке `# --- Администраторы ---`):
|
||
|
||
```python
|
||
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) ---`) добавить новый
|
||
блок:
|
||
|
||
```python
|
||
# --- Напоминалки для админов (дата/время, периодичность) ---
|
||
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: Коммит**
|
||
|
||
```bash
|
||
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`:
|
||
|
||
```python
|
||
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()
|
||
```
|
||
|
||
Заменить на (добавлена строка с кнопкой напоминалок перед разделом супер-админа):
|
||
|
||
```python
|
||
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`:
|
||
|
||
```python
|
||
def get_cancel_admin_keyboard() -> str:
|
||
"""Кнопка отмены для админских мастеров ввода."""
|
||
keyboard = VkKeyboard(one_time=False)
|
||
keyboard.add_button("Отмена ❌", color=VkKeyboardColor.NEGATIVE)
|
||
return keyboard.get_keyboard()
|
||
```
|
||
|
||
Добавить новую функцию:
|
||
|
||
```python
|
||
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: Коммит**
|
||
|
||
```bash
|
||
git add keyboards.py
|
||
git commit -m "Напоминалки: кнопка в админ-панели и клавиатура выбора периода"
|
||
```
|
||
|
||
---
|
||
|
||
## Task 4: `bot.py` — мастер создания и раздел списка/удаления
|
||
|
||
**Files:**
|
||
- Modify: `bot.py`
|
||
|
||
- [ ] **Step 1: Импорт и константы**
|
||
|
||
В `bot.py` найти блок импортов:
|
||
|
||
```python
|
||
from database import DatabaseManager
|
||
import config
|
||
import mailer
|
||
```
|
||
|
||
Заменить на:
|
||
|
||
```python
|
||
from database import DatabaseManager
|
||
from reminders import parse_reminder_datetime, compute_next_occurrence
|
||
from datetime import datetime
|
||
import config
|
||
import mailer
|
||
```
|
||
|
||
Найти константу:
|
||
|
||
```python
|
||
# Сколько секунд кэшировать список рантайм-админов из БД
|
||
ADMINS_CACHE_TTL = 30
|
||
```
|
||
|
||
Заменить на:
|
||
|
||
```python
|
||
# Сколько секунд кэшировать список рантайм-админов из БД
|
||
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` теперь
|
||
импортирован на уровне модуля):
|
||
|
||
```python
|
||
try:
|
||
from datetime import datetime
|
||
ts = datetime.now().strftime("%d.%m.%Y %H:%M")
|
||
```
|
||
|
||
Заменить на:
|
||
|
||
```python
|
||
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`):
|
||
|
||
```python
|
||
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):
|
||
```
|
||
|
||
Заменить на (добавлены новые методы между ними):
|
||
|
||
```python
|
||
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` блок обработки быстрых ответов/ссылок:
|
||
|
||
```python
|
||
elif text_lower in ["ссылки 🔗", "ссылки", "ссылка"]:
|
||
self._send_links_admin(user_id)
|
||
return
|
||
```
|
||
|
||
Заменить на (добавлены три новых elif сразу после):
|
||
|
||
```python
|
||
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` найти конец обработки администраторов (перед сбросом неизвестного состояния):
|
||
|
||
```python
|
||
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)
|
||
```
|
||
|
||
Заменить на (новый блок `# ---------- Напоминалки ----------` вставлен перед сбросом):
|
||
|
||
```python
|
||
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: Коммит**
|
||
|
||
```bash
|
||
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):`):
|
||
|
||
```python
|
||
self.send_message(admin_id, f"✅ Рассылка завершена. Доставлено получателям: {sent} из {len(recipients)}.")
|
||
|
||
def run(self):
|
||
```
|
||
|
||
Заменить на:
|
||
|
||
```python
|
||
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()`:
|
||
|
||
```python
|
||
def run(self):
|
||
"""Запуск бесконечного цикла прослушивания событий VK LongPoll."""
|
||
if not self.longpoll:
|
||
```
|
||
|
||
Заменить на:
|
||
|
||
```python
|
||
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: Коммит**
|
||
|
||
```bash
|
||
git add bot.py
|
||
git commit -m "Напоминалки: фоновый планировщик отправки в срок"
|
||
```
|
||
|
||
---
|
||
|
||
## Task 6: Сквозные тесты мастера в `test_flow.py`
|
||
|
||
**Files:**
|
||
- Modify: `tests/test_flow.py`
|
||
|
||
- [ ] **Step 1: Написать тесты полного сценария**
|
||
|
||
Добавить в конец `tests/test_flow.py`:
|
||
|
||
```python
|
||
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: Коммит**
|
||
|
||
```bash
|
||
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`, в каталоге проекта):
|
||
|
||
```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 делается им
|
||
самим (или по отдельному явному запросу) после того, как он лично протестирует бота.
|