Azimkin/WouldYouRather
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Remove obsolete manage command and related modules

Browse Source 
- Deleted `manage.py` and related classes
- Removed unused `PendingManageAction` and repository calls
- Updated references to obsolete manage features in tests:
  - Deleted `test_statistics.py`
  - Deleted `test_cron.py`
  - Deleted `test_kurigram_compat.py`
- Cleaned up outdated functionality for managing questions, schedules, and statistics via `/manage`.
This commit is contained in:
azimkin
2026-04-05 18:44:10 +02:00
parent 6c6ef263be
commit 5acad376b0
18 changed files with 1157 additions and 1068 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
STATISTICS_GUIDE.md
View File

@@ -177,11 +177,10 @@ handler.register(client)
### Изменение голоса
Если пользователь изменит свой выбор:
- Telegram отправит обновлённый `PollAnswer`
- Бот обработает **все текущие выборы**
- **Важно:** Предыдущие голоса не удаляются автоматически
- Telegram отправит обновлённое событие `UpdateMessagePoll` с новыми агрегированными результатами
- Бот вычислит дельту и запишет изменения
⚠️ **Ограничение:** Telegram не отправляет информацию об отменённых голосах, поэтому если пользователь изменит выбор, старый голос останется в базе. Это известная особенность API Telegram.
**Важно:** Система работает только с агрегированными данными. Для анонимных опросов Telegram не предоставляет информацию об индивидуальных голосах пользователей.
### Переопубликация вопросов
При повторной публикации одного вопроса:

4
bot/bot.py
View File

@@ -51,13 +51,15 @@ async def main():
repository = SQLiteWouldYouRatherRepository()
registry.bind(client, context)
ManageController(context).register(client)
controller = ManageController(context, repository=repository)
controller.register(client)
PollStatisticsHandler(repository).register(client)
scheduler = QuestionPublisherScheduler(
client=client,
repository=repository,
channel_ids=context.poll_channel_ids,
admin_ids=context.admin_ids,
)
logger.info("Bot client initialized and command handlers registered.")
async with client:

523
bot/manage.py
View File

