MathApp/rag/management/commands/rag_verify_solutions.py

"""
راستی‌آزماییِ راه‌حل‌هایی که vision از کتابِ درسی درآورده.

چرا لازم است: استخراج گاهی راه‌حلِ غلط می‌سازد و این خطا در بازیابی دیده
نمی‌شود — فقط وقتی کسی به `solution` اعتماد کند (ساختِ سؤال، یا خودِ دستیار
هنگامِ نقلِ جواب) خودش را نشان می‌دهد. نمونه‌های واقعی از همین کتاب‌ها:

    BMI = ۱۰۰/(۲٫۷۴)²          کتاب: ۳۳٫۲      درست: ۱۳٫۳
    ۱۶ آبی + ۲۴ صورتی          کتاب: «کل ۵۰»   درست: ۴۰
    بزرگ‌ترین عددِ ۴رقمیِ بی‌تکرار کتاب: ۸۵۳۲     درست: ۹۸۷۶
    g(x) = x³ − ۲              کتاب: «نمایی»   درست: چندجمله‌ایِ درجه ۳

`solution` هرگز بازنویسی نمی‌شود. کتاب باید کتاب بماند: اگر جوابِ مدل را
جایش می‌گذاشتیم و مدل اشتباه می‌کرد، منبعِ حقیقت بی‌صدا خراب می‌شد.

──────────────────────────────────────────────────────────────────────────
درس‌هایی که از اجرای اولِ این دستور گرفته شد و در طراحیِ فعلی نشسته‌اند:

۱. نتیجه باید سه‌حالته باشد. با بله/خیر، «نمی‌دانم» ناچار به «غلط است»
   ترجمه می‌شد: ۳۵٪ از گزارش‌ها مسئله‌هایی بودند که به شکل یا جدولِ کتاب
   ارجاع می‌دادند و اصلاً قابلِ قضاوت نبودند.

۲. «۳/۱۰۴» در این کتاب‌ها عددِ اعشاریِ ۳٫۱۰۴ است، نه کسرِ ۳÷۱۰۴. بدونِ گفتنِ
   این به مدل، ۲۴٪ از گزارش‌ها هشدارِ کاذب بود — بیشترشان در پایه ششم که
   پر از اعشار است.

۳. مسئله‌های وابسته به شکل اصلاً فرستاده نمی‌شوند. هزینه دارند و جوابشان
   بی‌معناست؛ همان‌جا برچسبِ unjudgeable می‌خورند.

۴. دو مدل، دو نقش. deepseek-reasoner ارزان است و تشخیصش خوب، ولی جوابِ
   خودش گاهی غلط است (مسئله‌ی سرعتِ نور را ۱۰^۱۷ داد به‌جای ۱۰^۱۴). پس
   او تور می‌اندازد و o4-mini فقط روی صیدها داوری می‌کند.
──────────────────────────────────────────────────────────────────────────

    python manage.py rag_verify_solutions --all
    python manage.py rag_verify_solutions --all --adjudicate
    python manage.py rag_verify_solutions --collection "ریاضی ششم" --limit 20
    python manage.py rag_verify_solutions --report
"""
import json
import re
import threading
import time
from concurrent.futures import ThreadPoolExecutor

from django.core.management.base import BaseCommand, CommandError
from django.db.models import Count, Q
from django.utils import timezone

from rag.models import LessonChunk
from rag.services import clients
from rag.services.config import rag_is_configured

SCOUT_MODEL = 'deepseek-reasoner'
JUDGE_MODEL = 'o4-mini'

# دو خطای کاملاً متفاوت که هر دو ۴۲۹‌اند و نباید یکسان با آن‌ها رفتار کرد:
#
#   • فشارِ لحظه‌ایِ موازی‌کاری — گذراست، عقب‌نشینی حلش می‌کند.
#   • سقفِ مصرفِ حساب — ساعت‌ها باز نمی‌شود. تلاشِ دوباره نه‌تنها بی‌فایده
#     است، پنجره‌ی محدودیت را هم تمدید می‌کند: یک اجرا ۱۱۹ بار دوباره تلاش
#     کرد، ۶۵ دقیقه طول کشید، هیچ کاری نکرد، و مهلت را از ۱۴ ساعت جلوتر برد.
class AccountLimitReached(Exception):
    """حسابِ سرویس جواب نمی‌دهد؛ ادامه دادن بی‌معناست."""


