Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
80 changes: 49 additions & 31 deletions apps/questions/services/check_questions_answers.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,18 +2,20 @@
from copy import deepcopy
from typing import Any, Iterable, Optional, Union

from courses.models import Task, TaskObject
from django.db import transaction
from django.db.models import QuerySet
from rest_framework import status

from courses.models import TaskObject
from progress.models import TaskObjUserResult, UserAnswersAttemptCounter
from questions.exceptions import (QustionConnectException,
UserAlreadyAnsweredException)
from questions.exceptions import QustionConnectException, UserAlreadyAnsweredException
from questions.mapping import TypeQuestionPoints
from questions.models.answers import AnswerConnect, AnswerSingle
from questions.models.questions import (InfoSlide, QuestionConnect,
QuestionSingleAnswer, QuestionWrite)
from questions.models.questions import (
InfoSlide,
QuestionConnect,
QuestionSingleAnswer,
QuestionWrite,
)
from rest_framework import status


class AbstractAnswersService(ABC):
Expand All @@ -27,7 +29,9 @@ class AbstractAnswersService(ABC):
def __init__(
self,
request_profile_id: int,
request_question: Union[InfoSlide, QuestionWrite, QuestionConnect, QuestionSingleAnswer],
request_question: Union[
InfoSlide, QuestionWrite, QuestionConnect, QuestionSingleAnswer
],
request_task_object: TaskObject,
request_data: dict,
) -> None:
Expand Down Expand Up @@ -90,9 +94,14 @@ def _process_answer_attempt_counter(self, question_answers: Any):
counter.attempts_made_before += 1

# Проверка после подсказки, если counter `after` max, то сохранение без баллов:
if counter.is_take_hint and counter.attempts_made_after >= self.request_question.attempts_after_hint:
if (
counter.is_take_hint
and counter.attempts_made_after >= self.request_question.attempts_after_hint
):
counter.delete()
self._create_tast_obj_result(TypeQuestionPoints.QUESTION_WO_POINTS, correct_answer=False)
self._create_tast_obj_result(
TypeQuestionPoints.QUESTION_WO_POINTS, correct_answer=False
)
question_answer_body = self._create_answer_response_body(question_answers)
question_answer_body["hint"] = self.request_question.hint_text
return question_answer_body, status.HTTP_201_CREATED
Expand All @@ -105,13 +114,20 @@ def _process_answer_attempt_counter(self, question_answers: Any):
counter.save()
response_body = deepcopy(self._UNSUCCESS_RESPONSE_BODY)