@@ -1,523 +0,0 @@
from __future__ import annotations
from dataclasses import dataclass
import math
from pyrogram import Client, filters
from pyrogram.errors import MessageNotModified
from pyrogram.handlers import CallbackQueryHandler, MessageHandler
from pyrogram.types import CallbackQuery, InlineKeyboardButton, InlineKeyboardMarkup, Message
from game import SQLiteWouldYouRatherRepository
from .command_system import CommandContext
from .scheduler import validate_cron_expression
@dataclass(frozen=True)
class PendingManageAction:
mode: str
question_id: int | None = None
schedule_id: int | None = None
class ManageController:
def __init__(self, context: CommandContext, page_size: int = 5) -> None:
self.context = context
self.page_size = page_size
self.repository = SQLiteWouldYouRatherRepository()
self.pending_actions: dict[int, PendingManageAction] = {}
def register(self, client: Client) -> None:
client.add_handler(
MessageHandler(self._on_manage_command, filters.command(["manage"]))
)
client.add_handler(
CallbackQueryHandler(self._on_callback_query, filters.regex(r"^mg:"))
)
client.add_handler(
MessageHandler(self._on_text_message, filters.text & ~filters.regex(r"^/"))
)
async def _on_manage_command(self, _: Client, message: Message) -> None:
user_id = message.from_user.id if message.from_user else None
if user_id is None or user_id not in self.context.admin_ids:
await message.reply_text("Команда доступна только администраторам.")
return
self.pending_actions.pop(user_id, None)
await message.reply_text(
"Панель управления вопросами:",
reply_markup=self._menu_keyboard(),
)
async def _on_callback_query(self, _: Client, callback: CallbackQuery) -> None:
user_id = callback.from_user.id if callback.from_user else None
if user_id is None or user_id not in self.context.admin_ids:
await callback.answer("Только для администраторов.", show_alert=True)
return
data = callback.data or ""
if data == "mg:menu":
self.pending_actions.pop(user_id, None)
await self._edit(callback, "Панель управления вопросами:", self._menu_keyboard())
await callback.answer()
return
if data.startswith("mg:view:"):
page = self._extract_page(data)
await self._render_questions_page(callback, page)
await callback.answer()
return
if data == "mg:add":
self.pending_actions[user_id] = PendingManageAction(mode="add")
await self._edit(
callback,
"Отправьте новый вопрос в формате:\n"
"<вопрос> | <вариант 1> | <вариант 2> [| <вариант 3> ...]",
self._back_to_menu_keyboard(),
)
await callback.answer()
return
if data == "mg:schedule:list":
await self._render_schedules(callback)
await callback.answer()
return
if data == "mg:schedule:add":
self.pending_actions[user_id] = PendingManageAction(mode="schedule_add")
await self._edit(
callback,
"Отправьте расписание в формате:\n"
"<название> | <cron 5 полей> | <active: 1 или 0>\n"
"Пример: Утро | 0 9 * * 1-5 | 1",
self._back_to_menu_keyboard(),
)
await callback.answer()
return
if data == "mg:schedule:edit:list":
await self._render_schedule_edit_list(callback)
await callback.answer()
return
if data.startswith("mg:schedule:edit:"):
try:
schedule_id = self._extract_id(data)
except ValueError:
await callback.answer("Некорректный идентификатор.", show_alert=True)
return
schedule = self.repository.get_schedule(schedule_id)
if schedule is None:
await callback.answer("Расписание не найдено.", show_alert=True)
return
self.pending_actions[user_id] = PendingManageAction(
mode="schedule_edit",
schedule_id=schedule_id,
)
await self._edit(
callback,
"Отправьте обновлённое расписание в формате:\n"
"<название> | <cron 5 полей> | <active: 1 или 0>\n"
f"Текущее: {schedule.name} | {schedule.cron} | {1 if schedule.is_active else 0}",
self._back_to_menu_keyboard(),
)
await callback.answer()
return
if data == "mg:schedule:delete:list":
await self._render_schedule_delete_list(callback)
await callback.answer()
return
if data.startswith("mg:schedule:delete:"):
try:
schedule_id = self._extract_id(data)
except ValueError:
await callback.answer("Некорректный идентификатор.", show_alert=True)
return
try:
schedule = self.repository.get_schedule(schedule_id)
if schedule is None:
await callback.answer("Расписание не найдено.", show_alert=True)
return
self.repository.delete_schedule(schedule_id)
await callback.answer(f"Расписание '{schedule.name}' удалено.", show_alert=True)
await self._render_schedule_delete_list(callback)
except ValueError as error:
await callback.answer(str(error), show_alert=True)
return
if data.startswith("mg:edit:"):
page = self._extract_page(data)
await self._render_edit_page(callback, page)
await callback.answer()
return
if data.startswith("mg:editq:"):
try:
question_id = self._extract_id(data)
except ValueError:
await callback.answer("Некорректный идентификатор.", show_alert=True)
return
self.pending_actions[user_id] = PendingManageAction(mode="edit", question_id=question_id)
await self._edit(
callback,
"Отправьте обновлённый вопрос в формате:\n"
"<вопрос> | <вариант 1> | <вариант 2> [| <вариант 3> ...]",
self._back_to_menu_keyboard(),
)
await callback.answer()
return
if data.startswith("mg:stats:"):
page = self._extract_page(data)
await self._render_stats_page(callback, page)
await callback.answer()
return
if data.startswith("mg:statq:"):
try:
question_id = self._extract_id(data)
except ValueError:
await callback.answer("Некорректный идентификатор.", show_alert=True)
return
await self._render_question_stats(callback, question_id)
await callback.answer()
return
await callback.answer("Неизвестное действие.")
async def _on_text_message(self, _: Client, message: Message) -> None:
user_id = message.from_user.id if message.from_user else None
if user_id is None or user_id not in self.context.admin_ids:
return
action = self.pending_actions.get(user_id)
if action is None:
return
text = message.text or ""
try:
if action.mode == "add":
question_text, answers = self._parse_question_input(text)
question_id = self.repository.add_question(question_text, answers)
self.pending_actions.pop(user_id, None)
await message.reply_text(
f"Вопрос добавлен (ID: {question_id}).",
reply_markup=self._menu_keyboard(),
)
return
if action.mode == "edit":
if action.question_id is None:
raise ValueError("Question ID is missing for edit mode.")
question_text, answers = self._parse_question_input(text)
self.repository.update_question(action.question_id, question_text, answers)
self.pending_actions.pop(user_id, None)
await message.reply_text(
f"Вопрос #{action.question_id} обновлён.",
reply_markup=self._menu_keyboard(),
)
return
if action.mode == "schedule_add":
name, cron, is_active = self._parse_schedule_input(text)
self.repository.add_schedule(name=name, cron=cron, is_active=is_active)
self.pending_actions.pop(user_id, None)
await message.reply_text(
"Расписание добавлено.",
reply_markup=self._menu_keyboard(),
)
return
if action.mode == "schedule_edit":
if action.schedule_id is None:
raise ValueError("Schedule ID is missing for edit mode.")
name, cron, is_active = self._parse_schedule_input(text)
self.repository.update_schedule(
schedule_id=action.schedule_id,
name=name,
cron=cron,
is_active=is_active,
)
self.pending_actions.pop(user_id, None)
await message.reply_text(
f"Расписание #{action.schedule_id} обновлено.",
reply_markup=self._menu_keyboard(),
)
return
except ValueError as error:
if action.mode in ("schedule_add", "schedule_edit"):
await message.reply_text(
f"{error}\n\nИспользуйте формат:\n"
"<название> | <cron 5 полей> | <active: 1 или 0>"
)
else:
await message.reply_text(
f"{error}\n\nИспользуйте формат:\n"
"<вопрос> | <вариант 1> | <вариант 2> [| <вариант 3> ...]"
)
async def _render_questions_page(self, callback: CallbackQuery, page: int) -> None:
total = self.repository.count_questions()
total_pages = max(1, math.ceil(total / self.page_size))
safe_page = min(max(page, 1), total_pages)
offset = (safe_page - 1) * self.page_size
rows = self.repository.list_questions(limit=self.page_size, offset=offset)
if not rows:
text = "Вопросов пока нет."
else:
lines = [f"Список вопросов (страница {safe_page}/{total_pages}):"]
for item in rows:
short_text = item.text if len(item.text) <= 70 else f"{item.text[:67]}..."
lines.append(f"#{item.id} ({item.answers_count} вариантов): {short_text}")
text = "\n".join(lines)
await self._edit(callback, text, self._pagination_keyboard("mg:view", safe_page, total_pages))
async def _render_edit_page(self, callback: CallbackQuery, page: int) -> None:
total = self.repository.count_questions()
total_pages = max(1, math.ceil(total / self.page_size))
safe_page = min(max(page, 1), total_pages)
offset = (safe_page - 1) * self.page_size
rows = self.repository.list_questions(limit=self.page_size, offset=offset)
if not rows:
await self._edit(callback, "Нет вопросов для изменения.", self._back_to_menu_keyboard())
return
lines = [f"Выберите вопрос для изменения ({safe_page}/{total_pages}):"]
keyboard_rows: list[list[InlineKeyboardButton]] = []
for item in rows:
short_text = item.text if len(item.text) <= 50 else f"{item.text[:47]}..."
lines.append(f"#{item.id}: {short_text}")
keyboard_rows.append(
[InlineKeyboardButton(f"✏️ #{item.id}", callback_data=f"mg:editq:{item.id}")]
)
keyboard_rows.extend(self._pagination_rows("mg:edit", safe_page, total_pages))
keyboard_rows.append([InlineKeyboardButton("⬅️ В меню", callback_data="mg:menu")])
await self._edit(callback, "\n".join(lines), InlineKeyboardMarkup(keyboard_rows))
async def _render_stats_page(self, callback: CallbackQuery, page: int) -> None:
total = self.repository.count_questions()
total_pages = max(1, math.ceil(total / self.page_size))
safe_page = min(max(page, 1), total_pages)
offset = (safe_page - 1) * self.page_size
rows = self.repository.list_questions(limit=self.page_size, offset=offset)
if not rows:
await self._edit(callback, "Нет вопросов для статистики.", self._back_to_menu_keyboard())
return
lines = [f"Выберите вопрос для статистики ({safe_page}/{total_pages}):"]
keyboard_rows: list[list[InlineKeyboardButton]] = []
for item in rows:
short_text = item.text if len(item.text) <= 50 else f"{item.text[:47]}..."
lines.append(f"#{item.id}: {short_text}")
keyboard_rows.append(
[InlineKeyboardButton(f"📊 #{item.id}", callback_data=f"mg:statq:{item.id}")]
)
keyboard_rows.extend(self._pagination_rows("mg:stats", safe_page, total_pages))
keyboard_rows.append([InlineKeyboardButton("⬅️ В меню", callback_data="mg:menu")])
await self._edit(callback, "\n".join(lines), InlineKeyboardMarkup(keyboard_rows))
async def _render_question_stats(self, callback: CallbackQuery, question_id: int) -> None:
question = self.repository.get_question(question_id)
if question is None:
await self._edit(
callback,
"Вопрос не найден.",
InlineKeyboardMarkup([[InlineKeyboardButton("⬅️ В статистику", callback_data="mg:stats:1")]]),
)
return
stats = self.repository.get_question_statistics(question_id)
total_votes = sum(item.count for item in stats)
lines = [f"Статистика для вопроса #{question_id}:", question.question.text, ""]
if not stats:
lines.append("Нет вариантов ответа.")
else:
for item in stats:
percent = (item.count / total_votes * 100) if total_votes else 0.0
lines.append(f"{item.answer_text}: {item.count} ({percent:.1f}%)")
lines.append("")
lines.append(f"Всего голосов: {total_votes}")
keyboard = InlineKeyboardMarkup(
[
[InlineKeyboardButton("⬅️ К вопросам статистики", callback_data="mg:stats:1")],
[InlineKeyboardButton("⬅️ В меню", callback_data="mg:menu")],
]
)
await self._edit(callback, "\n".join(lines), keyboard)
async def _render_schedules(self, callback: CallbackQuery) -> None:
schedules = self.repository.list_schedules()
if not schedules:
text = "Расписаний пока нет."
else:
lines = ["Текущие расписания:"]
for schedule in schedules:
status = "🟢 active" if schedule.is_active else "⚪️ inactive"
lines.append(
f"#{schedule.id} [{status}] {schedule.name}: `{schedule.cron}`"
)
text = "\n".join(lines)
keyboard = InlineKeyboardMarkup(
[
[InlineKeyboardButton("➕ Добавить расписание", callback_data="mg:schedule:add")],
[InlineKeyboardButton("✏️ Изменить расписание", callback_data="mg:schedule:edit:list")],
[InlineKeyboardButton("🗑️ Удалить расписание", callback_data="mg:schedule:delete:list")],
[InlineKeyboardButton("⬅️ В меню", callback_data="mg:menu")],
]
)
await self._edit(callback, text, keyboard)
async def _render_schedule_edit_list(self, callback: CallbackQuery) -> None:
schedules = self.repository.list_schedules()
if not schedules:
await self._edit(
callback,
"Нет расписаний для изменения.",
InlineKeyboardMarkup(
[[InlineKeyboardButton("⬅️ К расписаниям", callback_data="mg:schedule:list")]]
),
)
return
lines = ["Выберите расписание для изменения:"]
keyboard_rows: list[list[InlineKeyboardButton]] = []
for schedule in schedules:
status = "🟢" if schedule.is_active else "⚪️"
lines.append(f"#{schedule.id} {schedule.name}: {schedule.cron} ({status})")
keyboard_rows.append(
[
InlineKeyboardButton(
f"✏️ #{schedule.id} {schedule.name}",
callback_data=f"mg:schedule:edit:{schedule.id}",
)
]
)
keyboard_rows.append([InlineKeyboardButton("⬅️ К расписаниям", callback_data="mg:schedule:list")])
keyboard_rows.append([InlineKeyboardButton("⬅️ В меню", callback_data="mg:menu")])
await self._edit(callback, "\n".join(lines), InlineKeyboardMarkup(keyboard_rows))
async def _render_schedule_delete_list(self, callback: CallbackQuery) -> None:
schedules = self.repository.list_schedules()
if not schedules:
await self._edit(
callback,
"Нет расписаний для удаления.",
InlineKeyboardMarkup(
[[InlineKeyboardButton("⬅️ К расписаниям", callback_data="mg:schedule:list")]]
),
)
return
lines = ["⚠️ Выберите расписание для удаления:"]
keyboard_rows: list[list[InlineKeyboardButton]] = []
for schedule in schedules:
status = "🟢" if schedule.is_active else "⚪️"
lines.append(f"#{schedule.id} {schedule.name}: {schedule.cron} ({status})")
keyboard_rows.append(
[
InlineKeyboardButton(
f"🗑️ #{schedule.id} {schedule.name}",
callback_data=f"mg:schedule:delete:{schedule.id}",
)
]
)
keyboard_rows.append([InlineKeyboardButton("⬅️ К расписаниям", callback_data="mg:schedule:list")])
keyboard_rows.append([InlineKeyboardButton("⬅️ В меню", callback_data="mg:menu")])
await self._edit(callback, "\n".join(lines), InlineKeyboardMarkup(keyboard_rows))
def _menu_keyboard(self) -> InlineKeyboardMarkup:
return InlineKeyboardMarkup(
[
[InlineKeyboardButton("📋 Просмотреть вопросы", callback_data="mg:view:1")],
[InlineKeyboardButton("➕ Добавить вопрос", callback_data="mg:add")],
[InlineKeyboardButton("✏️ Изменить вопрос", callback_data="mg:edit:1")],
[InlineKeyboardButton("📊 Смотреть статистику", callback_data="mg:stats:1")],
[InlineKeyboardButton("⏰ Расписания", callback_data="mg:schedule:list")],
]
)
def _back_to_menu_keyboard(self) -> InlineKeyboardMarkup:
return InlineKeyboardMarkup([[InlineKeyboardButton("⬅️ В меню", callback_data="mg:menu")]])
def _pagination_keyboard(self, prefix: str, page: int, total_pages: int) -> InlineKeyboardMarkup:
rows = self._pagination_rows(prefix, page, total_pages)
rows.append([InlineKeyboardButton("⬅️ В меню", callback_data="mg:menu")])
return InlineKeyboardMarkup(rows)
def _pagination_rows(self, prefix: str, page: int, total_pages: int) -> list[list[InlineKeyboardButton]]:
buttons: list[InlineKeyboardButton] = []
if page > 1:
buttons.append(InlineKeyboardButton("⬅️", callback_data=f"{prefix}:{page - 1}"))
if page < total_pages:
buttons.append(InlineKeyboardButton("➡️", callback_data=f"{prefix}:{page + 1}"))
if not buttons:
return []
return [buttons]
async def _edit(
self,
callback: CallbackQuery,
text: str,
reply_markup: InlineKeyboardMarkup,
) -> None:
if callback.message:
try:
await callback.message.edit_text(text=text, reply_markup=reply_markup)
except MessageNotModified:
return
def _parse_question_input(self, text: str) -> tuple[str, list[str]]:
parts = [part.strip() for part in text.split("|")]
if len(parts) < 3:
raise ValueError("Нужно указать вопрос и минимум 2 варианта ответа.")
question_text = parts[0]
answers = [part for part in parts[1:] if part]
if not question_text:
raise ValueError("Текст вопроса не может быть пустым.")
if len(answers) < 2:
raise ValueError("Нужно указать минимум 2 непустых варианта.")
return question_text, answers
def _parse_schedule_input(self, text: str) -> tuple[str, str, bool]:
parts = [part.strip() for part in text.split("|")]
if len(parts) != 3:
raise ValueError("Нужно указать имя, cron и active-флаг (1 или 0).")
name, cron, active_token = parts
if not name:
raise ValueError("Имя расписания не может быть пустым.")
validate_cron_expression(cron)
if active_token not in ("0", "1"):
raise ValueError("active-флаг должен быть 1 (активно) или 0 (выключено).")
return name, cron, active_token == "1"
def _extract_page(self, data: str) -> int:
try:
return max(1, int(data.rsplit(":", 1)[1]))
except (IndexError, ValueError):
return 1
def _extract_id(self, data: str) -> int:
try:
value = int(data.rsplit(":", 1)[1])
except (IndexError, ValueError) as error:
raise ValueError("Invalid callback payload.") from error
return value

