255 lines
11 KiB
Python
255 lines
11 KiB
Python
"""
|
||
وارد/خارج کردنِ سوال‌ها به‌صورتِ فایل، از داخلِ پنلِ ادمین.
|
||
|
||
چرا: تا پیش از این تنها راهِ گذاشتنِ سوال روی یک سرور، زدنِ دستور در ترمینالِ
|
||
همان سرور بود. یعنی هر تغییرِ محتوایی به کسی که SSH دارد وابسته می‌شد، تاریخچه
|
||
نداشت، و برگشت‌پذیر نبود. محتوا داده است نه کد؛ باید با رابطِ کاربری ویرایش شود.
|
||
|
||
قالبِ فایل عمداً ساده است تا یک معلم هم بتواند در ویرایشگرِ متن درستش کند:
|
||
|
||
[
|
||
{
|
||
"lessonId": 42,
|
||
"prompt": "حاصل ۲+۳ چند است؟",
|
||
"type": "multiple_choice",
|
||
"choices": ["۴", "۵", "۶", "۷"],
|
||
"correct": 1
|
||
}
|
||
]
|
||
|
||
`correct` شماره‌ی گزینه‌ی درست است (از صفر). خروجیِ export دقیقاً همین قالب را
|
||
می‌دهد، پس چرخه‌ی «بگیر → ویرایش کن → برگردان» بسته است.
|
||
"""
|
||
import json
|
||
|
||
from django.db import transaction
|
||
|
||
from rag.services import diagram as diagram_spec
|
||
|
||
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 باید شماره‌ی یکی از گزینه‌ها باشد (از صفر).')
|
||
|
||
figure = row.get('figure')
|
||
if figure is not None:
|
||
# شکلِ نامعتبر بی‌صدا حذف نمی‌شود (برخلافِ دستیار): فایلِ محتوا باید خراب‌بودنش را نشان دهد.
|
||
figure = diagram_spec.validate(figure) if isinstance(figure, dict) else None
|
||
if figure is None:
|
||
raise ImportError_(f'{where}: «figure» شکلِ معتبر نیست (قالبِ rag.services.diagram).')
|
||
|
||
return {
|
||
'lesson_id': lesson_id,
|
||
'prompt': prompt,
|
||
'type': qtype,
|
||
'choices': choices,
|
||
'correct': correct,
|
||
'figure': figure,
|
||
}
|
||
|
||
|
||
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, skip_existing: bool = False,
|
||
) -> dict:
|
||
"""
|
||
همه‌ی ردیف‌ها یا با هم نوشته می‌شوند یا هیچ‌کدام — نیمه‌کاره نمی‌ماند.
|
||
|
||
replace=True یعنی سوال‌های قبلیِ همان درس‌ها اول پاک شوند؛ بدونِ آن
|
||
سوال‌ها اضافه می‌شوند.
|
||
|
||
skip_existing=True یعنی سوالی که همین صورت را در همین درس دارد دوباره
|
||
ساخته نشود. این چیزی است که وارد کردنِ فایلِ ثابتِ ریپو را در هر دیپلوی
|
||
بی‌خطر می‌کند: بدونِ آن، دیپلویِ دوم همه‌ی سوال‌ها را تکراری می‌سازد، و با
|
||
replace به‌جایش پاسخ‌های کاربران هم آبشاری پاک می‌شوند.
|
||
"""
|
||
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)
|
||
|
||
already = set()
|
||
if skip_existing:
|
||
already = set(
|
||
Question.objects.filter(lesson_id__in=lesson_ids)
|
||
.values_list('lesson_id', 'prompt')
|
||
)
|
||
|
||
created = skipped = 0
|
||
for row in rows:
|
||
if (row['lesson_id'], row['prompt']) in already:
|
||
skipped += 1
|
||
continue
|
||
question = Question.objects.create(
|
||
lesson_id=row['lesson_id'], type=row['type'], prompt=row['prompt'],
|
||
figure=row.get('figure'),
|
||
)
|
||
Choice.objects.bulk_create([
|
||
Choice(question=question, value=value, is_correct=(i == row['correct']))
|
||
for i, value in enumerate(row['choices'])
|
||
])
|
||
created += 1
|
||
if skip_existing:
|
||
# ردیفِ تکراری در خودِ فایل هم نباید دو بار ساخته شود.
|
||
already.add((row['lesson_id'], row['prompt']))
|
||
|
||
return {
|
||
'created': created, 'removed': removed,
|
||
'lessons': len(lesson_ids), 'skipped': skipped,
|
||
}
|
||
|
||
|
||
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)
|
||
row = {
|
||
'lessonId': question.lesson_id,
|
||
'prompt': question.prompt,
|
||
'type': question.type,
|
||
'choices': [c.value for c in choices],
|
||
'correct': correct,
|
||
}
|
||
if question.figure:
|
||
row['figure'] = question.figure
|
||
out.append(row)
|
||
return out
|
||
|
||
|
||
# ─── سوال‌ها بر پایه‌ی مسیرِ عنوان ───────────────────────────────────────────
|
||
#
|
||
# `lessonId` برای کارِ روزمره‌ی ادمین خوب است (کاربر درس را از همان صفحه
|
||
# برمی‌دارد و id جلوِ چشمش است)، ولی برای فایلی که در ریپو کامیت می‌شود و روی
|
||
# سرورِ دیگری لود خواهد شد اشتباهِ خطرناکی است: idها بینِ دو دیتابیس یکی
|
||
# نیستند. لوکالِ ما ۱۳۰۷ درس دارد با idهای ۳ تا ۲۰۰۵ (جای خالی، از پاک شدنِ
|
||
# دیتای دمو)، و همان کوریکولوم روی یک دیتابیسِ خالی idهای ۱ تا ۱۳۰۷ می‌گیرد.
|
||
# پس سوالِ درسِ ۳ روی سرور به درسِ بی‌ربطی می‌چسبد — بدونِ هیچ خطایی.
|
||
#
|
||
# مسیرِ عنوان (پایه › فصل › مبحث › درس) همان چیزی است که curriculum_io هم با
|
||
# آن هم‌سان‌سازی می‌کند، پس هر دو فایل به یک زبان حرف می‌زنند.
|
||
|
||
def export_questions_by_path(queryset) -> list[dict]:
|
||
"""همان قالبِ export_questions، ولی با مسیرِ عنوان به‌جای lessonId."""
|
||
out = []
|
||
for question in queryset.prefetch_related('choices').select_related(
|
||
'lesson__node__unit__grade'
|
||
):
|
||
choices = list(question.choices.all())
|
||
correct = next((i for i, c in enumerate(choices) if c.is_correct), 0)
|
||
lesson = question.lesson
|
||
node = lesson.node
|
||
row = {
|
||
'path': [node.unit.grade.title, node.unit.title, node.title, lesson.title],
|
||
'prompt': question.prompt,
|
||
'type': question.type,
|
||
'choices': [c.value for c in choices],
|
||
'correct': correct,
|
||
}
|
||
if question.figure:
|
||
row['figure'] = question.figure
|
||
out.append(row)
|
||
return out
|
||
|
||
|
||
def resolve_paths(rows: list[dict]) -> list[dict]:
|
||
"""
|
||
مسیرهای عنوان را به lesson_id تبدیل می‌کند.
|
||
|
||
اگر مسیری پیدا نشود خطا می‌دهد و هیچ ردیفی برنمی‌گرداند: مسیرِ گم‌شده یعنی
|
||
فایلِ سوال‌ها و فایلِ کوریکولوم با هم نمی‌خوانند، و وارد کردنِ بقیه یعنی
|
||
محتوایی ناقص که کسی متوجهش نمی‌شود.
|
||
"""
|
||
lessons = {
|
||
(lesson.node.unit.grade.title, lesson.node.unit.title,
|
||
lesson.node.title, lesson.title): lesson.pk
|
||
for lesson in Lesson.objects.select_related('node__unit__grade')
|
||
}
|
||
|
||
resolved, missing = [], []
|
||
for index, row in enumerate(rows):
|
||
path = row.get('path')
|
||
if not isinstance(path, list) or len(path) != 4:
|
||
raise ImportError_(
|
||
f'ردیف {index + 1}: «path» باید چهار عنوان باشد '
|
||
'(پایه، فصل، مبحث، درس).'
|
||
)
|
||
key = tuple(str(part).strip() for part in path)
|
||
lesson_id = lessons.get(key)
|
||
if lesson_id is None:
|
||
missing.append(' › '.join(key))
|
||
continue
|
||
resolved.append({**{k: v for k, v in row.items() if k != 'path'},
|
||
'lessonId': lesson_id})
|
||
|
||
if missing:
|
||
raise ImportError_(
|
||
'این درس‌ها در برنامه‌ی درسی پیدا نشدند (اول کوریکولوم را وارد کنید): '
|
||
+ '؛ '.join(missing[:5]) + ('…' if len(missing) > 5 else '')
|
||
)
|
||
return resolved
|