This commit is contained in:
2026-01-03 02:12:28 +03:00
parent 9e822789d2
commit 8273ede069
12 changed files with 285 additions and 25 deletions
+4
View File
@@ -2,6 +2,10 @@
token = "1234567890:ABCdefGHIjklMNOpqrsTUVwxyz"
creator_id = 123456789
[security]
test_hash_salt = "your_secret_salt_here_change_in_production"
test_hash_length = 16
[database]
host = "localhost"
port = 5432
-6
View File
@@ -1,6 +0,0 @@
def main():
print("Hello from trudex!")
if __name__ == "__main__":
main()
+3 -1
View File
@@ -78,10 +78,12 @@ async def main() -> None:
router.message.middleware(RejectNotAdminMiddleware())
router.message.middleware(RejectNotCreatorMiddleware())
container = make_async_container(DatabaseProvider())
container = make_async_container(DatabaseProvider(), context={Bot: bot, Config: config})
setup_dialogs(dp)
setup_dishka(container, dp, auto_inject=True)
bott = await container.get(Bot)
async with container() as request_container:
user_repo = await request_container.get(UserRepository)
await setup_bot_commands(bot, config, user_repo)
@@ -14,7 +14,9 @@ class AdminUsersSG(StatesGroup):
class AdminTestsSG(StatesGroup):
tests_list = State()
test_detail = State()
share_test = State()
edit_password = State()
edit_attempts = State()
edit_group = State()
edit_expires = State()
@@ -1,5 +1,6 @@
from datetime import date, datetime
from aiogram import Bot
from aiogram.types import CallbackQuery, Message
from aiogram_dialog import Dialog, DialogManager, StartMode, Window
from aiogram_dialog.widgets.input import MessageInput
@@ -14,6 +15,8 @@ from trudex.application.bot.admin_dialogs.states import (AdminMenuSG,
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.utils.config import Config
from trudex.infrastructure.utils.test_id_to_hash import generate_alpha_id
@inject
@@ -57,6 +60,7 @@ async def get_test_detail(test_dao: FromDishka[TestDAO], test_repo: FromDishka[T
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 = 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 "👥 Для всех"
@@ -67,6 +71,7 @@ async def get_test_detail(test_dao: FromDishka[TestDAO], test_repo: FromDishka[T
f"<b>Статус:</b> {status}\n"
f"<b>Вопросов:</b> {questions_count}\n"
f"{password_str}\n"
f"<b>Попыток:</b> {attempts_str}\n"
f"{expires_str}\n"
f"{group_str}\n\n"
f"<b>Создан:</b> {test.created_at.strftime('%d.%m.%Y %H:%M') if test.created_at else ''}"
@@ -101,10 +106,38 @@ async def on_back_to_list(_callback: CallbackQuery, _button: Button, manager: Di
await manager.switch_to(AdminTestsSG.tests_list)
async def on_share_test(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.switch_to(AdminTestsSG.share_test)
@inject
async def get_share_link(dialog_manager: DialogManager, config: FromDishka[Config], bot: FromDishka[Bot], **_kwargs):
test_id = dialog_manager.dialog_data.get("selected_test_id")
if not test_id:
return {"share_link": "Ошибка: тест не найден"}
test_hash = generate_alpha_id(
test_id,
config.security.test_hash_salt,
config.security.test_hash_length
)
bot_info = await bot.get_me()
bot_username = bot_info.username or "your_bot"
share_link = f"https://t.me/{bot_username}?start={test_hash}"
return {"share_link": share_link}
async def on_edit_password(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.switch_to(AdminTestsSG.edit_password)
async def on_edit_attempts(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.switch_to(AdminTestsSG.edit_attempts)
async def on_edit_group(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.switch_to(AdminTestsSG.edit_group)
@@ -146,6 +179,50 @@ async def on_remove_password(_callback: CallbackQuery, _button: Button, manager:
await manager.switch_to(AdminTestsSG.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(AdminTestsSG.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(AdminTestsSG.test_detail)
@inject
async def get_groups_for_edit(dialog_manager: DialogManager, group_dao: FromDishka[GroupDAO], **_kwargs):
groups = await group_dao.get_all()
@@ -242,7 +319,9 @@ tests_dialog = Dialog(
id="toggle_active",
on_click=on_toggle_active
),
Button(Const("🔗 Поделиться"), id="share", on_click=on_share_test),
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_list),
@@ -250,6 +329,13 @@ tests_dialog = Dialog(
state=AdminTestsSG.test_detail,
getter=get_test_detail,
),
Window(
Const("<b>🔗 Поделиться тестом</b>\n\n📎 <b>Ссылка на тест:</b>"),
Format("\n<code>{share_link}</code>\n\n💡 Отправьте эту ссылку пользователям для прохождения теста"),
Button(Const("◀️ Назад"), id="back", on_click=on_back_to_list),
state=AdminTestsSG.share_test,
getter=get_share_link,
),
Window(
Const("<b>🔑 Изменение пароля</b>\n\n💬 <b>Введите новый пароль</b> или удалите текущий:\n<i>(максимум 255 символов)</i>"),
MessageInput(on_password_input),
@@ -259,6 +345,15 @@ tests_dialog = Dialog(
),
state=AdminTestsSG.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_list),
),
state=AdminTestsSG.edit_attempts,
),
Window(
Const("<b>👥 Изменение группы</b>\n\n🎓 <b>Выберите группу</b> или удалите привязку:"),
ScrollingGroup(
@@ -70,25 +70,42 @@ async def on_password_input(message: Message, _widget: MessageInput, manager: Di
return
manager.dialog_data["password"] = password
groups = await group_dao.get_all()
if len(groups) == 0:
manager.dialog_data["for_group"] = None
await manager.switch_to(CreateTestSG.confirm_test_info)
else:
await manager.switch_to(CreateTestSG.input_expires_at)
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
groups = await group_dao.get_all()
await manager.switch_to(CreateTestSG.input_attempts)
if len(groups) == 0:
manager.dialog_data["for_group"] = None
await manager.switch_to(CreateTestSG.confirm_test_info)
else:
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)
@@ -125,10 +142,12 @@ 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_str = expires_at.strftime("%d.%m.%Y") if expires_at else "Без срока"
group_str = str(for_group) if for_group else "Для всех"
@@ -138,6 +157,7 @@ async def get_test_info(dialog_manager: DialogManager, **_kwargs):
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}"
)
@@ -147,8 +167,10 @@ async def get_test_info(dialog_manager: DialogManager, **_kwargs):
@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")
@@ -156,6 +178,7 @@ async def on_confirm_test(_callback: CallbackQuery, _button: Button, manager: Di
title=title,
description=description,
password=password,
attempts=attempts,
expires_at=expires_at,
for_group=for_group,
)
@@ -361,6 +384,7 @@ async def on_save_question(
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", [])
@@ -443,6 +467,12 @@ create_test_dialog = Dialog(
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),
@@ -118,6 +118,8 @@ async def get_delete_confirm_data(dialog_manager: DialogManager, **_kwargs):
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"] = "✅ Группа удалена"
@@ -15,7 +15,9 @@ class CreatorUsersSG(StatesGroup):
class CreatorTestsSG(StatesGroup):
tests_list = State()
test_detail = State()
share_test = State()
edit_password = State()
edit_attempts = State()
edit_group = State()
edit_expires = State()
@@ -36,6 +38,7 @@ 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()
@@ -1,5 +1,6 @@
from datetime import date, datetime
from aiogram import Bot
from aiogram.types import CallbackQuery, Message
from aiogram_dialog import Dialog, DialogManager, StartMode, Window
from aiogram_dialog.widgets.input import MessageInput
@@ -15,6 +16,8 @@ from trudex.application.bot.creator_dialogs.states import (CreateTestSG,
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.utils.config import Config
from trudex.infrastructure.utils.test_id_to_hash import generate_alpha_id
@inject
@@ -58,6 +61,7 @@ async def get_test_detail(test_dao: FromDishka[TestDAO], test_repo: FromDishka[T
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 = 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 "👥 Для всех"
@@ -68,6 +72,7 @@ async def get_test_detail(test_dao: FromDishka[TestDAO], test_repo: FromDishka[T
f"<b>Статус:</b> {status}\n"
f"<b>Вопросов:</b> {questions_count}\n"
f"{password_str}\n"
f"<b>Попыток:</b> {attempts_str}\n"
f"{expires_str}\n"
f"{group_str}\n\n"
f"<b>Создан:</b> {test.created_at.strftime('%d.%m.%Y %H:%M') if test.created_at else ''}"
@@ -102,10 +107,38 @@ async def on_back_to_list(_callback: CallbackQuery, _button: Button, manager: Di
await manager.switch_to(CreatorTestsSG.tests_list)
async def on_share_test(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.switch_to(CreatorTestsSG.share_test)
@inject
async def get_share_link(dialog_manager: DialogManager, config: FromDishka[Config], **_kwargs):
test_id = dialog_manager.dialog_data.get("selected_test_id")
if not test_id:
return {"share_link": "Ошибка: тест не найден"}
test_hash = generate_alpha_id(
test_id,
config.security.test_hash_salt,
config.security.test_hash_length
)
bot = Bot(config.bot.token)
bot_info = await bot.get_me()
await bot.session.close()
bot_username = bot_info.username or "your_bot"
share_link = f"https://t.me/{bot_username}?start={test_hash}"
return {"share_link": share_link}
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)
@@ -147,6 +180,50 @@ async def on_remove_password(_callback: CallbackQuery, _button: Button, manager:
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()
@@ -243,7 +320,9 @@ tests_dialog = Dialog(
id="toggle_active",
on_click=on_toggle_active
),
Button(Const("🔗 Поделиться"), id="share", on_click=on_share_test),
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_list),
@@ -251,6 +330,13 @@ tests_dialog = Dialog(
state=CreatorTestsSG.test_detail,
getter=get_test_detail,
),
Window(
Const("<b>🔗 Поделиться тестом</b>\n\n📎 <b>Ссылка на тест:</b>"),
Format("\n<code>{share_link}</code>\n\n💡 Отправьте эту ссылку пользователям для прохождения теста"),
Button(Const("◀️ Назад"), id="back", on_click=on_back_to_list),
state=CreatorTestsSG.share_test,
getter=get_share_link,
),
Window(
Const("<b>🔑 Изменение пароля</b>\n\n💬 <b>Введите новый пароль</b> или удалите текущий:\n<i>(максимум 255 символов)</i>"),
MessageInput(on_password_input),
@@ -260,6 +346,15 @@ tests_dialog = Dialog(
),
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_list),
),
state=CreatorTestsSG.edit_attempts,
),
Window(
Const("<b>👥 Изменение группы</b>\n\n🎓 <b>Выберите группу</b> или удалите привязку:"),
ScrollingGroup(
-4
View File
@@ -19,10 +19,6 @@ from trudex.infrastructure.utils.config import Config
class DatabaseProvider(Provider):
@provide(scope=Scope.APP)
def get_config(self) -> Config:
return Config.from_toml("config.toml")
@provide(scope=Scope.APP)
def get_session_maker(self, config: Config) -> async_sessionmaker[AsyncSession]:
return new_session_maker(config.database.url)
+12
View File
@@ -10,6 +10,12 @@ class BotConfig:
creator_id: int
@dataclass
class SecurityConfig:
test_hash_salt: str
test_hash_length: int = 16
@dataclass
class DatabaseConfig:
host: str
@@ -27,6 +33,7 @@ class DatabaseConfig:
class Config:
bot: BotConfig
database: DatabaseConfig
security: SecurityConfig
@classmethod
def from_toml(cls, path: str | Path) -> Self:
@@ -35,6 +42,7 @@ class Config:
bot_data: dict[str, str | int] = data["bot"]
db_data: dict[str, str | int] = data["database"]
security_data: dict[str, str | int] = data["security"]
return cls(
bot=BotConfig(
@@ -47,5 +55,9 @@ class Config:
user=str(db_data["user"]),
password=str(db_data["password"]),
database=str(db_data["database"])
),
security=SecurityConfig(
test_hash_salt=str(security_data["test_hash_salt"]),
test_hash_length=int(security_data.get("test_hash_length", 16))
)
)
@@ -0,0 +1,25 @@
import hashlib
import hmac
import string
def generate_alpha_id(n: int, secret_key: str, length: int = 16) -> str:
data = str(n).encode('utf-8')
key = secret_key.encode('utf-8')
digest = hmac.new(key, data, hashlib.sha256).digest()
num = int.from_bytes(digest, byteorder='big')
alphabet = string.ascii_letters
result = []
while num > 0:
num, rem = divmod(num, 52)
result.append(alphabet[rem])
encoded = "".join(result)
if len(encoded) < length:
encoded = encoded.ljust(length, alphabet[0])
return encoded[:length]