Обновлена логика прохождения информационных заданий и агрегированного прогресса курсов, добавлена валидация изображений для курсов, модулей и заданий

Author: Toksi86

courses/admin.py

@@ -19,7 +19,7 @@
     UserTaskAnswerFile,
     UserTaskAnswerOption,
 )
-from .models.content import looks_like_image_file
+from .models.file_validation import looks_like_image_file
 
 # Admin-only captions for sections in app index
 CourseModule._meta.verbose_name = "Модуль"
@@ -77,6 +77,15 @@ def _courses_get_app_list(self, request, app_label=None):
     admin.site._courses_order_patched = True
 
 
+def _validate_image_upload(form: forms.ModelForm, field_name: str, error_message: str) -> None:
+    uploaded_file = form.cleaned_data.get(field_name)
+    if uploaded_file and not looks_like_image_file(
+        mime_type=getattr(uploaded_file, "content_type", ""),
+        extension=getattr(uploaded_file, "name", "").rsplit(".", 1)[-1],
+    ):
+        form.add_error(field_name, error_message)
+
+
 class OrderUniqueInlineFormSet(BaseInlineFormSet):
     duplicate_field_error = "Такой порядковый номер уже используется в этом разделе."
     duplicate_form_error = "Найдены дублирующиеся значения. Исправьте строки ниже."
@@ -134,6 +143,25 @@ class Meta:
         model = Course
         fields = "__all__"
 