4
bot/manage/__init__.py Normal file
View File

@@ -0,0 +1,4 @@
from .controller import ManageController
from .state import PendingManageAction
__all__ = ["ManageController", "PendingManageAction"]

138
bot/manage/api.py Normal file
View File

@@ -0,0 +1,138 @@
from __future__ import annotations
from pyrogram import Client
from pyrogram.types import CallbackQuery, InlineKeyboardButton, InlineKeyboardMarkup
from game import SQLiteWouldYouRatherRepository
from game.api_service import QuestionApiService, parse_wyr_app_payload
from .shared import edit_message, menu_keyboard
from .state import PendingManageAction
class ApiMenu:
def __init__(
self,
repository: SQLiteWouldYouRatherRepository,
pending_actions: dict[int, PendingManageAction],
) -> None:
self.repository = repository
self.pending_actions = pending_actions
async def handle_callback(self, client: Client, callback: CallbackQuery, data: str, user_id: int) -> bool:
"""Handle API-related callbacks. Returns True if handled."""
if data == "mg:api:load":
self.pending_actions.pop(user_id, None)
await self._handle_api_load(callback, user_id)
return True
if data == "mg:api:next":
self.pending_actions.pop(user_id, None)
offset = int(self.repository.get_setting("api_offset", "0"))
self.repository.set_setting("api_offset", str(offset + 5))
await self._handle_api_load(callback, user_id)
return True
if data.startswith("mg:api:add:"):
index = int(data.split(":")[-1])
await self._handle_api_add(callback, user_id, index)
await callback.answer()
return True
if data == "mg:api:done":
offset = int(self.repository.get_setting("api_offset", "0"))
self.repository.set_setting("api_offset", str(offset + 5))
self.pending_actions.pop(user_id, None)
await edit_message(callback, "✅ Загрузка завершена.", menu_keyboard())
await callback.answer()
return True
return False
async def _handle_api_load(self, callback: CallbackQuery, user_id: int) -> None:
offset_str = self.repository.get_setting("api_offset", "0")
offset = int(offset_str)
page_size = 5
api_url = f"https://io.wyr.app/api/v1/statements/ru/daily?pageOffset={offset}&pageSize={page_size}"
service = QuestionApiService(url=api_url, handler=parse_wyr_app_payload)
try:
questions = service.fetch_questions()
if not questions:
await callback.answer("API не вернуло вопросов.", show_alert=True)
return
self.pending_actions[user_id] = PendingManageAction(
mode="api_select",
api_questions=questions,
added_indices=set()
)
await self._render_api_selection(callback, questions)
await callback.answer()
except Exception as e:
await callback.answer(f"Ошибка загрузки: {str(e)}", show_alert=True)
async def _render_api_selection(self, callback: CallbackQuery, questions: list) -> None:
user_id = callback.from_user.id if callback.from_user else None
action = self.pending_actions.get(user_id) if user_id else None
added_indices = action.added_indices if action else set()
text = "🌐 **Выберите вопросы для добавления в базу:**\n\n"
keyboard_rows = []
for i, q in enumerate(questions):
is_added = i in added_indices
status = "✅ [ДОБАВЛЕНО] " if is_added else ""
answers_text = "\n ".join([f"🔹 {a}" for a in q.answers])
text += f"{i+1}. {status}{q.text}\n {answers_text}\n\n"
if is_added:
keyboard_rows.append([InlineKeyboardButton(f"➖ Удалить #{i+1}", callback_data=f"mg:api:add:{i}")])
else:
keyboard_rows.append([InlineKeyboardButton(f"➕ Добавить #{i+1}", callback_data=f"mg:api:add:{i}")])
keyboard_rows.append([InlineKeyboardButton("➡️ Загрузить еще (пропустить эти)", callback_data="mg:api:next")])
keyboard_rows.append([InlineKeyboardButton("✅ Готово", callback_data="mg:api:done")])
await edit_message(callback, text, InlineKeyboardMarkup(keyboard_rows))
async def _handle_api_add(self, callback: CallbackQuery, user_id: int, index: int) -> None:
action = self.pending_actions.get(user_id)
if not action or action.mode != "api_select" or not action.api_questions:
await callback.answer("Сессия истекла.", show_alert=True)
return
if index < 0 or index >= len(action.api_questions):
await callback.answer("Некорректный индекс.", show_alert=True)
return
if index in action.added_indices:
question = action.api_questions[index]
try:
if self.repository.delete_question_by_text(question.text):
action.added_indices.remove(index)
await callback.answer(f"Вопрос #{index+1} удалён!")
await self._render_api_selection(callback, action.api_questions)
else:
await callback.answer("Не удалось найти вопрос для удаления.", show_alert=True)
except Exception as e:
await callback.answer(f"Ошибка удаления: {str(e)}", show_alert=True)
return
question = action.api_questions[index]
try:
self.repository.add_question(question.text, question.answers)
action.added_indices.add(index)
await callback.answer(f"Вопрос #{index+1} добавлен!")
await self._render_api_selection(callback, action.api_questions)
except Exception as e:
if "UNIQUE constraint failed" in str(e):
action.added_indices.add(index)
await callback.answer("Этот вопрос уже есть в базе.", show_alert=True)
await self._render_api_selection(callback, action.api_questions)
else:
await callback.answer(f"Ошибка: {str(e)}", show_alert=True)

99
bot/manage/controller.py Normal file
View File

@@ -0,0 +1,99 @@
from __future__ import annotations
from pyrogram import Client, filters
from pyrogram.handlers import CallbackQueryHandler, MessageHandler
from pyrogram.types import Message
from game import SQLiteWouldYouRatherRepository
from bot.command_system import CommandContext
from .api import ApiMenu
from .questions import QuestionMenu
from .schedules import ScheduleMenu
from .shared import edit_message, menu_keyboard
from .state import PendingManageAction
from .statistics import StatisticsMenu
class ManageController:
def __init__(
self,
context: CommandContext,
page_size: int = 5,
repository: SQLiteWouldYouRatherRepository | None = None,
) -> None:
self.context = context
self.page_size = page_size
self.repository = repository or SQLiteWouldYouRatherRepository()
self.pending_actions: dict[int, PendingManageAction] = {}
# Initialize menu handlers
self.question_menu = QuestionMenu(self.repository, self.pending_actions, page_size)
self.statistics_menu = StatisticsMenu(self.repository, self.pending_actions, page_size)
self.schedule_menu = ScheduleMenu(self.repository, self.pending_actions)
self.api_menu = ApiMenu(self.repository, self.pending_actions)
def register(self, client: Client) -> None:
client.add_handler(
MessageHandler(self._on_manage_command, filters.command(["manage"]))
)
client.add_handler(
CallbackQueryHandler(self._on_callback_query, filters.regex(r"^mg:"))
)
client.add_handler(
MessageHandler(self._on_text_message, filters.text & ~filters.regex(r"^/"))
)
async def _on_manage_command(self, _: Client, message: Message) -> None:
user_id = message.from_user.id if message.from_user else None
if user_id is None or user_id not in self.context.admin_ids:
await message.reply_text("Команда доступна только администраторам.")
return
await message.reply_text(
"Панель управления вопросами:",
reply_markup=menu_keyboard(),
)
async def _on_callback_query(self, client: Client, callback) -> None:
user_id = callback.from_user.id if callback.from_user else None
if user_id is None or user_id not in self.context.admin_ids:
await callback.answer("Только для администраторов.", show_alert=True)
return
data = callback.data or ""
# Main menu
if data == "mg:menu":
self.pending_actions.pop(user_id, None)
await edit_message(callback, "Панель управления вопросами:", menu_keyboard())
await callback.answer()
return
# Delegate to menu handlers
if await self.question_menu.handle_callback(client, callback, data, user_id):
return
if await self.statistics_menu.handle_callback(client, callback, data, user_id):
return
if await self.schedule_menu.handle_callback(client, callback, data, user_id):
return
if await self.api_menu.handle_callback(client, callback, data, user_id):
return
await callback.answer("Неизвестное действие.")
async def _on_text_message(self, client: Client, message: Message) -> None:
user_id = message.from_user.id if message.from_user else None
if user_id is None or user_id not in self.context.admin_ids:
return
# Delegate to menu handlers
if await self.question_menu.handle_text_message(message, user_id):
return
if await self.schedule_menu.handle_text_message(message, user_id):
return

175
bot/manage/questions.py Normal file
View File