# خودِ تشخیص در rag.services.clients است چون دستیارِ زنده هم لازمش دارد
# (/api/health باید بگوید «اعتبار تمام شده»، نه فقط «کلید تنظیم شده»).
from rag.services.clients import is_account_limit  # noqa: E402,F401


# مسئله‌ای که به شکلِ کتاب ارجاع می‌دهد و محتوایش را ننوشته، بدونِ تصویر
# قابلِ حل نیست؛ فرستادنش هم پول می‌سوزاند هم آمار را خراب می‌کند.
#
# ⚠️ این محافظ باید دو شرطی بماند — ارجاعِ صریح به بیرونِ متن، *و* نیامدنِ
# توصیف بعد از آن. نسخه‌ی اولش یک regexِ تخت روی «شکل|جدول|روبه‌رو|…» بود و
# «زاویه‌های روبه‌رو در لوزی» یا «کسرهای زیر را گویا کنید: ۲/√۳» را هم می‌گرفت:
# در نمونه‌ی ۵۰تایی ۸۴٪ از ردشده‌ها واقعاً قابلِ قضاوت بودند، یعنی ~۷۸۴ مسئله‌ی
# سالم بی‌صدا از چرخه‌ی ساختِ سوال بیرون مانده بود.
_FIGURE_REFERENCE = re.compile(
    r'(شکل|جدول|نمودار|تصویر|نقشه)\s*(ِ|های|‌های)?\s*'
    r'(زیر|بالا|روبه\u200cرو|روبرو|مقابل|فوق|داده\s*شده)'
    r'|با توجه به\s*(شکل|جدول|نمودار|تصویر)'
)
# نشانه‌ی اینکه محتوای شکل داخلِ متن آمده.
_FIGURE_DESCRIBED = re.compile(r'که\s|شامل|اعضای|مقادیر|برابر|=|\{|\d')
# کارهایی که ذاتاً تصویری‌اند و هیچ متنی نجاتشان نمی‌دهد.
_ART_TASK = re.compile(r'رنگ[‌ ]?آمیزی|رنگ کنید|رنگ\u200cآمیزی')


def needs_figure(problem: str) -> bool:
    """آیا این مسئله بدونِ دیدنِ صفحه‌ی کتاب واقعاً قابلِ قضاوت نیست؟"""
    text = problem or ''
    if _ART_TASK.search(text):
        return True
    match = _FIGURE_REFERENCE.search(text)
    if not match:
        return False
    # ۱۶۰ نویسه بعد از ارجاع: جایی که توصیفِ شکل اگر بیاید، می‌آید.
    return not _FIGURE_DESCRIBED.search(text[match.end():match.end() + 160])


PROMPT = """تو یک ریاضی‌دانِ دقیق هستی که راه‌حل‌های کتابِ درسیِ ایران را بازبینی می‌کند.

⚠️ سه قاعده‌ی مهم درباره‌ی نگارشِ این متن‌ها:

۱. علامتِ «/» بینِ دو عدد، در این کتاب‌ها تقریباً همیشه **جداکننده‌ی اعشار**
   است، نه خطِ کسر. یعنی «۳/۱۰۴» عددِ ۳٫۱۰۴ است و «۲/۵» عددِ ۲٫۵.
   کسرها معمولاً به شکلِ «۳ بر ۴» یا با کسرِ عمودی می‌آیند.
۲. صفر در ریاضیِ مدرسه‌ی ایران عددِ طبیعی **نیست**.
۳. متن با OCR از PDF درآمده؛ فاصله و نیم‌فاصله ممکن است به‌هم‌ریخته باشد.
   تفاوتِ صرفاً نگارشی، گِردکردن، یا ترتیبِ نوشتن **خطا نیست**.

مسئله:
{problem}

راه‌حلِ کتاب:
{solution}

کارَت:
۱. اول خودت مسئله را قدم‌به‌قدم حل کن.
۲. بعد راه‌حلِ کتاب را با جوابِ خودت بسنج.

یکی از این سه را برگردان:
• "agrees"      — راه‌حلِ کتاب از نظرِ ریاضی درست است (حتی اگر نگارشش فرق دارد)
• "disagrees"   — خطای قطعیِ ریاضی دارد. فقط وقتی مطمئنی.
• "unjudgeable" — بدونِ شکل/جدولِ کتاب یا داده‌ی جاافتاده نمی‌شود قضاوت کرد،
                  یا مسئله جوابِ عددیِ مشخصی ندارد (پرسشِ باز یا توضیحی).

اگر شک داری بینِ disagrees و unjudgeable، unjudgeable را انتخاب کن.

فقط این JSON را برگردان، بدونِ هیچ متنِ اضافه:
{{"verdict": "agrees|disagrees|unjudgeable", "answer": "جوابِ خودت، کوتاه", "why": "حداکثر ۱۲۰ نویسه"}}"""


