""" وارد کردنِ محتوای ریپو در دیپلوی، بدونِ برگرداندنِ کارِ ادمین (`seed_content`). دو منبع داریم: فایل‌های `content/seed/` که با هر دیپلوی می‌آیند، و پنلِ ادمین. تا ۲۰۲۶-۱۰-۰۳ سید هر چه در فایل بود و در دیتابیس پیدا نمی‌کرد دوباره می‌ساخت؛ با تستِ دورریختنی دیده شد که سوالِ پاک‌شده برمی‌گشت، سوالِ اصلاح‌شده نسخه‌ی غلطش را کنارش می‌گرفت، تغییرِ نامِ یک فصل ۳۳ درس و ۴۴ سوالِ تکراری می‌ساخت، و متنِ توضیحِ هر درس با متنِ فایل بازنویسی می‌شد. قاعده‌ی تازه، با دفترِ `SeedRecord`: - چیزی که در دیتابیس هست: ثبت می‌شود و دست نمی‌خورد (جز متنِ توضیح، پایین). - چیزی که نیست و قبلاً هم تحویل نشده: ساخته و ثبت می‌شود (محتوای تازه‌ی ریپو). - چیزی که نیست ولی قبلاً تحویل شده: ادمین پاکش کرده یا نامش را عوض کرده؛ ساخته نمی‌شود و فقط شمرده می‌شود (`retired`). زیرمجموعه‌اش هم ساخته نمی‌شود. - متنِ توضیحِ درس: اگر متنِ دیتابیس همان است که سید آخرین بار نوشت، متنِ تازه‌ی فایل جایش می‌نشیند؛ اگر ادمین عوضش کرده، متنِ ادمین می‌ماند. دیتابیسِ بی‌محتوا (سرورِ تازه، یا بازگرداندن پس از پاک شدنِ همه‌چیز) دفتر را نادیده می‌گیرد و همه‌چیز را از فایل می‌سازد. آن‌چه آزموده نشده: اولین دیپلو روی سرورِ فعلی دفتری خالی دارد؛ هر چه ادمین تا آن لحظه پاک کرده همان یک بارِ آخر دوباره ساخته می‌شود (رفتارِ قبلی)، و از آن به بعد نه. متنِ توضیحی که در آن لحظه با فایل فرق دارد، کارِ ادمین فرض می‌شود و هرگز از فایل بازنویسی نمی‌شود. """ import hashlib import json from django.db import transaction from . import question_io 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: existing = set(SeedRecord.objects.filter(key__in=list(self.write)).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, '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']) counts['units'] += 1 ledger.record(_key('unit', *upath), 'unit') 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') for row in n['lessons']: _sync_lesson(node, npath + (row['title'],), row, ledger, counts) _sync_questions(question_rows, ledger, counts) ledger.save() return counts def _sync_lesson(node, path, row, ledger, counts): 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 [], ) 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 = [] 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)) 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')