@@ -0,0 +1,175 @@
from __future__ import annotations
import math
from pyrogram import Client
from pyrogram.types import CallbackQuery, InlineKeyboardButton, InlineKeyboardMarkup, Message
from game import SQLiteWouldYouRatherRepository
from .shared import (
back_to_menu_keyboard,
edit_message,
extract_id,
extract_page,
menu_keyboard,
pagination_keyboard,
parse_question_input,
)
from .state import PendingManageAction
class QuestionMenu:
def __init__(
self,
repository: SQLiteWouldYouRatherRepository,
pending_actions: dict[int, PendingManageAction],
page_size: int = 5,
) -> None:
self.repository = repository
self.pending_actions = pending_actions
self.page_size = page_size
async def handle_callback(self, client: Client, callback: CallbackQuery, data: str, user_id: int) -> bool:
"""Handle question-related callbacks. Returns True if handled."""
if data.startswith("mg:view:"):
page = extract_page(data)
await self._render_questions_page(callback, page)
await callback.answer()
return True
if data == "mg:add":
self.pending_actions.pop(user_id, None)
self.pending_actions[user_id] = PendingManageAction(mode="add")
await edit_message(
callback,
"Отправьте новый вопрос в формате:\n"
"<вопрос> | <вариант 1> | <вариант 2> [| <вариант 3> ...]",
back_to_menu_keyboard(),
)
await callback.answer()
return True
if data.startswith("mg:edit:"):
page = extract_page(data)
await self._render_edit_page(callback, page)
await callback.answer()
return True
if data.startswith("mg:editq:"):
try:
question_id = extract_id(data)
except ValueError:
await callback.answer("Некорректный идентификатор.", show_alert=True)
return True
self.pending_actions[user_id] = PendingManageAction(mode="edit", question_id=question_id)
await edit_message(
callback,
"Отправьте обновлённый вопрос в формате:\n"
"<вопрос> | <вариант 1> | <вариант 2> [| <вариант 3> ...]",
back_to_menu_keyboard(),
)
await callback.answer()
return True
return False
async def handle_text_message(self, message: Message, user_id: int) -> bool:
"""Handle text messages for question actions. Returns True if handled."""
action = self.pending_actions.get(user_id)
if action is None:
return False
text = message.text or ""
if action.mode == "add":
try:
question_text, answers = parse_question_input(text)
question_id = self.repository.add_question(question_text, answers)
self.pending_actions.pop(user_id, None)
await message.reply_text(
f"Вопрос добавлен (ID: {question_id}).",
reply_markup=menu_keyboard(),
)
except ValueError as error:
await message.reply_text(
f"{error}\n\nИспользуйте формат:\n"
"<вопрос> | <вариант 1> | <вариант 2> [| <вариант 3> ...]"
)
return True
if action.mode == "edit":
if action.question_id is None:
await message.reply_text("Ошибка: ID вопроса не найден.")
return True
try:
question_text, answers = parse_question_input(text)
self.repository.update_question(action.question_id, question_text, answers)
self.pending_actions.pop(user_id, None)
await message.reply_text(
f"Вопрос #{action.question_id} обновлён.",
reply_markup=menu_keyboard(),
)
except ValueError as error:
await message.reply_text(
f"{error}\n\nИспользуйте формат:\n"
"<вопрос> | <вариант 1> | <вариант 2> [| <вариант 3> ...]"
)
return True
return False
async def _render_questions_page(self, callback: CallbackQuery, page: int) -> None:
total = self.repository.count_questions(unused_only=True)
total_pages = max(1, math.ceil(total / self.page_size))
safe_page = min(max(page, 1), total_pages)
offset = (safe_page - 1) * self.page_size
rows = self.repository.list_questions(
limit=self.page_size, offset=offset, unused_only=True
)
if not rows:
text = "В очереди пока нет вопросов."
else:
lines = [
f"В очереди: {total} вопросов",
f"Список (страница {safe_page}/{total_pages}):",
"",
]
for item in rows:
short_text = (
item.text if len(item.text) <= 100 else f"{item.text[:97]}..."
)
full_question = self.repository.get_question(item.id)
answers_text = ", ".join([a.text for a in full_question.answers]) if full_question else "нет вариантов"
lines.append(f"#{item.id}: {short_text}\n 🔹 {answers_text}\n")
text = "\n".join(lines)
await edit_message(callback, text, pagination_keyboard("mg:view", safe_page, total_pages))
async def _render_edit_page(self, callback: CallbackQuery, page: int) -> None:
total = self.repository.count_questions()
total_pages = max(1, math.ceil(total / self.page_size))
safe_page = min(max(page, 1), total_pages)
offset = (safe_page - 1) * self.page_size
rows = self.repository.list_questions(limit=self.page_size, offset=offset)
if not rows:
await edit_message(callback, "Нет вопросов для изменения.", back_to_menu_keyboard())
return
lines = [f"Выберите вопрос для изменения ({safe_page}/{total_pages}):"]
keyboard_rows: list[list[InlineKeyboardButton]] = []
for item in rows:
short_text = item.text if len(item.text) <= 50 else f"{item.text[:47]}..."
lines.append(f"#{item.id}: {short_text}")
keyboard_rows.append(
[InlineKeyboardButton(f"✏️ #{item.id}", callback_data=f"mg:editq:{item.id}")]
)
from .shared import pagination_rows
keyboard_rows.extend(pagination_rows("mg:edit", safe_page, total_pages))
keyboard_rows.append([InlineKeyboardButton("⬅️ В меню", callback_data="mg:menu")])
await edit_message(callback, "\n".join(lines), InlineKeyboardMarkup(keyboard_rows))

269
bot/manage/schedules.py Normal file
View File

@@ -0,0 +1,269 @@
from __future__ import annotations
from pyrogram import Client
from pyrogram.types import CallbackQuery, InlineKeyboardButton, InlineKeyboardMarkup, Message
from game import SQLiteWouldYouRatherRepository
from .shared import (
back_to_menu_keyboard,
edit_message,
extract_id,
menu_keyboard,
parse_schedule_input,
)
from .state import PendingManageAction
class ScheduleMenu:
def __init__(
self,
repository: SQLiteWouldYouRatherRepository,
pending_actions: dict[int, PendingManageAction],
) -> None:
self.repository = repository
self.pending_actions = pending_actions
async def handle_callback(self, client: Client, callback: CallbackQuery, data: str, user_id: int) -> bool:
"""Handle schedule-related callbacks. Returns True if handled."""
if data == "mg:schedule:help":
await self._render_help(callback)
await callback.answer()
return True
if data == "mg:schedule:list":
await self._render_schedules(callback)
await callback.answer()
return True
if data == "mg:schedule:add":
self.pending_actions.pop(user_id, None)
self.pending_actions[user_id] = PendingManageAction(mode="schedule_add")
await edit_message(
callback,
"Отправьте расписание в формате:\n"
"<название> | <cron 5 полей> | <active: 1 или 0>\n"
"Пример: Утро | 0 9 * * 1-5 | 1",
back_to_menu_keyboard(),
)
await callback.answer()
return True
if data == "mg:schedule:edit:list":
await self._render_schedule_edit_list(callback)
await callback.answer()
return True
if data.startswith("mg:schedule:edit:"):
try:
schedule_id = extract_id(data)
except ValueError:
await callback.answer("Некорректный идентификатор.", show_alert=True)
return True
schedule = self.repository.get_schedule(schedule_id)
if schedule is None:
await callback.answer("Расписание не найдено.", show_alert=True)
return True
self.pending_actions[user_id] = PendingManageAction(
mode="schedule_edit",
schedule_id=schedule_id,
)
await edit_message(
callback,
"Отправьте обновлённое расписание в формате:\n"
"<название> | <cron 5 полей> | <active: 1 или 0>\n"
f"Текущее: {schedule.name} | {schedule.cron} | {1 if schedule.is_active else 0}",
back_to_menu_keyboard(),
)
await callback.answer()
return True
if data == "mg:schedule:delete:list":
await self._render_schedule_delete_list(callback)
await callback.answer()
return True
if data.startswith("mg:schedule:delete:"):
try:
schedule_id = extract_id(data)
except ValueError:
await callback.answer("Некорректный идентификатор.", show_alert=True)
return True
try:
schedule = self.repository.get_schedule(schedule_id)
if schedule is None:
await callback.answer("Расписание не найдено.", show_alert=True)
return True
self.repository.delete_schedule(schedule_id)
await callback.answer(f"Расписание '{schedule.name}' удалено.", show_alert=True)
await self._render_schedule_delete_list(callback)
except ValueError as error:
await callback.answer(str(error), show_alert=True)
return True
return False
async def handle_text_message(self, message: Message, user_id: int) -> bool:
"""Handle text messages for schedule actions. Returns True if handled."""
action = self.pending_actions.get(user_id)
if action is None:
return False
text = message.text or ""
if action.mode == "schedule_add":
try:
name, cron, is_active = parse_schedule_input(text)
self.repository.add_schedule(name=name, cron=cron, is_active=is_active)
self.pending_actions.pop(user_id, None)
await message.reply_text(
"Расписание добавлено.",
reply_markup=menu_keyboard(),
)
except ValueError as error:
await message.reply_text(
f"{error}\n\nИспользуйте формат:\n"
"<название> | <cron 5 полей> | <active: 1 или 0>"
)
return True
if action.mode == "schedule_edit":
if action.schedule_id is None:
await message.reply_text("Ошибка: ID расписания не найден.")
return True
try:
name, cron, is_active = parse_schedule_input(text)
self.repository.update_schedule(
schedule_id=action.schedule_id,
name=name,
cron=cron,
is_active=is_active,
)
self.pending_actions.pop(user_id, None)
await message.reply_text(
f"Расписание #{action.schedule_id} обновлено.",
reply_markup=menu_keyboard(),
)
except ValueError as error:
await message.reply_text(
f"{error}\n\nИспользуйте формат:\n"
"<название> | <cron 5 полей> | <active: 1 или 0>"
)
return True
return False
async def _render_schedules(self, callback: CallbackQuery) -> None:
schedules = self.repository.list_schedules()
if not schedules:
text = "Расписаний пока нет."
else:
lines = ["Текущие расписания:"]
for schedule in schedules:
status = "🟢 active" if schedule.is_active else "⚪️ inactive"
lines.append(
f"#{schedule.id} [{status}] {schedule.name}: `{schedule.cron}`"
)
text = "\n".join(lines)
keyboard = InlineKeyboardMarkup(
[
[InlineKeyboardButton("➕ Добавить расписание", callback_data="mg:schedule:add")],
[InlineKeyboardButton("✏️ Изменить расписание", callback_data="mg:schedule:edit:list")],
[InlineKeyboardButton("🗑️ Удалить расписание", callback_data="mg:schedule:delete:list")],
[InlineKeyboardButton("❓ Помощь", callback_data="mg:schedule:help")],
[InlineKeyboardButton("⬅️ В меню", callback_data="mg:menu")],
]
)
await edit_message(callback, text, keyboard)
async def _render_schedule_edit_list(self, callback: CallbackQuery) -> None:
schedules = self.repository.list_schedules()
if not schedules:
await edit_message(
callback,
"Нет расписаний для изменения.",
InlineKeyboardMarkup(
[[InlineKeyboardButton("⬅️ К расписаниям", callback_data="mg:schedule:list")]]
),
)
return
lines = ["Выберите расписание для изменения:"]
keyboard_rows: list[list[InlineKeyboardButton]] = []
for schedule in schedules:
status = "🟢" if schedule.is_active else "⚪️"
lines.append(f"#{schedule.id} {schedule.name}: {schedule.cron} ({status})")
keyboard_rows.append(
[
InlineKeyboardButton(
f"✏️ #{schedule.id} {schedule.name}",
callback_data=f"mg:schedule:edit:{schedule.id}",
)
]
)
keyboard_rows.append([InlineKeyboardButton("⬅️ К расписаниям", callback_data="mg:schedule:list")])
keyboard_rows.append([InlineKeyboardButton("⬅️ В меню", callback_data="mg:menu")])
await edit_message(callback, "\n".join(lines), InlineKeyboardMarkup(keyboard_rows))
async def _render_schedule_delete_list(self, callback: CallbackQuery) -> None:
schedules = self.repository.list_schedules()
if not schedules:
await edit_message(
callback,
"Нет расписаний для удаления.",
InlineKeyboardMarkup(
[[InlineKeyboardButton("⬅️ К расписаниям", callback_data="mg:schedule:list")]]
),
)
return
lines = ["⚠️ Выберите расписание для удаления:"]
keyboard_rows: list[list[InlineKeyboardButton]] = []
for schedule in schedules:
status = "🟢" if schedule.is_active else "⚪️"
lines.append(f"#{schedule.id} {schedule.name}: {schedule.cron} ({status})")
keyboard_rows.append(
[
InlineKeyboardButton(
f"🗑️ #{schedule.id} {schedule.name}",
callback_data=f"mg:schedule:delete:{schedule.id}",
)
]
)
keyboard_rows.append([InlineKeyboardButton("⬅️ К расписаниям", callback_data="mg:schedule:list")])
keyboard_rows.append([InlineKeyboardButton("⬅️ В меню", callback_data="mg:menu")])
await edit_message(callback, "\n".join(lines), InlineKeyboardMarkup(keyboard_rows))
async def _render_help(self, callback: CallbackQuery) -> None:
help_text = (
"📖 <b>Помощь по расписаниям</b>\n\n"
"<b>Формат ввода:</b>\n"
"название | cron | active\n\n"
"<b>Cron-формат (5 полей, UTC):</b>\n"
"минута час день месяц день_недели\n\n"
"<b>Примеры:</b>\n"
"• <code>0 9 * * 1-5</code> — каждый рабочий день в 09:00 UTC\n"
"• <code>30 12 * * *</code> — каждый день в 12:30 UTC\n"
"• <code>0 18 * * 0</code> — каждое воскресенье в 18:00 UTC\n"
"• <code>*/15 * * * *</code> — каждые 15 минут\n"
"• <code>0 9,18 * * *</code> — дважды в день: 09:00 и 18:00\n\n"
"<b>День недели:</b> 0 и 7 = воскресенье, 1 = понедельник\n\n"
"<b>Active-флаг:</b>\n"
"• 1 — расписание активно (вопросы публикуются)\n"
"• 0 — расписание выключено"
)
keyboard = InlineKeyboardMarkup(
[
[InlineKeyboardButton("⬅️ К расписаниям", callback_data="mg:schedule:list")],
[InlineKeyboardButton("⬅️ В меню", callback_data="mg:menu")],
]
)
await edit_message(callback, help_text, keyboard)