def first_json(text: str):
    """
    اولین شیءِ JSON را از متن بیرون می‌کشد.

    مدل‌های استدلالی گاهی قبل یا بعد از JSON توضیح می‌چسبانند و json.loads خام
    روی همان می‌افتد. در محکِ اولیه همین باعث شد deepseek به‌اشتباه ۷ از ۸
    به‌نظر برسد در حالی که ۸ از ۸ بود.
    """
    depth = 0
    start = None
    for i, ch in enumerate(text or ''):
        if ch == '{':
            if depth == 0:
                start = i
            depth += 1
        elif ch == '}':
            depth -= 1
            if depth == 0 and start is not None:
                try:
                    return json.loads(text[start:i + 1])
                except json.JSONDecodeError:
                    start = None
    return None


class Command(BaseCommand):
    help = 'بازبینیِ درستیِ راه‌حل‌های استخراج‌شده از کتاب (هزینه دارد، کم).'

    def add_arguments(self, parser):
        parser.add_argument('--collection', default=None)
        parser.add_argument('--all', action='store_true')
        parser.add_argument('--limit', type=int, default=None)
        parser.add_argument('--workers', type=int, default=12)
        parser.add_argument('--adjudicate', action='store_true',
                            help='پاسِ دوم: مشکوک‌ها را با مدلِ دقیق‌تر بسنج.')
        parser.add_argument('--recheck', action='store_true')
        parser.add_argument('--report', action='store_true',
                            help='فقط گزارشِ وضعیت؛ هیچ فراخوانی‌ای نمی‌کند.')
        parser.add_argument('--sync-qdrant', action='store_true',
                            help='برچسب‌ها را در payloadِ Qdrant بنویس (رایگان).')

    def handle(self, *args, **opts):
        if opts['report']:
            return self._report()
        if opts['sync_qdrant']:
            return self._sync_qdrant(opts)
        if not rag_is_configured():
            raise CommandError('RAG پیکربندی نشده (کلیدها خالی‌اند).')
        if opts['adjudicate']:
            return self._adjudicate(opts)

        qs = LessonChunk.objects.exclude(problem='').exclude(solution='')
        if opts['collection']:
            qs = qs.filter(collection=opts['collection'])
        elif not opts['all']:
            raise CommandError('یا --collection بدهید یا --all.')
        if not opts['recheck']:
            qs = qs.filter(solution_checked_at__isnull=True)
        rows = list(qs.order_by('collection', 'source_page', 'order'))

        skipped = self._mark_figure_dependent(rows)
        rows = [r for r in rows if not needs_figure(r.problem or '')]
        if opts['limit']:
            rows = rows[:opts['limit']]
        if skipped:
            self.stdout.write(
                f'  {skipped} مسئله‌ی وابسته به شکل/جدول بدونِ فراخوانی '
                'برچسبِ unjudgeable گرفتند.'
            )
        if not rows:
            self.stdout.write(self.style.WARNING('چیزی برای بررسی نماند.'))
            return

        self.stdout.write(self.style.MIGRATE_HEADING(
            f'\n{len(rows)} chunk با {SCOUT_MODEL}  (~${len(rows) * 0.00014:.2f})\n'))
        self._run(rows, SCOUT_MODEL, opts['workers'])

    def _mark_figure_dependent(self, rows) -> int:
        """قبل از خرجِ یک ریال، آن‌هایی که ذاتاً قابلِ قضاوت نیستند کنار می‌روند."""
        marked = []
        for chunk in rows:
            if needs_figure(chunk.problem or ''):
                chunk.solution_verdict = LessonChunk.VERDICT_UNJUDGEABLE
                chunk.check_note = 'به شکل/جدولِ کتاب ارجاع می‌دهد؛ از متن قابلِ قضاوت نیست.'
                chunk.checked_by = 'pre-filter'
                chunk.solution_checked_at = timezone.now()
                marked.append(chunk)
        if marked:
            LessonChunk.objects.bulk_update(
                marked, ['solution_verdict', 'check_note', 'checked_by',
                         'solution_checked_at'])
        return len(marked)

    def _adjudicate(self, opts):
        qs = LessonChunk.objects.filter(
            solution_verdict=LessonChunk.VERDICT_DISAGREES).exclude(problem='')
        if opts['collection']:
            qs = qs.filter(collection=opts['collection'])
        if not opts['recheck']:
            qs = qs.exclude(checked_by=JUDGE_MODEL)
        rows = list(qs.order_by('collection', 'source_page')[:opts['limit'] or 10_000])
        if not rows:
            self.stdout.write(self.style.WARNING('مشکوکی برای داوری نمانده.'))
            return
        # مدلِ داور روی این پروکسی سخت‌گیرتر نرخ‌بندی می‌شود؛ نصفِ کارگرها.
        workers = max(1, min(opts['workers'], 4))
        self.stdout.write(self.style.MIGRATE_HEADING(
            f'\nداوریِ {len(rows)} مورد با {JUDGE_MODEL} ({workers} کارگر)  '
            f'(~${len(rows) * 0.0018:.2f})\n'))
        self._run(rows, JUDGE_MODEL, workers)

    def _run(self, rows, model, workers):
        """
        فراخوانی‌ها موازی، نوشتن دسته‌ای از نخِ اصلی.

        هر chunk چند ثانیه منتظرِ شبکه است، پس ترتیبی یعنی ساعت‌ها؛ ولی نوشتنِ
        هم‌زمان از چند نخ روی SQLite به قفلِ نوشتن می‌خورد.
        """
        client = clients.get_openai_client()

        stop = threading.Event()

        def ask(chunk):
            # اگر یک نخ به سقفِ حساب خورده، بقیه بی‌درنگ برمی‌گردند؛ ادامه دادن
            # فقط پنجره‌ی محدودیت را طولانی‌تر می‌کند.
            if stop.is_set():
                return chunk, None, None
            delay = 4.0
            last = None
            for _ in range(5):
                try:
                    completion = client.chat.completions.create(
                        model=model,
                        messages=[{'role': 'user', 'content': PROMPT.format(
                            problem=chunk.problem[:900], solution=chunk.solution[:900])}],
                    )
                    data = first_json(completion.choices[0].message.content)
                    verdict = (data or {}).get('verdict')
                    if verdict not in dict(LessonChunk.VERDICT_CHOICES):
                        raise ValueError(f'verdictِ نامعتبر: {verdict!r}')
                    return chunk, data, None
                except Exception as exc:  # noqa: BLE001
                    last = exc
                    text = str(exc).lower()
                    if is_account_limit(exc):
                        stop.set()
                        return chunk, None, AccountLimitReached(str(exc)[:200])
                    if '429' not in text and 'rate' not in text:
                        break
                    if stop.is_set():
                        break
                    time.sleep(delay)
                    delay *= 2
            return chunk, None, last

        counts = {v: 0 for v, _ in LessonChunk.VERDICT_CHOICES}
        failed = 0
        aborted: AccountLimitReached | None = None
        batch: list[LessonChunk] = []
        t0 = time.time()
        done = 0
        with ThreadPoolExecutor(max_workers=workers) as pool:
            for chunk, data, err in pool.map(ask, rows):
                done += 1
                if isinstance(err, AccountLimitReached):
                    aborted = err
                    continue
                if err is None and data is None:
                    continue          # بعد از توقف، نوبتش نرسید
                if err is not None:
                    failed += 1
                    self.stdout.write(self.style.ERROR(f'  ✗ #{chunk.pk}: {str(err)[:75]}'))
                else:
                    chunk.solution_verdict = data['verdict']
                    chunk.checked_answer = str(data.get('answer') or '')[:2000]
                    chunk.check_note = str(data.get('why') or '')[:300]
                    chunk.checked_by = model
                    chunk.solution_checked_at = timezone.now()
                    batch.append(chunk)
                    counts[data['verdict']] += 1
                    if data['verdict'] == LessonChunk.VERDICT_DISAGREES:
                        self.stdout.write(
                            f'  ⚠ {chunk.collection} ص{chunk.source_page}: '
                            f'{chunk.problem[:50]} → {chunk.checked_answer[:40]}')
                if len(batch) >= 40:
                    self._flush(batch)
                if done % 100 == 0:
                    rate = done / max(time.time() - t0, 1)
                    self.stdout.write(
                        f'  … {done}/{len(rows)}  (~{(len(rows)-done)/max(rate,.001)/60:.0f} دقیقه مانده)')
        self._flush(batch)
        self.stdout.write(self.style.SUCCESS(
            f"\n{counts['agrees']} درست، {counts['disagrees']} مشکوک، "
            f"{counts['unjudgeable']} غیرقابلِ قضاوت، {failed} خطا — {time.time()-t0:.0f}s"))
        if aborted is not None:
            # آنچه تا اینجا سنجیده شده ذخیره شده و اجرای بعدی از همین‌جا ادامه
            # می‌دهد، چون فقط بررسی‌نشده‌ها برداشته می‌شوند.
            raise CommandError(
                'سقفِ مصرفِ حسابِ سرویس پر شده و اجرا همین‌جا متوقف شد.\n'
                f'  پیامِ سرویس: {aborted}\n'
                '  تا باز شدنِ سقف دوباره تلاش نکنید؛ هر تلاش مهلت را جلوتر می‌برد.\n'
                '  وضعیت با: python manage.py rag_verify_solutions --report'
            )

    @staticmethod
    def _flush(batch):
        if batch:
            LessonChunk.objects.bulk_update(
                batch, ['solution_verdict', 'checked_answer', 'check_note',
                        'checked_by', 'solution_checked_at'])
            batch.clear()

    def _sync_qdrant(self, opts):
        """
        برچسب‌ها را روی نقاطِ موجود می‌نویسد، بدونِ بردار و بدونِ هزینه.

        نقطه‌ها بر اساسِ مقدارِ برچسب دسته می‌شوند، پس برای هر collection سه
        فراخوانی کافی است نه هزاران‌تا.
        """
        from rag.services.qdrant_store import set_payload_fields

        qs = LessonChunk.objects.exclude(solution_verdict='')
        if opts['collection']:
            qs = qs.filter(collection=opts['collection'])

        buckets: dict[tuple[str, str], list[str]] = {}
        for collection, verdict, point_id in qs.values_list(
                'collection', 'solution_verdict', 'qdrant_point_id'):
            buckets.setdefault((collection, verdict), []).append(str(point_id))

        if not buckets:
            self.stdout.write(self.style.WARNING('برچسبی برای همگام‌سازی نیست.'))
            return

        total = 0
        for (collection, verdict), ids in sorted(buckets.items()):
            set_payload_fields(collection, ids, {'solution_verdict': verdict})
            total += len(ids)
            self.stdout.write(f'  {collection:14s} {verdict:12s} {len(ids):4d} نقطه')
        self.stdout.write(self.style.SUCCESS(f'\n{total} نقطه در Qdrant به‌روز شد.'))

    def _report(self):
        rows = (LessonChunk.objects.exclude(problem='').exclude(solution='')
                .values('collection')
                .annotate(n=Count('id'),
                          ok=Count('id', filter=Q(solution_verdict='agrees')),
                          bad=Count('id', filter=Q(solution_verdict='disagrees')),
                          unk=Count('id', filter=Q(solution_verdict='unjudgeable')),
                          todo=Count('id', filter=Q(solution_checked_at__isnull=True)))
                .order_by('collection'))
        self.stdout.write(f"\n{'کتاب':14s} {'کل':>5s} {'درست':>6s} {'مشکوک':>7s} "
                          f"{'بی‌قضاوت':>9s} {'مانده':>6s}")
        for r in rows:
            self.stdout.write(f"{r['collection']:14s} {r['n']:5d} {r['ok']:6d} "
                              f"{r['bad']:7d} {r['unk']:9d} {r['todo']:6d}")