This commit is contained in:
2026-01-07 00:10:25 +03:00
parent 7d2a734b7d
commit ebdc9954de
15 changed files with 600 additions and 42 deletions
@@ -0,0 +1,24 @@
"""add warning_sent_at to test_attempt
Revision ID: c4d5e6f7a8b9
Revises: b1c2d3e4f5a6
Create Date: 2026-01-06
"""
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
revision: str = 'c4d5e6f7a8b9'
down_revision: str | None = 'b1c2d3e4f5a6'
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
op.add_column('test_attempts', sa.Column('warning_sent_at', sa.DateTime(), nullable=True))
def downgrade() -> None:
op.drop_column('test_attempts', 'warning_sent_at')
@@ -12,6 +12,7 @@ from quizzi.infrastructure.database.models import QuestionType
from quizzi.infrastructure.database.repo.test import TestRepository
from quizzi.infrastructure.database.repo.test_attempt import TestAttemptRepository
from quizzi.infrastructure.utils.rate_limiter import PasswordRateLimiter
from quizzi.infrastructure.utils.timezone import now_msk_naive
@inject
@@ -92,15 +93,22 @@ async def on_start_deeplink_test(
if active_attempt:
await attempt_repo.attempt_dao.delete(active_attempt.id)
if test.password:
if test.time_limit:
await manager.start(UserTestSG.confirm_time_limit, mode=StartMode.NORMAL, data={
"test_id": test_id,
"time_limit": test.time_limit,
"has_password": bool(test.password),
})
elif test.password:
allowed, wait_time = await rate_limiter.check(user_id)
if not allowed:
minutes = int(wait_time // 60) + 1
await _callback.answer(f"⏳ Слишком много попыток. Подождите {minutes} мин.", show_alert=True)
return
manager.dialog_data["time_limit"] = None
await manager.switch_to(UserDeeplinkSG.password_input)
else:
await start_test_without_password(manager, test_repo, attempt_repo, test_id, user_id)
await start_test_without_password(manager, test_repo, attempt_repo, test_id, user_id, None)
async def start_test_without_password(
@@ -109,6 +117,7 @@ async def start_test_without_password(
attempt_repo: TestAttemptRepository,
test_id: int,
user_id: int,
time_limit: int | None = None,
):
_, questions = await test_repo.get_test_with_questions(test_id)
@@ -116,6 +125,7 @@ async def start_test_without_password(
return
attempt = await attempt_repo.attempt_dao.create(user_id=user_id, test_id=test_id)
started_at = now_msk_naive()
first_question, _ = await test_repo.get_question_with_options(questions[0].id)
@@ -138,6 +148,8 @@ async def start_test_without_password(
"questions": [q.id for q in questions],
"current_question_index": 0,
"user_answers": {},
"time_limit": time_limit,
"started_at": started_at.isoformat(),
}
)
@@ -170,8 +182,9 @@ async def on_deeplink_password_input(
if message.text and message.text.strip() == test.password:
await message.answer("✅ Пароль верный")
time_limit = manager.dialog_data.get("time_limit")
await start_test_without_password(
manager, test_repo, attempt_repo, test_id, message.from_user.id
manager, test_repo, attempt_repo, test_id, message.from_user.id, time_limit
)
else:
allowed, wait_time = await rate_limiter.check(message.from_user.id)
@@ -13,11 +13,13 @@ class UserMenuSG(StatesGroup):
class UserTestSG(StatesGroup):
password_input = State()
confirm_time_limit = State()
question_single = State()
question_multiple = State()
question_input = State()
results = State()
detailed_results = State()
time_expired = State()
class UserDeeplinkSG(StatesGroup):
@@ -1,9 +1,11 @@
from datetime import datetime
from aiogram.enums import ContentType as AiogramContentType
from aiogram.types import CallbackQuery, Message
from aiogram_dialog import Dialog, DialogManager, StartMode, Window
from aiogram_dialog.api.entities import MediaAttachment, MediaId
from aiogram_dialog.widgets.input import MessageInput
from aiogram_dialog.widgets.kbd import Button, Column, Multiselect, Radio
from aiogram_dialog.widgets.kbd import Button, Column, Multiselect, Radio, Row
from aiogram_dialog.widgets.media import DynamicMedia
from aiogram_dialog.widgets.text import Const, Format
from dishka import FromDishka
@@ -28,6 +30,145 @@ async def get_state_for_question_type(question_type: str):
return UserTestSG.question_input
def get_remaining_time(started_at: datetime, time_limit: int) -> int | None:
if not time_limit:
return None
elapsed = (now_msk_naive() - started_at).total_seconds()
remaining = time_limit - elapsed
return max(0, int(remaining))
def format_time(seconds: int) -> str:
if seconds >= 3600:
hours = seconds // 3600
minutes = (seconds % 3600) // 60
secs = seconds % 60
return f"{hours}:{minutes:02d}:{secs:02d}"
else:
minutes = seconds // 60
secs = seconds % 60
return f"{minutes}:{secs:02d}"
def is_time_expired(started_at: datetime, time_limit: int | None) -> bool:
if not time_limit:
return False
remaining = get_remaining_time(started_at, time_limit)
return remaining is not None and remaining <= 0
async def finish_test_by_timeout(
manager: DialogManager,
attempt_repo: TestAttemptRepository,
answer_dao: UserAnswerDAO,
test_repo: TestRepository,
attempt_id: int,
questions: list[int],
user_answers: dict,
are_results_viewable: bool = False,
):
answered_question_ids = set()
answers = await attempt_repo.get_answers_for_attempt(attempt_id)
for answer in answers:
answered_question_ids.add(answer.question_id)
for question_id in questions:
if question_id in answered_question_ids:
continue
answer_data = user_answers.get(str(question_id))
if answer_data:
question, options = await test_repo.get_question_with_options(question_id)
if not question:
continue
if answer_data["type"] == "single":
selected_option_id = answer_data["answer"]
correct_options = [opt for opt in options if opt.is_correct]
is_correct = any(opt.id == selected_option_id for opt in correct_options)
selected_text = next((opt.text for opt in options if opt.id == selected_option_id), "")
await answer_dao.create(
attempt_id=attempt_id,
question_id=question_id,
selected_option_id=selected_option_id,
text_answer=selected_text,
is_correct=is_correct,
)
elif answer_data["type"] == "multiple":
selected_option_ids = set(answer_data["answer"])
selected_texts = sorted([opt.text for opt in options if opt.id in selected_option_ids])
correct_texts = sorted([opt.text for opt in options if opt.is_correct])
is_correct = selected_texts == correct_texts
await answer_dao.create(
attempt_id=attempt_id,
question_id=question_id,
text_answer="|".join(selected_texts),
is_correct=is_correct,
)
else:
await answer_dao.create(
attempt_id=attempt_id,
question_id=question_id,
text_answer=None,
is_correct=False,
)
correct_count = await attempt_repo.calculate_attempt_score(attempt_id)
total_questions = len(questions)
score = round((correct_count / total_questions * 100)) if total_questions > 0 else 0
is_passed = score >= 50
await attempt_repo.finish_attempt(attempt_id, score, is_passed)
manager.dialog_data["score"] = score
manager.dialog_data["correct_count"] = correct_count
manager.dialog_data["total_questions"] = total_questions
manager.dialog_data["is_passed"] = is_passed
manager.dialog_data["are_results_viewable"] = are_results_viewable
manager.dialog_data["time_expired"] = True
await manager.switch_to(UserTestSG.time_expired)
async def check_time_and_finish_if_expired(
manager: DialogManager,
test_dao: TestDAO,
test_repo: TestRepository,
attempt_repo: TestAttemptRepository,
answer_dao: UserAnswerDAO,
) -> bool:
start_data = manager.start_data or {}
assert isinstance(start_data, dict)
test_id = manager.dialog_data.get("test_id") or start_data.get("test_id")
attempt_id = manager.dialog_data.get("attempt_id") or start_data.get("attempt_id")
started_at_str = manager.dialog_data.get("started_at") or start_data.get("started_at")
time_limit = manager.dialog_data.get("time_limit") or start_data.get("time_limit")
if not time_limit or not started_at_str or not attempt_id:
return False
started_at = datetime.fromisoformat(started_at_str) if isinstance(started_at_str, str) else started_at_str
if is_time_expired(started_at, time_limit):
questions = manager.dialog_data.get("questions") or start_data.get("questions", [])
user_answers = manager.dialog_data.get("user_answers", {})
test = await test_dao.get_by_id(test_id) if test_id else None
are_results_viewable = test.are_results_viewable if test else False
await finish_test_by_timeout(
manager, attempt_repo, answer_dao, test_repo,
attempt_id, questions, user_answers, are_results_viewable
)
return True
return False
@inject
async def on_start_test(
_callback: CallbackQuery,
@@ -70,34 +211,119 @@ async def on_start_test(
if active_attempt:
await attempt_repo.attempt_dao.delete(active_attempt.id)
if test.password:
if test.time_limit:
await manager.start(UserTestSG.confirm_time_limit, mode=StartMode.NORMAL, data={
"test_id": test_id,
"time_limit": test.time_limit,
"has_password": bool(test.password),
})
elif test.password:
allowed, wait_time = await rate_limiter.check(user_id)
if not allowed:
minutes = int(wait_time // 60) + 1
await _callback.answer(f"⏳ Слишком много попыток. Подождите {minutes} мин.", show_alert=True)
return
await manager.start(UserTestSG.password_input, mode=StartMode.NORMAL, data={"test_id": test_id})
await manager.start(UserTestSG.password_input, mode=StartMode.NORMAL, data={
"test_id": test_id,
"time_limit": None,
})
else:
await start_test_directly(manager, test_repo, attempt_repo, test_id, user_id, None)
async def get_confirm_time_limit_data(dialog_manager: DialogManager, **_kwargs):
start_data = dialog_manager.start_data or {}
assert isinstance(start_data, dict)
time_limit = start_data.get("time_limit", 0)
minutes = time_limit // 60
return {
"time_limit_text": (
f"⚠️ <b>ВНИМАНИЕ!</b>\n\n"
f"У этого теста установлен лимит времени:\n"
f"<b>⏱️ {minutes} минут</b>\n\n"
f"После начала теста таймер нельзя остановить.\n"
f"Если время закончится, тест будет автоматически завершён.\n\n"
f"<b>Вы готовы начать?</b>"
)
}
@inject
async def on_confirm_time_limit(
_callback: CallbackQuery,
_button: Button,
manager: DialogManager,
test_dao: FromDishka[TestDAO],
test_repo: FromDishka[TestRepository],
attempt_repo: FromDishka[TestAttemptRepository],
rate_limiter: FromDishka[PasswordRateLimiter],
):
assert _callback.from_user is not None
start_data = manager.start_data or {}
assert isinstance(start_data, dict)
test_id = start_data.get("test_id")
time_limit = start_data.get("time_limit")
has_password = start_data.get("has_password", False)
user_id = _callback.from_user.id
if not test_id:
await _callback.answer("❌ Тест не найден")
return
if has_password:
test = await test_dao.get_by_id(test_id)
if test and test.password:
allowed, wait_time = await rate_limiter.check(user_id)
if not allowed:
minutes = int(wait_time // 60) + 1
await _callback.answer(f"⏳ Слишком много попыток. Подождите {minutes} мин.", show_alert=True)
return
await manager.start(UserTestSG.password_input, mode=StartMode.NORMAL, data={
"test_id": test_id,
"time_limit": time_limit,
})
return
await start_test_directly(manager, test_repo, attempt_repo, test_id, user_id, time_limit)
async def on_cancel_time_limit(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.done()
async def start_test_directly(
manager: DialogManager,
test_repo: TestRepository,
attempt_repo: TestAttemptRepository,
test_id: int,
user_id: int,
time_limit: int | None,
):
_, questions = await test_repo.get_test_with_questions(test_id)
if not questions:
await _callback.answer("❌ В тесте нет вопросов")
return
attempt = await attempt_repo.attempt_dao.create(user_id=user_id, test_id=test_id)
started_at = now_msk_naive()
first_question, _ = await test_repo.get_question_with_options(questions[0].id)
first_state = await get_state_for_question_type(first_question.question_type if first_question else QuestionType.SINGLE)
await manager.start(
first_state,
mode=StartMode.NORMAL,
mode=StartMode.RESET_STACK,
data={
"test_id": test_id,
"attempt_id": attempt.id,
"questions": [q.id for q in questions],
"current_question_index": 0,
"user_answers": {},
"time_limit": time_limit,
"started_at": started_at.isoformat(),
}
)
@@ -116,6 +342,7 @@ async def on_password_input(
start_data = manager.start_data or {}
assert isinstance(start_data, dict)
test_id = start_data.get("test_id")
time_limit = start_data.get("time_limit")
if not test_id:
await message.answer("❌ Тест не найден")
@@ -137,14 +364,18 @@ async def on_password_input(
return
attempt = await attempt_repo.attempt_dao.create(user_id=message.from_user.id, test_id=test_id)
started_at = now_msk_naive()
first_question, _ = await test_repo.get_question_with_options(questions[0].id)
first_state = await get_state_for_question_type(first_question.question_type if first_question else QuestionType.SINGLE)
manager.dialog_data["test_id"] = test_id
manager.dialog_data["attempt_id"] = attempt.id
manager.dialog_data["questions"] = [q.id for q in questions]
manager.dialog_data["current_question_index"] = 0
manager.dialog_data["user_answers"] = {}
manager.dialog_data["time_limit"] = time_limit
manager.dialog_data["started_at"] = started_at.isoformat()
await manager.switch_to(first_state)
else:
@@ -184,6 +415,23 @@ async def get_question_data(
start_data = dialog_manager.start_data or {}
assert isinstance(start_data, dict)
time_limit = dialog_manager.dialog_data.get("time_limit") or start_data.get("time_limit")
started_at_str = dialog_manager.dialog_data.get("started_at") or start_data.get("started_at")
timer_str = ""
if time_limit and started_at_str:
started_at = datetime.fromisoformat(started_at_str) if isinstance(started_at_str, str) else started_at_str
remaining = get_remaining_time(started_at, time_limit)
if remaining is not None:
if remaining <= 0:
return {
"question_text": "⏰ <b>Время истекло!</b>",
"options": [],
"media": None,
"time_expired": True,
}
timer_str = f" ⏱️ {format_time(remaining)}"
current_index = dialog_manager.dialog_data.get("current_question_index")
if current_index is None:
current_index = start_data.get("current_question_index", 0)
@@ -209,13 +457,26 @@ async def get_question_data(
)
return {
"question_text": f"<b>📝 Вопрос {progress}</b>\n\n<blockquote>{question.text}</blockquote>",
"question_text": f"<b>📝 Вопрос {progress}{timer_str}</b>\n\n<blockquote>{question.text}</blockquote>",
"options": [(opt.text, str(opt.id)) for opt in options],
"media": media,
}
async def on_single_answer_selected(_callback: CallbackQuery, _widget, manager: DialogManager, item_id: str):
@inject
async def on_single_answer_selected(
_callback: CallbackQuery,
_widget,
manager: DialogManager,
item_id: str,
test_dao: FromDishka[TestDAO],
test_repo: FromDishka[TestRepository],
attempt_repo: FromDishka[TestAttemptRepository],
answer_dao: FromDishka[UserAnswerDAO],
):
if await check_time_and_finish_if_expired(manager, test_dao, test_repo, attempt_repo, answer_dao):
return
start_data = manager.start_data or {}
assert isinstance(start_data, dict)
current_index = manager.dialog_data.get("current_question_index")
@@ -232,7 +493,20 @@ async def on_single_answer_selected(_callback: CallbackQuery, _widget, manager:
manager.dialog_data["user_answers"] = user_answers
async def on_multiple_answer_changed(_event, widget, manager: DialogManager, _data: str):
@inject
async def on_multiple_answer_changed(
_event,
widget,
manager: DialogManager,
_data: str,
test_dao: FromDishka[TestDAO],
test_repo: FromDishka[TestRepository],
attempt_repo: FromDishka[TestAttemptRepository],
answer_dao: FromDishka[UserAnswerDAO],
):
if await check_time_and_finish_if_expired(manager, test_dao, test_repo, attempt_repo, answer_dao):
return
start_data = manager.start_data or {}
assert isinstance(start_data, dict)
current_index = manager.dialog_data.get("current_question_index")
@@ -261,6 +535,9 @@ async def on_text_answer_input(
answer_dao: FromDishka[UserAnswerDAO],
test_dao: FromDishka[TestDAO],
):
if await check_time_and_finish_if_expired(manager, test_dao, test_repo, attempt_repo, answer_dao):
return
start_data = manager.start_data or {}
assert isinstance(start_data, dict)
current_index = manager.dialog_data.get("current_question_index")
@@ -321,6 +598,9 @@ async def on_next_question(
answer_dao: FromDishka[UserAnswerDAO],
test_dao: FromDishka[TestDAO],
):
if await check_time_and_finish_if_expired(manager, test_dao, test_repo, attempt_repo, answer_dao):
return
start_data = manager.start_data or {}
assert isinstance(start_data, dict)
current_index = manager.dialog_data.get("current_question_index")
@@ -441,6 +721,29 @@ async def get_results_data(dialog_manager: DialogManager, **_kwargs):
return {"results_text": results_text, "are_results_viewable": are_results_viewable}
async def get_time_expired_data(dialog_manager: DialogManager, **_kwargs):
score = dialog_manager.dialog_data.get("score", 0)
correct_count = dialog_manager.dialog_data.get("correct_count", 0)
total_questions = dialog_manager.dialog_data.get("total_questions", 0)
is_passed = dialog_manager.dialog_data.get("is_passed", False)
are_results_viewable = dialog_manager.dialog_data.get("are_results_viewable", False)
if is_passed:
status = "✅ <b>Тест пройден!</b>"
else:
status = "❌ <b>Тест не пройден</b>"
results_text = (
f"⏰ <b>Время истекло!</b>\n\n"
f"{status}\n\n"
f"📊 <b>Результат:</b> {score}%\n"
f"✏️ <b>Правильных ответов:</b> {correct_count} из {total_questions}\n\n"
f"<i>Неотвеченные вопросы засчитаны как неправильные.</i>"
)
return {"results_text": results_text, "are_results_viewable": are_results_viewable}
async def on_back_to_menu(_callback: CallbackQuery, _button: Button, manager: DialogManager):
await manager.start(UserMenuSG.main, mode=StartMode.RESET_STACK)
@@ -494,6 +797,15 @@ async def get_detailed_results_data(
take_test_dialog = Dialog(
Window(
Format("{time_limit_text}"),
Row(
Button(Const("✅ Начать"), id="confirm", on_click=on_confirm_time_limit),
Button(Const("❌ Отмена"), id="cancel", on_click=on_cancel_time_limit),
),
state=UserTestSG.confirm_time_limit,
getter=get_confirm_time_limit_data,
),
Window(
Const("<b>🔑 Введите пароль для доступа к тесту:</b>"),
MessageInput(on_password_input),
@@ -555,6 +867,20 @@ take_test_dialog = Dialog(
state=UserTestSG.results,
getter=get_results_data,
),
Window(
Format("{results_text}"),
Column(
Button(
Const("📋 Подробные результаты"),
id="detailed",
on_click=on_show_detailed_results,
when="are_results_viewable",
),
Button(Const("◀️ В главное меню"), id="back", on_click=on_back_to_menu),
),
state=UserTestSG.time_expired,
getter=get_time_expired_data,
),
Window(
Format("{detailed_text}"),
Button(Const("◀️ Назад"), id="back", on_click=on_back_to_results),
+1
View File
@@ -74,6 +74,7 @@ class TestAttempt:
test_id: int
started_at: datetime
finished_at: datetime | None = None
warning_sent_at: datetime | None = None
score: int = 0
is_passed: bool = False
@@ -10,7 +10,6 @@ from quizzi.infrastructure.database.models import Test
class _UNSET:
"""Sentinel для различения None и "не передано"."""
pass
@@ -46,6 +46,7 @@ class TestAttemptDAO:
self,
attempt_id: int,
finished_at: datetime | None = None,
warning_sent_at: datetime | None = None,
score: int | None = None,
is_passed: bool | None = None,
) -> DomainTestAttempt | None:
@@ -58,6 +59,8 @@ class TestAttemptDAO:
if finished_at is not None:
attempt.finished_at = finished_at
if warning_sent_at is not None:
attempt.warning_sent_at = warning_sent_at
if score is not None:
attempt.score = score
if is_passed is not None:
@@ -9,7 +9,6 @@ from quizzi.infrastructure.database.models import User
class _UNSET:
"""Sentinel для различения None и "не передано"."""
pass
@@ -13,6 +13,7 @@ class TestAttemptDTO:
test_id=self.model.test_id,
started_at=self.model.started_at,
finished_at=self.model.finished_at,
warning_sent_at=self.model.warning_sent_at,
score=self.model.score,
is_passed=self.model.is_passed,
)
@@ -105,6 +105,7 @@ class TestAttempt(Base):
test_id: Mapped[int] = mapped_column(ForeignKey("tests.id"), index=True)
started_at: Mapped[datetime] = mapped_column(server_default=func.now())
finished_at: Mapped[datetime | None] = mapped_column(default=None)
warning_sent_at: Mapped[datetime | None] = mapped_column(default=None)
score: Mapped[int] = mapped_column(Integer, default=0)
is_passed: Mapped[bool] = mapped_column(default=False)
@@ -125,7 +125,6 @@ class TestRepository:
async def get_questions_with_options_by_ids(
self, question_ids: list[int]
) -> dict[int, tuple[Question, list[Option]]]:
"""Загружает вопросы с опциями по списку ID за один запрос."""
if not question_ids:
return {}
@@ -1,3 +1,4 @@
from datetime import datetime
from typing import final
from sqlalchemy import func, select
@@ -9,7 +10,10 @@ from quizzi.infrastructure.database.dao.test_attempt import TestAttemptDAO
from quizzi.infrastructure.database.dao.user_answer import UserAnswerDAO
from quizzi.infrastructure.database.dto.test_attempt import TestAttemptDTO
from quizzi.infrastructure.database.dto.user_answer import UserAnswerDTO
from quizzi.infrastructure.database.models import Question as QuestionModel
from quizzi.infrastructure.database.models import Test as TestModel
from quizzi.infrastructure.database.models import TestAttempt as TestAttemptModel
from quizzi.infrastructure.database.models import User as UserModel
from quizzi.infrastructure.database.models import UserAnswer as UserAnswerModel
from quizzi.infrastructure.utils.timezone import now_msk_naive
@@ -173,8 +177,6 @@ class TestAttemptRepository:
}
async def get_most_difficult_questions(self, test_id: int, limit: int = 10) -> list[tuple[int, float]]:
from quizzi.infrastructure.database.models import Question as QuestionModel
result = await self.session.execute(
select(
UserAnswerModel.question_id,
@@ -209,8 +211,6 @@ class TestAttemptRepository:
}
async def get_finished_attempts_with_tests(self, user_id: int) -> list[tuple[TestAttempt, str]]:
from quizzi.infrastructure.database.models import Test as TestModel
result = await self.session.execute(
select(TestAttemptModel, TestModel.title)
.join(TestModel, TestAttemptModel.test_id == TestModel.id)
@@ -222,8 +222,6 @@ class TestAttemptRepository:
return [(TestAttemptDTO(row[0]).to_domain(), row[1]) for row in rows]
async def get_test_attempts_with_users(self, test_id: int) -> list[tuple[TestAttempt, str]]:
from quizzi.infrastructure.database.models import User as UserModel
result = await self.session.execute(
select(TestAttemptModel, UserModel.name, UserModel.first_name)
.join(UserModel, TestAttemptModel.user_id == UserModel.id)
@@ -233,3 +231,56 @@ class TestAttemptRepository:
)
rows = result.all()
return [(TestAttemptDTO(row[0]).to_domain(), row[1] or row[2]) for row in rows]
async def get_expired_active_attempts(self, now: datetime) -> list[tuple[TestAttempt, int]]:
result = await self.session.execute(
select(TestAttemptModel, TestModel.time_limit)
.join(TestModel, TestAttemptModel.test_id == TestModel.id)
.where(TestAttemptModel.finished_at.is_(None))
.where(TestModel.time_limit.isnot(None))
)
rows = result.all()
expired = []
for attempt_model, time_limit in rows:
if time_limit:
elapsed = (now - attempt_model.started_at).total_seconds()
if elapsed >= time_limit:
expired.append((TestAttemptDTO(attempt_model).to_domain(), time_limit))
return expired
async def get_attempts_needing_warning(self, now: datetime) -> list[tuple[TestAttempt, int, int]]:
result = await self.session.execute(
select(
TestAttemptModel,
TestModel.time_limit,
func.count(QuestionModel.id).label("questions_count")
)
.join(TestModel, TestAttemptModel.test_id == TestModel.id)
.join(QuestionModel, QuestionModel.test_id == TestModel.id)
.where(TestAttemptModel.finished_at.is_(None))
.where(TestAttemptModel.warning_sent_at.is_(None))
.where(TestModel.time_limit.isnot(None))
.group_by(TestAttemptModel.id, TestModel.time_limit)
)
rows = result.all()
needing_warning = []
for attempt_model, time_limit, questions_count in rows:
if time_limit and questions_count > 0:
elapsed = (now - attempt_model.started_at).total_seconds()
time_remaining = time_limit - elapsed
threshold = time_limit * 0.1
if time_remaining <= threshold and time_remaining > 0:
needing_warning.append((
TestAttemptDTO(attempt_model).to_domain(),
time_limit,
questions_count
))
return needing_warning
async def mark_warning_sent(self, attempt_id: int, sent_at: datetime) -> None:
await self.attempt_dao.update(attempt_id=attempt_id, warning_sent_at=sent_at)
+19 -2
View File
@@ -1,6 +1,7 @@
import logging
from collections.abc import AsyncIterable
from aiogram import Bot
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from dishka import AsyncContainer, Provider, Scope, provide
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
@@ -16,7 +17,7 @@ from quizzi.infrastructure.database.dao.user_answer import UserAnswerDAO
from quizzi.infrastructure.database.repo.test import TestRepository
from quizzi.infrastructure.database.repo.test_attempt import TestAttemptRepository
from quizzi.infrastructure.database.repo.user import UserRepository
from quizzi.infrastructure.scheduling.tasks import deactivate_expired_tests
from quizzi.infrastructure.scheduling.tasks import deactivate_expired_tests, finish_expired_test_attempts, send_time_warning_notifications
from quizzi.infrastructure.utils.config import Config
from quizzi.infrastructure.utils.rate_limiter import PasswordRateLimiter
@@ -81,7 +82,7 @@ class DatabaseProvider(Provider):
class SchedulerProvider(Provider):
@provide(scope = Scope.APP)
def get_scheduler(self, container: AsyncContainer) -> AsyncIOScheduler:
def get_scheduler(self, container: AsyncContainer, bot: Bot) -> AsyncIOScheduler:
logging.getLogger('apscheduler').setLevel(logging.WARNING)
scheduler = AsyncIOScheduler()
@@ -93,4 +94,20 @@ class SchedulerProvider(Provider):
id='deactivate_expired_tests',
)
scheduler.add_job(
finish_expired_test_attempts,
'interval',
minutes=1,
args=[container, bot],
id='finish_expired_test_attempts',
)
scheduler.add_job(
send_time_warning_notifications,
'interval',
seconds=10,
args=[container, bot],
id='send_time_warning_notifications',
)
return scheduler
@@ -1,8 +1,14 @@
import logging
from aiogram import Bot
from aiogram.exceptions import TelegramAPIError
from dishka import AsyncContainer
from sqlalchemy.exc import SQLAlchemyError
from quizzi.infrastructure.database.dao.test import TestDAO
from quizzi.infrastructure.database.dao.user_answer import UserAnswerDAO
from quizzi.infrastructure.database.repo.test import TestRepository
from quizzi.infrastructure.database.repo.test_attempt import TestAttemptRepository
from quizzi.infrastructure.utils.timezone import now_msk_naive
logger = logging.getLogger(__name__)
@@ -17,3 +23,120 @@ async def deactivate_expired_tests(container: AsyncContainer) -> None:
for test in expired_tests:
await test_dao.update(test.id, is_active=False)
logger.info("Деактивирован истёкший тест: id=%d, title=%s", test.id, test.title)
async def finish_expired_test_attempts(container: AsyncContainer, bot: Bot) -> None:
async with container() as request_container:
attempt_repo = await request_container.get(TestAttemptRepository)
test_repo = await request_container.get(TestRepository)
answer_dao = await request_container.get(UserAnswerDAO)
now = now_msk_naive()
expired_attempts = await attempt_repo.get_expired_active_attempts(now)
for attempt, _ in expired_attempts:
try:
test, questions_with_options = await test_repo.get_full_test(attempt.test_id)
if not test:
continue
question_ids = [q.id for q, _ in questions_with_options]
answered_question_ids = set()
answers = await attempt_repo.get_answers_for_attempt(attempt.id)
for answer in answers:
answered_question_ids.add(answer.question_id)
for question_id in question_ids:
if question_id not in answered_question_ids:
await answer_dao.create(
attempt_id=attempt.id,
question_id=question_id,
text_answer=None,
is_correct=False,
)
correct_count = await attempt_repo.calculate_attempt_score(attempt.id)
total_questions = len(question_ids)
score = round((correct_count / total_questions * 100)) if total_questions > 0 else 0
is_passed = score >= 50
await attempt_repo.finish_attempt(attempt.id, score, is_passed)
status = "пройден ✅" if is_passed else "не пройден ❌"
try:
await bot.send_message(
attempt.user_id,
f"⏰ <b>Время на прохождение теста истекло!</b>\n\n"
f"📝 <b>Тест:</b> {test.title}\n"
f"📊 <b>Результат:</b> {score}%\n"
f"🏆 <b>Статус:</b> {status}\n\n"
f"<i>Неотвеченные вопросы засчитаны как неправильные.</i>"
)
except TelegramAPIError as e:
logger.warning("Не удалось отправить уведомление пользователю %d: %s", attempt.user_id, e)
logger.info(
"Завершена просроченная попытка: attempt_id=%d, user_id=%d, test_id=%d, score=%d%%",
attempt.id, attempt.user_id, attempt.test_id, score
)
except SQLAlchemyError as e:
logger.error("Ошибка при завершении попытки %d: %s", attempt.id, e)
async def send_time_warning_notifications(container: AsyncContainer, bot: Bot) -> None:
async with container() as request_container:
attempt_repo = await request_container.get(TestAttemptRepository)
test_repo = await request_container.get(TestRepository)
now = now_msk_naive()
attempts_needing_warning = await attempt_repo.get_attempts_needing_warning(now)
for attempt, time_limit, questions_count in attempts_needing_warning:
try:
answers = await attempt_repo.get_answers_for_attempt(attempt.id)
if answers:
avg_time_per_question = time_limit / questions_count
time_since_start = (now - attempt.started_at).total_seconds()
expected_answers = int(time_since_start / avg_time_per_question)
if len(answers) >= expected_answers:
continue
test, _ = await test_repo.get_full_test(attempt.test_id)
if not test:
continue
elapsed = (now - attempt.started_at).total_seconds()
remaining_seconds = int(time_limit - elapsed)
remaining_minutes = remaining_seconds // 60
remaining_secs = remaining_seconds % 60
if remaining_minutes > 0:
time_str = f"{remaining_minutes} мин {remaining_secs} сек"
else:
time_str = f"{remaining_secs} сек"
try:
await bot.send_message(
attempt.user_id,
f"⚠️ <b>Внимание! Время заканчивается!</b>\n\n"
f"📝 <b>Тест:</b> {test.title}\n"
f"⏱️ <b>Осталось:</b> {time_str}\n\n"
f"<i>Поторопитесь с ответами!</i>"
)
except TelegramAPIError as e:
logger.warning("Не удалось отправить предупреждение пользователю %d: %s", attempt.user_id, e)
await attempt_repo.mark_warning_sent(attempt.id, now)
logger.info(
"Отправлено предупреждение о времени: attempt_id=%d, user_id=%d, remaining=%ds",
attempt.id, attempt.user_id, remaining_seconds
)
except SQLAlchemyError as e:
logger.error("Ошибка при отправке предупреждения для попытки %d: %s", attempt.id, e)
@@ -9,7 +9,6 @@ def now_msk() -> datetime:
def now_msk_naive() -> datetime:
"""Возвращает текущее время в МСК без timezone info (для сохранения в БД)."""
return datetime.now(MSK_TZ).replace(tzinfo=None)