94
bot/manage/shared.py Normal file
View File

@@ -0,0 +1,94 @@
from __future__ import annotations
from pyrogram.errors import MessageNotModified
from pyrogram.types import CallbackQuery, InlineKeyboardButton, InlineKeyboardMarkup
from bot.scheduler import validate_cron_expression
def menu_keyboard() -> InlineKeyboardMarkup:
return InlineKeyboardMarkup(
[
[InlineKeyboardButton("📋 Просмотреть вопросы", callback_data="mg:view:1")],
[InlineKeyboardButton("➕ Добавить вопрос", callback_data="mg:add")],
[InlineKeyboardButton("🌐 Загрузить из API", callback_data="mg:api:load")],
[InlineKeyboardButton("✏️ Изменить вопрос", callback_data="mg:edit:1")],
[InlineKeyboardButton("📊 Смотреть статистику", callback_data="mg:stats:1")],
[InlineKeyboardButton("⏰ Расписания", callback_data="mg:schedule:list")],
]
)
def back_to_menu_keyboard() -> InlineKeyboardMarkup:
return InlineKeyboardMarkup([[InlineKeyboardButton("⬅️ В меню", callback_data="mg:menu")]])
def pagination_keyboard(prefix: str, page: int, total_pages: int) -> InlineKeyboardMarkup:
rows = pagination_rows(prefix, page, total_pages)
rows.append([InlineKeyboardButton("⬅️ В меню", callback_data="mg:menu")])
return InlineKeyboardMarkup(rows)
def pagination_rows(prefix: str, page: int, total_pages: int) -> list[list[InlineKeyboardButton]]:
buttons: list[InlineKeyboardButton] = []
if page > 1:
buttons.append(InlineKeyboardButton("⬅️", callback_data=f"{prefix}:{page - 1}"))
if page < total_pages:
buttons.append(InlineKeyboardButton("➡️", callback_data=f"{prefix}:{page + 1}"))
if not buttons:
return []
return [buttons]
async def edit_message(
callback: CallbackQuery,
text: str,
reply_markup: InlineKeyboardMarkup,
) -> None:
if callback.message:
try:
await callback.message.edit_text(text=text, reply_markup=reply_markup)
except MessageNotModified:
return
def parse_question_input(text: str) -> tuple[str, list[str]]:
parts = [part.strip() for part in text.split("|")]
if len(parts) < 3:
raise ValueError("Нужно указать вопрос и минимум 2 варианта ответа.")
question_text = parts[0]
answers = [part for part in parts[1:] if part]
if not question_text:
raise ValueError("Текст вопроса не может быть пустым.")
if len(answers) < 2:
raise ValueError("Нужно указать минимум 2 непустых варианта.")
return question_text, answers
def parse_schedule_input(text: str) -> tuple[str, str, bool]:
parts = [part.strip() for part in text.split("|")]
if len(parts) != 3:
raise ValueError("Нужно указать имя, cron и active-флаг (1 или 0).")
name, cron, active_token = parts
if not name:
raise ValueError("Имя расписания не может быть пустым.")
validate_cron_expression(cron)
if active_token not in ("0", "1"):
raise ValueError("active-флаг должен быть 1 (активно) или 0 (выключено).")
return name, cron, active_token == "1"
def extract_page(data: str) -> int:
try:
return max(1, int(data.rsplit(":", 1)[1]))
except (IndexError, ValueError):
return 1
def extract_id(data: str) -> int:
try:
value = int(data.rsplit(":", 1)[1])
except (IndexError, ValueError) as error:
raise ValueError("Invalid callback payload.") from error
return value

14
bot/manage/state.py Normal file
View File

@@ -0,0 +1,14 @@
from __future__ import annotations
from dataclasses import dataclass, field
from game.api_service import ApiQuestion
@dataclass
class PendingManageAction:
mode: str
question_id: int | None = None
schedule_id: int | None = None
api_questions: list[ApiQuestion] | None = None
added_indices: set[int] = field(default_factory=set)

104
bot/manage/statistics.py Normal file
View File

@@ -0,0 +1,104 @@
from __future__ import annotations
import math
from pyrogram import Client
from pyrogram.types import CallbackQuery, InlineKeyboardButton, InlineKeyboardMarkup
from game import SQLiteWouldYouRatherRepository
from .shared import (
back_to_menu_keyboard,
edit_message,
extract_id,
extract_page,
pagination_rows,
)
from .state import PendingManageAction
class StatisticsMenu:
def __init__(
self,
repository: SQLiteWouldYouRatherRepository,
pending_actions: dict[int, PendingManageAction],
page_size: int = 5,
) -> None:
self.repository = repository
self.pending_actions = pending_actions
self.page_size = page_size
async def handle_callback(self, client: Client, callback: CallbackQuery, data: str, user_id: int) -> bool:
"""Handle statistics-related callbacks. Returns True if handled."""
if data.startswith("mg:stats:"):
page = extract_page(data)
await self._render_stats_page(callback, page)
await callback.answer()
return True
if data.startswith("mg:statq:"):
try:
question_id = extract_id(data)
except ValueError:
await callback.answer("Некорректный идентификатор.", show_alert=True)
return True
await self._render_question_stats(callback, question_id)
await callback.answer()
return True
return False
async def _render_stats_page(self, callback: CallbackQuery, page: int) -> None:
total = self.repository.count_questions()
total_pages = max(1, math.ceil(total / self.page_size))
safe_page = min(max(page, 1), total_pages)
offset = (safe_page - 1) * self.page_size
rows = self.repository.list_questions(limit=self.page_size, offset=offset)
if not rows:
await edit_message(callback, "Нет вопросов для статистики.", back_to_menu_keyboard())
return
lines = [f"Выберите вопрос для статистики ({safe_page}/{total_pages}):"]
keyboard_rows: list[list[InlineKeyboardButton]] = []
for item in rows:
short_text = item.text if len(item.text) <= 50 else f"{item.text[:47]}..."
lines.append(f"#{item.id}: {short_text}")
keyboard_rows.append(
[InlineKeyboardButton(f"📊 #{item.id}", callback_data=f"mg:statq:{item.id}")]
)
keyboard_rows.extend(pagination_rows("mg:stats", safe_page, total_pages))
keyboard_rows.append([InlineKeyboardButton("⬅️ В меню", callback_data="mg:menu")])
await edit_message(callback, "\n".join(lines), InlineKeyboardMarkup(keyboard_rows))
async def _render_question_stats(self, callback: CallbackQuery, question_id: int) -> None:
question = self.repository.get_question(question_id)
if question is None:
await edit_message(
callback,
"Вопрос не найден.",
InlineKeyboardMarkup([[InlineKeyboardButton("⬅️ В статистику", callback_data="mg:stats:1")]]),
)
return
stats = self.repository.get_question_statistics(question_id)
total_votes = sum(item.count for item in stats)
lines = [f"Статистика для вопроса #{question_id}:", question.question.text, ""]
if not stats:
lines.append("Нет вариантов ответа.")
else:
for item in stats:
percent = (item.count / total_votes * 100) if total_votes else 0.0
lines.append(f"{item.answer_text}: {item.count} ({percent:.1f}%)")
lines.append("")
lines.append(f"Всего голосов: {total_votes}")
keyboard = InlineKeyboardMarkup(
[
[InlineKeyboardButton("⬅️ К вопросам статистики", callback_data="mg:stats:1")],
[InlineKeyboardButton("⬅️ В меню", callback_data="mg:menu")],
]
)
await edit_message(callback, "\n".join(lines), keyboard)