+    def clean(self):
+        cleaned_data = super().clean()
+        _validate_image_upload(
+            self,
+            "avatar_upload",
+            "В поле аватара можно загрузить только файл изображения.",
+        )
+        _validate_image_upload(
+            self,
+            "card_cover_upload",
+            "В поле обложки карточки можно загрузить только файл изображения.",
+        )
+        _validate_image_upload(
+            self,
+            "header_cover_upload",
+            "В поле обложки шапки можно загрузить только файл изображения.",
+        )
+        return cleaned_data
+
 
 class CourseTaskAdminForm(forms.ModelForm):
     image_upload = forms.FileField(
@@ -153,14 +181,11 @@ def clean(self):
         cleaned_data = super().clean()
         image_upload = cleaned_data.get("image_upload")
         attachment_upload = cleaned_data.get("attachment_upload")
-        if image_upload and not looks_like_image_file(
-            mime_type=getattr(image_upload, "content_type", ""),
-            extension=getattr(image_upload, "name", "").rsplit(".", 1)[-1],
-        ):
-            self.add_error(
-                "image_upload",
-                "В поле изображения можно загрузить только файл изображения.",
-            )
+        _validate_image_upload(
+            self,
+            "image_upload",
+            "В поле изображения можно загрузить только файл изображения.",
+        )
 
         # Preserve the fact that a file was provided so model validation
         # doesn't add a second "required image" error for the same field.
@@ -179,6 +204,15 @@ class Meta:
         model = CourseModule
         fields = "__all__"
 
+    def clean(self):
+        cleaned_data = super().clean()
+        _validate_image_upload(
+            self,
+            "avatar_upload",
+            "В поле аватара можно загрузить только файл изображения.",
+        )
+        return cleaned_data
+
 
 class CourseModuleInline(admin.TabularInline):
     model = CourseModule

courses/models/answers.py

@@ -96,23 +96,45 @@ def clean(self):
         if self.task_id is None:
             return
 
-        if self.task.task_kind != CourseTaskKind.QUESTION:
-            errors["task"] = "Ответ можно отправлять только на вопросное задание."
-
-        answer_type = self.task.answer_type
-        is_text_filled = CourseTask._require_non_blank(self.answer_text)
-        if answer_type in (
-            CourseTaskAnswerType.TEXT,
-            CourseTaskAnswerType.TEXT_AND_FILES,
-        ) and not is_text_filled:
-            errors["answer_text"] = "Для выбранного типа ответа требуется заполнить текст."
-
-        if answer_type in (
-            CourseTaskAnswerType.SINGLE_CHOICE,
-            CourseTaskAnswerType.MULTIPLE_CHOICE,
-            CourseTaskAnswerType.FILES,
-        ) and is_text_filled:
-            errors["answer_text"] = "Для выбранного типа ответа текст не используется."
+        if self.task.task_kind == CourseTaskKind.INFORMATIONAL:
+            if CourseTask._require_non_blank(self.answer_text):
+                errors["answer_text"] = (
+                    "Для информационного задания текст ответа не требуется."
+                )
+            if self.status != UserTaskAnswerStatus.SUBMITTED:
+                errors["status"] = (
+                    "Для информационного задания допустим только статус submitted."
+                )
+            if CourseTask._require_non_blank(self.review_comment):
+                errors["review_comment"] = (
+                    "Для информационного задания комментарий проверки не используется."
+                )
+            if self.reviewed_by_id or self.reviewed_at:
+                errors["reviewed_by"] = (
+                    "Для информационного задания поля reviewed_by и reviewed_at не используются."
+                )
+                errors["reviewed_at"] = (
+                    "Для информационного задания поля reviewed_by и reviewed_at не используются."
+                )
+        else:
+            answer_type = self.task.answer_type
+            is_text_filled = CourseTask._require_non_blank(self.answer_text)
+            if answer_type in (
+                CourseTaskAnswerType.TEXT,
+                CourseTaskAnswerType.TEXT_AND_FILES,
+            ) and not is_text_filled:
+                errors["answer_text"] = (
+                    "Для выбранного типа ответа требуется заполнить текст."
+                )
+
+            if answer_type in (
+                CourseTaskAnswerType.SINGLE_CHOICE,
+                CourseTaskAnswerType.MULTIPLE_CHOICE,
+                CourseTaskAnswerType.FILES,
+            ) and is_text_filled:
+                errors["answer_text"] = (
+                    "Для выбранного типа ответа текст не используется."
+                )
 
         if (
             self.task.check_type == CourseTaskCheckType.WITHOUT_REVIEW

courses/models/content.py

@@ -15,25 +15,7 @@
     CourseTaskQuestionType,
 )
 from .course import Course
-
-ALLOWED_IMAGE_EXTENSIONS = {
-    "bmp",
-    "gif",
-    "jpg",
-    "jpeg",
-    "png",
-    "svg",
-    "webp",
-}
-
-
-def looks_like_image_file(*, mime_type: str | None = None, extension: str | None = None) -> bool:
-    normalized_mime_type = (mime_type or "").strip().lower()
-    if normalized_mime_type.startswith("image/"):
-        return True
-
-    normalized_extension = (extension or "").strip().lower().lstrip(".")
-    return normalized_extension in ALLOWED_IMAGE_EXTENSIONS
+from .file_validation import looks_like_image_file
 
 
 class CourseModule(models.Model):
@@ -95,6 +77,17 @@ class Meta:
     def __str__(self):
         return f"CourseModule<{self.id}> - {self.title}"
 
+    def clean(self):
+        super().clean()
+
+        if self.avatar_file_id is not None and not looks_like_image_file(
+            mime_type=self.avatar_file.mime_type,
+            extension=self.avatar_file.extension,
+        ):
+            raise ValidationError(
+                {"avatar_file": "В поле аватара можно выбрать только файл изображения."}
+            )
+
     def save(self, *args, **kwargs):
         self.full_clean()
         super().save(*args, **kwargs)

courses/models/course.py

@@ -8,6 +8,7 @@
 from partner_programs.models import PartnerProgram
 
 from .choices import CourseAccessType, CourseContentStatus
+from .file_validation import looks_like_image_file
 
 
 class Course(models.Model):
@@ -137,6 +138,38 @@ def is_completed_by_date(self) -> bool:
     def clean(self):
         super().clean()
 
+        if self.avatar_file_id is not None and not looks_like_image_file(
+            mime_type=self.avatar_file.mime_type,
+            extension=self.avatar_file.extension,
+        ):
+            raise ValidationError(
+                {"avatar_file": "В поле аватара можно выбрать только файл изображения."}
+            )
+
+        if self.card_cover_file_id is not None and not looks_like_image_file(
+            mime_type=self.card_cover_file.mime_type,
+            extension=self.card_cover_file.extension,
+        ):
+            raise ValidationError(
+                {
+                    "card_cover_file": (
+                        "В поле обложки карточки можно выбрать только файл изображения."
+                    )
+                }
+            )
+
+        if self.header_cover_file_id is not None and not looks_like_image_file(
+            mime_type=self.header_cover_file.mime_type,
+            extension=self.header_cover_file.extension,
+        ):
+            raise ValidationError(
+                {
+                    "header_cover_file": (
+                        "В поле обложки шапки можно выбрать только файл изображения."
+                    )
+                }
+            )
+
         if (
             self.access_type == CourseAccessType.PROGRAM_MEMBERS
             and self.partner_program_id is None

courses/models/file_validation.py

@@ -0,0 +1,22 @@
+ALLOWED_IMAGE_EXTENSIONS = {
+    "bmp",
+    "gif",
+    "jpg",
+    "jpeg",
+    "png",
+    "svg",
+    "webp",
+}
+
+
+def looks_like_image_file(
+    *,
+    mime_type: str | None = None,
+    extension: str | None = None,
+) -> bool:
+    normalized_mime_type = (mime_type or "").strip().lower()
+    if normalized_mime_type.startswith("image/"):
+        return True
+
+    normalized_extension = (extension or "").strip().lower().lstrip(".")
+    return normalized_extension in ALLOWED_IMAGE_EXTENSIONS

courses/services/answers.py

@@ -193,9 +193,51 @@ def submit_user_task_answer(
             {"task": "Отправка ответа доступна только для опубликованных заданий."}
         )
 
-    if task.task_kind != CourseTaskKind.QUESTION:
-        raise ValidationError(
-            {"task": "Отправка ответа доступна только для вопросных заданий."}
+    submitted_at = timezone.now()
+    manager = UserTaskAnswer.objects
+    if transaction.get_connection().in_atomic_block:
+        manager = manager.select_for_update()
+    answer = manager.filter(user=user, task=task).first()
+    if answer is None:
+        answer = UserTaskAnswer(user=user, task=task)
+
+    if task.task_kind == CourseTaskKind.INFORMATIONAL:
+        answer.answer_text = ""
+        answer.submitted_at = submitted_at
+        answer.review_comment = ""
+        answer.reviewed_by = None
+        answer.reviewed_at = None
+        answer.status = UserTaskAnswerStatus.SUBMITTED
+        answer.is_correct = True
+
+        try:
+            answer.save(validate=False)
+        except DjangoValidationError as exc:
+            if hasattr(exc, "message_dict"):
+                raise ValidationError(exc.message_dict) from exc
+            raise ValidationError({"detail": exc.messages}) from exc
+        except IntegrityError:
+            retry_manager = UserTaskAnswer.objects
+            if transaction.get_connection().in_atomic_block:
+                retry_manager = retry_manager.select_for_update()
+            answer = retry_manager.get(user=user, task=task)
+            answer.answer_text = ""
+            answer.submitted_at = submitted_at
+            answer.review_comment = ""
+            answer.reviewed_by = None
+            answer.reviewed_at = None
+            answer.status = UserTaskAnswerStatus.SUBMITTED
+            answer.is_correct = True
+            answer.save(validate=False)
+
+        answer.selected_options.all().delete()
+        answer.files.all().delete()
+        next_task = get_next_published_task(task)
+        return SubmitAnswerResult(
+            answer=answer,
+            is_correct=True,
+            can_continue=True,
+            next_task_id=next_task.id if next_task else None,
         )
 
     if not task.answer_type:
@@ -213,13 +255,6 @@ def submit_user_task_answer(
     )
 
     normalized_text = (payload.answer_text or "").strip()
-    submitted_at = timezone.now()
-    manager = UserTaskAnswer.objects
-    if transaction.get_connection().in_atomic_block:
-        manager = manager.select_for_update()
-    answer = manager.filter(user=user, task=task).first()
-    if answer is None:
-        answer = UserTaskAnswer(user=user, task=task)
 
     answer.answer_text = normalized_text
     answer.submitted_at = submitted_at