This commit is contained in:
2026-01-04 03:36:13 +03:00
parent 8d708f2cce
commit 12beb5a986
19 changed files with 226 additions and 2103 deletions
+15 -22
View File
@@ -9,23 +9,18 @@ from apscheduler.schedulers.asyncio import AsyncIOScheduler
from dishka import make_async_container
from dishka.integrations.aiogram import setup_dishka
from trudex.application.bot.admin_dialogs.broadcast import broadcast_dialog as admin_broadcast_dialog
from trudex.application.bot.admin_dialogs.create_test import admin_create_test_dialog
from trudex.application.bot.admin_dialogs.groups import groups_dialog as admin_groups_dialog
from trudex.application.bot.admin_dialogs.main_menu import admin_menu_dialog
from trudex.application.bot.admin_dialogs.templates import templates_dialog as admin_templates_dialog
from trudex.application.bot.admin_dialogs.tests import tests_dialog as admin_tests_dialog
from trudex.application.bot.admin_dialogs.users import users_dialog as admin_users_dialog
from trudex.application.bot.creator_dialogs.broadcast import broadcast_dialog as creator_broadcast_dialog
from trudex.application.bot.creator_dialogs.create_test import create_test_dialog
from trudex.application.bot.creator_dialogs.groups import groups_dialog as creator_groups_dialog
from trudex.application.bot.admin_dialogs.users import admin_users_dialog
from trudex.application.bot.creator_dialogs.main_menu import creator_menu_dialog
from trudex.application.bot.creator_dialogs.templates import templates_dialog as creator_templates_dialog
from trudex.application.bot.creator_dialogs.tests import tests_dialog as creator_tests_dialog
from trudex.application.bot.creator_dialogs.users import users_dialog as creator_users_dialog
from trudex.application.bot.creator_dialogs.users import creator_users_dialog
from trudex.application.bot.handlers import router
from trudex.application.bot.middlewares.reject_not_admin import RejectNotAdminMiddleware
from trudex.application.bot.middlewares.reject_not_creator import RejectNotCreatorMiddleware
from trudex.application.bot.shared_dialogs.broadcast import shared_broadcast_dialog
from trudex.application.bot.shared_dialogs.create_test import shared_create_test_dialog
from trudex.application.bot.shared_dialogs.groups import shared_groups_dialog
from trudex.application.bot.shared_dialogs.templates import shared_templates_dialog
from trudex.application.bot.shared_dialogs.tests import shared_tests_dialog
from trudex.application.bot.user_dialogs.deeplink import deeplink_dialog
from trudex.application.bot.user_dialogs.main_menu import user_menu_dialog
from trudex.application.bot.user_dialogs.registration import registration_dialog
@@ -57,20 +52,18 @@ async def main() -> None:
take_test_dialog,
registration_dialog,
deeplink_dialog,
# Shared dialogs
shared_tests_dialog,
shared_groups_dialog,
shared_broadcast_dialog,
shared_templates_dialog,
shared_create_test_dialog,
# Admin dialogs
admin_menu_dialog,
admin_users_dialog,
admin_tests_dialog,
admin_groups_dialog,
admin_broadcast_dialog,
admin_templates_dialog,
admin_create_test_dialog,
# Creator dialogs
creator_menu_dialog,
creator_users_dialog,
creator_tests_dialog,
creator_groups_dialog,
creator_broadcast_dialog,
creator_templates_dialog,
create_test_dialog,
)
router.message.middleware(RejectNotAdminMiddleware())
@@ -1,186 +0,0 @@
from aiogram.types import CallbackQuery, Message
from aiogram_dialog import Dialog, DialogManager, StartMode, Window
from aiogram_dialog.widgets.input import MessageInput
from aiogram_dialog.widgets.kbd import Button, Column, Row, ScrollingGroup, Select
from aiogram_dialog.widgets.text import Const, Format
from dishka import FromDishka
from dishka.integrations.aiogram_dialog import inject
from trudex.application.bot.admin_dialogs.states import AdminGroupsSG, AdminMenuSG
from trudex.infrastructure.database.dao.group import GroupDAO
async def on_group_click(_callback: CallbackQuery, _widget, _manager: DialogManager, _item_id: str):
await _callback.answer("ℹ️ Для удаления используйте кнопку 'Удалить группу'")
@inject
async def get_groups_data(dialog_manager: DialogManager, group_dao: FromDishka[GroupDAO], **_kwargs):
groups = await group_dao.get_all()
success_message = dialog_manager.dialog_data.pop("success_message", None)
message_text = "<b>👥 Управление группами</b>\n\n"
if success_message:
message_text += f"{success_message}\n\n"
message_text += f"📊 <b>Всего групп:</b> {len(groups)}\n\n<b>Список групп:</b>"
return {
"groups": [(str(g.id), str(g.number)) for g in groups],
"groups_count": len(groups),
"message_text": message_text,
}
async def on_add_group(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.switch_to(AdminGroupsSG.add_group_input_number)
async def on_delete_group(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.switch_to(AdminGroupsSG.delete_groups_list)
async def on_back_to_menu(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.start(AdminMenuSG.main, mode=StartMode.RESET_STACK)
@inject
async def on_group_number_input(message: Message, _widget: MessageInput, manager: DialogManager, group_dao: FromDishka[GroupDAO]):
if not message.text:
await message.answer("❌ Номер группы не может быть пустым")
return
number_str = message.text.strip()
if not number_str.isdigit():
await message.answer("❌ Номер группы должен содержать только цифры")
return
number = int(number_str)
if number < 1000 or number > 9999:
await message.answer("❌ Номер группы должен быть четырехзначным (1000-9999)")
return
existing = await group_dao.get_by_number(number)
if existing:
await message.answer(f"❌ Группа с номером {number} уже существует")
return
try:
await group_dao.create(number=number)
manager.dialog_data["success_message"] = f"✅ Группа {number} создана"
except Exception as e:
await message.answer(f"❌ Ошибка создания группы: {e}")
return
await manager.switch_to(AdminGroupsSG.groups_list)
async def on_cancel_add(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.switch_to(AdminGroupsSG.groups_list)
@inject
async def get_delete_groups_data(dialog_manager: DialogManager, group_dao: FromDishka[GroupDAO], **_kwargs):
groups = await group_dao.get_all()
return {
"groups": [(str(g.id), f"{g.number}") for g in groups],
"groups_count": len(groups),
}
@inject
async def on_select_group_to_delete(_callback: CallbackQuery, _widget, manager: DialogManager, item_id: str, group_dao: FromDishka[GroupDAO]):
group = await group_dao.get_by_id(int(item_id))
if not group:
await _callback.answer("❌ Группа не найдена", show_alert=True)
return
manager.dialog_data["delete_group_id"] = group.id
manager.dialog_data["delete_group_number"] = group.number
await manager.switch_to(AdminGroupsSG.delete_confirm)
async def get_delete_confirm_data(dialog_manager: DialogManager, **_kwargs):
number = dialog_manager.dialog_data.get("delete_group_number", "")
return {
"group_info": str(number)
}
@inject
async def on_confirm_delete(_callback: CallbackQuery, _button: Button, manager: DialogManager, group_dao: FromDishka[GroupDAO]):
group_id = manager.dialog_data.get("delete_group_id")
assert isinstance(group_id, int)
await group_dao.delete(group_id)
manager.dialog_data["success_message"] = "✅ Группа удалена"
await manager.switch_to(AdminGroupsSG.groups_list)
async def on_cancel_delete(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.switch_to(AdminGroupsSG.delete_groups_list)
groups_dialog = Dialog(
Window(
Format("{message_text}"),
ScrollingGroup(
Select(
Format("{item[1]}"),
id="groups",
item_id_getter=lambda x: x[0],
items="groups",
on_click=on_group_click,
),
id="groups_scroll",
width=2,
height=7,
),
Column(
Button(Const(" Добавить группу"), id="add", on_click=on_add_group),
Button(Const("🗑 Удалить группу"), id="delete", on_click=on_delete_group),
Button(Const("◀️ Назад"), id="back", on_click=on_back_to_menu),
),
state=AdminGroupsSG.groups_list,
getter=get_groups_data,
),
Window(
Const("<b>➕ Добавление группы</b>\n\n🔢 <b>Введите номер группы</b> (четырехзначное число 1000-9999):"),
MessageInput(on_group_number_input),
Button(Const("◀️ Отмена"), id="cancel", on_click=on_cancel_add),
state=AdminGroupsSG.add_group_input_number,
),
Window(
Format("<b>🗑 Удаление группы</b>\n\n<b>Выберите группу для удаления:</b>\n\n📊 <b>Всего групп:</b> {groups_count}"),
ScrollingGroup(
Select(
Format("{item[1]}"),
id="delete_groups",
item_id_getter=lambda x: x[0],
items="groups",
on_click=on_select_group_to_delete,
),
id="delete_groups_scroll",
width=2,
height=7,
),
Button(Const("◀️ Назад"), id="back", on_click=on_cancel_add),
state=AdminGroupsSG.delete_groups_list,
getter=get_delete_groups_data,
),
Window(
Format("<b>⚠️ Подтверждение удаления</b>\n\n<b>Точно хотите удалить группу?</b>\n\n👥 {group_info}"),
Row(
Button(Const("✅ Да, удалить"), id="confirm", on_click=on_confirm_delete),
Button(Const("❌ Отмена"), id="cancel", on_click=on_cancel_delete),
),
state=AdminGroupsSG.delete_confirm,
getter=get_delete_confirm_data,
),
)
@@ -1,30 +1,35 @@
from aiogram.types import CallbackQuery
from aiogram_dialog import Dialog, DialogManager, StartMode, Window
from aiogram_dialog import Dialog, DialogManager, Window
from aiogram_dialog.widgets.kbd import Button, Column
from aiogram_dialog.widgets.text import Const
from trudex.application.bot.admin_dialogs.states import (AdminBroadcastSG, AdminGroupsSG, AdminMenuSG,
AdminTemplatesSG, AdminTestsSG, AdminUsersSG)
from trudex.application.bot.admin_dialogs.states import AdminMenuSG, AdminUsersSG
from trudex.application.bot.shared_dialogs.states import (
SharedBroadcastSG,
SharedGroupsSG,
SharedTemplatesSG,
SharedTestsSG,
)
async def on_tests_clicked(_callback: CallbackQuery, _button: Button, manager: DialogManager) -> None:
await manager.start(AdminTestsSG.tests_list, mode=StartMode.RESET_STACK)
await manager.start(SharedTestsSG.tests_list)
async def on_users_clicked(_callback: CallbackQuery, _button: Button, manager: DialogManager) -> None:
await manager.start(AdminUsersSG.users_list, mode=StartMode.RESET_STACK)
await manager.start(AdminUsersSG.users_list)
async def on_groups_clicked(_callback: CallbackQuery, _button: Button, manager: DialogManager) -> None:
await manager.start(AdminGroupsSG.groups_list, mode=StartMode.RESET_STACK)
await manager.start(SharedGroupsSG.groups_list)
async def on_broadcast_clicked(_callback: CallbackQuery, _button: Button, manager: DialogManager) -> None:
await manager.start(AdminBroadcastSG.broadcast_input, mode=StartMode.RESET_STACK)
await manager.start(SharedBroadcastSG.broadcast_input)
async def on_templates_clicked(_callback: CallbackQuery, _button: Button, manager: DialogManager) -> None:
await manager.start(AdminTemplatesSG.main, mode=StartMode.RESET_STACK)
await manager.start(SharedTemplatesSG.main)
admin_menu_dialog = Dialog(
@@ -5,57 +5,7 @@ class AdminMenuSG(StatesGroup):
main = State()
class AdminTemplatesSG(StatesGroup):
main = State()
export_list = State()
spec = State()
import_file = State()
class AdminUsersSG(StatesGroup):
users_list = State()
users_input = State()
user_detail = State()
class AdminTestsSG(StatesGroup):
tests_list = State()
test_detail = State()
share_test = State()
edit_menu = State()
edit_password = State()
edit_attempts = State()
edit_group = State()
edit_expires = State()
statistics = State()
attempt_detail = State()
class AdminBroadcastSG(StatesGroup):
broadcast_input = State()
broadcast_confirm = State()
class AdminGroupsSG(StatesGroup):
groups_list = State()
add_group_input_number = State()
delete_groups_list = State()
delete_confirm = State()
class AdminCreateTestSG(StatesGroup):
input_title = State()
input_description = State()
input_password = State()
input_attempts = State()
input_expires_at = State()
input_for_group = State()
confirm_test_info = State()
add_question = State()
input_question_text = State()
select_question_type = State()
input_correct_answer = State()
input_options = State()
mark_correct_options = State()
confirm_question = State()
test_created = State()
@@ -1,25 +1,26 @@
from aiogram.types import CallbackQuery, Message
from aiogram_dialog import Dialog, DialogManager, StartMode, Window
from aiogram_dialog import Dialog, DialogManager, Window
from aiogram_dialog.widgets.input import MessageInput
from aiogram_dialog.widgets.kbd import Button, Column, ScrollingGroup, Select, SwitchTo
from aiogram_dialog.widgets.text import Const, Format
from dishka import FromDishka
from dishka.integrations.aiogram_dialog import inject
from trudex.application.bot.admin_dialogs.states import AdminMenuSG, AdminUsersSG
from trudex.application.bot.admin_dialogs.states import AdminUsersSG
from trudex.infrastructure.database.dao.user import UserDAO
@inject
async def get_users_data(user_dao: FromDishka[UserDAO], **_kwargs):
users = await user_dao.get_all()
users_sorted = sorted(users, key=lambda u: u.created_at or u.id, reverse=True)
return {
"users": [
(f"{u.name or u.first_name} (@{u.username or 'нет'})", u.id)
for u in users
(f"{'👑 ' if u.is_admin else ''}{u.name or u.first_name} (@{u.username or 'нет'})", u.id)
for u in users_sorted
],
"count": len(users),
"count": len(users_sorted),
}
@@ -34,7 +35,6 @@ async def get_user_detail_data(dialog_manager: DialogManager, user_dao: FromDish
return {"user_info": "Пользователь не найден"}
username_str = f"@{user.username}" if user.username else ""
last_name_str = user.last_name or ""
name_str = user.name or ""
group_str = str(user.group) if user.group else ""
admin_status = "✅ Да" if user.is_admin else "❌ Нет"
@@ -42,8 +42,7 @@ async def get_user_detail_data(dialog_manager: DialogManager, user_dao: FromDish
user_info = (
f"<b>👤 Информация о пользователе</b>\n\n"
f"<b>ID:</b> <code>{user.id}</code>\n"
f"<b>Имя:</b> {user.first_name}\n"
f"<b>Фамилия:</b> {last_name_str}\n"
f"<b>Ник:</b> {user.first_name}\n"
f"<b>Имя и фамилия:</b> {name_str}\n"
f"<b>Username:</b> {username_str}\n"
f"<b>Группа:</b> {group_str}\n"
@@ -83,10 +82,10 @@ async def on_user_input(message: Message, _widget: MessageInput, manager: Dialog
async def on_back_to_main(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.start(AdminMenuSG.main, mode=StartMode.RESET_STACK)
await manager.done()
users_dialog = Dialog(
admin_users_dialog = Dialog(
Window(
Format("<b>👥 Пользователи</b>\n\nВсего: {count}"),
ScrollingGroup(
@@ -1,73 +0,0 @@
from aiogram.types import CallbackQuery, Message
from aiogram_dialog import Dialog, DialogManager, StartMode, Window
from aiogram_dialog.widgets.input import MessageInput
from aiogram_dialog.widgets.kbd import Button, Row
from aiogram_dialog.widgets.text import Const
from dishka import FromDishka
from dishka.integrations.aiogram_dialog import inject
from trudex.application.bot.creator_dialogs.states import CreatorBroadcastSG, CreatorMenuSG
from trudex.infrastructure.database.dao.user import UserDAO
from trudex.infrastructure.utils.broadcast import broadcast_message
async def on_broadcast_input(message: Message, _widget: MessageInput, manager: DialogManager):
manager.dialog_data["broadcast_message_id"] = message.message_id
manager.dialog_data["broadcast_chat_id"] = message.chat.id
await manager.switch_to(CreatorBroadcastSG.broadcast_confirm)
@inject
async def on_broadcast_confirm(_callback: CallbackQuery, _button: Button, manager: DialogManager, user_dao: FromDishka[UserDAO]):
message_id = manager.dialog_data.get("broadcast_message_id")
chat_id = manager.dialog_data.get("broadcast_chat_id")
if not message_id or not chat_id or not _callback.message:
await _callback.answer("Ошибка: сообщение не найдено")
return
await _callback.message.answer("⏳ Рассылка началась...")
bot = _callback.bot
if not bot:
await _callback.answer("Ошибка: бот не найден")
return
stats = await broadcast_message(bot, message_id, chat_id, user_dao)
stats_text = (
f"✅ <b>Рассылка завершена</b>\n\n"
f"Всего пользователей: {stats.total}\n"
f"Успешно отправлено: {stats.success}\n"
f"Не удалось отправить: {stats.failed}"
)
await _callback.message.answer(stats_text)
await manager.done()
async def on_broadcast_cancel(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await _callback.answer("Рассылка отменена")
await manager.start(CreatorMenuSG.main, mode=StartMode.RESET_STACK)
async def on_back_to_main(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.start(CreatorMenuSG.main, mode=StartMode.RESET_STACK)
broadcast_dialog = Dialog(
Window(
Const("<b>📢 Рассылка</b>\n\nОтправьте сообщение, которое хотите разослать всем пользователям:"),
MessageInput(on_broadcast_input),
Button(Const("◀️ Отмена"), id="back", on_click=on_back_to_main),
state=CreatorBroadcastSG.broadcast_input,
),
Window(
Const("<b>⚠️ Подтверждение рассылки</b>\n\nВы уверены, что хотите отправить это сообщение всем пользователям?"),
Row(
Button(Const("✅ Да"), id="broadcast_confirm", on_click=on_broadcast_confirm),
Button(Const("❌ Нет"), id="broadcast_cancel", on_click=on_broadcast_cancel),
),
state=CreatorBroadcastSG.broadcast_confirm,
),
)
@@ -1,574 +0,0 @@
from datetime import date, datetime, time
from aiogram.types import CallbackQuery, ContentType, Message
from aiogram_dialog import Dialog, DialogManager, StartMode, Window
from aiogram_dialog.widgets.input import MessageInput
from aiogram_dialog.widgets.kbd import Button, Calendar, Cancel, Column, Row, ScrollingGroup, Select
from aiogram_dialog.widgets.text import Const, Format
from dishka import FromDishka
from dishka.integrations.aiogram_dialog import inject
from trudex.application.bot.creator_dialogs.states import CreateTestSG, CreatorTestsSG
from trudex.infrastructure.database.dao.group import GroupDAO
from trudex.infrastructure.database.dao.option import OptionDAO
from trudex.infrastructure.database.dao.question import QuestionDAO
from trudex.infrastructure.database.dao.test import TestDAO
from trudex.infrastructure.database.repo.test import TestRepository
from trudex.infrastructure.utils.timezone import to_msk
async def on_title_input(message: Message, _widget: MessageInput, manager: DialogManager):
if not message.text:
await message.answer("❌ Название не может быть пустым")
return
title = message.text.strip()
if not title:
await message.answer("❌ Название не может быть пустым")
return
if len(title) > 255:
await message.answer("❌ Название слишком длинное (максимум 255 символов)")
return
manager.dialog_data["title"] = title
await manager.switch_to(CreateTestSG.input_description)
async def on_description_input(message: Message, _widget: MessageInput, manager: DialogManager):
if not message.text:
await message.answer("❌ Описание не может быть пустым")
return
description = message.text.strip()
if not description:
await message.answer("❌ Описание не может быть пустым")
return
if len(description) > 2000:
await message.answer("❌ Описание слишком длинное (максимум 2000 символов)")
return
manager.dialog_data["description"] = description
await manager.switch_to(CreateTestSG.input_password)
@inject
async def on_password_input(message: Message, _widget: MessageInput, manager: DialogManager, _group_dao: FromDishka[GroupDAO]):
if not message.text:
await message.answer("❌ Пароль не может быть пустым")
return
password = message.text.strip()
if not password:
await message.answer("❌ Пароль не может быть пустым")
return
if len(password) > 255:
await message.answer("❌ Пароль слишком длинный (максимум 255 символов)")
return
manager.dialog_data["password"] = password
await manager.switch_to(CreateTestSG.input_attempts)
@inject
async def on_skip_password(_callback: CallbackQuery, _button: Button, manager: DialogManager, _group_dao: FromDishka[GroupDAO]):
manager.dialog_data["password"] = None
await manager.switch_to(CreateTestSG.input_attempts)
async def on_attempts_input(message: Message, _widget: MessageInput, manager: DialogManager):
if not message.text:
await message.answer("❌ Количество попыток не может быть пустым")
return
attempts_str = message.text.strip()
if not attempts_str.isdigit():
await message.answer("❌ Количество попыток должно быть числом")
return
attempts = int(attempts_str)
if attempts < 1:
await message.answer("❌ Количество попыток должно быть больше 0")
return
if attempts > 100:
await message.answer("❌ Количество попыток не может быть больше 100")
return
manager.dialog_data["attempts"] = attempts
await manager.switch_to(CreateTestSG.input_expires_at)
async def on_skip_attempts(_callback: CallbackQuery, _button: Button, manager: DialogManager):
manager.dialog_data["attempts"] = None
await manager.switch_to(CreateTestSG.input_expires_at)
async def on_date_selected(_callback, _widget, manager: DialogManager, selected_date: date):
manager.dialog_data["expires_at"] = datetime.combine(selected_date, time.min)
await manager.switch_to(CreateTestSG.input_for_group)
async def on_skip_expires(_callback: CallbackQuery, _button: Button, manager: DialogManager):
manager.dialog_data["expires_at"] = None
await manager.switch_to(CreateTestSG.input_for_group)
@inject
async def get_groups_for_test(group_dao: FromDishka[GroupDAO], **_kwargs):
groups = await group_dao.get_all()
return {
"groups": [(str(g.number), str(g.number)) for g in groups],
}
async def on_group_selected(_callback: CallbackQuery, _widget, manager: DialogManager, item_id: str):
manager.dialog_data["for_group"] = int(item_id)
await manager.switch_to(CreateTestSG.confirm_test_info)
async def on_skip_group(_callback: CallbackQuery, _button: Button, manager: DialogManager):
manager.dialog_data["for_group"] = None
await manager.switch_to(CreateTestSG.confirm_test_info)
async def get_test_info(dialog_manager: DialogManager, **_kwargs):
title = dialog_manager.dialog_data.get("title", "")
description = dialog_manager.dialog_data.get("description", "")
password = dialog_manager.dialog_data.get("password")
attempts = dialog_manager.dialog_data.get("attempts")
expires_at = dialog_manager.dialog_data.get("expires_at")
for_group = dialog_manager.dialog_data.get("for_group")
password_str = f"🔒 {password}" if password else "Без пароля"
attempts_str = f"🔄 {attempts}" if attempts else "♾️ Без ограничений"
expires_at_msk = to_msk(expires_at)
expires_str = expires_at_msk.strftime("%d.%m.%Y") if expires_at_msk else "Без срока"
group_str = str(for_group) if for_group else "Для всех"
return {
"info": (
f"<b>📝 Информация о тесте</b>\n\n"
f"<b>Название:</b> {title}\n"
f"<b>Описание:</b> {description}\n"
f"<b>Пароль:</b> {password_str}\n"
f"<b>Попыток:</b> {attempts_str}\n"
f"<b>Истекает:</b> {expires_str}\n"
f"<b>Для группы:</b> {group_str}"
)
}
@inject
async def on_confirm_test(_callback: CallbackQuery, _button: Button, manager: DialogManager, test_dao: FromDishka[TestDAO]):
title = manager.dialog_data.get("title")
assert isinstance(title, str)
description = manager.dialog_data.get("description")
password = manager.dialog_data.get("password")
attempts = manager.dialog_data.get("attempts")
expires_at = manager.dialog_data.get("expires_at")
for_group = manager.dialog_data.get("for_group")
test = await test_dao.create(
title=title,
description=description,
password=password,
attempts=attempts,
expires_at=expires_at,
for_group=for_group,
)
manager.dialog_data["test_id"] = test.id
manager.dialog_data["questions"] = []
await manager.switch_to(CreateTestSG.add_question)
async def on_add_question(_callback: CallbackQuery, _button: Button, manager: DialogManager):
manager.dialog_data["current_question"] = {}
await manager.switch_to(CreateTestSG.input_question_text)
async def on_question_input(message: Message, _widget: MessageInput, manager: DialogManager):
current_question = manager.dialog_data.get("current_question", {})
if message.content_type == ContentType.PHOTO:
photo = message.photo[-1] if message.photo else None
if photo:
text = (message.caption or "").strip()
if not text:
await message.answer("❌ Изображение должно содержать подпись с текстом вопроса")
return
if len(text) > 2000:
await message.answer("❌ Текст вопроса слишком длинный (максимум 2000 символов)")
return
current_question["tg_file_id"] = photo.file_id
current_question["text"] = text
elif message.content_type == ContentType.TEXT and message.text:
text = message.text.strip()
if not text:
await message.answer("❌ Текст вопроса не может быть пустым")
return
if len(text) > 2000:
await message.answer("❌ Текст вопроса слишком длинный (максимум 2000 символов)")
return
current_question["text"] = text
current_question["tg_file_id"] = None
else:
await message.answer("❌ Отправьте текст или фото с подписью")
return
manager.dialog_data["current_question"] = current_question
await manager.switch_to(CreateTestSG.select_question_type)
async def get_question_type_data(**_kwargs):
return {
"question_types": [
("single", "📌 Один правильный ответ"),
("multiple", " Несколько правильных ответов"),
("input", "✏️ Ввод текста"),
]
}
async def on_question_type_selected(_callback: CallbackQuery, _widget, manager: DialogManager, item_id: str):
current_question = manager.dialog_data.get("current_question", {})
current_question["question_type"] = item_id
manager.dialog_data["current_question"] = current_question
if item_id == "input":
await manager.switch_to(CreateTestSG.input_correct_answer)
else:
manager.dialog_data["current_options"] = []
await manager.switch_to(CreateTestSG.input_options)
async def on_correct_answer_input(message: Message, _widget: MessageInput, manager: DialogManager):
if not message.text:
await message.answer("❌ Правильный ответ не может быть пустым")
return
answer = message.text.strip()
if not answer:
await message.answer("❌ Правильный ответ не может быть пустым")
return
if len(answer) > 255:
await message.answer("❌ Ответ слишком длинный (максимум 255 символов)")
return
current_question = manager.dialog_data.get("current_question", {})
current_question["correct_answer"] = answer
manager.dialog_data["current_question"] = current_question
await manager.switch_to(CreateTestSG.confirm_question)
async def on_option_input(message: Message, _widget: MessageInput, manager: DialogManager):
if not message.text:
await message.answer("❌ Вариант ответа не может быть пустым")
return
option_text = message.text.strip()
if not option_text:
await message.answer("❌ Вариант ответа не может быть пустым")
return
if len(option_text) > 255:
await message.answer("❌ Вариант ответа слишком длинный (максимум 255 символов)")
return
current_options = manager.dialog_data.get("current_options", [])
if len(current_options) >= 10:
await message.answer("❌ Максимум 10 вариантов ответа")
return
current_options.append({"text": option_text, "is_correct": False})
manager.dialog_data["current_options"] = current_options
await message.answer(f"✅ Вариант {len(current_options)} добавлен")
async def on_finish_options(_callback: CallbackQuery, _button: Button, manager: DialogManager):
current_options = manager.dialog_data.get("current_options", [])
if len(current_options) < 2:
await _callback.answer("❌ Добавьте минимум 2 варианта ответа", show_alert=True)
return
await manager.switch_to(CreateTestSG.mark_correct_options)
async def get_options_data(dialog_manager: DialogManager, **_kwargs):
current_options = dialog_manager.dialog_data.get("current_options", [])
formatted_options = []
for i, opt in enumerate(current_options):
marker = "" if opt["is_correct"] else ""
formatted_options.append((str(i), f"{marker} {opt['text']}"))
return {
"options": formatted_options,
"options_count": len(current_options),
}
async def on_option_toggle(_callback: CallbackQuery, _widget, manager: DialogManager, item_id: str):
current_options = manager.dialog_data.get("current_options", [])
current_question = manager.dialog_data.get("current_question", {})
question_type = current_question.get("question_type", "single")
option_idx = int(item_id)
if question_type == "single":
for opt in current_options:
opt["is_correct"] = False
current_options[option_idx]["is_correct"] = True
else:
current_options[option_idx]["is_correct"] = not current_options[option_idx]["is_correct"]
manager.dialog_data["current_options"] = current_options
await _callback.answer()
async def on_confirm_correct(_callback: CallbackQuery, _button: Button, manager: DialogManager):
current_options = manager.dialog_data.get("current_options", [])
if not any(opt["is_correct"] for opt in current_options):
await _callback.answer("❌ Отметьте хотя бы один правильный ответ", show_alert=True)
return
await manager.switch_to(CreateTestSG.confirm_question)
async def get_question_preview(dialog_manager: DialogManager, **_kwargs):
current_question = dialog_manager.dialog_data.get("current_question", {})
current_options = dialog_manager.dialog_data.get("current_options", [])
text = current_question.get("text", "")
question_type = current_question.get("question_type", "single")
has_image = current_question.get("tg_file_id") is not None
type_names = {
"single": "📌 Один правильный ответ",
"multiple": "📋 Несколько правильных ответов",
"input": "✏️ Ввод текста",
}
preview = f"<b>📝 Предпросмотр вопроса</b>\n\n"
preview += f"<b>Текст:</b> {text}\n"
preview += f"<b>Тип:</b> {type_names[question_type]}\n"
preview += f"<b>Изображение:</b> {'✅ Да' if has_image else '❌ Нет'}\n\n"
if question_type == "input":
correct_answer = current_question.get("correct_answer", "")
preview += f"<b>Правильный ответ:</b> <code>{correct_answer}</code>"
else:
preview += "<b>Варианты ответов:</b>\n"
for i, opt in enumerate(current_options, 1):
marker = "" if opt["is_correct"] else ""
preview += f"{i}. {marker} {opt['text']}\n"
return {"preview": preview}
@inject
async def on_save_question(
_callback: CallbackQuery,
_button: Button,
manager: DialogManager,
question_dao: FromDishka[QuestionDAO],
option_dao: FromDishka[OptionDAO],
test_repo: FromDishka[TestRepository],
):
test_id = manager.dialog_data.get("test_id")
assert isinstance(test_id, int)
current_question = manager.dialog_data.get("current_question", {})
current_options = manager.dialog_data.get("current_options", [])
questions_count = await test_repo.count_questions_in_test(test_id)
question = await question_dao.create(
test_id=test_id,
text=current_question.get("text", ""),
position=questions_count,
question_type=current_question.get("question_type", "single"),
tg_file_id=current_question.get("tg_file_id"),
)
if current_question.get("question_type") == "input":
await option_dao.create(
question_id=question.id,
text=current_question.get("correct_answer", ""),
is_correct=True,
)
else:
for opt in current_options:
await option_dao.create(
question_id=question.id,
text=opt["text"],
is_correct=opt["is_correct"],
)
questions = manager.dialog_data.get("questions", [])
questions.append(question.id)
manager.dialog_data["questions"] = questions
manager.dialog_data.pop("current_question", None)
manager.dialog_data.pop("current_options", None)
await _callback.answer("✅ Вопрос добавлен")
await manager.switch_to(CreateTestSG.add_question)
async def on_cancel_question(_callback: CallbackQuery, _button: Button, manager: DialogManager):
manager.dialog_data.pop("current_question", None)
manager.dialog_data.pop("current_options", None)
await manager.switch_to(CreateTestSG.add_question)
async def get_questions_count(dialog_manager: DialogManager, **_kwargs):
questions = dialog_manager.dialog_data.get("questions", [])
return {"questions_count": len(questions)}
async def on_finish_test(_callback: CallbackQuery, _button: Button, manager: DialogManager):
questions = manager.dialog_data.get("questions", [])
if len(questions) == 0:
await _callback.answer("❌ Добавьте хотя бы один вопрос", show_alert=True)
return
await _callback.answer("✅ Тест создан")
await manager.start(CreatorTestsSG.tests_list, mode=StartMode.RESET_STACK)
async def on_cancel(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.start(CreatorTestsSG.tests_list, mode=StartMode.RESET_STACK)
create_test_dialog = Dialog(
Window(
Const("<b>📝 Создание теста</b>\n\n💬 <b>Введите название теста:</b>\n<i>(максимум 255 символов)</i>"),
MessageInput(on_title_input),
Cancel(Const("◀️ Отмена")),
state=CreateTestSG.input_title,
),
Window(
Const("<b>📝 Создание теста</b>\n\n📄 <b>Введите описание теста:</b>\n<i>(максимум 2000 символов)</i>"),
MessageInput(on_description_input),
state=CreateTestSG.input_description,
),
Window(
Const("<b>🔒 Пароль</b>\n\n🔑 <b>Введите пароль для доступа к тесту</b> или пропустите этот шаг:\n<i>(максимум 255 символов)</i>"),
MessageInput(on_password_input),
Button(Const("⏭️ Без пароля"), id="skip_password", on_click=on_skip_password),
state=CreateTestSG.input_password,
),
Window(
Const("<b>🔄 Количество попыток</b>\n\n🔢 <b>Введите количество попыток</b> (1-100) или пропустите для неограниченного количества:"),
MessageInput(on_attempts_input),
Button(Const("⏭️ Без ограничений"), id="skip_attempts", on_click=on_skip_attempts),
state=CreateTestSG.input_attempts,
),
Window(
Const("<b>📅 Срок действия</b>\n\n🗓 <b>Выберите дату истечения теста</b> или пропустите:"),
Calendar(id="calendar", on_click=on_date_selected),
Button(Const("⏭️ Без срока"), id="skip_expires", on_click=on_skip_expires),
state=CreateTestSG.input_expires_at,
),
Window(
Const("<b>👥 Группа</b>\n\n🎓 <b>Выберите группу</b> или пропустите для всех:"),
ScrollingGroup(
Select(
Format("{item[1]}"),
id="groups",
item_id_getter=lambda x: x[0],
items="groups",
on_click=on_group_selected,
),
id="groups_scroll",
width=2,
height=7,
),
Button(Const("⏭️ Для всех"), id="skip_group", on_click=on_skip_group),
state=CreateTestSG.input_for_group,
getter=get_groups_for_test,
),
Window(
Format("{info}\n\n<b>✅ Подтвердите создание теста:</b>"),
Row(
Button(Const("✅ Создать"), id="confirm", on_click=on_confirm_test),
Button(Const("❌ Отмена"), id="cancel", on_click=on_cancel),
),
state=CreateTestSG.confirm_test_info,
getter=get_test_info,
),
Window(
Format("<b>➕ Добавление вопросов</b>\n\n📊 <b>Вопросов добавлено:</b> {questions_count}\n\n💡 Добавьте вопросы к тесту:"),
Column(
Button(Const(" Добавить вопрос"), id="add_question", on_click=on_add_question),
Button(Const("✅ Завершить создание"), id="finish", on_click=on_finish_test),
),
state=CreateTestSG.add_question,
getter=get_questions_count,
),
Window(
Const("<b>❓ Текст вопроса</b>\n\n📝 <b>Отправьте текст вопроса</b> или 📷 <b>фото с подписью:</b>\n<i>(максимум 2000 символов)</i>"),
MessageInput(on_question_input, content_types=[ContentType.TEXT, ContentType.PHOTO]),
Button(Const("◀️ Назад"), id="back", on_click=on_cancel_question),
state=CreateTestSG.input_question_text,
),
Window(
Const("<b>📋 Тип вопроса</b>\n\n🎯 <b>Выберите тип вопроса:</b>"),
Column(Select(
Format("{item[1]}"),
id="question_type",
item_id_getter=lambda x: x[0],
items="question_types",
on_click=on_question_type_selected,
)),
Button(Const("◀️ Назад"), id="back", on_click=on_cancel_question),
state=CreateTestSG.select_question_type,
getter=get_question_type_data,
),
Window(
Const("<b>✏️ Правильный ответ</b>\n\n💬 <b>Введите правильный ответ</b> (регистр и пробелы игнорируются):\n<i>(максимум 255 символов)</i>"),
MessageInput(on_correct_answer_input),
Button(Const("◀️ Назад"), id="back", on_click=on_cancel_question),
state=CreateTestSG.input_correct_answer,
),
Window(
Format("<b>📝 Варианты ответов</b>\n\n📊 <b>Добавлено вариантов:</b> {options_count}/10\n\n💬 <b>Введите вариант ответа:</b>\n<i>(максимум 255 символов)</i>"),
MessageInput(on_option_input),
Button(Const("✅ Завершить добавление вариантов"), id="finish_options", on_click=on_finish_options),
Button(Const("◀️ Назад"), id="back", on_click=on_cancel_question),
state=CreateTestSG.input_options,
getter=get_options_data,
),
Window(
Const("<b>✅ Правильные ответы</b>\n\n<b>Отметьте правильные варианты ответов:</b>"),
Column(Select(
Format("{item[1]}"),
id="options",
item_id_getter=lambda x: x[0],
items="options",
on_click=on_option_toggle,
)),
Button(Const("✅ Подтвердить выбор"), id="confirm", on_click=on_confirm_correct),
Button(Const("◀️ Назад"), id="back", on_click=on_cancel_question),
state=CreateTestSG.mark_correct_options,
getter=get_options_data,
),
Window(
Format("{preview}\n\n<b>💾 Сохранить вопрос?</b>"),
Row(
Button(Const("✅ Сохранить"), id="save", on_click=on_save_question),
Button(Const("❌ Отмена"), id="cancel", on_click=on_cancel_question),
),
state=CreateTestSG.confirm_question,
getter=get_question_preview,
),
)
@@ -1,30 +1,35 @@
from aiogram.types import CallbackQuery
from aiogram_dialog import Dialog, DialogManager, StartMode, Window
from aiogram_dialog import Dialog, DialogManager, Window
from aiogram_dialog.widgets.kbd import Button, Column
from aiogram_dialog.widgets.text import Const
from trudex.application.bot.creator_dialogs.states import (CreatorBroadcastSG, CreatorGroupsSG, CreatorMenuSG,
CreatorTemplatesSG, CreatorTestsSG, CreatorUsersSG)
from trudex.application.bot.creator_dialogs.states import CreatorMenuSG, CreatorUsersSG
from trudex.application.bot.shared_dialogs.states import (
SharedBroadcastSG,
SharedGroupsSG,
SharedTemplatesSG,
SharedTestsSG,
)
async def on_tests_clicked(_callback: CallbackQuery, _button: Button, manager: DialogManager) -> None:
await manager.start(CreatorTestsSG.tests_list, mode=StartMode.RESET_STACK)
await manager.start(SharedTestsSG.tests_list)
async def on_users_clicked(_callback: CallbackQuery, _button: Button, manager: DialogManager) -> None:
await manager.start(CreatorUsersSG.users_list, mode=StartMode.RESET_STACK)
await manager.start(CreatorUsersSG.users_list)
async def on_groups_clicked(_callback: CallbackQuery, _button: Button, manager: DialogManager) -> None:
await manager.start(CreatorGroupsSG.groups_list, mode=StartMode.RESET_STACK)
await manager.start(SharedGroupsSG.groups_list)
async def on_broadcast_clicked(_callback: CallbackQuery, _button: Button, manager: DialogManager) -> None:
await manager.start(CreatorBroadcastSG.broadcast_input, mode=StartMode.RESET_STACK)
await manager.start(SharedBroadcastSG.broadcast_input)
async def on_templates_clicked(_callback: CallbackQuery, _button: Button, manager: DialogManager) -> None:
await manager.start(CreatorTemplatesSG.main, mode=StartMode.RESET_STACK)
await manager.start(SharedTemplatesSG.main)
creator_menu_dialog = Dialog(
@@ -5,59 +5,9 @@ class CreatorMenuSG(StatesGroup):
main = State()
class CreatorTemplatesSG(StatesGroup):
main = State()
export_list = State()
spec = State()
import_file = State()
class CreatorUsersSG(StatesGroup):
users_list = State()
users_input = State()
user_detail = State()
make_admin_confirm = State()
remove_admin_confirm = State()
class CreatorTestsSG(StatesGroup):
tests_list = State()
test_detail = State()
share_test = State()
edit_menu = State()
edit_password = State()
edit_attempts = State()
edit_group = State()
edit_expires = State()
statistics = State()
attempt_detail = State()
class CreatorBroadcastSG(StatesGroup):
broadcast_input = State()
broadcast_confirm = State()
class CreatorGroupsSG(StatesGroup):
groups_list = State()
add_group_input_number = State()
delete_groups_list = State()
delete_confirm = State()
class CreateTestSG(StatesGroup):
input_title = State()
input_description = State()
input_password = State()
input_attempts = State()
input_expires_at = State()
input_for_group = State()
confirm_test_info = State()
add_question = State()
input_question_text = State()
select_question_type = State()
input_correct_answer = State()
input_options = State()
mark_correct_options = State()
confirm_question = State()
test_created = State()
@@ -1,425 +0,0 @@
import json
from aiogram import Bot
from aiogram.types import BufferedInputFile, CallbackQuery, ContentType, Message
from aiogram_dialog import Dialog, DialogManager, StartMode, Window
from aiogram_dialog.widgets.input import MessageInput
from aiogram_dialog.widgets.kbd import Button, Row, ScrollingGroup, Select
from aiogram_dialog.widgets.text import Const, Format
from dishka import FromDishka
from dishka.integrations.aiogram_dialog import inject
from trudex.application.bot.creator_dialogs.states import CreatorMenuSG, CreatorTemplatesSG, CreatorTestsSG
from trudex.domain.schemas import QuestionType
from trudex.domain.test_parser import ParsedTest, TestParser
from trudex.infrastructure.database.dao.option import OptionDAO
from trudex.infrastructure.database.dao.question import QuestionDAO
from trudex.infrastructure.database.dao.test import TestDAO
from trudex.infrastructure.database.repo.test import TestRepository
TEMPLATES_INFO = (
"<b>📦 Шаблоны тестов</b>\n\n"
"Шаблоны позволяют экспортировать и импортировать тесты в формате JSON.\n\n"
"🔹 <b>Экспорт</b> — сохраните тест как файл для резервной копии или передачи\n"
"🔹 <b>Импорт</b> — загрузите тест из файла\n"
"🔹 <b>Спецификация</b> — описание формата JSON для создания тестов вручную"
)
SPEC_INFO = """<b>📋 Спецификация формата JSON</b>
<b>Структура файла:</b>
<code>{
"title": "Название теста",
"description": "Описание теста",
"password": null,
"attempts": null,
"expires_at": null,
"for_group": null,
"questions": [...]
}</code>
<b>Поля теста:</b>
• <code>title</code> — название (обязательно, до 255 символов)
• <code>description</code> — описание (до 2000 символов)
• <code>password</code> — пароль для доступа или <code>null</code>
• <code>attempts</code> — лимит попыток (1-100) или <code>null</code>
• <code>expires_at</code> — срок действия в ISO формате или <code>null</code>
• <code>for_group</code> — номер группы или <code>null</code> для всех
<b>Типы вопросов:</b>
• <code>single</code> — один правильный ответ
• <code>multiple</code> — несколько правильных ответов
• <code>input</code> — ввод текста (регистр и пробелы игнорируются)
<b>Формат вопроса (single/multiple):</b>
<code>{
"text": "Текст вопроса",
"question_type": "single",
"options": [
{"text": "Вариант 1", "is_correct": true},
{"text": "Вариант 2", "is_correct": false}
]
}</code>
<b>Формат вопроса (input):</b>
<code>{
"text": "Текст вопроса",
"question_type": "input",
"correct_answer": "правильный ответ"
}</code>
<b>⚠️ Важно:</b>
• Для <code>single</code> — ровно один <code>is_correct: true</code>
• Для <code>multiple</code> — один или более <code>is_correct: true</code>
• Минимум 2 варианта ответа для single/multiple"""
TEMPLATE_SINGLE = {
"title": "Пример теста с одиночным выбором",
"description": "Демонстрация формата single вопросов",
"password": None,
"attempts": None,
"expires_at": None,
"for_group": None,
"questions": [
{
"text": "Какой язык программирования используется для разработки Telegram ботов?",
"question_type": "single",
"options": [
{"text": "Python", "is_correct": True},
{"text": "HTML", "is_correct": False},
{"text": "CSS", "is_correct": False},
],
},
],
}
TEMPLATE_MULTIPLE = {
"title": "Пример теста с множественным выбором",
"description": "Демонстрация формата multiple вопросов",
"password": None,
"attempts": None,
"expires_at": None,
"for_group": None,
"questions": [
{
"text": "Выберите языки программирования:",
"question_type": "multiple",
"options": [
{"text": "Python", "is_correct": True},
{"text": "JavaScript", "is_correct": True},
{"text": "HTML", "is_correct": False},
{"text": "CSS", "is_correct": False},
],
},
],
}
TEMPLATE_INPUT = {
"title": "Пример теста с вводом текста",
"description": "Демонстрация формата input вопросов",
"password": None,
"attempts": None,
"expires_at": None,
"for_group": None,
"questions": [
{
"text": "Как называется библиотека для создания Telegram ботов на Python?",
"question_type": "input",
"correct_answer": "aiogram",
},
],
}
TEMPLATE_FULL = {
"title": "Полный пример теста",
"description": "Тест со всеми типами вопросов и настройками",
"password": "secret123",
"attempts": 3,
"expires_at": "2026-12-31T23:59:59",
"for_group": 1234,
"questions": [
{
"text": "Выберите правильный ответ:",
"question_type": "single",
"options": [
{"text": "Вариант A", "is_correct": False},
{"text": "Вариант B", "is_correct": True},
{"text": "Вариант C", "is_correct": False},
],
},
{
"text": "Выберите все правильные ответы:",
"question_type": "multiple",
"options": [
{"text": "Ответ 1", "is_correct": True},
{"text": "Ответ 2", "is_correct": True},
{"text": "Ответ 3", "is_correct": False},
],
},
{
"text": "Введите ответ:",
"question_type": "input",
"correct_answer": "ответ",
},
],
}
async def on_export_clicked(_callback: CallbackQuery, _button: Button, manager: DialogManager) -> None:
await manager.switch_to(CreatorTemplatesSG.export_list)
async def on_import_clicked(_callback: CallbackQuery, _button: Button, manager: DialogManager) -> None:
await manager.switch_to(CreatorTemplatesSG.import_file)
async def on_spec_clicked(_callback: CallbackQuery, _button: Button, manager: DialogManager) -> None:
await manager.switch_to(CreatorTemplatesSG.spec)
async def on_back_clicked(_callback: CallbackQuery, _button: Button, manager: DialogManager) -> None:
await manager.start(CreatorMenuSG.main, mode=StartMode.RESET_STACK)
async def on_back_to_templates(_callback: CallbackQuery, _button: Button, manager: DialogManager) -> None:
await manager.switch_to(CreatorTemplatesSG.main)
@inject
async def get_tests_for_export(test_dao: FromDishka[TestDAO], **_kwargs):
tests = await test_dao.get_all()
return {
"tests": [(f"📝 {t.title}", t.id) for t in tests],
"count": len(tests),
}
@inject
async def on_test_selected_for_export(
_callback: CallbackQuery,
_widget: Select, # type: ignore[type-arg]
_manager: DialogManager,
item_id: str,
test_repo: FromDishka[TestRepository],
) -> None:
test_id = int(item_id)
test, questions_with_options = await test_repo.get_full_test(test_id)
if not test:
await _callback.answer("❌ Тест не найден")
return
export_data: dict = {
"title": test.title,
"description": test.description,
"password": test.password,
"attempts": test.attempts,
"expires_at": test.expires_at.isoformat() if test.expires_at else None,
"for_group": test.for_group,
"questions": [],
}
questions_list: list = export_data["questions"]
for question, options in questions_with_options:
question_data: dict = {
"text": question.text,
"question_type": question.question_type,
}
if question.question_type == QuestionType.INPUT:
correct_options = [o for o in options if o.is_correct]
if correct_options:
question_data["correct_answer"] = correct_options[0].text
else:
question_data["options"] = [
{"text": o.text, "is_correct": o.is_correct}
for o in options
]
questions_list.append(question_data)
json_str = json.dumps(export_data, ensure_ascii=False, indent=2)
safe_title = "".join(c if c.isalnum() or c in "-_" else "_" for c in test.title)[:50]
filename = f"{safe_title}.json"
assert _callback.message is not None
await _callback.message.answer_document(
document=BufferedInputFile(json_str.encode("utf-8"), filename=filename),
caption=f"📤 <b>Экспорт теста:</b> {test.title}",
)
async def send_template(callback: CallbackQuery, template: dict, name: str) -> None:
json_str = json.dumps(template, ensure_ascii=False, indent=2)
filename = f"template_{name}.json"
assert callback.message is not None
await callback.message.answer_document(
document=BufferedInputFile(json_str.encode("utf-8"), filename=filename),
caption=f"📄 <b>Шаблон:</b> {template['title']}",
)
async def on_template_single(_callback: CallbackQuery, _button: Button, _manager: DialogManager) -> None:
await send_template(_callback, TEMPLATE_SINGLE, "single")
async def on_template_multiple(_callback: CallbackQuery, _button: Button, _manager: DialogManager) -> None:
await send_template(_callback, TEMPLATE_MULTIPLE, "multiple")
async def on_template_input(_callback: CallbackQuery, _button: Button, _manager: DialogManager) -> None:
await send_template(_callback, TEMPLATE_INPUT, "input")
async def on_template_full(_callback: CallbackQuery, _button: Button, _manager: DialogManager) -> None:
await send_template(_callback, TEMPLATE_FULL, "full")
async def create_test_from_parsed(
parsed: ParsedTest,
test_dao: TestDAO,
question_dao: QuestionDAO,
option_dao: OptionDAO,
) -> int:
test = await test_dao.create(
title=parsed.title,
description=parsed.description,
password=parsed.password,
attempts=parsed.attempts,
expires_at=parsed.expires_at,
for_group=parsed.for_group,
is_active=False,
)
for position, q in enumerate(parsed.questions):
question = await question_dao.create(
test_id=test.id,
text=q.text,
position=position,
question_type=q.question_type,
)
for opt in q.options:
await option_dao.create(
question_id=question.id,
text=opt.text,
is_correct=opt.is_correct,
)
return test.id
@inject
async def on_import_file(
message: Message,
_widget: MessageInput,
manager: DialogManager,
bot_inst: FromDishka[Bot],
test_dao: FromDishka[TestDAO],
question_dao: FromDishka[QuestionDAO],
option_dao: FromDishka[OptionDAO],
) -> None:
if not message.document:
await message.answer("❌ Отправьте JSON файл")
return
if message.document.file_size and message.document.file_size > 1024 * 1024:
await message.answer("❌ Файл слишком большой (максимум 1 МБ)")
return
file = await bot_inst.get_file(message.document.file_id)
if not file.file_path:
await message.answer("❌ Не удалось загрузить файл")
return
file_bytes = await bot_inst.download_file(file.file_path)
if not file_bytes:
await message.answer("❌ Не удалось загрузить файл")
return
try:
json_str = file_bytes.read().decode("utf-8")
except UnicodeDecodeError:
await message.answer("❌ Файл должен быть в кодировке UTF-8")
return
parser = TestParser()
result = parser.parse(json_str)
if isinstance(result, list):
if not result:
await message.answer("❌ Неизвестная ошибка валидации")
return
error_lines = ["❌ <b>Ошибки валидации:</b>\n"]
for err in result[:10]:
path_str = f" (<code>{err.path}</code>)" if err.path else ""
error_lines.append(f"{err.message}{path_str}")
if len(result) > 10:
error_lines.append(f"\n... и ещё {len(result) - 10} ошибок")
await message.answer("\n".join(error_lines))
return
await create_test_from_parsed(result, test_dao, question_dao, option_dao)
await message.answer(
f"✅ <b>Тест импортирован!</b>\n\n"
f"📝 <b>Название:</b> {result.title}\n"
f"❓ <b>Вопросов:</b> {len(result.questions)}\n\n"
f"Тест создан в деактивированном состоянии."
)
await manager.start(CreatorTestsSG.tests_list, mode=StartMode.RESET_STACK)
templates_dialog = Dialog(
Window(
Const(TEMPLATES_INFO),
Row(
Button(Const("📤 Экспорт"), id="export", on_click=on_export_clicked),
Button(Const("📥 Импорт"), id="import", on_click=on_import_clicked),
),
Button(Const("📋 Спецификация"), id="spec", on_click=on_spec_clicked),
Button(Const("◀️ Назад"), id="back", on_click=on_back_clicked),
state=CreatorTemplatesSG.main,
),
Window(
Format("<b>📤 Экспорт теста</b>\n\nВыберите тест для экспорта:\n\nВсего: {count}"),
ScrollingGroup(
Select(
Format("{item[0]}"),
id="test_select",
item_id_getter=lambda x: x[1],
items="tests",
on_click=on_test_selected_for_export, # type: ignore[arg-type]
),
id="tests_scroll",
width=1,
height=7,
),
Button(Const("◀️ Назад"), id="back", on_click=on_back_to_templates),
state=CreatorTemplatesSG.export_list,
getter=get_tests_for_export,
),
Window(
Const(SPEC_INFO),
Row(
Button(Const("📌 Single"), id="tpl_single", on_click=on_template_single),
Button(Const("📋 Multiple"), id="tpl_multiple", on_click=on_template_multiple),
),
Row(
Button(Const("✏️ Input"), id="tpl_input", on_click=on_template_input),
Button(Const("📦 Полный"), id="tpl_full", on_click=on_template_full),
),
Button(Const("◀️ Назад"), id="back", on_click=on_back_to_templates),
state=CreatorTemplatesSG.spec,
),
Window(
Const("<b>📥 Импорт теста</b>\n\nОтправьте JSON файл с тестом.\n\n<i>Формат файла описан в разделе «Спецификация»</i>"),
MessageInput(on_import_file, content_types=[ContentType.DOCUMENT]),
Button(Const("◀️ Назад"), id="back", on_click=on_back_to_templates),
state=CreatorTemplatesSG.import_file,
),
)
@@ -1,561 +0,0 @@
import asyncio
import functools
from datetime import date, datetime, time
from aiogram import Bot
from aiogram.enums import ContentType
from aiogram.types import BufferedInputFile, CallbackQuery, Message
from aiogram_dialog import Dialog, DialogManager, StartMode, Window
from aiogram_dialog.api.entities import MediaAttachment
from aiogram_dialog.widgets.input import MessageInput
from aiogram_dialog.widgets.kbd import Button, Calendar, Column, Row, ScrollingGroup, Select
from aiogram_dialog.widgets.media import DynamicMedia
from aiogram_dialog.widgets.text import Const, Format
from dishka import FromDishka
from dishka.integrations.aiogram_dialog import inject
from trudex.application.bot.creator_dialogs.states import CreateTestSG, CreatorMenuSG, CreatorTestsSG
from trudex.infrastructure.database.dao.group import GroupDAO
from trudex.infrastructure.database.dao.test import TestDAO
from trudex.infrastructure.database.repo.test import TestRepository
from trudex.infrastructure.database.repo.test_attempt import TestAttemptRepository
from trudex.infrastructure.utils.config import Config
from trudex.infrastructure.utils.qr_generator import generate_qr_bytes
from trudex.infrastructure.utils.test_id_to_hash import encode_id
from trudex.infrastructure.utils.timezone import to_msk
@inject
async def get_tests_data(test_dao: FromDishka[TestDAO], **_kwargs):
tests = await test_dao.get_all()
return {
"tests": [
(f"{'🟢' if t.is_active else '🔴'} {t.title}", t.id)
for t in tests
],
"count": len(tests),
}
async def on_test_selected(_callback: CallbackQuery, _widget: Select, manager: DialogManager, item_id: str):
manager.dialog_data["selected_test_id"] = int(item_id)
await manager.switch_to(CreatorTestsSG.test_detail)
@inject
async def get_test_detail(test_dao: FromDishka[TestDAO], test_repo: FromDishka[TestRepository], dialog_manager: DialogManager, **_kwargs):
test_id = dialog_manager.dialog_data.get("selected_test_id")
if not test_id:
return {
"test_info": "Тест не найден",
"is_active": False,
"button_text": "◀️ Назад",
"results_button_text": "👁 Показать результаты",
}
test = await test_dao.get_by_id(test_id)
questions_count = await test_repo.count_questions_in_test(test_id)
if not test:
return {
"test_info": "Тест не найден",
"is_active": False,
"button_text": "◀️ Назад",
"results_button_text": "👁 Показать результаты",
}
status = "🟢 Активен" if test.is_active else "🔴 Деактивирован"
password_str = f"🔒 {test.password}" if test.password else "🔓 Без пароля"
attempts_str = f"🔄 {test.attempts}" if test.attempts else "♾️ Без ограничений"
expires_str = f"📅 {to_msk(test.expires_at).strftime('%d.%m.%Y %H:%M')}" if test.expires_at else "📅 Без срока"
group_str = f"🎓 Группа {test.for_group}" if test.for_group else "👥 Для всех"
results_str = "👁 Результаты видны" if test.are_results_viewable else "🔒 Результаты скрыты"
test_info = (
f"<b>📝 Информация о тесте</b>\n\n"
f"<b>Название:</b>\n<blockquote>{test.title}</blockquote>\n"
f"<b>Описание:</b>\n<blockquote>{test.description or ''}</blockquote>\n\n"
f"<b>Статус:</b> {status}\n"
f"<b>Вопросов:</b> {questions_count}\n"
f"<b>Пароль:</b> {password_str}\n"
f"<b>Попытки:</b> {attempts_str}\n"
f"<b>Срок:</b> {expires_str}\n"
f"<b>Группа:</b> {group_str}\n"
f"<b>Видимость:</b> {results_str}\n\n"
f"<b>Создан:</b> {to_msk(test.created_at).strftime('%d.%m.%Y %H:%M') if test.created_at else ''}"
)
button_text = "🔴 Деактивировать" if test.is_active else "🟢 Активировать"
results_button_text = "🔒 Скрыть результаты" if test.are_results_viewable else "👁 Показать результаты"
return {
"test_info": test_info,
"is_active": test.is_active,
"button_text": button_text,
"results_button_text": results_button_text,
}
@inject
async def on_toggle_active(_callback: CallbackQuery, _button: Button, manager: DialogManager, test_dao: FromDishka[TestDAO]):
test_id = manager.dialog_data.get("selected_test_id")
if not test_id:
await _callback.answer("❌ Тест не найден")
return
test = await test_dao.get_by_id(test_id)
if test:
await test_dao.update(test_id, is_active=not test.is_active)
action = "деактивирован" if test.is_active else "активирован"
await _callback.answer(f"✅ Тест {action}")
await manager.switch_to(CreatorTestsSG.test_detail)
@inject
async def on_toggle_results_viewable(_callback: CallbackQuery, _button: Button, manager: DialogManager, test_dao: FromDishka[TestDAO]):
test_id = manager.dialog_data.get("selected_test_id")
if not test_id:
await _callback.answer("❌ Тест не найден")
return
test = await test_dao.get_by_id(test_id)
if test:
await test_dao.update(test_id, are_results_viewable=not test.are_results_viewable)
action = "скрыты" if test.are_results_viewable else "видны"
await _callback.answer(f"✅ Результаты теперь {action}")
await manager.switch_to(CreatorTestsSG.test_detail)
async def on_back_to_list(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.switch_to(CreatorTestsSG.tests_list)
async def on_statistics(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.switch_to(CreatorTestsSG.statistics)
@inject
async def get_statistics_data(
dialog_manager: DialogManager,
attempt_repo: FromDishka[TestAttemptRepository],
**_kwargs
):
test_id = dialog_manager.dialog_data.get("selected_test_id")
if not test_id:
return {"attempts": [], "count": 0}
attempts_with_users = await attempt_repo.get_test_attempts_with_users(test_id)
results = []
for attempt, user_name in attempts_with_users:
status = "" if attempt.is_passed else ""
finished_at_msk = to_msk(attempt.finished_at)
date_str = finished_at_msk.strftime("%d.%m.%Y %H:%M") if finished_at_msk else ""
results.append((f"{status} {user_name}{attempt.score}% ({date_str})", attempt.id))
return {
"attempts": results,
"count": len(results),
}
async def on_attempt_selected(_callback: CallbackQuery, _widget: Select, manager: DialogManager, item_id: str):
manager.dialog_data["selected_attempt_id"] = int(item_id)
await manager.switch_to(CreatorTestsSG.attempt_detail)
async def on_back_to_statistics(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.switch_to(CreatorTestsSG.statistics)
@inject
async def get_attempt_detail(
dialog_manager: DialogManager,
attempt_repo: FromDishka[TestAttemptRepository],
test_repo: FromDishka[TestRepository],
**_kwargs
):
attempt_id = dialog_manager.dialog_data.get("selected_attempt_id")
if not attempt_id:
return {"attempt_info": "❌ Результат не найден"}
attempt, answers = await attempt_repo.get_attempt_with_answers(attempt_id)
if not attempt:
return {"attempt_info": "❌ Результат не найден"}
status = "✅ Пройден" if attempt.is_passed else "❌ Не пройден"
finished_at_msk = to_msk(attempt.finished_at)
date_str = finished_at_msk.strftime("%d.%m.%Y %H:%M") if finished_at_msk else ""
lines = [
f"<b>📊 Результат прохождения</b>\n",
f"📈 <b>Результат:</b> {attempt.score}%",
f"📅 <b>Дата:</b> {date_str}",
f"🏆 <b>Статус:</b> {status}\n",
"<b>📋 Ответы:</b>\n",
]
for i, answer in enumerate(answers, 1):
question, options = await test_repo.get_question_with_options(answer.question_id)
if not question:
continue
correct_options = [opt for opt in options if opt.is_correct]
correct_texts = [opt.text for opt in correct_options]
status_icon = "" if answer.is_correct else ""
user_answer = answer.text_answer or ""
if "|" in user_answer:
user_answer = ", ".join(user_answer.split("|"))
lines.append(f"{status_icon} <b>Вопрос {i}</b>")
lines.append(f"<blockquote>{question.text}</blockquote>")
lines.append(f"👤 <i>Ответ:</i> {user_answer or ''}")
lines.append(f"✓ <i>Правильно:</i> {', '.join(correct_texts)}\n")
return {"attempt_info": "\n".join(lines)}
@inject
async def on_share_test(_callback: CallbackQuery, _button: Button, manager: DialogManager, config: FromDishka[Config], bot_inst: FromDishka[Bot]):
test_id = manager.dialog_data.get("selected_test_id")
if not test_id:
return {
"share_link": "Ошибка: тест не найден"
}
test_hash = encode_id(
test_id,
config.security.encode_key,
config.security.encoded_string_length
)
bot_info = await bot_inst.get_me()
bot_username = bot_info.username or "your_bot"
share_link = f"https://t.me/{bot_username}?start={test_hash}"
loop = asyncio.get_running_loop()
qr_bytes = await loop.run_in_executor(
None,
functools.partial(generate_qr_bytes, share_link)
)
assert _callback.message is not None
await _callback.message.answer_photo(
photo=BufferedInputFile(qr_bytes, filename="qr.png"),
caption=f"<b>🔗 Поделиться тестом</b>\n\n📎 <b>Ссылка на тест:</b>\n<code>{share_link}</code>\n\n💡 Отправьте эту ссылку или QR-код пользователям для прохождения теста"
)
async def on_edit_menu(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.switch_to(CreatorTestsSG.edit_menu)
async def on_back_to_detail(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.switch_to(CreatorTestsSG.test_detail)
async def on_back_to_edit_menu(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.switch_to(CreatorTestsSG.edit_menu)
async def on_edit_password(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.switch_to(CreatorTestsSG.edit_password)
async def on_edit_attempts(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.switch_to(CreatorTestsSG.edit_attempts)
async def on_edit_group(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.switch_to(CreatorTestsSG.edit_group)
async def on_edit_expires(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.switch_to(CreatorTestsSG.edit_expires)
@inject
async def on_password_input(message: Message, _widget: MessageInput, manager: DialogManager, test_dao: FromDishka[TestDAO]):
test_id = manager.dialog_data.get("selected_test_id")
if not test_id:
await message.answer("❌ Тест не найден")
return
if not message.text:
await message.answer("❌ Пароль не может быть пустым")
return
password = message.text.strip()
if len(password) > 255:
await message.answer("❌ Пароль слишком длинный (максимум 255 символов)")
return
await test_dao.update(test_id, password=password)
await message.answer("✅ Пароль обновлен")
await manager.switch_to(CreatorTestsSG.test_detail)
@inject
async def on_remove_password(_callback: CallbackQuery, _button: Button, manager: DialogManager, test_dao: FromDishka[TestDAO]):
test_id = manager.dialog_data.get("selected_test_id")
if not test_id:
await _callback.answer("❌ Тест не найден")
return
await test_dao.update(test_id, password=None)
await _callback.answer("✅ Пароль удален")
await manager.switch_to(CreatorTestsSG.test_detail)
@inject
async def on_attempts_input_edit(message: Message, _widget: MessageInput, manager: DialogManager, test_dao: FromDishka[TestDAO]):
test_id = manager.dialog_data.get("selected_test_id")
if not test_id:
await message.answer("❌ Тест не найден")
return
if not message.text:
await message.answer("❌ Количество попыток не может быть пустым")
return
attempts_str = message.text.strip()
if not attempts_str.isdigit():
await message.answer("❌ Количество попыток должно быть числом")
return
attempts = int(attempts_str)
if attempts < 1:
await message.answer("❌ Количество попыток должно быть больше 0")
return
if attempts > 100:
await message.answer("❌ Количество попыток не может быть больше 100")
return
await test_dao.update(test_id, attempts=attempts)
await message.answer("✅ Количество попыток обновлено")
await manager.switch_to(CreatorTestsSG.test_detail)
@inject
async def on_remove_attempts(_callback: CallbackQuery, _button: Button, manager: DialogManager, test_dao: FromDishka[TestDAO]):
test_id = manager.dialog_data.get("selected_test_id")
if not test_id:
await _callback.answer("❌ Тест не найден")
return
await test_dao.update(test_id, attempts=None)
await _callback.answer("✅ Ограничение попыток удалено")
await manager.switch_to(CreatorTestsSG.test_detail)
@inject
async def get_groups_for_edit(dialog_manager: DialogManager, group_dao: FromDishka[GroupDAO], **_kwargs):
groups = await group_dao.get_all()
return {
"groups": [(str(g.number), str(g.number)) for g in groups],
}
@inject
async def on_group_selected_for_test(_callback: CallbackQuery, _widget, manager: DialogManager, item_id: str, test_dao: FromDishka[TestDAO]):
test_id = manager.dialog_data.get("selected_test_id")
if not test_id:
await _callback.answer("❌ Тест не найден")
return
await test_dao.update(test_id, for_group=int(item_id))
await _callback.answer("✅ Группа обновлена")
await manager.switch_to(CreatorTestsSG.test_detail)
@inject
async def on_remove_group(_callback: CallbackQuery, _button: Button, manager: DialogManager, test_dao: FromDishka[TestDAO]):
test_id = manager.dialog_data.get("selected_test_id")
if not test_id:
await _callback.answer("❌ Тест не найден")
return
await test_dao.update(test_id, for_group=None)
await _callback.answer("✅ Тест теперь доступен для всех групп")
await manager.switch_to(CreatorTestsSG.test_detail)
@inject
async def on_date_selected_for_test(_callback, _widget, manager: DialogManager, selected_date: date, test_dao: FromDishka[TestDAO]):
test_id = manager.dialog_data.get("selected_test_id")
if not test_id:
await _callback.answer("❌ Тест не найден")
return
expires_at = datetime.combine(selected_date, time.min)
await test_dao.update(test_id, expires_at=expires_at)
await _callback.answer("✅ Срок действия обновлен")
await manager.switch_to(CreatorTestsSG.test_detail)
@inject
async def on_remove_expires(_callback: CallbackQuery, _button: Button, manager: DialogManager, test_dao: FromDishka[TestDAO]):
test_id = manager.dialog_data.get("selected_test_id")
if not test_id:
await _callback.answer("❌ Тест не найден")
return
await test_dao.update(test_id, expires_at=None)
await _callback.answer("✅ Срок действия удален")
await manager.switch_to(CreatorTestsSG.test_detail)
async def on_add_test_clicked(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.start(CreateTestSG.input_title, mode=StartMode.RESET_STACK)
async def on_back_clicked(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.start(CreatorMenuSG.main, mode=StartMode.RESET_STACK)
tests_dialog = Dialog(
Window(
Format("<b>📝 Тесты</b>\n\nВсего: {count}"),
ScrollingGroup(
Select(
Format("{item[0]}"),
id="test_select",
item_id_getter=lambda x: x[1],
items="tests",
on_click=on_test_selected,
),
id="tests_scroll",
width=1,
height=7,
),
Column(
Button(Const(" Добавить тест"), id="add_test", on_click=on_add_test_clicked),
Button(Const("◀️ Назад"), id="back", on_click=on_back_clicked),
),
state=CreatorTestsSG.tests_list,
getter=get_tests_data,
),
Window(
Format("{test_info}"),
Column(
Button(
Format("{button_text}"),
id="toggle_active",
on_click=on_toggle_active
),
Button(
Format("{results_button_text}"),
id="toggle_results",
on_click=on_toggle_results_viewable
),
Button(Const("📊 Статистика"), id="statistics", on_click=on_statistics),
Button(Const("🔗 Поделиться"), id="share", on_click=on_share_test),
Button(Const("✏️ Изменить"), id="edit_menu", on_click=on_edit_menu),
Button(Const("◀️ Назад"), id="back", on_click=on_back_to_list),
),
state=CreatorTestsSG.test_detail,
getter=get_test_detail,
),
Window(
Const("<b>✏️ Изменить тест</b>\n\nВыберите, что хотите изменить:"),
Column(
Button(Const("🔑 Пароль"), id="edit_password", on_click=on_edit_password),
Button(Const("🔄 Попытки"), id="edit_attempts", on_click=on_edit_attempts),
Button(Const("👥 Группа"), id="edit_group", on_click=on_edit_group),
Button(Const("📅 Срок действия"), id="edit_expires", on_click=on_edit_expires),
Button(Const("◀️ Назад"), id="back", on_click=on_back_to_detail),
),
state=CreatorTestsSG.edit_menu,
),
Window(
Const("<b>🔑 Изменение пароля</b>\n\n💬 <b>Введите новый пароль</b> или удалите текущий:\n<i>(максимум 255 символов)</i>"),
MessageInput(on_password_input),
Column(
Button(Const("🗑 Удалить пароль"), id="remove_password", on_click=on_remove_password),
Button(Const("◀️ Назад"), id="back", on_click=on_back_to_edit_menu),
),
state=CreatorTestsSG.edit_password,
),
Window(
Const("<b>🔄 Изменение количества попыток</b>\n\n🔢 <b>Введите новое количество попыток</b> (1-100) или удалите ограничение:"),
MessageInput(on_attempts_input_edit),
Column(
Button(Const("🗑 Без ограничений"), id="remove_attempts", on_click=on_remove_attempts),
Button(Const("◀️ Назад"), id="back", on_click=on_back_to_edit_menu),
),
state=CreatorTestsSG.edit_attempts,
),
Window(
Const("<b>👥 Изменение группы</b>\n\n🎓 <b>Выберите группу</b> или удалите привязку:"),
ScrollingGroup(
Select(
Format("{item[1]}"),
id="groups",
item_id_getter=lambda x: x[0],
items="groups",
on_click=on_group_selected_for_test,
),
id="groups_scroll",
width=2,
height=7,
),
Column(
Button(Const("🗑 Для всех групп"), id="remove_group", on_click=on_remove_group),
Button(Const("◀️ Назад"), id="back", on_click=on_back_to_edit_menu),
),
state=CreatorTestsSG.edit_group,
getter=get_groups_for_edit,
),
Window(
Const("<b>📅 Изменение срока действия</b>\n\n🗓 <b>Выберите новую дату</b> или удалите срок:"),
Calendar(id="calendar", on_click=on_date_selected_for_test),
Column(
Button(Const("🗑 Удалить срок"), id="remove_expires", on_click=on_remove_expires),
Button(Const("◀️ Назад"), id="back", on_click=on_back_to_edit_menu),
),
state=CreatorTestsSG.edit_expires,
),
Window(
Format("<b>📊 Статистика теста</b>\n\nПрошли тест: {count}"),
ScrollingGroup(
Select(
Format("{item[0]}"),
id="attempt_select",
item_id_getter=lambda x: x[1],
items="attempts",
on_click=on_attempt_selected,
),
id="attempts_scroll",
width=1,
height=7,
),
Column(
Button(Const("🔄 Обновить"), id="refresh", on_click=on_statistics),
Button(Const("◀️ Назад"), id="back", on_click=on_back_to_detail),
),
state=CreatorTestsSG.statistics,
getter=get_statistics_data,
),
Window(
Format("{attempt_info}"),
Button(Const("◀️ Назад"), id="back", on_click=on_back_to_statistics),
state=CreatorTestsSG.attempt_detail,
getter=get_attempt_detail,
),
)
@@ -2,14 +2,14 @@ import asyncio
from aiogram import Bot
from aiogram.types import CallbackQuery, Message
from aiogram_dialog import Dialog, DialogManager, StartMode, Window
from aiogram_dialog import Dialog, DialogManager, Window
from aiogram_dialog.widgets.input import MessageInput
from aiogram_dialog.widgets.kbd import Button, Column, Row, ScrollingGroup, Select, SwitchTo
from aiogram_dialog.widgets.text import Const, Format
from dishka import FromDishka
from dishka.integrations.aiogram_dialog import inject
from trudex.application.bot.creator_dialogs.states import CreatorMenuSG, CreatorUsersSG
from trudex.application.bot.creator_dialogs.states import CreatorUsersSG
from trudex.infrastructure.database.dao.user import UserDAO
from trudex.infrastructure.database.repo.user import UserRepository
from trudex.infrastructure.utils.bot_commands import setup_bot_commands
@@ -19,7 +19,7 @@ from trudex.infrastructure.utils.config import Config
@inject
async def get_users_data(user_dao: FromDishka[UserDAO], **_kwargs):
users = await user_dao.get_all()
users_sorted = sorted(users, key=lambda u: u.created_at, reverse=True)
users_sorted = sorted(users, key=lambda u: u.created_at or u.id, reverse=True)
return {
"users": [
@@ -206,10 +206,10 @@ async def on_confirm_no(_callback: CallbackQuery, _button: Button, manager: Dial
async def on_back_to_main(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.start(CreatorMenuSG.main, mode=StartMode.RESET_STACK)
await manager.done()
users_dialog = Dialog(
creator_users_dialog = Dialog(
Window(
Format("<b>👥 Пользователи</b>\n\nВсего: {count}"),
ScrollingGroup(
@@ -1,12 +1,12 @@
from aiogram.types import CallbackQuery, Message
from aiogram_dialog import Dialog, DialogManager, StartMode, Window
from aiogram_dialog import Dialog, DialogManager, Window
from aiogram_dialog.widgets.input import MessageInput
from aiogram_dialog.widgets.kbd import Button, Row
from aiogram_dialog.widgets.text import Const
from dishka import FromDishka
from dishka.integrations.aiogram_dialog import inject
from trudex.application.bot.admin_dialogs.states import AdminBroadcastSG, AdminMenuSG
from trudex.application.bot.shared_dialogs.states import SharedBroadcastSG
from trudex.infrastructure.database.dao.user import UserDAO
from trudex.infrastructure.utils.broadcast import broadcast_message
@@ -14,7 +14,7 @@ from trudex.infrastructure.utils.broadcast import broadcast_message
async def on_broadcast_input(message: Message, _widget: MessageInput, manager: DialogManager):
manager.dialog_data["broadcast_message_id"] = message.message_id
manager.dialog_data["broadcast_chat_id"] = message.chat.id
await manager.switch_to(AdminBroadcastSG.broadcast_confirm)
await manager.switch_to(SharedBroadcastSG.broadcast_confirm)
@inject
@@ -48,19 +48,19 @@ async def on_broadcast_confirm(_callback: CallbackQuery, _button: Button, manage
async def on_broadcast_cancel(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await _callback.answer("Рассылка отменена")
await manager.start(AdminMenuSG.main, mode=StartMode.RESET_STACK)
await manager.done()
async def on_back_to_main(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.start(AdminMenuSG.main, mode=StartMode.RESET_STACK)
await manager.done()
broadcast_dialog = Dialog(
shared_broadcast_dialog = Dialog(
Window(
Const("<b>📢 Рассылка</b>\n\nОтправьте сообщение, которое хотите разослать всем пользователям:"),
MessageInput(on_broadcast_input),
Button(Const("◀️ Отмена"), id="back", on_click=on_back_to_main),
state=AdminBroadcastSG.broadcast_input,
state=SharedBroadcastSG.broadcast_input,
),
Window(
Const("<b>⚠️ Подтверждение рассылки</b>\n\nВы уверены, что хотите отправить это сообщение всем пользователям?"),
@@ -68,6 +68,6 @@ broadcast_dialog = Dialog(
Button(Const("✅ Да"), id="broadcast_confirm", on_click=on_broadcast_confirm),
Button(Const("❌ Нет"), id="broadcast_cancel", on_click=on_broadcast_cancel),
),
state=AdminBroadcastSG.broadcast_confirm,
state=SharedBroadcastSG.broadcast_confirm,
),
)
@@ -8,7 +8,7 @@ from aiogram_dialog.widgets.text import Const, Format
from dishka import FromDishka
from dishka.integrations.aiogram_dialog import inject
from trudex.application.bot.admin_dialogs.states import AdminCreateTestSG, AdminTestsSG
from trudex.application.bot.shared_dialogs.states import SharedCreateTestSG, SharedTestsSG
from trudex.infrastructure.database.dao.group import GroupDAO
from trudex.infrastructure.database.dao.option import OptionDAO
from trudex.infrastructure.database.dao.question import QuestionDAO
@@ -32,7 +32,7 @@ async def on_title_input(message: Message, _widget: MessageInput, manager: Dialo
return
manager.dialog_data["title"] = title
await manager.switch_to(AdminCreateTestSG.input_description)
await manager.switch_to(SharedCreateTestSG.input_description)
async def on_description_input(message: Message, _widget: MessageInput, manager: DialogManager):
@@ -50,7 +50,7 @@ async def on_description_input(message: Message, _widget: MessageInput, manager:
return
manager.dialog_data["description"] = description
await manager.switch_to(AdminCreateTestSG.input_password)
await manager.switch_to(SharedCreateTestSG.input_password)
@inject
@@ -69,13 +69,13 @@ async def on_password_input(message: Message, _widget: MessageInput, manager: Di
return
manager.dialog_data["password"] = password
await manager.switch_to(AdminCreateTestSG.input_attempts)
await manager.switch_to(SharedCreateTestSG.input_attempts)
@inject
async def on_skip_password(_callback: CallbackQuery, _button: Button, manager: DialogManager, _group_dao: FromDishka[GroupDAO]):
manager.dialog_data["password"] = None
await manager.switch_to(AdminCreateTestSG.input_attempts)
await manager.switch_to(SharedCreateTestSG.input_attempts)
async def on_attempts_input(message: Message, _widget: MessageInput, manager: DialogManager):
@@ -100,41 +100,38 @@ async def on_attempts_input(message: Message, _widget: MessageInput, manager: Di
return
manager.dialog_data["attempts"] = attempts
await manager.switch_to(AdminCreateTestSG.input_expires_at)
await manager.switch_to(SharedCreateTestSG.input_expires_at)
async def on_skip_attempts(_callback: CallbackQuery, _button: Button, manager: DialogManager):
manager.dialog_data["attempts"] = None
await manager.switch_to(AdminCreateTestSG.input_expires_at)
await manager.switch_to(SharedCreateTestSG.input_expires_at)
async def on_date_selected(_callback, _widget, manager: DialogManager, selected_date: date):
manager.dialog_data["expires_at"] = datetime.combine(selected_date, time.min)
await manager.switch_to(AdminCreateTestSG.input_for_group)
await manager.switch_to(SharedCreateTestSG.input_for_group)
async def on_skip_expires(_callback: CallbackQuery, _button: Button, manager: DialogManager):
manager.dialog_data["expires_at"] = None
await manager.switch_to(AdminCreateTestSG.input_for_group)
await manager.switch_to(SharedCreateTestSG.input_for_group)
@inject
async def get_groups_for_test(group_dao: FromDishka[GroupDAO], **_kwargs):
groups = await group_dao.get_all()
return {
"groups": [(str(g.number), str(g.number)) for g in groups],
}
return {"groups": [(str(g.number), str(g.number)) for g in groups]}
async def on_group_selected(_callback: CallbackQuery, _widget, manager: DialogManager, item_id: str):
manager.dialog_data["for_group"] = int(item_id)
await manager.switch_to(AdminCreateTestSG.confirm_test_info)
await manager.switch_to(SharedCreateTestSG.confirm_test_info)
async def on_skip_group(_callback: CallbackQuery, _button: Button, manager: DialogManager):
manager.dialog_data["for_group"] = None
await manager.switch_to(AdminCreateTestSG.confirm_test_info)
await manager.switch_to(SharedCreateTestSG.confirm_test_info)
async def get_test_info(dialog_manager: DialogManager, **_kwargs):
@@ -185,12 +182,12 @@ async def on_confirm_test(_callback: CallbackQuery, _button: Button, manager: Di
manager.dialog_data["test_id"] = test.id
manager.dialog_data["questions"] = []
await manager.switch_to(AdminCreateTestSG.add_question)
await manager.switch_to(SharedCreateTestSG.add_question)
async def on_add_question(_callback: CallbackQuery, _button: Button, manager: DialogManager):
manager.dialog_data["current_question"] = {}
await manager.switch_to(AdminCreateTestSG.input_question_text)
await manager.switch_to(SharedCreateTestSG.input_question_text)
async def on_question_input(message: Message, _widget: MessageInput, manager: DialogManager):
@@ -223,7 +220,7 @@ async def on_question_input(message: Message, _widget: MessageInput, manager: Di
return
manager.dialog_data["current_question"] = current_question
await manager.switch_to(AdminCreateTestSG.select_question_type)
await manager.switch_to(SharedCreateTestSG.select_question_type)
async def get_question_type_data(**_kwargs):
@@ -242,10 +239,10 @@ async def on_question_type_selected(_callback: CallbackQuery, _widget, manager:
manager.dialog_data["current_question"] = current_question
if item_id == "input":
await manager.switch_to(AdminCreateTestSG.input_correct_answer)
await manager.switch_to(SharedCreateTestSG.input_correct_answer)
else:
manager.dialog_data["current_options"] = []
await manager.switch_to(AdminCreateTestSG.input_options)
await manager.switch_to(SharedCreateTestSG.input_options)
async def on_correct_answer_input(message: Message, _widget: MessageInput, manager: DialogManager):
@@ -265,7 +262,7 @@ async def on_correct_answer_input(message: Message, _widget: MessageInput, manag
current_question = manager.dialog_data.get("current_question", {})
current_question["correct_answer"] = answer
manager.dialog_data["current_question"] = current_question
await manager.switch_to(AdminCreateTestSG.confirm_question)
await manager.switch_to(SharedCreateTestSG.confirm_question)
async def on_option_input(message: Message, _widget: MessageInput, manager: DialogManager):
@@ -300,7 +297,7 @@ async def on_finish_options(_callback: CallbackQuery, _button: Button, manager:
await _callback.answer("❌ Добавьте минимум 2 варианта ответа", show_alert=True)
return
await manager.switch_to(AdminCreateTestSG.mark_correct_options)
await manager.switch_to(SharedCreateTestSG.mark_correct_options)
async def get_options_data(dialog_manager: DialogManager, **_kwargs):
@@ -340,7 +337,7 @@ async def on_confirm_correct(_callback: CallbackQuery, _button: Button, manager:
await _callback.answer("❌ Отметьте хотя бы один правильный ответ", show_alert=True)
return
await manager.switch_to(AdminCreateTestSG.confirm_question)
await manager.switch_to(SharedCreateTestSG.confirm_question)
async def get_question_preview(dialog_manager: DialogManager, **_kwargs):
@@ -357,7 +354,7 @@ async def get_question_preview(dialog_manager: DialogManager, **_kwargs):
"input": "✏️ Ввод текста",
}
preview = f"<b>📝 Предпросмотр вопроса</b>\n\n"
preview = "<b>📝 Предпросмотр вопроса</b>\n\n"
preview += f"<b>Текст:</b> {text}\n"
preview += f"<b>Тип:</b> {type_names[question_type]}\n"
preview += f"<b>Изображение:</b> {'✅ Да' if has_image else '❌ Нет'}\n\n"
@@ -420,13 +417,13 @@ async def on_save_question(
manager.dialog_data.pop("current_options", None)
await _callback.answer("✅ Вопрос добавлен")
await manager.switch_to(AdminCreateTestSG.add_question)
await manager.switch_to(SharedCreateTestSG.add_question)
async def on_cancel_question(_callback: CallbackQuery, _button: Button, manager: DialogManager):
manager.dialog_data.pop("current_question", None)
manager.dialog_data.pop("current_options", None)
await manager.switch_to(AdminCreateTestSG.add_question)
await manager.switch_to(SharedCreateTestSG.add_question)
async def get_questions_count(dialog_manager: DialogManager, **_kwargs):
@@ -442,42 +439,42 @@ async def on_finish_test(_callback: CallbackQuery, _button: Button, manager: Dia
return
await _callback.answer("✅ Тест создан")
await manager.start(AdminTestsSG.tests_list, mode=StartMode.RESET_STACK)
await manager.start(SharedTestsSG.tests_list, mode=StartMode.RESET_STACK)
async def on_cancel(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.start(AdminTestsSG.tests_list, mode=StartMode.RESET_STACK)
await manager.start(SharedTestsSG.tests_list, mode=StartMode.RESET_STACK)
admin_create_test_dialog = Dialog(
shared_create_test_dialog = Dialog(
Window(
Const("<b>📝 Создание теста</b>\n\n💬 <b>Введите название теста:</b>\n<i>(максимум 255 символов)</i>"),
MessageInput(on_title_input),
Cancel(Const("◀️ Отмена")),
state=AdminCreateTestSG.input_title,
state=SharedCreateTestSG.input_title,
),
Window(
Const("<b>📝 Создание теста</b>\n\n📄 <b>Введите описание теста:</b>\n<i>(максимум 2000 символов)</i>"),
MessageInput(on_description_input),
state=AdminCreateTestSG.input_description,
state=SharedCreateTestSG.input_description,
),
Window(
Const("<b>🔒 Пароль</b>\n\n🔑 <b>Введите пароль для доступа к тесту</b> или пропустите этот шаг:\n<i>(максимум 255 символов)</i>"),
MessageInput(on_password_input),
Button(Const("⏭️ Без пароля"), id="skip_password", on_click=on_skip_password),
state=AdminCreateTestSG.input_password,
state=SharedCreateTestSG.input_password,
),
Window(
Const("<b>🔄 Количество попыток</b>\n\n🔢 <b>Введите количество попыток</b> (1-100) или пропустите для неограниченного количества:"),
MessageInput(on_attempts_input),
Button(Const("⏭️ Без ограничений"), id="skip_attempts", on_click=on_skip_attempts),
state=AdminCreateTestSG.input_attempts,
state=SharedCreateTestSG.input_attempts,
),
Window(
Const("<b>📅 Срок действия</b>\n\n🗓 <b>Выберите дату истечения теста</b> или пропустите:"),
Calendar(id="calendar", on_click=on_date_selected),
Button(Const("⏭️ Без срока"), id="skip_expires", on_click=on_skip_expires),
state=AdminCreateTestSG.input_expires_at,
state=SharedCreateTestSG.input_expires_at,
),
Window(
Const("<b>👥 Группа</b>\n\n🎓 <b>Выберите группу</b> или пропустите для всех:"),
@@ -494,7 +491,7 @@ admin_create_test_dialog = Dialog(
height=7,
),
Button(Const("⏭️ Для всех"), id="skip_group", on_click=on_skip_group),
state=AdminCreateTestSG.input_for_group,
state=SharedCreateTestSG.input_for_group,
getter=get_groups_for_test,
),
Window(
@@ -503,7 +500,7 @@ admin_create_test_dialog = Dialog(
Button(Const("✅ Создать"), id="confirm", on_click=on_confirm_test),
Button(Const("❌ Отмена"), id="cancel", on_click=on_cancel),
),
state=AdminCreateTestSG.confirm_test_info,
state=SharedCreateTestSG.confirm_test_info,
getter=get_test_info,
),
Window(
@@ -512,14 +509,14 @@ admin_create_test_dialog = Dialog(
Button(Const(" Добавить вопрос"), id="add_question", on_click=on_add_question),
Button(Const("✅ Завершить создание"), id="finish", on_click=on_finish_test),
),
state=AdminCreateTestSG.add_question,
state=SharedCreateTestSG.add_question,
getter=get_questions_count,
),
Window(
Const("<b>❓ Текст вопроса</b>\n\n📝 <b>Отправьте текст вопроса</b> или 📷 <b>фото с подписью:</b>\n<i>(максимум 2000 символов)</i>"),
MessageInput(on_question_input, content_types=[ContentType.TEXT, ContentType.PHOTO]),
Button(Const("◀️ Назад"), id="back", on_click=on_cancel_question),
state=AdminCreateTestSG.input_question_text,
state=SharedCreateTestSG.input_question_text,
),
Window(
Const("<b>📋 Тип вопроса</b>\n\n🎯 <b>Выберите тип вопроса:</b>"),
@@ -531,21 +528,21 @@ admin_create_test_dialog = Dialog(
on_click=on_question_type_selected,
)),
Button(Const("◀️ Назад"), id="back", on_click=on_cancel_question),
state=AdminCreateTestSG.select_question_type,
state=SharedCreateTestSG.select_question_type,
getter=get_question_type_data,
),
Window(
Const("<b>✏️ Правильный ответ</b>\n\n💬 <b>Введите правильный ответ</b> (регистр и пробелы игнорируются):\n<i>(максимум 255 символов)</i>"),
MessageInput(on_correct_answer_input),
Button(Const("◀️ Назад"), id="back", on_click=on_cancel_question),
state=AdminCreateTestSG.input_correct_answer,
state=SharedCreateTestSG.input_correct_answer,
),
Window(
Format("<b>📝 Варианты ответов</b>\n\n📊 <b>Добавлено вариантов:</b> {options_count}/10\n\n💬 <b>Введите вариант ответа:</b>\n<i>(максимум 255 символов)</i>"),
MessageInput(on_option_input),
Button(Const("✅ Завершить добавление вариантов"), id="finish_options", on_click=on_finish_options),
Button(Const("◀️ Назад"), id="back", on_click=on_cancel_question),
state=AdminCreateTestSG.input_options,
state=SharedCreateTestSG.input_options,
getter=get_options_data,
),
Window(
@@ -559,7 +556,7 @@ admin_create_test_dialog = Dialog(
)),
Button(Const("✅ Подтвердить выбор"), id="confirm", on_click=on_confirm_correct),
Button(Const("◀️ Назад"), id="back", on_click=on_cancel_question),
state=AdminCreateTestSG.mark_correct_options,
state=SharedCreateTestSG.mark_correct_options,
getter=get_options_data,
),
Window(
@@ -568,7 +565,7 @@ admin_create_test_dialog = Dialog(
Button(Const("✅ Сохранить"), id="save", on_click=on_save_question),
Button(Const("❌ Отмена"), id="cancel", on_click=on_cancel_question),
),
state=AdminCreateTestSG.confirm_question,
state=SharedCreateTestSG.confirm_question,
getter=get_question_preview,
),
)
@@ -1,12 +1,12 @@
from aiogram.types import CallbackQuery, Message
from aiogram_dialog import Dialog, DialogManager, StartMode, Window
from aiogram_dialog import Dialog, DialogManager, Window
from aiogram_dialog.widgets.input import MessageInput
from aiogram_dialog.widgets.kbd import Button, Column, Row, ScrollingGroup, Select
from aiogram_dialog.widgets.text import Const, Format
from dishka import FromDishka
from dishka.integrations.aiogram_dialog import inject
from trudex.application.bot.creator_dialogs.states import CreatorGroupsSG, CreatorMenuSG
from trudex.application.bot.shared_dialogs.states import SharedGroupsSG
from trudex.infrastructure.database.dao.group import GroupDAO
@@ -33,16 +33,15 @@ async def get_groups_data(group_dao: FromDishka[GroupDAO], dialog_manager: Dialo
async def on_add_group(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.switch_to(CreatorGroupsSG.add_group_input_number)
await manager.switch_to(SharedGroupsSG.add_group_input_number)
async def on_delete_group(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.switch_to(CreatorGroupsSG.delete_groups_list)
await manager.switch_to(SharedGroupsSG.delete_groups_list)
async def on_back_to_menu(_callback: CallbackQuery, _button: Button, manager: DialogManager):
from trudex.application.bot.creator_dialogs.states import CreatorMenuSG
await manager.start(CreatorMenuSG.main, mode=StartMode.RESET_STACK)
await manager.done()
@inject
@@ -68,18 +67,13 @@ async def on_group_number_input(message: Message, _widget: MessageInput, manager
await message.answer(f"❌ Группа с номером {number} уже существует")
return
try:
await group_dao.create(number=number)
manager.dialog_data["success_message"] = f"✅ Группа {number} создана"
except Exception as e:
await message.answer(f"❌ Ошибка создания группы: {e}")
return
await manager.switch_to(CreatorGroupsSG.groups_list)
await group_dao.create(number=number)
manager.dialog_data["success_message"] = f"✅ Группа {number} создана"
await manager.switch_to(SharedGroupsSG.groups_list)
async def on_cancel_add(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.switch_to(CreatorGroupsSG.groups_list)
await manager.switch_to(SharedGroupsSG.groups_list)
@inject
@@ -101,34 +95,29 @@ async def on_select_group_to_delete(_callback: CallbackQuery, _widget, manager:
manager.dialog_data["delete_group_id"] = group.id
manager.dialog_data["delete_group_number"] = group.number
await manager.switch_to(CreatorGroupsSG.delete_confirm)
await manager.switch_to(SharedGroupsSG.delete_confirm)
async def get_delete_confirm_data(dialog_manager: DialogManager, **_kwargs):
number = dialog_manager.dialog_data.get("delete_group_number", "")
return {
"group_info": str(number)
}
return {"group_info": str(number)}
@inject
async def on_confirm_delete(_callback: CallbackQuery, _button: Button, manager: DialogManager, group_dao: FromDishka[GroupDAO]):
group_id = manager.dialog_data.get("delete_group_id")
assert isinstance(group_id, int)
await group_dao.delete(group_id)
manager.dialog_data["success_message"] = "✅ Группа удалена"
await manager.switch_to(CreatorGroupsSG.groups_list)
await manager.switch_to(SharedGroupsSG.groups_list)
async def on_cancel_delete(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.switch_to(CreatorGroupsSG.delete_groups_list)
await manager.switch_to(SharedGroupsSG.delete_groups_list)
groups_dialog = Dialog(
shared_groups_dialog = Dialog(
Window(
Format("{message_text}"),
ScrollingGroup(
@@ -148,14 +137,14 @@ groups_dialog = Dialog(
Button(Const("🗑 Удалить группу"), id="delete", on_click=on_delete_group),
Button(Const("◀️ Назад"), id="back", on_click=on_back_to_menu),
),
state=CreatorGroupsSG.groups_list,
state=SharedGroupsSG.groups_list,
getter=get_groups_data,
),
Window(
Const("<b>➕ Добавление группы</b>\n\n🔢 <b>Введите номер группы</b> (четырехзначное число 1000-9999):"),
MessageInput(on_group_number_input),
Button(Const("◀️ Отмена"), id="cancel", on_click=on_cancel_add),
state=CreatorGroupsSG.add_group_input_number,
state=SharedGroupsSG.add_group_input_number,
),
Window(
Format("<b>🗑 Удаление группы</b>\n\n<b>Выберите группу для удаления:</b>\n\n📊 <b>Всего групп:</b> {groups_count}"),
@@ -172,7 +161,7 @@ groups_dialog = Dialog(
height=7,
),
Button(Const("◀️ Назад"), id="back", on_click=on_cancel_add),
state=CreatorGroupsSG.delete_groups_list,
state=SharedGroupsSG.delete_groups_list,
getter=get_delete_groups_data,
),
Window(
@@ -181,7 +170,7 @@ groups_dialog = Dialog(
Button(Const("✅ Да, удалить"), id="confirm", on_click=on_confirm_delete),
Button(Const("❌ Отмена"), id="cancel", on_click=on_cancel_delete),
),
state=CreatorGroupsSG.delete_confirm,
state=SharedGroupsSG.delete_confirm,
getter=get_delete_confirm_data,
),
)
@@ -0,0 +1,51 @@
from aiogram.fsm.state import State, StatesGroup
class SharedTemplatesSG(StatesGroup):
main = State()
export_list = State()
spec = State()
import_file = State()
class SharedTestsSG(StatesGroup):
tests_list = State()
test_detail = State()
share_test = State()
edit_menu = State()
edit_password = State()
edit_attempts = State()
edit_group = State()
edit_expires = State()
statistics = State()
attempt_detail = State()
class SharedBroadcastSG(StatesGroup):
broadcast_input = State()
broadcast_confirm = State()
class SharedGroupsSG(StatesGroup):
groups_list = State()
add_group_input_number = State()
delete_groups_list = State()
delete_confirm = State()
class SharedCreateTestSG(StatesGroup):
input_title = State()
input_description = State()
input_password = State()
input_attempts = State()
input_expires_at = State()
input_for_group = State()
confirm_test_info = State()
add_question = State()
input_question_text = State()
select_question_type = State()
input_correct_answer = State()
input_options = State()
mark_correct_options = State()
confirm_question = State()
test_created = State()
@@ -9,7 +9,7 @@ from aiogram_dialog.widgets.text import Const, Format
from dishka import FromDishka
from dishka.integrations.aiogram_dialog import inject
from trudex.application.bot.admin_dialogs.states import AdminMenuSG, AdminTemplatesSG, AdminTestsSG
from trudex.application.bot.shared_dialogs.states import SharedTemplatesSG, SharedTestsSG
from trudex.domain.schemas import QuestionType
from trudex.domain.test_parser import ParsedTest, TestParser
from trudex.infrastructure.database.dao.option import OptionDAO
@@ -17,6 +17,7 @@ from trudex.infrastructure.database.dao.question import QuestionDAO
from trudex.infrastructure.database.dao.test import TestDAO
from trudex.infrastructure.database.repo.test import TestRepository
TEMPLATES_INFO = (
"<b>📦 Шаблоны тестов</b>\n\n"
"Шаблоны позволяют экспортировать и импортировать тесты в формате JSON.\n\n"
@@ -73,6 +74,7 @@ SPEC_INFO = """<b>📋 Спецификация формата JSON</b>
Для <code>multiple</code> один или более <code>is_correct: true</code>
Минимум 2 варианта ответа для single/multiple"""
TEMPLATE_SINGLE = {
"title": "Пример теста с одиночным выбором",
"description": "Демонстрация формата single вопросов",
@@ -166,23 +168,23 @@ TEMPLATE_FULL = {
async def on_export_clicked(_callback: CallbackQuery, _button: Button, manager: DialogManager) -> None:
await manager.switch_to(AdminTemplatesSG.export_list)
await manager.switch_to(SharedTemplatesSG.export_list)
async def on_import_clicked(_callback: CallbackQuery, _button: Button, manager: DialogManager) -> None:
await manager.switch_to(AdminTemplatesSG.import_file)
await manager.switch_to(SharedTemplatesSG.import_file)
async def on_spec_clicked(_callback: CallbackQuery, _button: Button, manager: DialogManager) -> None:
await manager.switch_to(AdminTemplatesSG.spec)
await manager.switch_to(SharedTemplatesSG.spec)
async def on_back_clicked(_callback: CallbackQuery, _button: Button, manager: DialogManager) -> None:
await manager.start(AdminMenuSG.main, mode=StartMode.RESET_STACK)
await manager.done()
async def on_back_to_templates(_callback: CallbackQuery, _button: Button, manager: DialogManager) -> None:
await manager.switch_to(AdminTemplatesSG.main)
await manager.switch_to(SharedTemplatesSG.main)
@inject
@@ -224,7 +226,7 @@ async def on_test_selected_for_export(
for question, options in questions_with_options:
question_data: dict = {
"text": question.text,
"question_type": question.question_type,
"question_type": question.question_type.value,
}
if question.question_type == QuestionType.INPUT:
@@ -371,10 +373,10 @@ async def on_import_file(
f"Тест создан в деактивированном состоянии."
)
await manager.start(AdminTestsSG.tests_list, mode=StartMode.RESET_STACK)
await manager.start(SharedTestsSG.tests_list, mode=StartMode.RESET_STACK)
templates_dialog = Dialog(
shared_templates_dialog = Dialog(
Window(
Const(TEMPLATES_INFO),
Row(
@@ -383,7 +385,7 @@ templates_dialog = Dialog(
),
Button(Const("📋 Спецификация"), id="spec", on_click=on_spec_clicked),
Button(Const("◀️ Назад"), id="back", on_click=on_back_clicked),
state=AdminTemplatesSG.main,
state=SharedTemplatesSG.main,
),
Window(
Format("<b>📤 Экспорт теста</b>\n\nВыберите тест для экспорта:\n\nВсего: {count}"),
@@ -400,7 +402,7 @@ templates_dialog = Dialog(
height=7,
),
Button(Const("◀️ Назад"), id="back", on_click=on_back_to_templates),
state=AdminTemplatesSG.export_list,
state=SharedTemplatesSG.export_list,
getter=get_tests_for_export,
),
Window(
@@ -414,12 +416,12 @@ templates_dialog = Dialog(
Button(Const("📦 Полный"), id="tpl_full", on_click=on_template_full),
),
Button(Const("◀️ Назад"), id="back", on_click=on_back_to_templates),
state=AdminTemplatesSG.spec,
state=SharedTemplatesSG.spec,
),
Window(
Const("<b>📥 Импорт теста</b>\n\nОтправьте JSON файл с тестом.\n\n<i>Формат файла описан в разделе «Спецификация»</i>"),
MessageInput(on_import_file, content_types=[ContentType.DOCUMENT]),
Button(Const("◀️ Назад"), id="back", on_click=on_back_to_templates),
state=AdminTemplatesSG.import_file,
state=SharedTemplatesSG.import_file,
),
)
@@ -11,7 +11,7 @@ from aiogram_dialog.widgets.text import Const, Format
from dishka import FromDishka
from dishka.integrations.aiogram_dialog import inject
from trudex.application.bot.admin_dialogs.states import AdminCreateTestSG, AdminMenuSG, AdminTestsSG
from trudex.application.bot.shared_dialogs.states import SharedCreateTestSG, SharedTestsSG
from trudex.infrastructure.database.dao.group import GroupDAO
from trudex.infrastructure.database.dao.test import TestDAO
from trudex.infrastructure.database.repo.test import TestRepository
@@ -37,7 +37,7 @@ async def get_tests_data(test_dao: FromDishka[TestDAO], **_kwargs):
async def on_test_selected(_callback: CallbackQuery, _widget: Select, manager: DialogManager, item_id: str):
manager.dialog_data["selected_test_id"] = int(item_id)
await manager.switch_to(AdminTestsSG.test_detail)
await manager.switch_to(SharedTestsSG.test_detail)
@inject
@@ -108,7 +108,7 @@ async def on_toggle_active(_callback: CallbackQuery, _button: Button, manager: D
await test_dao.update(test_id, is_active=not test.is_active)
action = "деактивирован" if test.is_active else "активирован"
await _callback.answer(f"✅ Тест {action}")
await manager.switch_to(AdminTestsSG.test_detail)
await manager.switch_to(SharedTestsSG.test_detail)
@inject
@@ -124,15 +124,15 @@ async def on_toggle_results_viewable(_callback: CallbackQuery, _button: Button,
await test_dao.update(test_id, are_results_viewable=not test.are_results_viewable)
action = "скрыты" if test.are_results_viewable else "видны"
await _callback.answer(f"✅ Результаты теперь {action}")
await manager.switch_to(AdminTestsSG.test_detail)
await manager.switch_to(SharedTestsSG.test_detail)
async def on_back_to_list(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.switch_to(AdminTestsSG.tests_list)
await manager.switch_to(SharedTestsSG.tests_list)
async def on_statistics(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.switch_to(AdminTestsSG.statistics)
await manager.switch_to(SharedTestsSG.statistics)
@inject
@@ -163,11 +163,11 @@ async def get_statistics_data(
async def on_attempt_selected(_callback: CallbackQuery, _widget: Select, manager: DialogManager, item_id: str):
manager.dialog_data["selected_attempt_id"] = int(item_id)
await manager.switch_to(AdminTestsSG.attempt_detail)
await manager.switch_to(SharedTestsSG.attempt_detail)
async def on_back_to_statistics(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.switch_to(AdminTestsSG.statistics)
await manager.switch_to(SharedTestsSG.statistics)
@inject
@@ -192,7 +192,7 @@ async def get_attempt_detail(
date_str = finished_at_msk.strftime("%d.%m.%Y %H:%M") if finished_at_msk else ""
lines = [
f"<b>📊 Результат прохождения</b>\n",
"<b>📊 Результат прохождения</b>\n",
f"📈 <b>Результат:</b> {attempt.score}%",
f"📅 <b>Дата:</b> {date_str}",
f"🏆 <b>Статус:</b> {status}\n",
@@ -226,8 +226,9 @@ async def on_share_test(_callback: CallbackQuery, _button: Button, manager: Dial
test_id = manager.dialog_data.get("selected_test_id")
if not test_id:
await _callback.answer("Ошибка: тест не найден")
return
return {
"share_link": "Ошибка: тест не найден"
}
test_hash = encode_id(
test_id,
@@ -254,31 +255,31 @@ async def on_share_test(_callback: CallbackQuery, _button: Button, manager: Dial
async def on_edit_menu(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.switch_to(AdminTestsSG.edit_menu)
await manager.switch_to(SharedTestsSG.edit_menu)
async def on_back_to_detail(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.switch_to(AdminTestsSG.test_detail)
await manager.switch_to(SharedTestsSG.test_detail)
async def on_back_to_edit_menu(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.switch_to(AdminTestsSG.edit_menu)
await manager.switch_to(SharedTestsSG.edit_menu)
async def on_edit_password(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.switch_to(AdminTestsSG.edit_password)
await manager.switch_to(SharedTestsSG.edit_password)
async def on_edit_attempts(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.switch_to(AdminTestsSG.edit_attempts)
await manager.switch_to(SharedTestsSG.edit_attempts)
async def on_edit_group(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.switch_to(AdminTestsSG.edit_group)
await manager.switch_to(SharedTestsSG.edit_group)
async def on_edit_expires(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.switch_to(AdminTestsSG.edit_expires)
await manager.switch_to(SharedTestsSG.edit_expires)
@inject
@@ -299,7 +300,7 @@ async def on_password_input(message: Message, _widget: MessageInput, manager: Di
await test_dao.update(test_id, password=password)
await message.answer("✅ Пароль обновлен")
await manager.switch_to(AdminTestsSG.test_detail)
await manager.switch_to(SharedTestsSG.test_detail)
@inject
@@ -311,7 +312,7 @@ async def on_remove_password(_callback: CallbackQuery, _button: Button, manager:
await test_dao.update(test_id, password=None)
await _callback.answer("✅ Пароль удален")
await manager.switch_to(AdminTestsSG.test_detail)
await manager.switch_to(SharedTestsSG.test_detail)
@inject
@@ -343,7 +344,7 @@ async def on_attempts_input_edit(message: Message, _widget: MessageInput, manage
await test_dao.update(test_id, attempts=attempts)
await message.answer("✅ Количество попыток обновлено")
await manager.switch_to(AdminTestsSG.test_detail)
await manager.switch_to(SharedTestsSG.test_detail)
@inject
@@ -355,7 +356,7 @@ async def on_remove_attempts(_callback: CallbackQuery, _button: Button, manager:
await test_dao.update(test_id, attempts=None)
await _callback.answer("✅ Ограничение попыток удалено")
await manager.switch_to(AdminTestsSG.test_detail)
await manager.switch_to(SharedTestsSG.test_detail)
@inject
@@ -376,7 +377,7 @@ async def on_group_selected_for_test(_callback: CallbackQuery, _widget, manager:
await test_dao.update(test_id, for_group=int(item_id))
await _callback.answer("✅ Группа обновлена")
await manager.switch_to(AdminTestsSG.test_detail)
await manager.switch_to(SharedTestsSG.test_detail)
@inject
@@ -388,7 +389,7 @@ async def on_remove_group(_callback: CallbackQuery, _button: Button, manager: Di
await test_dao.update(test_id, for_group=None)
await _callback.answer("✅ Тест теперь доступен для всех групп")
await manager.switch_to(AdminTestsSG.test_detail)
await manager.switch_to(SharedTestsSG.test_detail)
@inject
@@ -401,7 +402,7 @@ async def on_date_selected_for_test(_callback, _widget, manager: DialogManager,
expires_at = datetime.combine(selected_date, time.min)
await test_dao.update(test_id, expires_at=expires_at)
await _callback.answer("✅ Срок действия обновлен")
await manager.switch_to(AdminTestsSG.test_detail)
await manager.switch_to(SharedTestsSG.test_detail)
@inject
@@ -413,18 +414,18 @@ async def on_remove_expires(_callback: CallbackQuery, _button: Button, manager:
await test_dao.update(test_id, expires_at=None)
await _callback.answer("✅ Срок действия удален")
await manager.switch_to(AdminTestsSG.test_detail)
await manager.switch_to(SharedTestsSG.test_detail)
async def on_add_test_clicked(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.start(AdminCreateTestSG.input_title, mode=StartMode.RESET_STACK)
await manager.start(SharedCreateTestSG.input_title, mode=StartMode.RESET_STACK)
async def on_back_clicked(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.start(AdminMenuSG.main, mode=StartMode.RESET_STACK)
await manager.done()
tests_dialog = Dialog(
shared_tests_dialog = Dialog(
Window(
Format("<b>📝 Тесты</b>\n\nВсего: {count}"),
ScrollingGroup(
@@ -443,7 +444,7 @@ tests_dialog = Dialog(
Button(Const(" Добавить тест"), id="add_test", on_click=on_add_test_clicked),
Button(Const("◀️ Назад"), id="back", on_click=on_back_clicked),
),
state=AdminTestsSG.tests_list,
state=SharedTestsSG.tests_list,
getter=get_tests_data,
),
Window(
@@ -464,7 +465,7 @@ tests_dialog = Dialog(
Button(Const("✏️ Изменить"), id="edit_menu", on_click=on_edit_menu),
Button(Const("◀️ Назад"), id="back", on_click=on_back_to_list),
),
state=AdminTestsSG.test_detail,
state=SharedTestsSG.test_detail,
getter=get_test_detail,
),
Window(
@@ -476,7 +477,7 @@ tests_dialog = Dialog(
Button(Const("📅 Срок действия"), id="edit_expires", on_click=on_edit_expires),
Button(Const("◀️ Назад"), id="back", on_click=on_back_to_detail),
),
state=AdminTestsSG.edit_menu,
state=SharedTestsSG.edit_menu,
),
Window(
Const("<b>🔑 Изменение пароля</b>\n\n💬 <b>Введите новый пароль</b> или удалите текущий:\n<i>(максимум 255 символов)</i>"),
@@ -485,7 +486,7 @@ tests_dialog = Dialog(
Button(Const("🗑 Удалить пароль"), id="remove_password", on_click=on_remove_password),
Button(Const("◀️ Назад"), id="back", on_click=on_back_to_edit_menu),
),
state=AdminTestsSG.edit_password,
state=SharedTestsSG.edit_password,
),
Window(
Const("<b>🔄 Изменение количества попыток</b>\n\n🔢 <b>Введите новое количество попыток</b> (1-100) или удалите ограничение:"),
@@ -494,7 +495,7 @@ tests_dialog = Dialog(
Button(Const("🗑 Без ограничений"), id="remove_attempts", on_click=on_remove_attempts),
Button(Const("◀️ Назад"), id="back", on_click=on_back_to_edit_menu),
),
state=AdminTestsSG.edit_attempts,
state=SharedTestsSG.edit_attempts,
),
Window(
Const("<b>👥 Изменение группы</b>\n\n🎓 <b>Выберите группу</b> или удалите привязку:"),
@@ -514,7 +515,7 @@ tests_dialog = Dialog(
Button(Const("🗑 Для всех групп"), id="remove_group", on_click=on_remove_group),
Button(Const("◀️ Назад"), id="back", on_click=on_back_to_edit_menu),
),
state=AdminTestsSG.edit_group,
state=SharedTestsSG.edit_group,
getter=get_groups_for_edit,
),
Window(
@@ -524,7 +525,7 @@ tests_dialog = Dialog(
Button(Const("🗑 Удалить срок"), id="remove_expires", on_click=on_remove_expires),
Button(Const("◀️ Назад"), id="back", on_click=on_back_to_edit_menu),
),
state=AdminTestsSG.edit_expires,
state=SharedTestsSG.edit_expires,
),
Window(
Format("<b>📊 Статистика теста</b>\n\nПрошли тест: {count}"),
@@ -544,13 +545,13 @@ tests_dialog = Dialog(
Button(Const("🔄 Обновить"), id="refresh", on_click=on_statistics),
Button(Const("◀️ Назад"), id="back", on_click=on_back_to_detail),
),
state=AdminTestsSG.statistics,
state=SharedTestsSG.statistics,
getter=get_statistics_data,
),
Window(
Format("{attempt_info}"),
Button(Const("◀️ Назад"), id="back", on_click=on_back_to_statistics),
state=AdminTestsSG.attempt_detail,
state=SharedTestsSG.attempt_detail,
getter=get_attempt_detail,
),
)