Initial commit

This commit is contained in:
2026-01-02 17:39:56 +03:00
parent 3e51b1f95e
commit ac03de4db5
17 changed files with 690 additions and 416 deletions
@@ -0,0 +1,31 @@
"""test model add password and expires_at fields
Revision ID: d3bd5df63c1b
Revises: f63140aa50c0
Create Date: 2026-01-02 17:05:33.443875
"""
from collections.abc import Sequence
from alembic import op
import sqlalchemy as sa
revision: str = 'd3bd5df63c1b'
down_revision: str | None = 'f63140aa50c0'
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.add_column('tests', sa.Column('password', sa.String(length=255), nullable=True))
op.add_column('tests', sa.Column('expires_at', sa.DateTime(), nullable=True))
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.drop_column('tests', 'expires_at')
op.drop_column('tests', 'password')
# ### end Alembic commands ###
+25 -8
View File
@@ -8,12 +8,19 @@ from aiogram_dialog import setup_dialogs
from dishka import make_async_container from dishka import make_async_container
from dishka.integrations.aiogram import setup_dishka 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.main_menu import admin_menu_dialog from trudex.application.bot.admin_dialogs.main_menu import admin_menu_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.main_menu import creator_menu_dialog from trudex.application.bot.creator_dialogs.main_menu import creator_menu_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.handlers import router from trudex.application.bot.handlers import router
from trudex.application.bot.middlewares.reject_not_admin import RejectNotAdminMiddleware from trudex.application.bot.middlewares.reject_not_admin import RejectNotAdminMiddleware
from trudex.application.bot.middlewares.reject_not_creator import RejectNotCreatorMiddleware from trudex.application.bot.middlewares.reject_not_creator import RejectNotCreatorMiddleware
from trudex.application.bot.user_dialogs.main_menu import user_menu_dialog from trudex.application.bot.user_dialogs.main_menu import user_menu_dialog
from trudex.infrastructure.database.repo.user import UserRepository
from trudex.infrastructure.di import DatabaseProvider from trudex.infrastructure.di import DatabaseProvider
from trudex.infrastructure.utils.bot_commands import setup_bot_commands from trudex.infrastructure.utils.bot_commands import setup_bot_commands
from trudex.infrastructure.utils.config import Config from trudex.infrastructure.utils.config import Config
@@ -33,23 +40,33 @@ async def main() -> None:
) )
dp = Dispatcher() dp = Dispatcher()
dp.message.middleware(RejectNotAdminMiddleware())
dp.message.middleware(RejectNotCreatorMiddleware())
dp.include_router(router)
dp.include_router(user_menu_dialog) dp.include_routers(
dp.include_router(admin_menu_dialog) router,
dp.include_router(creator_menu_dialog) user_menu_dialog,
admin_menu_dialog,
admin_users_dialog,
admin_tests_dialog,
admin_broadcast_dialog,
creator_menu_dialog,
creator_users_dialog,
creator_tests_dialog,
creator_broadcast_dialog,
)
router.message.middleware(RejectNotAdminMiddleware())
router.message.middleware(RejectNotCreatorMiddleware())
container = make_async_container(DatabaseProvider()) container = make_async_container(DatabaseProvider())
setup_dishka(container, dp, auto_inject=True)
setup_dialogs(dp) setup_dialogs(dp)
setup_dishka(container, dp, auto_inject=True)
async with container() as request_container: async with container() as request_container:
from trudex.infrastructure.database.repo.user import UserRepository
user_repo = await request_container.get(UserRepository) user_repo = await request_container.get(UserRepository)
await setup_bot_commands(bot, config, user_repo) await setup_bot_commands(bot, config, user_repo)
await bot.delete_webhook(drop_pending_updates=True)
logging.info("Бот запущен") logging.info("Бот запущен")
try: try:
@@ -0,0 +1,73 @@
from aiogram.types import CallbackQuery, Message
from aiogram_dialog import Dialog, DialogManager, Window, StartMode
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.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(AdminBroadcastSG.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.done()
async def on_back_to_main(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.start(AdminMenuSG.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=AdminBroadcastSG.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=AdminBroadcastSG.broadcast_confirm,
),
)
@@ -1,139 +1,21 @@
from aiogram.types import CallbackQuery, Message from aiogram.types import CallbackQuery
from aiogram_dialog import Dialog, DialogManager, Window from aiogram_dialog import Dialog, DialogManager, Window, StartMode
from aiogram_dialog.widgets.input import MessageInput from aiogram_dialog.widgets.kbd import Button, Column
from aiogram_dialog.widgets.kbd import Back, Button, Cancel, Column, Row, ScrollingGroup, Select, SwitchTo from aiogram_dialog.widgets.text import Const
from aiogram_dialog.widgets.text import Const, Format
from dishka import FromDishka
from dishka.integrations.aiogram import CONTAINER_NAME
from dishka.integrations.aiogram_dialog import inject
from trudex.application.bot.admin_dialogs.states import AdminMenuSG from trudex.application.bot.admin_dialogs.states import AdminMenuSG, AdminUsersSG, AdminTestsSG, AdminBroadcastSG
from trudex.infrastructure.database.dao.user import UserDAO
from trudex.infrastructure.utils.broadcast import broadcast_message
@inject async def on_tests_clicked(_callback: CallbackQuery, _button: Button, manager: DialogManager) -> None:
async def get_users_data(user_dao: FromDishka[UserDAO], **_kwargs): await manager.start(AdminTestsSG.tests_list, mode=StartMode.RESET_STACK)
users = await user_dao.get_all()
return {
"users": [
(f"{u.first_name} (@{u.username or 'нет'})", u.id)
for u in users
],
"count": len(users),
}
@inject
async def get_user_detail_data(dialog_manager: DialogManager, user_dao: FromDishka[UserDAO], **_kwargs):
user_id = dialog_manager.dialog_data.get("selected_user_id")
if not user_id:
return {"user_info": "Пользователь не выбран"}
user = await user_dao.get_by_id(user_id)
if not user:
return {"user_info": "Пользователь не найден"}
username_str = f"@{user.username}" if user.username else ""
last_name_str = user.last_name or ""
group_str = str(user.group) if user.group else ""
admin_status = "✅ Да" if user.is_admin else "❌ Нет"
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>Username:</b> {username_str}\n"
f"<b>Группа:</b> {group_str}\n"
f"<b>Администратор:</b> {admin_status}"
)
return {"user_info": user_info}
async def on_user_selected(_callback: CallbackQuery, _widget: Select, manager: DialogManager, item_id: str):
manager.dialog_data["selected_user_id"] = int(item_id)
await manager.switch_to(AdminMenuSG.user_detail)
async def on_input_mode(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.switch_to(AdminMenuSG.users_input)
async def on_user_input(message: Message, _widget: MessageInput, manager: DialogManager):
container = manager.middleware_data[CONTAINER_NAME]
user_dao = await container.get(UserDAO)
text = (message.text or "").strip()
user = None
if text.startswith("@"):
username = text[1:]
all_users = await user_dao.get_all()
user = next((u for u in all_users if u.username == username), None)
elif text.isdigit():
user = await user_dao.get_by_id(int(text))
if not user:
await message.answer("❌ Пользователь не найден в базе данных.")
return
manager.dialog_data["selected_user_id"] = user.id
await manager.switch_to(AdminMenuSG.user_detail)
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(AdminMenuSG.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.switch_to(AdminMenuSG.main)
async def on_tests_clicked(_callback: CallbackQuery, _button: Button, _manager: DialogManager) -> None:
await _callback.answer("Управление тестами")
async def on_users_clicked(_callback: CallbackQuery, _button: Button, manager: DialogManager) -> None: async def on_users_clicked(_callback: CallbackQuery, _button: Button, manager: DialogManager) -> None:
await manager.switch_to(AdminMenuSG.users_list) await manager.start(AdminUsersSG.users_list, mode=StartMode.RESET_STACK)
async def on_broadcast_clicked(_callback: CallbackQuery, _button: Button, manager: DialogManager) -> None: async def on_broadcast_clicked(_callback: CallbackQuery, _button: Button, manager: DialogManager) -> None:
await manager.switch_to(AdminMenuSG.broadcast_input) await manager.start(AdminBroadcastSG.broadcast_input, mode=StartMode.RESET_STACK)
admin_menu_dialog = Dialog( admin_menu_dialog = Dialog(
@@ -146,51 +28,4 @@ admin_menu_dialog = Dialog(
), ),
state=AdminMenuSG.main, state=AdminMenuSG.main,
), ),
Window(
Format("<b>👥 Пользователи</b>\n\nВсего: {count}"),
ScrollingGroup(
Select(
Format("{item[0]}"),
id="user_select",
item_id_getter=lambda x: x[1],
items="users",
on_click=on_user_selected,
),
id="users_scroll",
width=1,
height=7,
),
Column(
Button(Const("✏️ Ввести ID/Username"), id="input_mode", on_click=on_input_mode),
Back(Const("◀️ Назад")),
),
state=AdminMenuSG.users_list,
getter=get_users_data,
),
Window(
Const("<b>Введите ID или @username пользователя:</b>"),
MessageInput(on_user_input),
SwitchTo(Const("◀️ Назад"), id="back_to_list", state=AdminMenuSG.users_list),
state=AdminMenuSG.users_input,
),
Window(
Format("{user_info}"),
SwitchTo(Const("◀️ Назад"), id="back_to_list", state=AdminMenuSG.users_list),
state=AdminMenuSG.user_detail,
getter=get_user_detail_data,
),
Window(
Const("<b>📢 Рассылка</b>\n\nОтправьте сообщение, которое хотите разослать всем пользователям:"),
MessageInput(on_broadcast_input),
SwitchTo(Const("◀️ Отмена"), id="cancel_broadcast", state=AdminMenuSG.main),
state=AdminMenuSG.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=AdminMenuSG.broadcast_confirm,
),
) )
@@ -3,8 +3,18 @@ from aiogram.fsm.state import State, StatesGroup
class AdminMenuSG(StatesGroup): class AdminMenuSG(StatesGroup):
main = State() main = State()
class AdminUsersSG(StatesGroup):
users_list = State() users_list = State()
users_input = State() users_input = State()
user_detail = State() user_detail = State()
class AdminTestsSG(StatesGroup):
tests_list = State()
class AdminBroadcastSG(StatesGroup):
broadcast_input = State() broadcast_input = State()
broadcast_confirm = State() broadcast_confirm = State()
@@ -0,0 +1,60 @@
from aiogram.types import CallbackQuery
from aiogram_dialog import Dialog, DialogManager, Window, StartMode
from aiogram_dialog.widgets.kbd import Button, Column, 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 AdminTestsSG, AdminMenuSG
from trudex.infrastructure.database.dao.test import TestDAO
@inject
async def get_tests_data(test_dao: FromDishka[TestDAO], **_kwargs):
tests = await test_dao.get_all()
return {
"tests": [
(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 _callback.answer("Тест выбран")
async def on_add_test_clicked(_callback: CallbackQuery, _button: Button, _manager: DialogManager):
await _callback.answer("Добавление теста")
async def on_back_clicked(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.start(AdminMenuSG.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=AdminTestsSG.tests_list,
getter=get_tests_data,
),
)
@@ -0,0 +1,124 @@
from aiogram.types import CallbackQuery, Message
from aiogram_dialog import Dialog, DialogManager, Window, StartMode
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 import CONTAINER_NAME
from dishka.integrations.aiogram_dialog import inject
from trudex.application.bot.admin_dialogs.states import AdminUsersSG, AdminMenuSG
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()
return {
"users": [
(f"{u.first_name} (@{u.username or 'нет'})", u.id)
for u in users
],
"count": len(users),
}
@inject
async def get_user_detail_data(dialog_manager: DialogManager, user_dao: FromDishka[UserDAO], **_kwargs):
user_id = dialog_manager.dialog_data.get("selected_user_id")
if not user_id:
return {"user_info": "Пользователь не выбран"}
user = await user_dao.get_by_id(user_id)
if not user:
return {"user_info": "Пользователь не найден"}
username_str = f"@{user.username}" if user.username else ""
last_name_str = user.last_name or ""
group_str = str(user.group) if user.group else ""
admin_status = "✅ Да" if user.is_admin else "❌ Нет"
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>Username:</b> {username_str}\n"
f"<b>Группа:</b> {group_str}\n"
f"<b>Администратор:</b> {admin_status}"
)
return {"user_info": user_info}
async def on_user_selected(_callback: CallbackQuery, _widget: Select, manager: DialogManager, item_id: str):
manager.dialog_data["selected_user_id"] = int(item_id)
await manager.switch_to(AdminUsersSG.user_detail)
async def on_input_mode(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.switch_to(AdminUsersSG.users_input)
async def on_user_input(message: Message, _widget: MessageInput, manager: DialogManager):
container = manager.middleware_data[CONTAINER_NAME]
user_dao = await container.get(UserDAO)
text = (message.text or "").strip()
user = None
if text.startswith("@"):
username = text[1:]
all_users = await user_dao.get_all()
user = next((u for u in all_users if u.username == username), None)
elif text.isdigit():
user = await user_dao.get_by_id(int(text))
if not user:
await message.answer("❌ Пользователь не найден в базе данных.")
return
manager.dialog_data["selected_user_id"] = user.id
await manager.switch_to(AdminUsersSG.user_detail)
async def on_back_to_main(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.start(AdminMenuSG.main, mode=StartMode.RESET_STACK)
users_dialog = Dialog(
Window(
Format("<b>👥 Пользователи</b>\n\nВсего: {count}"),
ScrollingGroup(
Select(
Format("{item[0]}"),
id="user_select",
item_id_getter=lambda x: x[1],
items="users",
on_click=on_user_selected,
),
id="users_scroll",
width=1,
height=7,
),
Column(
Button(Const("✏️ Ввести ID/Username"), id="input_mode", on_click=on_input_mode),
Button(Const("◀️ Назад"), id="back", on_click=on_back_to_main),
),
state=AdminUsersSG.users_list,
getter=get_users_data,
),
Window(
Const("<b>Введите ID или @username пользователя:</b>"),
MessageInput(on_user_input),
SwitchTo(Const("◀️ Назад"), id="back_to_list", state=AdminUsersSG.users_list),
state=AdminUsersSG.users_input,
),
Window(
Format("{user_info}"),
SwitchTo(Const("◀️ Назад"), id="back_to_list", state=AdminUsersSG.users_list),
state=AdminUsersSG.user_detail,
getter=get_user_detail_data,
),
)
@@ -0,0 +1,73 @@
from aiogram.types import CallbackQuery, Message
from aiogram_dialog import Dialog, DialogManager, Window
from aiogram_dialog.widgets.input import MessageInput
from aiogram_dialog.widgets.kbd import Button, Row, Cancel
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
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.done()
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,182 +1,21 @@
from aiogram.types import CallbackQuery, Message from aiogram.types import CallbackQuery
from aiogram_dialog import Dialog, DialogManager, Window from aiogram_dialog import Dialog, DialogManager, Window, StartMode
from aiogram_dialog.widgets.input import MessageInput from aiogram_dialog.widgets.kbd import Button, Column
from aiogram_dialog.widgets.kbd import Back, Button, Cancel, Column, Row, ScrollingGroup, Select, SwitchTo from aiogram_dialog.widgets.text import Const
from aiogram_dialog.widgets.text import Const, Format
from dishka import FromDishka
from dishka.integrations.aiogram import CONTAINER_NAME
from dishka.integrations.aiogram_dialog import inject
from trudex.application.bot.creator_dialogs.states import CreatorMenuSG from trudex.application.bot.creator_dialogs.states import CreatorMenuSG, CreatorUsersSG, CreatorTestsSG, CreatorBroadcastSG
from trudex.infrastructure.database.dao.user import UserDAO
from trudex.infrastructure.utils.broadcast import broadcast_message
@inject async def on_tests_clicked(_callback: CallbackQuery, _button: Button, manager: DialogManager) -> None:
async def get_users_data(user_dao: FromDishka[UserDAO], **_kwargs): await manager.start(CreatorTestsSG.tests_list, mode=StartMode.RESET_STACK)
users = await user_dao.get_all()
return {
"users": [
(f"{u.first_name} (@{u.username or 'нет'})", u.id)
for u in users
],
"count": len(users),
}
@inject
async def get_user_detail_data(dialog_manager: DialogManager, user_dao: FromDishka[UserDAO], **_kwargs):
user_id = dialog_manager.dialog_data.get("selected_user_id")
if not user_id:
return {"user_info": "Пользователь не выбран", "is_admin": True, "show_make_admin": False}
user = await user_dao.get_by_id(user_id)
if not user:
return {"user_info": "Пользователь не найден", "is_admin": True, "show_make_admin": False}
username_str = f"@{user.username}" if user.username else ""
last_name_str = user.last_name or ""
group_str = str(user.group) if user.group else ""
admin_status = "✅ Да" if user.is_admin else "❌ Нет"
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>Username:</b> {username_str}\n"
f"<b>Группа:</b> {group_str}\n"
f"<b>Администратор:</b> {admin_status}"
)
return {
"user_info": user_info,
"is_admin": user.is_admin,
"show_make_admin": not user.is_admin,
}
@inject
async def get_confirm_data(dialog_manager: DialogManager, user_dao: FromDishka[UserDAO], **_kwargs):
user_id = dialog_manager.dialog_data.get("selected_user_id")
if not user_id:
return {"user_info": "Пользователь не выбран"}
user = await user_dao.get_by_id(user_id)
if not user:
return {"user_info": "Пользователь не найден"}
username_str = f"@{user.username}" if user.username else ""
return {
"user_info": f"<b>{user.first_name}</b>\n{username_str}\nID: <code>{user.id}</code>"
}
async def on_user_selected(_callback: CallbackQuery, _widget: Select, manager: DialogManager, item_id: str):
manager.dialog_data["selected_user_id"] = int(item_id)
await manager.switch_to(CreatorMenuSG.user_detail)
async def on_input_mode(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.switch_to(CreatorMenuSG.users_input)
async def on_user_input(message: Message, _widget: MessageInput, manager: DialogManager):
container = manager.middleware_data[CONTAINER_NAME]
user_dao = await container.get(UserDAO)
text = (message.text or "").strip()
user = None
if text.startswith("@"):
username = text[1:]
all_users = await user_dao.get_all()
user = next((u for u in all_users if u.username == username), None)
elif text.isdigit():
user = await user_dao.get_by_id(int(text))
if not user:
await message.answer("❌ Пользователь не найден в базе данных.")
return
manager.dialog_data["selected_user_id"] = user.id
await manager.switch_to(CreatorMenuSG.user_detail)
async def on_make_admin_clicked(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.switch_to(CreatorMenuSG.make_admin_confirm)
async def on_confirm_yes(_callback: CallbackQuery, _button: Button, manager: DialogManager):
container = manager.middleware_data[CONTAINER_NAME]
user_dao = await container.get(UserDAO)
user_id = manager.dialog_data.get("selected_user_id")
if not user_id:
await _callback.answer("Ошибка: пользователь не выбран")
return
await user_dao.update(user_id=user_id, is_admin=True)
await _callback.answer("✅ Пользователь назначен администратором")
await manager.switch_to(CreatorMenuSG.user_detail)
async def on_confirm_no(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await _callback.answer("Отменено")
await manager.switch_to(CreatorMenuSG.user_detail)
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(CreatorMenuSG.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.switch_to(CreatorMenuSG.main)
async def on_tests_clicked(_callback: CallbackQuery, _button: Button, _manager: DialogManager) -> None:
await _callback.answer("Тесты")
async def on_users_clicked(_callback: CallbackQuery, _button: Button, manager: DialogManager) -> None: async def on_users_clicked(_callback: CallbackQuery, _button: Button, manager: DialogManager) -> None:
await manager.switch_to(CreatorMenuSG.users_list) await manager.start(CreatorUsersSG.users_list, mode=StartMode.RESET_STACK)
async def on_broadcast_clicked(_callback: CallbackQuery, _button: Button, manager: DialogManager) -> None: async def on_broadcast_clicked(_callback: CallbackQuery, _button: Button, manager: DialogManager) -> None:
await manager.switch_to(CreatorMenuSG.broadcast_input) await manager.start(CreatorBroadcastSG.broadcast_input, mode=StartMode.RESET_STACK)
creator_menu_dialog = Dialog( creator_menu_dialog = Dialog(
@@ -189,64 +28,4 @@ creator_menu_dialog = Dialog(
), ),
state=CreatorMenuSG.main, state=CreatorMenuSG.main,
), ),
Window(
Format("<b>👥 Пользователи</b>\n\nВсего: {count}"),
ScrollingGroup(
Select(
Format("{item[0]}"),
id="user_select",
item_id_getter=lambda x: x[1],
items="users",
on_click=on_user_selected,
),
id="users_scroll",
width=1,
height=7,
),
Column(
Button(Const("✏️ Ввести ID/Username"), id="input_mode", on_click=on_input_mode),
Cancel(Const("◀️ Назад")),
),
state=CreatorMenuSG.users_list,
getter=get_users_data,
),
Window(
Const("<b>Введите ID или @username пользователя:</b>"),
MessageInput(on_user_input),
SwitchTo(Const("◀️ Назад"), id="back_to_list", state=CreatorMenuSG.users_list),
state=CreatorMenuSG.users_input,
),
Window(
Format("{user_info}"),
Column(
Button(Const("👑 Сделать администратором"), id="make_admin", on_click=on_make_admin_clicked, when="show_make_admin"),
SwitchTo(Const("◀️ Назад"), id="back_to_list", state=CreatorMenuSG.users_list),
),
state=CreatorMenuSG.user_detail,
getter=get_user_detail_data,
),
Window(
Const("<b>⚠️ Подтверждение</b>\n\nВы уверены, что хотите назначить этого пользователя администратором?\n"),
Format("{user_info}"),
Row(
Button(Const("✅ Да"), id="confirm_yes", on_click=on_confirm_yes),
Button(Const("❌ Нет"), id="confirm_no", on_click=on_confirm_no),
),
state=CreatorMenuSG.make_admin_confirm,
getter=get_confirm_data,
),
Window(
Const("<b>📢 Рассылка</b>\n\nОтправьте сообщение, которое хотите разослать всем пользователям:"),
MessageInput(on_broadcast_input),
SwitchTo(Const("◀️ Отмена"), id="cancel_broadcast", state=CreatorMenuSG.main),
state=CreatorMenuSG.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=CreatorMenuSG.broadcast_confirm,
),
) )
@@ -3,9 +3,19 @@ from aiogram.fsm.state import State, StatesGroup
class CreatorMenuSG(StatesGroup): class CreatorMenuSG(StatesGroup):
main = State() main = State()
class CreatorUsersSG(StatesGroup):
users_list = State() users_list = State()
users_input = State() users_input = State()
user_detail = State() user_detail = State()
make_admin_confirm = State() make_admin_confirm = State()
class CreatorTestsSG(StatesGroup):
tests_list = State()
class CreatorBroadcastSG(StatesGroup):
broadcast_input = State() broadcast_input = State()
broadcast_confirm = State() broadcast_confirm = State()
@@ -0,0 +1,60 @@
from aiogram.types import CallbackQuery
from aiogram_dialog import Dialog, DialogManager, Window, StartMode
from aiogram_dialog.widgets.kbd import Button, Column, 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 CreatorTestsSG, CreatorMenuSG
from trudex.infrastructure.database.dao.test import TestDAO
@inject
async def get_tests_data(test_dao: FromDishka[TestDAO], **_kwargs):
tests = await test_dao.get_all()
return {
"tests": [
(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 _callback.answer("Тест выбран")
async def on_add_test_clicked(_callback: CallbackQuery, _button: Button, _manager: DialogManager):
await _callback.answer("Добавление теста")
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,
),
)
@@ -0,0 +1,180 @@
from aiogram.types import CallbackQuery, Message
from aiogram_dialog import Dialog, DialogManager, Window, StartMode
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 import CONTAINER_NAME
from dishka.integrations.aiogram_dialog import inject
from trudex.application.bot.creator_dialogs.states import CreatorUsersSG, CreatorMenuSG
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()
return {
"users": [
(f"{u.first_name} (@{u.username or 'нет'})", u.id)
for u in users
],
"count": len(users),
}
@inject
async def get_user_detail_data(dialog_manager: DialogManager, user_dao: FromDishka[UserDAO], **_kwargs):
user_id = dialog_manager.dialog_data.get("selected_user_id")
if not user_id:
return {"user_info": "Пользователь не выбран", "is_admin": True, "show_make_admin": False}
user = await user_dao.get_by_id(user_id)
if not user:
return {"user_info": "Пользователь не найден", "is_admin": True, "show_make_admin": False}
username_str = f"@{user.username}" if user.username else ""
last_name_str = user.last_name or ""
group_str = str(user.group) if user.group else ""
admin_status = "✅ Да" if user.is_admin else "❌ Нет"
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>Username:</b> {username_str}\n"
f"<b>Группа:</b> {group_str}\n"
f"<b>Администратор:</b> {admin_status}"
)
return {
"user_info": user_info,
"is_admin": user.is_admin,
"show_make_admin": not user.is_admin,
}
@inject
async def get_confirm_data(dialog_manager: DialogManager, user_dao: FromDishka[UserDAO], **_kwargs):
user_id = dialog_manager.dialog_data.get("selected_user_id")
if not user_id:
return {"user_info": "Пользователь не выбран"}
user = await user_dao.get_by_id(user_id)
if not user:
return {"user_info": "Пользователь не найден"}
username_str = f"@{user.username}" if user.username else ""
return {
"user_info": f"<b>{user.first_name}</b>\n{username_str}\nID: <code>{user.id}</code>"
}
async def on_user_selected(_callback: CallbackQuery, _widget: Select, manager: DialogManager, item_id: str):
manager.dialog_data["selected_user_id"] = int(item_id)
await manager.switch_to(CreatorUsersSG.user_detail)
async def on_input_mode(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.switch_to(CreatorUsersSG.users_input)
async def on_user_input(message: Message, _widget: MessageInput, manager: DialogManager):
container = manager.middleware_data[CONTAINER_NAME]
user_dao = await container.get(UserDAO)
text = (message.text or "").strip()
user = None
if text.startswith("@"):
username = text[1:]
all_users = await user_dao.get_all()
user = next((u for u in all_users if u.username == username), None)
elif text.isdigit():
user = await user_dao.get_by_id(int(text))
if not user:
await message.answer("❌ Пользователь не найден в базе данных.")
return
manager.dialog_data["selected_user_id"] = user.id
await manager.switch_to(CreatorUsersSG.user_detail)
async def on_make_admin_clicked(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.switch_to(CreatorUsersSG.make_admin_confirm)
async def on_confirm_yes(_callback: CallbackQuery, _button: Button, manager: DialogManager):
container = manager.middleware_data[CONTAINER_NAME]
user_dao = await container.get(UserDAO)
user_id = manager.dialog_data.get("selected_user_id")
if not user_id:
await _callback.answer("Ошибка: пользователь не выбран")
return
await user_dao.update(user_id=user_id, is_admin=True)
await _callback.answer("✅ Пользователь назначен администратором")
await manager.switch_to(CreatorUsersSG.user_detail)
async def on_confirm_no(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await _callback.answer("Отменено")
await manager.switch_to(CreatorUsersSG.user_detail)
async def on_back_to_main(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.start(CreatorMenuSG.main, mode=StartMode.RESET_STACK)
users_dialog = Dialog(
Window(
Format("<b>👥 Пользователи</b>\n\nВсего: {count}"),
ScrollingGroup(
Select(
Format("{item[0]}"),
id="user_select",
item_id_getter=lambda x: x[1],
items="users",
on_click=on_user_selected,
),
id="users_scroll",
width=1,
height=7,
),
Column(
Button(Const("✏️ Ввести ID/Username"), id="input_mode", on_click=on_input_mode),
Button(Const("◀️ Назад"), id="back", on_click=on_back_to_main),
),
state=CreatorUsersSG.users_list,
getter=get_users_data,
),
Window(
Const("<b>Введите ID или @username пользователя:</b>"),
MessageInput(on_user_input),
SwitchTo(Const("◀️ Назад"), id="back_to_list", state=CreatorUsersSG.users_list),
state=CreatorUsersSG.users_input,
),
Window(
Format("{user_info}"),
Column(
Button(Const("👑 Сделать администратором"), id="make_admin", on_click=on_make_admin_clicked, when="show_make_admin"),
SwitchTo(Const("◀️ Назад"), id="back_to_list", state=CreatorUsersSG.users_list),
),
state=CreatorUsersSG.user_detail,
getter=get_user_detail_data,
),
Window(
Const("<b>⚠️ Подтверждение</b>\n\nВы уверены, что хотите назначить этого пользователя администратором?\n\n"),
Format("{user_info}"),
Row(
Button(Const("✅ Да"), id="confirm_yes", on_click=on_confirm_yes),
Button(Const("❌ Нет"), id="confirm_no", on_click=on_confirm_no),
),
state=CreatorUsersSG.make_admin_confirm,
getter=get_confirm_data,
),
)
+10 -4
View File
@@ -29,13 +29,19 @@ async def start_handler(message: Message, user_dao: FromDishka[UserDAO], dialog_
@router.message(Command("admin")) @router.message(Command("admin"))
async def admin_command(_message: Message, dialog_manager: DialogManager) -> None: async def admin_command(message: Message, dialog_manager: DialogManager) -> None:
await dialog_manager.start(AdminMenuSG.main, mode=StartMode.RESET_STACK) try:
await dialog_manager.start(AdminMenuSG.main, mode=StartMode.RESET_STACK)
except Exception as e:
await message.answer(f"Ошибка запуска диалога: {e}")
@router.message(Command("creator")) @router.message(Command("creator"))
async def creator_command(_message: Message, dialog_manager: DialogManager) -> None: async def creator_command(message: Message, dialog_manager: DialogManager) -> None:
await dialog_manager.start(CreatorMenuSG.main, mode=StartMode.RESET_STACK) try:
await dialog_manager.start(CreatorMenuSG.main, mode=StartMode.RESET_STACK)
except Exception as e:
await message.answer(f"Ошибка запуска диалога: {e}")
@router.error() @router.error()
+2
View File
@@ -20,6 +20,8 @@ class Test:
title: str title: str
description: str | None = None description: str | None = None
for_group: int | None = None for_group: int | None = None
password: str | None = None
expires_at: datetime | None = None
is_active: bool = True is_active: bool = True
created_at: datetime | None = None created_at: datetime | None = None
updated_at: datetime | None = None updated_at: datetime | None = None
@@ -27,12 +27,16 @@ class TestDAO:
title: str, title: str,
description: str | None = None, description: str | None = None,
for_group: int | None = None, for_group: int | None = None,
password: str | None = None,
expires_at: str | None = None,
is_active: bool = True, is_active: bool = True,
) -> DomainTest: ) -> DomainTest:
test = Test( test = Test(
title=title, title=title,
description=description, description=description,
for_group=for_group, for_group=for_group,
password=password,
expires_at=expires_at,
is_active=is_active, is_active=is_active,
) )
self.session.add(test) self.session.add(test)
@@ -46,6 +50,8 @@ class TestDAO:
title: str | None = None, title: str | None = None,
description: str | None = None, description: str | None = None,
for_group: int | None = None, for_group: int | None = None,
password: str | None = None,
expires_at: str | None = None,
is_active: bool | None = None, is_active: bool | None = None,
) -> DomainTest | None: ) -> DomainTest | None:
result = await self.session.execute( result = await self.session.execute(
@@ -61,6 +67,10 @@ class TestDAO:
test.description = description test.description = description
if for_group is not None: if for_group is not None:
test.for_group = for_group test.for_group = for_group
if password is not None:
test.password = password
if expires_at is not None:
test.expires_at = expires_at
if is_active is not None: if is_active is not None:
test.is_active = is_active test.is_active = is_active
@@ -12,6 +12,8 @@ class TestDTO:
title=self.model.title, title=self.model.title,
description=self.model.description, description=self.model.description,
for_group=self.model.for_group, for_group=self.model.for_group,
password=self.model.password,
expires_at=self.model.expires_at,
is_active=self.model.is_active, is_active=self.model.is_active,
created_at=self.model.created_at, created_at=self.model.created_at,
updated_at=self.model.updated_at, updated_at=self.model.updated_at,
@@ -38,6 +38,8 @@ class Test(Base):
title: Mapped[str] = mapped_column(String(255)) title: Mapped[str] = mapped_column(String(255))
description: Mapped[str | None] = mapped_column(Text) description: Mapped[str | None] = mapped_column(Text)
for_group: Mapped[int | None] = mapped_column(default=None) for_group: Mapped[int | None] = mapped_column(default=None)
password: Mapped[str | None] = mapped_column(String(255), default=None)
expires_at: Mapped[datetime | None] = mapped_column(default=None)
is_active: Mapped[bool] = mapped_column(default=True) is_active: Mapped[bool] = mapped_column(default=True)
created_at: Mapped[datetime] = mapped_column(server_default=func.now()) created_at: Mapped[datetime] = mapped_column(server_default=func.now())
updated_at: Mapped[datetime] = mapped_column(server_default=func.now(), onupdate=func.now()) updated_at: Mapped[datetime] = mapped_column(server_default=func.now(), onupdate=func.now())