MathApp/content/question_io.py

"""
وارد/خارج کردنِ سوال‌ها به‌صورتِ فایل، از داخلِ پنلِ ادمین.

چرا: تا پیش از این تنها راهِ گذاشتنِ سوال روی یک سرور، زدنِ دستور در ترمینالِ
همان سرور بود. یعنی هر تغییرِ محتوایی به کسی که SSH دارد وابسته می‌شد، تاریخچه
نداشت، و برگشت‌پذیر نبود. محتوا داده است نه کد؛ باید با رابطِ کاربری ویرایش شود.

قالبِ فایل عمداً ساده است تا یک معلم هم بتواند در ویرایشگرِ متن درستش کند:

    [
      {
        "lessonId": 42,
        "prompt": "حاصل ۲+۳ چند است؟",
        "type": "multiple_choice",
        "choices": ["۴", "۵", "۶", "۷"],
        "correct": 1
      }
    ]

`correct` شماره‌ی گزینه‌ی درست است (از صفر). خروجیِ export دقیقاً همین قالب را
می‌دهد، پس چرخه‌ی «بگیر → ویرایش کن → برگردان» بسته است.
"""
import json

from django.db import transaction

from .models import Choice, Lesson, Question

VALID_TYPES = {value for value, _ in Question.TYPE_CHOICES}


class ImportError_(Exception):
    """خطای قابلِ نمایش به کاربر (نه باگ)."""


def _validate_row(row, index: int) -> dict:
    where = f'ردیف {index + 1}'
    if not isinstance(row, dict):
        raise ImportError_(f'{where}: باید یک شیء JSON باشد.')

    lesson_id = row.get('lessonId')
    if not isinstance(lesson_id, int):
        raise ImportError_(f'{where}: lessonId باید عدد باشد.')

    prompt = (row.get('prompt') or '').strip()
    if not prompt:
        raise ImportError_(f'{where}: prompt خالی است.')

    qtype = row.get('type') or 'multiple_choice'
    if qtype not in VALID_TYPES:
        raise ImportError_(f'{where}: نوعِ «{qtype}» معتبر نیست. مجاز: {"، ".join(sorted(VALID_TYPES))}')

    choices = row.get('choices')
    if not isinstance(choices, list) or len(choices) < 2:
        raise ImportError_(f'{where}: حداقل دو گزینه لازم است.')
    choices = [str(c).strip() for c in choices]
    if any(not c for c in choices):
        raise ImportError_(f'{where}: گزینه‌ی خالی دارد.')
    if len(set(choices)) != len(choices):
        raise ImportError_(f'{where}: گزینه‌های تکراری دارد.')

    correct = row.get('correct')
    if not isinstance(correct, int) or not (0 <= correct < len(choices)):
        raise ImportError_(f'{where}: correct باید شماره‌ی یکی از گزینه‌ها باشد (از صفر).')

    return {
        'lesson_id': lesson_id,
        'prompt': prompt,
        'type': qtype,
        'choices': choices,
        'correct': correct,
    }


def parse(raw: bytes) -> list[dict]:
    try:
        data = json.loads(raw.decode('utf-8'))
    except UnicodeDecodeError:
        raise ImportError_('فایل باید UTF-8 باشد.')
    except json.JSONDecodeError as exc:
        raise ImportError_(f'JSON نامعتبر است: {exc.msg} (خط {exc.lineno})')

    if isinstance(data, dict):
        data = data.get('questions', data)
    if not isinstance(data, list):
        raise ImportError_('محتوای فایل باید یک آرایه از سوال‌ها باشد.')
    if not data:
        raise ImportError_('فایل خالی است.')

    return [_validate_row(row, i) for i, row in enumerate(data)]


@transaction.atomic
def import_questions(rows: list[dict], *, replace: bool = False) -> dict:
    """
    همه‌ی ردیف‌ها یا با هم نوشته می‌شوند یا هیچ‌کدام — نیمه‌کاره نمی‌ماند.

    replace=True یعنی سوال‌های قبلیِ همان درس‌ها اول پاک شوند؛ بدونِ آن
    سوال‌ها اضافه می‌شوند.
    """
    lesson_ids = {row['lesson_id'] for row in rows}
    existing = set(Lesson.objects.filter(pk__in=lesson_ids).values_list('pk', flat=True))
    missing = sorted(lesson_ids - existing)
    if missing:
        raise ImportError_(
            'این درس‌ها وجود ندارند: ' + '، '.join(str(m) for m in missing[:10])
            + ('…' if len(missing) > 10 else '')
        )

    removed = 0
    if replace:
        # delete() مجموعِ همه‌ی ردیف‌های آبشاری (گزینه‌ها، پاسخ‌ها، گزارش‌ها) را
        # برمی‌گرداند؛ آنچه به کاربر گزارش می‌شود باید فقط تعدادِ سوال باشد.
        _, per_model = Question.objects.filter(lesson_id__in=lesson_ids).delete()
        removed = per_model.get('content.Question', 0)

    created = 0
    for row in rows:
        question = Question.objects.create(
            lesson_id=row['lesson_id'], type=row['type'], prompt=row['prompt'],
        )
        Choice.objects.bulk_create([
            Choice(question=question, value=value, is_correct=(i == row['correct']))
            for i, value in enumerate(row['choices'])
        ])
        created += 1

    return {'created': created, 'removed': removed, 'lessons': len(lesson_ids)}


def export_questions(queryset) -> list[dict]:
    """خروجی دقیقاً همان قالبی‌ست که import می‌خواند."""
    out = []
    for question in queryset.prefetch_related('choices').select_related('lesson'):
        choices = list(question.choices.all())
        correct = next((i for i, c in enumerate(choices) if c.is_correct), 0)
        out.append({
            'lessonId': question.lesson_id,
            'prompt': question.prompt,
            'type': question.type,
            'choices': [c.value for c in choices],
            'correct': correct,
        })
    return out