""" وارد کردنِ محتوای ریپو در دیپلوی، بدونِ برگرداندنِ کارِ ادمین (`seed_content`). دو منبع داریم: فایل‌های `content/seed/` که با هر دیپلوی می‌آیند، و پنلِ ادمین. تا ۲۰۲۶-۱۰-۰۳ سید هر چه در فایل بود و در دیتابیس پیدا نمی‌کرد دوباره می‌ساخت؛ با تستِ دورریختنی دیده شد که سوالِ پاک‌شده برمی‌گشت، سوالِ اصلاح‌شده نسخه‌ی غلطش را کنارش می‌گرفت، تغییرِ نامِ یک فصل ۳۳ درس و ۴۴ سوالِ تکراری می‌ساخت، و متنِ توضیحِ هر درس با متنِ فایل بازنویسی می‌شد. قاعده‌ی تازه، با دفترِ `SeedRecord`: - چیزی که در دیتابیس هست: ثبت می‌شود و دست نمی‌خورد (جز متنِ توضیح، پایین). - چیزی که نیست و قبلاً هم تحویل نشده: ساخته و ثبت می‌شود (محتوای تازه‌ی ریپو). - چیزی که نیست ولی قبلاً تحویل شده: ادمین پاکش کرده یا نامش را عوض کرده؛ ساخته نمی‌شود و فقط شمرده می‌شود (`retired`). زیرمجموعه‌اش هم ساخته نمی‌شود. - متنِ توضیحِ درس: اگر متنِ دیتابیس همان است که سید آخرین بار نوشت، متنِ تازه‌ی فایل جایش می‌نشیند؛ اگر ادمین عوضش کرده، متنِ ادمین می‌ماند. دیتابیسِ بی‌محتوا (سرورِ تازه، یا بازگرداندن پس از پاک شدنِ همه‌چیز) دفتر را نادیده می‌گیرد و همه‌چیز را از فایل می‌سازد. آن‌چه آزموده نشده: اولین دیپلو روی سرورِ فعلی دفتری خالی دارد؛ هر چه ادمین تا آن لحظه پاک کرده همان یک بارِ آخر دوباره ساخته می‌شود (رفتارِ قبلی)، و از آن به بعد نه. متنِ توضیحی که در آن لحظه با فایل فرق دارد، کارِ ادمین فرض می‌شود و هرگز از فایل بازنویسی نمی‌شود. """ import hashlib import json from django.db import transaction from . import question_io from .curriculum_io import _subject_or_error from .models import Grade, Lesson, Node, Question, SeedRecord, Unit def _key(kind: str, *parts) -> str: return hashlib.sha1('\x1f'.join((kind, *map(str, parts))).encode('utf-8')).hexdigest() def _text_hash(text: str) -> str: return hashlib.sha1((text or '').encode('utf-8')).hexdigest() class _Ledger: def __init__(self, active: bool): self.known = dict(SeedRecord.objects.values_list('key', 'value')) if active else {} self.write: dict[str, tuple[str, str]] = {} def delivered(self, key: str) -> bool: return key in self.known or key in self.write def value(self, key: str): if key in self.write: return self.write[key][1] return self.known.get(key) def record(self, key: str, kind: str, value: str = '') -> None: if self.known.get(key) != value or key not in self.known: self.write[key] = (kind, value) def save(self) -> None: # دسته‌دسته: SQLite بیش از ۳۲٬۷۶۶ متغیر در یک پرس‌وجو نمی‌پذیرد و دفترِ یک seedِ تازه از آن بیشتر است. keys = list(self.write) existing = set() for i in range(0, len(keys), 500): existing.update(SeedRecord.objects.filter(key__in=keys[i:i + 500]).values_list('key', flat=True)) SeedRecord.objects.bulk_create([ SeedRecord(key=k, kind=kind, value=value) for k, (kind, value) in self.write.items() if k not in existing ]) for k, (_kind, value) in self.write.items(): if k in existing: SeedRecord.objects.filter(key=k).update(value=value) @transaction.atomic def sync(grades: list[dict], question_rows: list[dict]) -> dict: """grades: خروجیِ curriculum_io.parse؛ question_rows: ردیف‌های مسیردارِ questions.json.""" ledger = _Ledger(active=Grade.objects.exists()) counts = {'grades': 0, 'units': 0, 'nodes': 0, 'lessons': 0, 'updated': 0, 'reordered': 0, 'kept_admin_text': 0, 'retired': 0, 'created': 0, 'skipped': 0} for g in grades: gpath = (g['title'],) grade = Grade.objects.filter(title=g['title']).first() if grade is None: if ledger.delivered(_key('grade', *gpath)): counts['retired'] += 1 continue grade = Grade.objects.create(title=g['title'], icon=g['icon'], color=g['color']) counts['grades'] += 1 ledger.record(_key('grade', *gpath), 'grade') for u in g['units']: upath = gpath + (u['title'],) unit = Unit.objects.filter(grade=grade, title=u['title']).first() if unit is None: if ledger.delivered(_key('unit', *upath)): counts['retired'] += 1 continue unit = Unit.objects.create( grade=grade, title=u['title'], order=u['order'], subject=_subject_or_error(u['subject']), ) counts['units'] += 1 ledger.record(_key('unit', *upath), 'unit') if unit.order != u['order']: # ترتیبِ فصل‌ها ترتیبِ کتاب است، نه سلیقه‌ی ادمین؛ دوباره از فایل گذاشته می‌شود. unit.order = u['order'] unit.save(update_fields=['order']) counts['reordered'] += 1 if u.get('blurb') and not unit.blurb: unit.blurb = u['blurb'] unit.save(update_fields=['blurb']) for n in u['nodes']: npath = upath + (n['title'],) node = Node.objects.filter(unit=unit, title=n['title']).first() if node is None: if ledger.delivered(_key('node', *npath)): counts['retired'] += 1 continue node = Node.objects.create(unit=unit, title=n['title'], order=n['order']) counts['nodes'] += 1 ledger.record(_key('node', *npath), 'node') if node.order != n['order']: node.order = n['order'] node.save(update_fields=['order']) counts['reordered'] += 1 for position, row in enumerate(n['lessons'], start=1): _sync_lesson(node, npath + (row['title'],), row, ledger, counts, position) _sync_questions(question_rows, ledger, counts) ledger.save() return counts def _sync_lesson(node, path, row, ledger, counts, position): intro_key = _key('intro', *path) lesson = Lesson.objects.filter(node=node, title=row['title']).first() if lesson is None: if ledger.delivered(_key('lesson', *path)): counts['retired'] += 1 return lesson = Lesson.objects.create( node=node, title=row['title'], intro_text=row['intro_text'], example=row['example'], steps=row['steps'], figures=row.get('figures') or [], order=position, ) counts['lessons'] += 1 ledger.record(_key('lesson', *path), 'lesson') ledger.record(intro_key, 'intro', _text_hash(lesson.intro_text)) return ledger.record(_key('lesson', *path), 'lesson') changed = [] if lesson.order != position: lesson.order = position changed.append('order') counts['reordered'] += 1 file_text, db_text = row['intro_text'], lesson.intro_text last_seeded = ledger.value(intro_key) if file_text and db_text != file_text: untouched = (last_seeded is not None and _text_hash(db_text) == last_seeded) or not db_text if untouched: lesson.intro_text = file_text changed.append('intro_text') ledger.record(intro_key, 'intro', _text_hash(file_text)) else: counts['kept_admin_text'] += 1 elif db_text == file_text: ledger.record(intro_key, 'intro', _text_hash(db_text)) # مثال و قدم‌ها فقط در خانه‌ی خالی (همان قاعده‌ی curriculum_io). if row.get('example') and not lesson.example: lesson.example = row['example'] changed.append('example') if row.get('steps') and not lesson.steps: lesson.steps = row['steps'] changed.append('steps') if row.get('figures') and not lesson.figures: lesson.figures = row['figures'] changed.append('figures') if changed: lesson.save(update_fields=changed) counts['updated'] += 1 def _sync_questions(rows, ledger, counts): if not rows: return lessons = { (l.node.unit.grade.title, l.node.unit.title, l.node.title, l.title): l.pk for l in Lesson.objects.select_related('node__unit__grade') } existing = set( Question.objects.filter(lesson_id__in=lessons.values()).values_list('lesson_id', 'prompt') ) to_create, missing = [], [] for index, row in enumerate(rows): path = row.get('path') if not isinstance(path, list) or len(path) != 4: raise question_io.ImportError_( f'ردیف {index + 1}: «path» باید چهار عنوان باشد (پایه، فصل، مبحث، درس).') path = tuple(str(p).strip() for p in path) prompt = (row.get('prompt') or '').strip() qkey = _key('question', *path, prompt) lesson_id = lessons.get(path) if lesson_id is None: # درسی که سید قبلاً داده و حالا نیست (ادمین پاک یا نامش را عوض کرده): سوالش هم نمی‌آید. if ledger.delivered(_key('lesson', *path)) or ledger.delivered(qkey): counts['retired'] += 1 continue missing.append(' › '.join(path)) continue if (lesson_id, prompt) in existing: ledger.record(qkey, 'question') counts['skipped'] += 1 continue if ledger.delivered(qkey): counts['retired'] += 1 continue to_create.append(({**{k: v for k, v in row.items() if k != 'path'}, 'lessonId': lesson_id}, qkey)) existing.add((lesson_id, prompt)) _refresh_practice(rows, lessons, ledger, counts) if missing: raise question_io.ImportError_( 'این درس‌ها در برنامه‌ی درسی پیدا نشدند (اول کوریکولوم را وارد کنید): ' + '؛ '.join(missing[:5]) + ('…' if len(missing) > 5 else '') ) if not to_create: return parsed = question_io.parse(json.dumps([r for r, _ in to_create]).encode('utf-8')) result = question_io.import_questions(parsed) counts['created'] += result['created'] for _row, qkey in to_create: ledger.record(qkey, 'question') PRACTICE_PREFIX = 'تمرین: ' def _refresh_practice(rows, lessons, ledger, counts): """ جزیره‌های «تمرین: …» را فایل می‌سازد و ادمین سوال‌هایش را کم‌وبیش دست نمی‌زند؛ پس اصلاحِ فایل (گزینه‌ی هم‌ارزِ جواب، جوابِ غلط) باید به سوالِ موجود برسد، وگرنه سید فقط «سوالِ تازه» می‌سازد و نسخه‌ی معیوب می‌ماند: - همان صورت، گزینه‌ها/پاسخِ درست فرق دارد → گزینه‌ها جا به جا (همان idها) به‌روز می‌شود تا پاسخ‌های ثبت‌شده نشکنند؛ - سوالی که خودِ سید قبلاً تحویل داده (دفتر) و حالا در فایل نیست و کسی جوابش نداده → پاک می‌شود. سوالِ دستیِ ادمین (در دفتر نیست) هرگز دست نمی‌خورد. آزموده: content/tests_seed_content.py. """ practice_lessons = { lid for path, lid in lessons.items() if path[2].startswith(PRACTICE_PREFIX) } if not practice_lessons: return want = {} file_keys = set() for row in rows: path = tuple(str(p).strip() for p in row.get('path', [])) if len(path) != 4 or not path[2].startswith(PRACTICE_PREFIX): continue lid = lessons.get(path) prompt = (row.get('prompt') or '').strip() if lid is None: continue want[(lid, prompt)] = row file_keys.add(_key('question', *path, prompt)) by_id = {} for lid, path in ((v, k) for k, v in lessons.items() if v in practice_lessons): by_id[lid] = path qs = Question.objects.filter(lesson_id__in=practice_lessons).prefetch_related('choices') stale = [] for q in qs.iterator(chunk_size=500): row = want.get((q.lesson_id, q.prompt)) if row is None: if _key('question', *by_id[q.lesson_id], q.prompt) in ledger.known: stale.append(q.pk) continue new_choices = [str(c).strip() for c in row['choices']] old = sorted(q.choices.all(), key=lambda c: c.pk) if len(old) != len(new_choices): continue for index, choice in enumerate(old): value, correct = new_choices[index], index == row['correct'] if choice.value != value or choice.is_correct != correct: choice.value, choice.is_correct = value, correct choice.save(update_fields=['value', 'is_correct']) counts['updated'] += 1 if stale: removable = list(Question.objects.filter(pk__in=stale, answers__isnull=True).values_list('pk', flat=True)) if removable: Question.objects.filter(pk__in=removable).delete() counts['retired'] += len(removable)