MathApp/content/seed_sync.py

"""
وارد کردنِ محتوای ریپو در دیپلوی، بدونِ برگرداندنِ کارِ ادمین (`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'],
        )
        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 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')