if self.request_question.attempts_after_hint and self.request_question.hint_text:
if (
self.request_question.attempts_after_hint
and self.request_question.hint_text
):
response_body["hint"] = self.request_question.hint_text
return response_body, status.HTTP_400_BAD_REQUEST
else:
counter.delete()
self._create_tast_obj_result(TypeQuestionPoints.QUESTION_WO_POINTS, correct_answer=False)
question_answer_body = self._create_answer_response_body(question_answers)
self._create_tast_obj_result(
TypeQuestionPoints.QUESTION_WO_POINTS, correct_answer=False
)
question_answer_body = self._create_answer_response_body(
question_answers
)
if self.request_question.hint_text:
question_answer_body["hint"] = self.request_question.hint_text
return question_answer_body, status.HTTP_201_CREATED
Expand All @@ -125,7 +141,9 @@ def _handle_no_question_validation(
required_data: Optional[Iterable] = None,
) -> tuple[dict, int]:
"""Если у TaskObject отключена проверка (необходимо ответить хоть что-то/сопоставить все даже неправильно)."""
if (required_data and len(self.request_data) < len(required_data)) or not self.request_data:
if (
required_data and len(self.request_data) < len(required_data)
) or not self.request_data:
return self._WRONG_ANSWER_WO_VALIDATION, status.HTTP_400_BAD_REQUEST
self._create_tast_obj_result(point_type)
return self._SUCCESS_RESPONSE_BODY, status.HTTP_201_CREATED
Expand All @@ -145,7 +163,7 @@ def _create_tast_obj_result(
points = 0
else:
skill = self.request_task_object.task.skill
tasks_count = skill.tasks.count() or 1 # защита от деления на 0
tasks_count = Task.published.filter(skill=skill).count()
points = round(80 / tasks_count)

TaskObjUserResult.objects.create(
Expand Down Expand Up @@ -174,10 +192,7 @@ class SingleCorrectAnswerService(AbstractAnswersService):
def _check_correct_answer(self) -> tuple[dict[str, Any], int]:
"""Проверка на корректность ответа + формирование процесса с `couter` (если вопросом это предполагается)."""
question_answer: QuerySet[AnswerSingle] = (
self.request_question
.single_answers
.filter(is_correct=True)
.first()
self.request_question.single_answers.filter(is_correct=True).first()
)

if self.request_data.get("answer_id") == question_answer.id:
Expand All @@ -202,7 +217,9 @@ class QuestionExcludeAnswerService(AbstractAnswersService):

def _check_correct_answer(self) -> tuple[dict[str, Any], int]:
correct_answers_ids: set[int] = set(
self.request_question.single_answers.filter(is_correct=False).values_list("id", flat=True)
self.request_question.single_answers.filter(is_correct=False).values_list(
"id", flat=True
)
)
given_answer_ids: set[int] = set(self.request_data)

Expand All @@ -228,18 +245,21 @@ class QuestionConnectAnswerService(AbstractAnswersService):
_QUSTION_POINTS: TypeQuestionPoints = TypeQuestionPoints.QUESTION_CONNECT

def _check_correct_answer(self) -> tuple[dict[str, Any], int]:
all_answer_options: QuerySet[AnswerConnect] = self.request_question.connect_answers.all()
all_answer_options: QuerySet[AnswerConnect] = (
self.request_question.connect_answers.all()
)

answers_ids: list[int] = set(all_answer_options.values_list("id", flat=True))
user_answer_ids: set[int] = (
{answer["right_id"] for answer in self.request_data}
| {answer["left_id"] for answer in self.request_data}
)
user_answer_ids: set[int] = {
answer["right_id"] for answer in self.request_data
} | {answer["left_id"] for answer in self.request_data}

if answers_ids != user_answer_ids:
raise QustionConnectException("Wrong ids or need more answers.")

answer_is_correct: bool = all([answer["right_id"] == answer["left_id"] for answer in self.request_data])
answer_is_correct: bool = all(
[answer["right_id"] == answer["left_id"] for answer in self.request_data]
)
if answer_is_correct:
self._create_tast_obj_result(self._QUSTION_POINTS)
self._delete_self_counter()
Expand All @@ -253,17 +273,14 @@ def _check_correct_answer(self) -> tuple[dict[str, Any], int]:
def _create_answer_response_body(self, question_answer: AnswerSingle):
response_body = deepcopy(self._UNSUCCESS_RESPONSE_BODY)
response_body["answer_ids"] = [
{
"left_id": answer.id,
"right_id": answer.id
}
for answer in question_answer
{"left_id": answer.id, "right_id": answer.id} for answer in question_answer
]
return response_body


class QuestionWriteAnswerService(AbstractAnswersService):
"""Проверка (запись) ответа пользователя на вопрос с вводом ответа."""

_QUSTION_POINTS: TypeQuestionPoints = TypeQuestionPoints.QUESTION_WRITE

def create_answer(self) -> tuple[dict[str, Any], int]:
Expand All @@ -282,6 +299,7 @@ def _create_answer_response_body(self) -> dict[str, Any]:

class InfoSlideAnswerService(AbstractAnswersService):
"""Проверка (запись) ответа пользователя инфо слайд."""

_QUSTION_POINTS: TypeQuestionPoints = TypeQuestionPoints.INFO_SLIDE

def create_answer(self) -> tuple[dict[str, Any], int]:
Expand Down