84
bot/poll_handler.py
View File

@@ -2,8 +2,9 @@ from __future__ import annotations
import logging
from pyrogram import Client, raw
from pyrogram.handlers import RawUpdateHandler
from pyrogram import Client
from pyrogram.handlers import PollHandler
from pyrogram.types import Poll
from game import SQLiteWouldYouRatherRepository
@@ -11,30 +12,36 @@ logger = logging.getLogger(__name__)
class PollStatisticsHandler:
"""Обработчик ответов на опросы для записи статистики."""
"""
Обработчик обновлений опросов для записи статистики.
Для анонимных опросов Telegram не отправляет индивидуальные голоса,
только агрегированные результаты. Мы отслеживаем изменения в количестве
голосов и записываем дельту.
"""
def __init__(self, repository: SQLiteWouldYouRatherRepository | None = None) -> None:
self.repository = repository or SQLiteWouldYouRatherRepository()
# Кэш последних известных результатов для вычисления дельты
# poll_id -> {option_index: voter_count}
self._last_results: dict[str, dict[int, int]] = {}
def register(self, client: Client) -> None:
"""Регистрирует обработчик в клиенте Pyrogram."""
client.add_handler(RawUpdateHandler(self._on_raw_update))
client.add_handler(PollHandler(self._on_poll_update))
logger.info("Poll statistics handler registered.")
async def _on_raw_update(self, client: Client, update, users, chats) -> None:
"""Обрабатывает raw update от Telegram."""
# Проверяем, является ли это обновлением голосования в опросе
if not isinstance(update, raw.types.UpdateMessagePollVote):
return
async def _on_poll_update(self, client: Client, poll: Poll) -> None:
"""
Обрабатывает обновление опроса (изменение результатов).
Telegram отправляет это событие каждый раз, когда кто-то голосует
в анонимном опросе, но содержит только агрегированные результаты.
"""
try:
await self._process_poll_vote(update)
except Exception:
logger.exception("Failed to process poll vote: %s", update)
async def _process_poll_vote(self, update: raw.types.UpdateMessagePollVote) -> None:
"""Обрабатывает голос пользователя в опросе."""
poll_id = str(update.poll_id)
poll_id = poll.id
if not poll_id:
return
# Получаем question_id по poll_id
question_id = self.repository.get_question_id_by_poll(poll_id)
@@ -42,28 +49,20 @@ class PollStatisticsHandler:
logger.debug("Poll %s not found in mappings, skipping.", poll_id)
return
# Обрабатываем выбранные опции (каждая опция - это bytes объект)
option_bytes_list = update.options or []
# Получаем текущие результаты
current_results = {}
for idx, option in enumerate(poll.options):
current_results[idx] = option.voter_count
# Получаем user_id из peer
peer = update.peer
if isinstance(peer, raw.types.PeerUser):
user_id = peer.user_id
else:
user_id = "unknown"
# Получаем предыдущие результаты или инициализируем нулями
previous_results = self._last_results.get(poll_id, {})
for option_data in option_bytes_list:
# В Telegram API каждая опция - это один байт с индексом
# option_data это bytes объект, берём первый байт как индекс
try:
if isinstance(option_data, bytes):
option_index = option_data[0] if len(option_data) > 0 else 0
else:
option_index = int(option_data)
except (ValueError, TypeError, IndexError) as e:
logger.warning("Invalid option data: %s (%s)", option_data, e)
continue
# Вычисляем и записываем изменения
for option_index, current_count in current_results.items():
previous_count = previous_results.get(option_index, 0)
delta = current_count - previous_count
if delta > 0:
# Получаем answer_id по индексу опции
answer_id = self.repository.get_answer_id_by_option_index(
question_id, option_index
@@ -76,12 +75,19 @@ class PollStatisticsHandler:
)
continue
# Записываем статистику
self.repository.record_answer(question_id, answer_id, increment=1)
# Записываем дельту в статистику
self.repository.record_answer(question_id, answer_id, increment=delta)
logger.debug(
"Recorded answer: question=%s, answer=%s, option=%s, user=%s",
"Recorded %d new vote(s): question=%s, answer=%s, option=%s (total: %d)",
delta,
question_id,
answer_id,
option_index,
user_id,
current_count,
)
# Сохраняем текущие результаты для следующего сравнения
self._last_results[poll_id] = current_results
except Exception:
logger.exception("Failed to process poll update: %s", poll)

12
bot/scheduler.py
View File

@@ -89,10 +89,12 @@ class QuestionPublisherScheduler:
client: Client,
repository: SQLiteWouldYouRatherRepository,
channel_ids: list[int],
admin_ids: set[int] | None = None,
) -> None:
self.client = client
self.repository = repository
self.channel_ids = channel_ids
self.admin_ids = admin_ids or set()
self._task: asyncio.Task[None] | None = None
def start(self) -> None:
@@ -187,3 +189,13 @@ class QuestionPublisherScheduler:
channel_id,
schedule_name,
)
# Проверяем количество оставшихся вопросов
unused_count = self.repository.count_questions(unused_only=True)
if unused_count == 2:
admin_message = "⚠️ В базе осталось всего 2 неиспользованных вопроса. Пожалуйста, пополните базу!"
for admin_id in self.admin_ids:
try:
await self.client.send_message(admin_id, admin_message)
except Exception:
logger.exception("Failed to notify admin %s about low questions.", admin_id)

65
game/api_service.py
View File

@@ -31,6 +31,8 @@ def parse_default_question_payload(payload: Any) -> ApiQuestion:
return parse_default_question_payload(payload["data"])
question_text = _extract_question_text(payload)
if not question_text.endswith("?"):
question_text += "?"
answers = _extract_answers(payload)
if len(answers) < 2:
raise ValueError("Question must have at least 2 answers.")
@@ -38,9 +40,44 @@ def parse_default_question_payload(payload: Any) -> ApiQuestion:
return ApiQuestion(text=question_text, answers=answers)
def parse_wyr_app_payload(payload: Any) -> list[ApiQuestion]:
if not isinstance(payload, dict):
raise ValueError("API response must be a JSON object.")
statements = payload.get("statements", [])
if not isinstance(statements, list):
return []
questions = []
for item in statements:
phrase = item.get("phrase")
if not isinstance(phrase, list) or len(phrase) < 2:
continue
answers = [p.get("text", "") for p in phrase if p.get("text")]
if not answers:
continue
text = item.get("title")
if not text:
# Если заголовка нет, пробуем составить его из ответов для уникальности
text = " ИЛИ ".join(answers)
# Добавляем "?" в конец, если его нет
text = text.strip()
if not text.endswith("?"):
text += "?"
if len(answers) >= 2:
questions.append(ApiQuestion(text=text, answers=answers))
return questions
def build_handler_registry() -> dict[str, QuestionPayloadHandler]:
return {
"default": parse_default_question_payload,
"wyr_app": parse_wyr_app_payload,
}
@@ -55,18 +92,36 @@ class QuestionApiService:
self.url = url
self.handler = handler or parse_default_question_payload
self.timeout = timeout
self.headers = headers or {"Accept": "application/json"}
default_headers = {
"Accept": "application/json",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
}
if headers:
default_headers.update(headers)
self.headers = default_headers
def fetch_question(self) -> ApiQuestion:
payload = self._fetch_json()
return self.handler(payload)
payload = self._fetch_json(self.url)
result = self.handler(payload)
if isinstance(result, list):
if not result:
raise ValueError("API returned no questions.")
return result[0]
return result
def fetch_questions(self, url: str | None = None) -> list[ApiQuestion]:
payload = self._fetch_json(url or self.url)
result = self.handler(payload)
if isinstance(result, list):
return result
return [result]
def fetch_and_store(self, repository: SQLiteWouldYouRatherRepository) -> int:
question = self.fetch_question()
return repository.add_question(question.text, question.answers)
def _fetch_json(self) -> Any:
request = Request(self.url, headers=self.headers, method="GET")
def _fetch_json(self, url: str) -> Any:
request = Request(url, headers=self.headers, method="GET")
try:
with urlopen(request, timeout=self.timeout) as response:
charset = response.headers.get_content_charset() or "utf-8"

120
game/database.py
View File

@@ -109,14 +109,51 @@ class SQLiteWouldYouRatherRepository:
id INTEGER PRIMARY KEY AUTOINCREMENT,
poll_id TEXT NOT NULL UNIQUE,
question_id INTEGER NOT NULL,
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (question_id) REFERENCES questions(id) ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS kv_settings (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
);
"""
)
# Migrate old schema: fix answers table UNIQUE constraint
self._migrate_answers_table(connection)
def _migrate_answers_table(self, connection: sqlite3.Connection) -> None:
"""Fix old schema where answers had UNIQUE(question_id) instead of UNIQUE(question_id, text)."""
# Check if we need to migrate by looking at the constraint
cursor = connection.execute("PRAGMA index_list(answers)")
indexes = cursor.fetchall()
for idx in indexes:
idx_name = idx[1] # name column
if idx_name == "sqlite_autoindex_answers_1":
# Check what columns are in this index
cursor2 = connection.execute(f"PRAGMA index_info({idx_name})")
columns = [row[2] for row in cursor2.fetchall()]
# If only question_id is indexed, we need to migrate
if columns == ["question_id"]:
# Recreate the table with correct schema
connection.executescript("""
CREATE TABLE answers_new (
id INTEGER PRIMARY KEY AUTOINCREMENT,
question_id INTEGER NOT NULL,
text TEXT NOT NULL,
FOREIGN KEY (question_id) REFERENCES questions(id) ON DELETE CASCADE,
UNIQUE (question_id, text)
);
INSERT INTO answers_new SELECT * FROM answers;
DROP TABLE answers;
ALTER TABLE answers_new RENAME TO answers;
""")
return
def add_question(self, text: str, answers: Iterable[str]) -> int:
prepared_text = text.strip()
if not prepared_text.endswith("?"):
prepared_text += "?"
prepared_answers = [answer.strip() for answer in answers if answer.strip()]
if not prepared_text:
raise ValueError("Question text cannot be empty.")
@@ -124,6 +161,7 @@ class SQLiteWouldYouRatherRepository:
raise ValueError("Question must have at least 2 answers.")
with self._connect() as connection:
try:
cursor = connection.execute(
"INSERT INTO questions(text, used) VALUES (?, 0)",
(prepared_text,),
@@ -134,6 +172,34 @@ class SQLiteWouldYouRatherRepository:
[(question_id, answer_text) for answer_text in prepared_answers],
)
return question_id
except sqlite3.IntegrityError as e:
import logging
import traceback
logger = logging.getLogger(__name__)
logger.error(
"Failed to add question due to integrity constraint: %s\nStacktrace:\n%s",
str(e),
traceback.format_exc()
)
raise ValueError(f"Question already exists or duplicate answer: {e}") from e
except Exception as e:
import logging
import traceback
logger = logging.getLogger(__name__)
logger.error(
"Unexpected error adding question: %s\nStacktrace:\n%s",
str(e),
traceback.format_exc()
)
raise
def delete_question_by_text(self, text: str) -> bool:
prepared_text = text.strip()
if not prepared_text.endswith("?"):
prepared_text += "?"
with self._connect() as connection:
cursor = connection.execute("DELETE FROM questions WHERE text = ?", (prepared_text,))
return cursor.rowcount > 0
def get_question(self, question_id: int) -> QuestionWithAnswers | None:
with self._connect() as connection:
@@ -164,20 +230,24 @@ class SQLiteWouldYouRatherRepository:
],
)
def count_questions(self) -> int:
def count_questions(self, unused_only: bool = False) -> int:
with self._connect() as connection:
row = connection.execute("SELECT COUNT(*) AS total FROM questions").fetchone()
query = "SELECT COUNT(*) AS total FROM questions"
if unused_only:
query += " WHERE used = 0"
row = connection.execute(query).fetchone()
return int(row["total"])
def list_questions(self, limit: int = 10, offset: int = 0) -> list[QuestionListItem]:
def list_questions(
self, limit: int = 10, offset: int = 0, unused_only: bool = False
) -> list[QuestionListItem]:
if limit < 1:
raise ValueError("Limit must be >= 1.")
if offset < 0:
raise ValueError("Offset must be >= 0.")
with self._connect() as connection:
rows = connection.execute(
"""
query = """
SELECT
q.id,
q.text,
@@ -185,12 +255,19 @@ class SQLiteWouldYouRatherRepository:
COUNT(a.id) AS answers_count
FROM questions q
LEFT JOIN answers a ON a.question_id = q.id
"""
params: list[int] = []
if unused_only:
query += " WHERE q.used = 0"
query += """
GROUP BY q.id, q.text, q.used
ORDER BY q.id
LIMIT ? OFFSET ?
""",
(limit, offset),
).fetchall()
"""
params.extend([limit, offset])
rows = connection.execute(query, params).fetchall()
return [
QuestionListItem(
@@ -209,17 +286,8 @@ class SQLiteWouldYouRatherRepository:
).fetchone()
if row is None:
has_questions = connection.execute(
"SELECT 1 FROM questions LIMIT 1"
).fetchone()
if has_questions is None:
return None
connection.execute("UPDATE questions SET used = 0")
row = connection.execute(
"SELECT id FROM questions ORDER BY RANDOM() LIMIT 1"
).fetchone()
question_id = int(row["id"])
connection.execute("UPDATE questions SET used = 1 WHERE id = ?", (question_id,))
@@ -275,6 +343,8 @@ class SQLiteWouldYouRatherRepository:
def update_question(self, question_id: int, text: str, answers: Iterable[str]) -> None:
prepared_text = text.strip()
if not prepared_text.endswith("?"):
prepared_text += "?"
prepared_answers = [answer.strip() for answer in answers if answer.strip()]
if not prepared_text:
raise ValueError("Question text cannot be empty.")
@@ -427,3 +497,17 @@ class SQLiteWouldYouRatherRepository:
def reset_used_flags(self) -> None:
with self._connect() as connection:
connection.execute("UPDATE questions SET used = 0")
def get_setting(self, key: str, default: str | None = None) -> str | None:
with self._connect() as connection:
row = connection.execute(
"SELECT value FROM kv_settings WHERE key = ?", (key,)
).fetchone()
return row["value"] if row else default
def set_setting(self, key: str, value: str) -> None:
with self._connect() as connection:
connection.execute(
"INSERT OR REPLACE INTO kv_settings (key, value) VALUES (?, ?)",
(key, value),
)

183
test_cron.py
View File

@@ -1,183 +0,0 @@
#!/usr/bin/env python3
"""
Утилита для тестирования cron выражений.
Использование:
python test_cron.py "0 9 * * 1-5"
python test_cron.py "*/15 * * * *"
"""
import sys
from datetime import datetime, timedelta, timezone
# Прямой импорт только нужных классов, избегая полный импорт bot
sys.path.insert(0, ".")
# Копируем необходимые классы прямо сюда для независимости
from dataclasses import dataclass
@dataclass(frozen=True)
class _CronField:
allowed: set[int]
def matches(self, value: int) -> bool:
return value in self.allowed
class CronExpression:
def __init__(self, expression: str) -> None:
parts = expression.strip().split()
if len(parts) != 5:
raise ValueError("Cron must have 5 fields: minute hour day month weekday.")
self.expression = expression.strip()
self._minute = self._parse_field(parts[0], 0, 59)
self._hour = self._parse_field(parts[1], 0, 23)
self._day = self._parse_field(parts[2], 1, 31)
self._month = self._parse_field(parts[3], 1, 12)
self._weekday = self._parse_field(parts[4], 0, 7)
def matches(self, dt: datetime) -> bool:
weekday = (dt.weekday() + 1) % 7
return (
self._minute.matches(dt.minute)
and self._hour.matches(dt.hour)
and self._day.matches(dt.day)
and self._month.matches(dt.month)
and (weekday in self._weekday.allowed or (7 in self._weekday.allowed and weekday == 0))
)
def _parse_field(self, token: str, minimum: int, maximum: int) -> _CronField:
allowed: set[int] = set()
for part in token.split(","):
part = part.strip()
if not part:
raise ValueError("Invalid empty cron token.")
step = 1
base = part
if "/" in part:
base, step_part = part.split("/", 1)
if not step_part.isdigit() or int(step_part) < 1:
raise ValueError(f"Invalid step value in cron token: {part}")
step = int(step_part)
if base == "*":
start, end = minimum, maximum
elif "-" in base:
start_str, end_str = base.split("-", 1)
start = int(start_str)
end = int(end_str)
else:
start = int(base)
end = int(base)
if start < minimum or end > maximum or start > end:
raise ValueError(f"Cron value out of range: {part}")
for value in range(start, end + 1, step):
allowed.add(value)
if not allowed:
raise ValueError("Cron field has no values.")
return _CronField(allowed=allowed)
def test_cron(expression: str, hours: int = 24):
"""
Тестирует cron выражение на следующие N часов.
Args:
expression: Cron выражение для теста
hours: Количество часов для проверки (по умолчанию 24)
"""
print(f"Тестирование cron выражения: '{expression}'")
print(f"Проверка следующих {hours} часов...")
print("-" * 60)
try:
cron = CronExpression(expression)
except ValueError as e:
print(f"❌ Ошибка в cron выражении: {e}")
return
now = datetime.now(timezone.utc).replace(second=0, microsecond=0)
matches = []
# Проверяем каждую минуту на протяжении указанного времени
total_minutes = hours * 60
for i in range(total_minutes):
check_time = now + timedelta(minutes=i)
if cron.matches(check_time):
matches.append(check_time)
if not matches:
print(f"⚠️ Нет совпадений в ближайшие {hours} часов")
else:
print(f"✅ Найдено {len(matches)} совпадений:\n")
for i, match_time in enumerate(matches[:20], 1): # Показываем первые 20
# Форматируем в читаемый вид
day_name = ["Пн", "Вт", "Ср", "Чт", "Пт", "Сб", "Вс"][match_time.weekday()]
formatted = match_time.strftime("%Y-%m-%d %H:%M UTC")
print(f" {i:2d}. {formatted} ({day_name})")
if len(matches) > 20:
print(f"\n ... и ещё {len(matches) - 20} совпадений")
print("-" * 60)
# Статистика
if matches:
first = matches[0]
time_until = (first - now).total_seconds() / 60
print(f"\nСледующее срабатывание через {int(time_until)} минут:")
day_name = ["Пн", "Вт", "Ср", "Чт", "Пт", "Сб", "Вс"][first.weekday()]
print(f" {first.strftime('%Y-%m-%d %H:%M UTC')} ({day_name})")
if len(matches) > 1:
# Вычисляем среднюю частоту
total_time = (matches[-1] - matches[0]).total_seconds() / 60
avg_interval = total_time / (len(matches) - 1) if len(matches) > 1 else 0
print(f"\nСредний интервал: {avg_interval:.1f} минут")
def show_examples():
"""Показывает примеры использования."""
print("=" * 60)
print("Примеры cron выражений:")
print("=" * 60)
examples = [
("0 9 * * *", "Каждый день в 9:00 UTC"),
("0 9 * * 1-5", "Каждый будний день в 9:00 UTC"),
("*/15 * * * *", "Каждые 15 минут"),
("0 */4 * * *", "Каждые 4 часа"),
("0 8,12,18 * * *", "В 8:00, 12:00 и 18:00"),
("0 12 * * 6,0", "Суббота и воскресенье в 12:00"),
("30 14 1 * *", "1-го числа каждого месяца в 14:30"),
]
for cron, description in examples:
print(f"\n{cron:20s} - {description}")
print("\n" + "=" * 60)
def main():
if len(sys.argv) < 2:
print("Использование: python test_cron.py \"<cron выражение>\" [часов]")
print("Пример: python test_cron.py \"0 9 * * 1-5\" 48")
print()
show_examples()
sys.exit(1)
expression = sys.argv[1]
hours = int(sys.argv[2]) if len(sys.argv) > 2 else 24
print()
test_cron(expression, hours)
print()
if __name__ == "__main__":
main()

83
test_kurigram_compat.py
View File

@@ -1,83 +0,0 @@
#!/usr/bin/env python3
"""
Проверка корректности работы с poll_id в kurigram.
Этот скрипт проверяет:
1. Что poll_id конвертируется в строку корректно
2. Что UpdateMessagePollVote содержит правильные поля
"""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent))
from pyrogram import raw
def test_poll_vote_structure():
"""Тестирует структуру UpdateMessagePollVote."""
print("=" * 60)
print("Проверка структуры UpdateMessagePollVote")
print("=" * 60)
# Проверяем, что можем создать update
try:
# Симулируем update от Telegram
update = raw.types.UpdateMessagePollVote(
poll_id=12345678901234567890, # Большое число, как в Telegram
user_id=123456789,
options=[0, 1], # Выбраны варианты 0 и 1
qts=0
)
print(f"\n✅ UpdateMessagePollVote создан успешно")
print(f" poll_id: {update.poll_id} (type: {type(update.poll_id).__name__})")
print(f" user_id: {update.user_id}")
print(f" options: {update.options}")
# Проверяем конвертацию в строку
poll_id_str = str(update.poll_id)
print(f"\n✅ Конвертация в строку: '{poll_id_str}'")
# Проверяем обратную конвертацию
poll_id_int = int(poll_id_str)
assert poll_id_int == update.poll_id, "Обратная конвертация должна работать"
print(f"✅ Обратная конвертация работает")
print("\n" + "=" * 60)
print("✅ ВСЕ ПРОВЕРКИ ПРОЙДЕНЫ")
print("=" * 60)
except Exception as e:
print(f"\n❌ ОШИБКА: {e}")
import traceback
traceback.print_exc()
sys.exit(1)
def test_poll_id_types():
"""Проверяет различные типы poll_id."""
print("\n" + "=" * 60)
print("Проверка различных типов poll_id")
print("=" * 60)
test_ids = [
123,
12345678901234567890,
"12345678901234567890",
]
for test_id in test_ids:
poll_id_str = str(test_id)
print(f"\nИсходное: {test_id} ({type(test_id).__name__})")
print(f"Строка: '{poll_id_str}'")
print(f"Длина: {len(poll_id_str)} символов")
print("\n✅ Все типы обрабатываются корректно")
if __name__ == "__main__":
test_poll_vote_structure()
test_poll_id_types()
print("\n🎉 Совместимость с kurigram подтверждена!\n")

177
test_statistics.py
View File

@@ -1,177 +0,0 @@
#!/usr/bin/env python3
"""
Тест системы статистики ответов.
Этот скрипт проверяет, что:
1. Маппинг опросов сохраняется корректно
2. Ответы записываются в статистику
3. Статистика корректно отображается
"""
import sys
import tempfile
from pathlib import Path
# Добавляем текущую директорию в путь
sys.path.insert(0, str(Path(__file__).parent))
from game.database import SQLiteWouldYouRatherRepository
def test_poll_statistics():
"""Тестирует полный цикл работы со статистикой опросов."""
print("=" * 60)
print("Тест системы статистики ответов")
print("=" * 60)
# Создаём временную базу данных
with tempfile.NamedTemporaryFile(suffix='.sqlite3', delete=False) as tmp:
db_path = tmp.name
try:
repo = SQLiteWouldYouRatherRepository(db_path)
print(f"\n✅ База данных создана: {db_path}")
# 1. Добавляем тестовый вопрос
print("\n--- Шаг 1: Добавление вопроса ---")
question_id = repo.add_question(
"Что бы вы выбрали?",
["Вариант 1", "Вариант 2", "Вариант 3"]
)
print(f"✅ Вопрос добавлен с ID: {question_id}")
# Получаем вопрос с ответами
question = repo.get_question(question_id)
assert question is not None, "Вопрос должен существовать"
print(f" Текст: {question.question.text}")
print(f" Ответов: {len(question.answers)}")
for i, answer in enumerate(question.answers):
print(f" {i}. ID={answer.id}, Текст='{answer.text}'")
# 2. Создаём маппинг для "опроса Telegram"
print("\n--- Шаг 2: Создание маппинга poll_id -> question_id ---")
fake_poll_id = "test_poll_12345"
repo.save_poll_mapping(fake_poll_id, question_id)
print(f"✅ Маппинг сохранён: {fake_poll_id} -> {question_id}")
# Проверяем маппинг
retrieved_question_id = repo.get_question_id_by_poll(fake_poll_id)
assert retrieved_question_id == question_id, "Маппинг должен работать"
print(f"✅ Маппинг работает корректно")
# 3. Симулируем ответы пользователей
print("\n--- Шаг 3: Запись ответов пользователей ---")
# 5 пользователей выбрали вариант 0 (Вариант 1)
answer_id_0 = repo.get_answer_id_by_option_index(question_id, 0)
assert answer_id_0 is not None, "Ответ 0 должен существовать"
for i in range(5):
repo.record_answer(question_id, answer_id_0, increment=1)
print(f"✅ Записано 5 голосов за вариант 0 (ID={answer_id_0})")
# 3 пользователя выбрали вариант 1 (Вариант 2)
answer_id_1 = repo.get_answer_id_by_option_index(question_id, 1)
assert answer_id_1 is not None, "Ответ 1 должен существовать"
for i in range(3):
repo.record_answer(question_id, answer_id_1, increment=1)
print(f"✅ Записано 3 голоса за вариант 1 (ID={answer_id_1})")
# 7 пользователей выбрали вариант 2 (Вариант 3)
answer_id_2 = repo.get_answer_id_by_option_index(question_id, 2)
assert answer_id_2 is not None, "Ответ 2 должен существовать"
for i in range(7):
repo.record_answer(question_id, answer_id_2, increment=1)
print(f"✅ Записано 7 голосов за вариант 2 (ID={answer_id_2})")
# 4. Проверяем статистику
print("\n--- Шаг 4: Проверка статистики ---")
stats = repo.get_question_statistics(question_id)
total_votes = sum(stat.count for stat in stats)
print(f"✅ Всего голосов: {total_votes}")
print("\nРаспределение голосов:")
expected = {answer_id_0: 5, answer_id_1: 3, answer_id_2: 7}
for stat in stats:
percentage = (stat.count / total_votes * 100) if total_votes > 0 else 0
print(f" '{stat.answer_text}': {stat.count} ({percentage:.1f}%)")
# Проверяем правильность
if stat.answer_id in expected:
expected_count = expected[stat.answer_id]
assert stat.count == expected_count, \
f"Ожидалось {expected_count}, получено {stat.count}"
print("\n✅ Все проверки прошли успешно!")
# 5. Дополнительная проверка - несуществующий poll_id
print("\n--- Шаг 5: Проверка несуществующего poll_id ---")
unknown_poll_id = repo.get_question_id_by_poll("non_existent_poll")
assert unknown_poll_id is None, "Должен вернуть None для несуществующего опроса"
print("✅ Корректная обработка несуществующего poll_id")
print("\n" + "=" * 60)
print("✅ ВСЕ ТЕСТЫ ПРОЙДЕНЫ УСПЕШНО!")
print("=" * 60)
finally:
# Удаляем временный файл (может не получиться на Windows, если SQLite держит файл)
try:
Path(db_path).unlink(missing_ok=True)
print(f"\n🧹 Временная база данных удалена")
except PermissionError:
print(f"\n⚠️ Временный файл {db_path} будет удалён автоматически")
def test_edge_cases():
"""Тестирует граничные случаи."""
print("\n" + "=" * 60)
print("Тест граничных случаев")
print("=" * 60)
with tempfile.NamedTemporaryFile(suffix='.sqlite3', delete=False) as tmp:
db_path = tmp.name
try:
repo = SQLiteWouldYouRatherRepository(db_path)
# Добавляем вопрос
question_id = repo.add_question("Тест?", ["A", "B"])
# Тест 1: Получение answer_id с неверным индексом
print("\n--- Тест: Неверный индекс опции ---")
answer_id = repo.get_answer_id_by_option_index(question_id, 999)
assert answer_id is None, "Должен вернуть None для несуществующего индекса"
print("✅ Корректная обработка неверного индекса")
# Тест 2: Повторная запись того же маппинга (должно работать через REPLACE)
print("\n--- Тест: Повторная запись маппинга ---")
repo.save_poll_mapping("poll_123", question_id)
repo.save_poll_mapping("poll_123", question_id) # Повторно
retrieved = repo.get_question_id_by_poll("poll_123")
assert retrieved == question_id
print("✅ Повторная запись маппинга работает")
print("\n✅ Граничные случаи обработаны корректно!")
finally:
try:
Path(db_path).unlink(missing_ok=True)
except PermissionError:
pass # Игнорируем на Windows
if __name__ == "__main__":
try:
test_poll_statistics()
test_edge_cases()
print("\n🎉 Система статистики полностью функциональна!\n")
except AssertionError as e:
print(f"\n❌ ОШИБКА: {e}\n")
sys.exit(1)
except Exception as e:
print(f"\n❌ НЕОЖИДАННАЯ ОШИБКА: {e}\n")
import traceback
traceback.print_exc()
sys.exit(1)