import json
import logging
import time

from django.conf import settings
from django.db import IntegrityError, transaction
from django.http import Http404, StreamingHttpResponse
from django.shortcuts import get_object_or_404
from django.utils import timezone
from drf_spectacular.types import OpenApiTypes
from drf_spectacular.utils import OpenApiParameter, extend_schema
from rest_framework.exceptions import APIException, Throttled
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from rest_framework.throttling import ScopedRateThrottle
from rest_framework.views import APIView

from accounts.models import User as AccountUser
from . import provider_status
from .models import AskIdempotency, Conversation, Message, MessageFeedback
from .serializers import (
    QuotaResponseSerializer,
    AskRequestSerializer,
    AskResponseSerializer,
    ConversationMessagesResponseSerializer,
    FeedbackRequestSerializer,
    FeedbackResponseSerializer,
)

logger = logging.getLogger(__name__)

def _learner_age(user) -> int | None:
    """سنِ کاربر از پروفایلش (accounts)؛ لحنِ دستیار برای ۱۸+ بزرگسالانه است."""
    from accounts.jalali import age_of
    from accounts.models import Profile

    profile = Profile.objects.filter(user_id=user.id).only('birth_date', 'birth_year').first()
    return age_of(profile.birth_date, profile.birth_year) if profile else None


def _cache_grade(grade_title: str, age: int | None) -> str:
    """پایه در کلیدِ کشِ جواب، با لحن: جوابِ بزرگسالانه به کودک سرو نشود و برعکس."""
    from rag.services.generation import tone_for

    return f'{grade_title}|adult' if tone_for(grade_title, age) == 'adult' else grade_title


def quota_state(user_id: int) -> dict:
    """
    {'remainingQuota', 'credits'}: سوال‌های رایگانِ باقیِ امروز و اعتبارِ بسته، هر بار تازه از
    دیتابیس. همه‌ی پاسخ‌ها از همین می‌خوانند تا شمارنده‌ی اپ هیچ‌وقت از موجودیِ واقعی جدا نیفتد.
    """
    from subscription.services import credit_balance

    used_free = Message.objects.filter(
        conversation__user_id=user_id, role='user', created_at__date=timezone.localdate(), charge='free',
    ).count()
    return {'remainingQuota': max(daily_quota(user_id) - used_free, 0), 'credits': credit_balance(user_id)}


def _uncharge(user_message) -> None:
    """سوالی که به API نرسید (جوابِ کش‌شده): نه از سهمیه کم شود نه از اعتبار."""
    from subscription.services import refund_credit

    if user_message.charge == 'credit':
        refund_credit(user_message.id)
    if user_message.charge in ('free', 'credit'):
        user_message.charge = 'none'
        user_message.save(update_fields=['charge'])


def daily_quota(user_id: int) -> int:
    """سقفِ سوالِ روزانه‌ی این کاربر؛ رایگان یا از پلنِ اشتراکش (subscription.services)."""
    from subscription.services import ai_daily_quota

    return ai_daily_quota(user_id)


def _annotation(retrieval_meta: dict, generation_meta: dict, *, from_cache: bool) -> dict:
    """
    متادیتای بازیابی/تولید → فیلدهای Message.

    همه‌ی مقادیر به str/int امن تبدیل می‌شوند و هر کلیدِ غایب مقدارِ پیش‌فرض
    می‌گیرد: این متادیتا از cache هم می‌آید و cacheهای قدیمی (که پیش از افزودنِ
    این فیلدها ست شده‌اند) کلیدها را ندارند — نبودشان نباید ثبتِ پیام را
    بشکند، چون آن‌وقت خودِ پاسخِ کاربر هم از بین می‌رود.
    """
    retrieval_meta = retrieval_meta or {}
    generation_meta = generation_meta or {}
    filter_value = retrieval_meta.get('filterValue')
    return {
        'model_name': str(generation_meta.get('model') or '')[:100],
        'prompt_version': str(generation_meta.get('promptVersion') or '')[:50],
        'grade_title': str(retrieval_meta.get('gradeTitle') or '')[:50],
        'retrieval_collection': str(retrieval_meta.get('collection') or '')[:200],
        'retrieval_filter_value': ('' if filter_value is None else str(filter_value))[:255],
        'chunk_count': int(retrieval_meta.get('chunkCount') or 0),
        'prompt_tokens': int(generation_meta.get('promptTokens') or 0),
        'completion_tokens': int(generation_meta.get('completionTokens') or 0),
        'latency_ms': int(generation_meta.get('latencyMs') or 0),
        'from_cache': bool(from_cache),
        # شکلِ اعتبارسنجی‌شده در rag.services.diagram؛ کشِ قدیمی آن را ندارد.
        'diagram': generation_meta.get('diagram') if isinstance(generation_meta.get('diagram'), dict) else None,
    }


def _is_db_ref_id(value) -> bool:
    """
    ⚠️ افزوده شده: lessonId/chapterId/questionId در قراردادِ API به‌صورتِ
    رشته (CharField) پذیرفته می‌شن، اما وقتی به‌عنوانِ ارجاعِ سبکِ (لایه‌ی
    DB) روی Conversation/Message ذخیره می‌شن، فیلدِ مقصد IntegerField‌ه.
    این تابع فقط تشخیص می‌ده که آیا مقدار برای ذخیره‌سازی در همچین
    فیلدی امن/عددی هست یا نه؛ روی مرزِ generate_tutor_answer اثری نداره
    (اونجا همیشه مقدارِ خامِ اصلی پاس داده می‌شه).

    پیامِ دستیار هم از همین مسیر (`_db_ref_or_none`) رد می‌شود؛ پیش از
    ۲۰۲۶-۰۹-۲۵ نمی‌شد و questionIdِ غیرعددی *بعد از* تولیدِ جواب ۵۰۰ می‌داد و
    سهمِ روزانه مصرف‌شده می‌ماند.
    """
    if value in (None, ''):
        return False
    # isascii: '²'.isdigit() هم True است ولی int('²') خطا می‌دهد.
    text = str(value).strip().lstrip('-')
    return text.isascii() and text.isdigit()


def _db_ref_or_none(value):
    """مقدارِ عددی برای ستونِ IntegerField، وگرنه None (نه خطای ORM)."""
    return value if _is_db_ref_id(value) else None


def _conversation_history(conversation, before_message) -> list[dict]:
    """
    چند پیامِ آخرِ همین گفتگو، پیش از سوالِ تازه، به ترتیبِ زمان.

    بدونِ این، هر سوال جدا از قبلی‌ها جواب داده می‌شد: «نفهمیدم، ساده‌تر بگو»
    برای مدل هیچ مرجعی نداشت.
    """
    from rag.services.generation import HISTORY_TURNS

    rows = list(
        Message.objects.filter(conversation=conversation, id__lt=before_message.id)
        .order_by('-id').values('role', 'text')[:HISTORY_TURNS]
    )
    return list(reversed(rows))


def _retrieval_query(question: str, history: list[dict], focus: str = '') -> str:
    """
    متنی که embedding می‌شود. «یه مثال دیگه بزن» به‌تنهایی به هیچ تکه‌ای از
    کتاب نزدیک نیست؛ سوالِ قبلیِ کاربر (یا سوالِ تمرینیِ اپ) موضوع را معلوم می‌کند.
    """
    if focus:
        return f'{focus}\n{question}'
    previous = next((t['text'] for t in reversed(history) if t['role'] == 'user'), '')
    return f'{previous}\n{question}' if previous else question


def _question_focus(v: dict, user) -> str:
    """اگر اپ questionId فرستاده، متنِ همان سوال و جوابِ دانش‌آموز (rag.generation)."""
    question_id = v.get('questionId')
    if not _is_db_ref_id(question_id):
        return ''
    from rag.services.generation import question_focus

    return question_focus(question_id, user.id)


class AIDailyCapReached(APIException):
    status_code = 503
    default_detail = 'دستیار امروز به سقفِ استفاده رسیده؛ لطفاً فردا دوباره بپرس.'
    default_code = 'ai_daily_cap'


def _check_global_daily_cap() -> None:
    """
    سقفِ روزانه‌ی کلِ سامانه (AI_GLOBAL_DAILY_MAX): سوال‌های امروز که به مدل رفته‌اند یا با کش
    برگشته‌اند ولی هنوز «free»/«credit» مانده‌اند. در دیتابیس شمرده می‌شود، پس بینِ workerها
    مشترک است. داخلِ همان تراکنشِ IMMEDIATEِ فازِ ۱ صدا زده می‌شود؛ روی Postgres دو درخواستِ
    هم‌زمان ممکن است هر دو از لبه رد شوند (چند سوال بیشتر از سقف، نه بیشتر).
    """
    from django.conf import settings

    start = timezone.localtime().replace(hour=0, minute=0, second=0, microsecond=0)
    used = Message.objects.filter(role='user', created_at__gte=start, charge__in=('free', 'credit')).count()
    if used >= settings.AI_GLOBAL_DAILY_MAX:
        logger.error('AI global daily cap reached: cap=%s used=%s', settings.AI_GLOBAL_DAILY_MAX, used)
        raise AIDailyCapReached()


class AIProviderUnavailable(APIException):
    """
    ⚠️ افزوده شده: وقتی generate_tutor_answer به یک AI provider واقعی
    (OpenAI/GapGPT/...) وصل بشه، شکستِ فراخوانیِ شبکه‌ای (timeout، خطای
    provider و غیره) باید یک خطای واضح و قابلِ retry برای کلاینت باشه، نه
    یک 500 خام. جزئیاتِ فنیِ خطای اصلی در لاگ سرور ثبت می‌شه (نگاه کنید به
    AskAIView.post)، نه در پاسخِ کلاینت.
    """

    status_code = 503
    default_detail = 'سرویس هوش مصنوعی موقتاً در دسترس نیست. لطفاً دوباره تلاش کنید.'
    default_code = 'ai_provider_unavailable'


def generate_tutor_answer(
    user_question: str, grade_id, lesson_id, context=None
) -> tuple[str, list[str]]:
    """
    مرزِ (boundary) تولیدِ پاسخِ AI Tutor.

    امضا: (user_question, grade_id, lesson_id, context). context یک پارامترِ
    عمومیِ اختیاریِ application-generated است که این تابع صرفاً مصرف می‌کنه،
    بدونِ اینکه بدونه از کجا retrieve شده. جریان:

        gradeId + subject + lessonNumber + userQuestion → لایه‌ی RAG (اپ rag،
        Qdrant) → ~۴ chunkِ انتخاب‌شده به‌عنوانِ context → این تابع (لایه‌ی
        تولید) → provider (OpenAI از طریقِ GapGPT)

    AskAIView خودش retrieval را انجام می‌ده و نتیجه را به‌عنوانِ context
    (لیستی از dictهای {title, summary, score}) پاس می‌ده؛ این تابع فقط
    تولید می‌کنه. مالکیتِ تاریخچه‌ی مکالمه همیشه دستِ دیتابیسِ خودمونه، نه
    provider.

    اگه RAG پیکربندی نشده باشه، پاسخِ stub فقط در DEBUG یا با RAG_FORCE_MOCK
    (تست‌ها) برمی‌گرده. بیرون از آن خطا می‌دهد و AskAIView آن را ۵۰۳ می‌کند.
    تا ۲۰۲۶-۱۰-۰۳ stub در پروداکشن هم برمی‌گشت: اگر کلیدی از .env سرور می‌افتاد،
    دانش‌آموز با ۲۰۰ متنِ «این یک پاسخ نمونه (stub)… OPENAI_API_KEY را تنظیم
    کنید» می‌گرفت و سهمیه‌ی روزانه‌اش هم کم می‌شد (با تستِ دورریختنی بازتولید شد).
    """
    from django.conf import settings

    from rag.services.config import rag_is_configured

    if rag_is_configured():
        from rag.services.generation import generate_answer

        context_chunks = context if isinstance(context, list) else None
        # استثناهای شبکه/provider اینجا خفه نمی‌شن؛ AskAIView اون‌ها رو به
        # 503 AIProviderUnavailable تبدیل می‌کنه (و اسلاتِ رزروشده رو آزاد).
        # این تابع قراردادِ دوتایی‌اش را نگه می‌دارد؛ متادیتای تولید فقط در
        # مسیرِ اصلیِ AskAIView (که آن را ذخیره می‌کند) لازم است.
        answer, follow_ups, _meta = generate_answer(user_question, context_chunks)
        return answer, follow_ups

    if not (settings.DEBUG or settings.RAG_FORCE_MOCK):
        raise RuntimeError('RAG پیکربندی نشده (OPENAI_API_KEY/QDRANT_API_KEY)؛ stub فقط در DEBUG.')

    context_note = f' (با context: {context})' if context else ''
    answer = (
        f'این یک پاسخ نمونه (stub) برای سوال «{user_question}»{context_note} است. '
        'برای اتصال به مدل واقعی هوش مصنوعی، OPENAI_API_KEY و QDRANT_API_KEY '
        'را در .env تنظیم کنید (لایه‌ی RAG در اپ rag).'
    )
    follow_ups = ['یک مثال دیگر بزن', 'بیشتر توضیح بده']
    return answer, follow_ups


class AskAIView(APIView):
    """POST /api/ai/ask — سوال از AI (استریم اختیاری؛ در این فاز استریم پیاده نشده)"""

    permission_classes = [IsAuthenticated]
    # daily_quota فقط سقفِ روزانه‌ست و جلوی burstِ کوتاه‌مدت رو نمی‌گیره؛
    # این throttle_scope مجزا اون burst رو محدود می‌کنه.
    throttle_scope = 'ai_ask'
    throttle_classes = [ScopedRateThrottle]

    @extend_schema(
        request=AskRequestSerializer,
        responses=AskResponseSerializer,
        parameters=[
            OpenApiParameter(
                'Idempotency-Key', OpenApiTypes.STR, OpenApiParameter.HEADER, required=False,
                description=(
                    'کلید اختیاری idempotency (مثل payment و content). اگه ست بشه، '
                    'رتری/دابل‌کلیکِ همون سوال با همون کلید، دوباره پیامِ کاربر/دستیارِ '
                    'تکراری نمی‌سازه و اسلاتِ کوتای روزانه‌ی دیگری مصرف نمی‌کنه؛ به‌جاش '
                    'نتیجه‌ی همون درخواستِ قبلاً پردازش‌شده برگردونده می‌شه.'
                ),
            ),
        ],
        tags=['AI Tutor (ai_tutor)'],
    )
    def post(self, request):
        serializer = AskRequestSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        v = serializer.validated_data
        user = request.user
        idempotency_key = request.headers.get('Idempotency-Key') or None

        # درخواست تکراری با همین کلید: نتیجه‌ی قبلی برگردانده می‌شود، بدون مصرف دوباره‌ی کوتا.
        if idempotency_key:
            existing = AskIdempotency.objects.filter(
                user_id=user.id, idempotency_key=idempotency_key,
            ).select_related('conversation', 'user_message', 'assistant_message').first()
            if existing is not None:
                return Response(self._build_ask_response(existing).data)

        # سه فاز جدا: فاز ۱ کوتای روزانه را زیر قفل ردیف کاربر چک/رزرو می‌کند،
        # فاز ۲ (فراخوانی AI) کاملاً بیرون از هر قفل/تراکنشی اجرا می‌شود تا
        # تأخیر شبکه قفل کاربر را نگه ندارد، فاز ۳ پیام دستیار و idempotency
        # را ثبت می‌کند. اگر فاز ۲/۳ شکست بخورد، رزرو فاز ۱ آزاد می‌شود.
        # محافظِ دستیار، لایه‌ی ۱ (rag/services/guardrails.py): ترفندِ آشکار برای عوض کردنِ
        # قواعد، یا جمله‌ای درباره‌ی آسیب به خود، اصلاً به مدل نمی‌رسد؛ پس هزینه‌ای ندارد و از
        # سهمیه و اعتبار هم کم نمی‌شود. سقفِ burst (throttle_scope=ai_ask) جلوی تکرارش را می‌گیرد.
        from rag.services import guardrails
        from subscription.services import credit_balance, spend_credit

        screened = guardrails.screen(v['userQuestion'])

        with transaction.atomic():
            AccountUser.objects.select_for_update().get(pk=user.id)

            # هر سوالی که به API می‌رسد پول است: اول از سهمیه‌ی رایگانِ امروز، بعد از اعتبارِ بسته
            # (۲۰۲۶-۱۰-۰۲). قفلِ ردیفِ کاربر و IMMEDIATEِ SQLite جلوی خرجِ دوباره‌ی یک اعتبار را می‌گیرند.
            today = timezone.localdate()
            used_today = Message.objects.filter(
                conversation__user_id=user.id, role='user', created_at__date=today, charge='free',
            ).count()
            quota = daily_quota(user.id)
            if not screened:
                _check_global_daily_cap()
            if screened:
                charge = 'none'
            elif used_today < quota:
                charge = 'free'
            elif credit_balance(user.id) > 0:
                charge = 'credit'
            else:
                from subscription.services import has_active_subscription

                n = str(quota).translate(str.maketrans('0123456789', '۰۱۲۳۴۵۶۷۸۹'))
                raise Throttled(detail=(
                    f'سقفِ امروزِ سوال از دستیار ({n} سوال) پر شد. با خریدِ بسته‌ی سوال همین الان ادامه بده.'
                    if has_active_subscription(user.id) else
                    f'سوال‌های رایگانِ امروز ({n} سوال) تمام شد. با خریدِ بسته‌ی سوال همین الان ادامه بده.'
                ))

            conversation_id = v.get('conversationId')
            conversation_is_new = not conversation_id
            if conversation_id:
                # ورودی غیرعددی هم باید 404 بدهد، نه خطای خام ORM.
                try:
                    conversation = get_object_or_404(Conversation, pk=conversation_id, user_id=user.id)
                except (ValueError, TypeError):
                    raise Http404
            else:
                # lessonId/chapterId روی Conversation فیلدهای IntegerField هستند؛
                # مقدار غیرعددی نادیده گرفته می‌شود (None ذخیره می‌شود) اما مقدار
                # خام همچنان بدون تغییر به generate_tutor_answer می‌رسد.
                raw_lesson_id = v.get('lessonId')
                raw_chapter_id = v.get('chapterId')
                conversation = Conversation.objects.create(
                    user_id=user.id,
                    lesson_id=raw_lesson_id if _is_db_ref_id(raw_lesson_id) else None,
                    chapter_id=raw_chapter_id if _is_db_ref_id(raw_chapter_id) else None,
                    grade_id=v.get('gradeId'),
                    subject=(v.get('subject') or '').strip(),
                    lesson_number=v.get('lessonNumber'),
                )

            raw_question_id = v.get('questionId')
            user_message = Message.objects.create(
                conversation=conversation,
                role='user',
                text=v['userQuestion'],
                question_id=raw_question_id if _is_db_ref_id(raw_question_id) else None,
                charge=charge,
            )
            if charge == 'credit':
                spend_credit(user.id, user_message.id)
        # قفل ردیف کاربر اینجا آزاد شده؛ فراخوانی AI بیرون از هر تراکنشی اجرا می‌شود.
        # لایه‌ی RAG (اپ rag) اینجا context را می‌سازد: پرسش → embedding →
        # جستجوی Qdrant (محدود به subject+grade+lessonNumber) → چند خلاصه.
        # اگر RAG پیکربندی نشده باشد context=None می‌ماند و generate_tutor_answer
        # همان پاسخِ stub را می‌دهد. خطای شبکه/provider در همین try به 503
        # تبدیل و اسلاتِ رزروشده آزاد می‌شود.
        # مقادیرِ ذخیره‌شده روی مکالمه هم fallback‌اند: در پیامِ دوم به بعد، اپ
        # فقط conversationId می‌فرستد و دیگر lessonId/gradeId نمی‌آید.
        grade_id = v.get('gradeId') or conversation.grade_id
        lesson_id = v.get('lessonId') or conversation.lesson_id
        subject = (v.get('subject') or conversation.subject or '').strip()
        grade_title = (v.get('grade') or '').strip()
        lesson_number = v.get('lessonNumber')
        if lesson_number is None:
            lesson_number = conversation.lesson_number

        # اگر اپ فقط lessonId (یا فقط conversationId) داده — حالتِ طبیعی، چون
        # می‌داند دانش‌آموز روی کدام درس است نه اینکه نامِ collectionِ Qdrant
        # چیست — خودمان subject/پایه/فصل را از زنجیره‌ی
        # Lesson→Node→Unit→Grade درمی‌آوریم.
        #
        # ⚠️ شرط عمداً `grade_title` را هم می‌بیند: پیش‌تر فقط subject و
        # lesson_number را چک می‌کرد، و چون Conversation فیلدی برای عنوانِ پایه
        # ندارد، از پیامِ دومِ هر مکالمه grade_title خالی می‌ماند و
        # resolve_target هیچ collectionی نمی‌ساخت — یعنی فقط جوابِ اول به کتاب
        # مستند بود و بقیه بی‌سروصدا عمومی می‌شدند.
        if lesson_id and not (subject and grade_title and lesson_number is not None):
            from rag.services.lesson_context import resolve_from_lesson

            derived = resolve_from_lesson(lesson_id)
            subject = subject or derived.get('subject', '')
            grade_title = grade_title or derived.get('gradeTitle', '')
            if lesson_number is None:
                lesson_number = derived.get('chapter')
            # تا آخرِ همین مکالمه دوباره حساب نشود.
            if subject and not conversation.subject:
                conversation.subject = subject
                conversation.lesson_number = lesson_number
                conversation.save(update_fields=['subject', 'lesson_number'])

        # نه درسی آمده نه subject — حالتِ عادیِ صفحه‌ی «از پی بپرس» که از نوارِ
        # پایین باز می‌شود. پایه‌ی ذخیره‌شده‌ی خودِ کاربر کتاب را معلوم می‌کند؛
        # بدونِ این، جواب‌های آن صفحه هیچ‌وقت به کتاب مستند نمی‌شدند.
        if not subject:
            from rag.services.lesson_context import resolve_from_grade

            grade_id = grade_id or getattr(user, 'grade_id', None)
            derived = resolve_from_grade(grade_id)
            subject = derived.get('subject', '')
            grade_title = grade_title or derived.get('gradeTitle', '')
            if subject and not conversation.subject:
                conversation.subject = subject
                conversation.grade_id = conversation.grade_id or grade_id
                conversation.save(update_fields=['subject', 'grade_id'])

        age = _learner_age(user)

        # --- مسیرِ استریم (SSE) --- فقط وقتی stream=true و RAG پیکربندی‌شده.
        # قراردادِ JSONِ غیراستریم دست‌نخورده می‌ماند؛ Swagger که stream
        # نمی‌فرسته همون مسیرِ قبلی رو می‌گیره.
        if v.get('stream'):
            from rag.services.config import rag_is_configured

            if rag_is_configured():
                return self._streaming_response(
                    request=request, v=v, user=user, conversation=conversation,
                    user_message=user_message, conversation_is_new=conversation_is_new,
                    idempotency_key=idempotency_key, used_today=used_today,
                    subject=subject, grade_id=grade_id, grade_title=grade_title,
                    lesson_number=lesson_number, screened=screened, quota=quota, age=age,
                )

        used_chunks: list[dict] = []
        retrieval_meta: dict = {}
        generation_meta: dict = {}
        answer_from_cache = False
        context = None
        try:
            from rag.services.config import rag_is_configured

            if screened:
                answer_text, follow_ups = screened[1], []
                generation_meta = {'promptVersion': f'guardrail-{screened[0]}'}
            elif rag_is_configured():
                from rag.services import answer_cache
                from rag.services.generation import generate_answer
                from rag.services.retrieval import resolve_target, retrieve_context

                history = _conversation_history(conversation, user_message)
                focus = _question_focus(v, user)
                cacheable = answer_cache.cacheable(history) and not focus
                _col, _fk, _fv = resolve_target(subject, grade_title, grade_id, lesson_number)
                cache_key = answer_cache.make_key(_col, _fv, v['userQuestion'], _cache_grade(grade_title, age))
                cached = answer_cache.get(cache_key) if cacheable else None
                if cached is not None:
                    answer_text = cached['answer']
                    follow_ups = list(cached['followUps'])
                    used_chunks = cached['usedChunks']
                    retrieval_meta = cached.get('retrievalMeta') or {}
                    generation_meta = cached.get('generationMeta') or {}
                    answer_from_cache = True
                    _uncharge(user_message)
                else:
                    used_chunks, retrieval_meta = retrieve_context(
                        _retrieval_query(v['userQuestion'], history, focus),
                        subject=subject, grade_id=grade_id,
                        grade_title=grade_title, lesson_number=lesson_number,
                    )
                    used_chunks = [
                        {'title': c.get('title', ''), 'summary': c.get('summary', '')}
                        for c in used_chunks
                    ]
                    answer_text, follow_ups, generation_meta = generate_answer(
                        v['userQuestion'], used_chunks or None,
                        retrieval_meta.get('gradeTitle') or grade_title or '',
                        history=history, focus=focus, age=age,
                    )
                    provider_status.record_success()
                    # ردِ «غیرِ ریاضی» نه منبعِ کتاب نشان می‌دهد نه کش می‌شود.
                    if generation_meta.get('refused'):
                        used_chunks = []
                    elif cacheable:
                        answer_cache.set(cache_key, {
                            'answer': answer_text, 'followUps': follow_ups,
                            'usedChunks': used_chunks, 'retrievalMeta': retrieval_meta,
                            'generationMeta': generation_meta,
                        })
            else:
                answer_text, follow_ups = generate_tutor_answer(
                    v['userQuestion'], grade_id, lesson_id, context
                )
        except Exception as exc:
            logger.exception('generate_tutor_answer failed for user_id=%s', user.id)
            provider_status.record_failure(exc)
            self._release_reserved_slot(conversation, user_message, conversation_is_new)
            raise AIProviderUnavailable()

        try:
            with transaction.atomic():
                assistant_message = Message.objects.create(
                    conversation=conversation,
                    role='assistant',
                    text=answer_text,
                    question_id=_db_ref_or_none(v.get('questionId')),
                    **_annotation(retrieval_meta, generation_meta, from_cache=answer_from_cache),
                )

                if idempotency_key:
                    AskIdempotency.objects.create(
                        user_id=user.id,
                        idempotency_key=idempotency_key,
                        conversation=conversation,
                        user_message=user_message,
                        assistant_message=assistant_message,
                        suggested_follow_ups=follow_ups,
                    )
        except IntegrityError:
            if not idempotency_key:
                raise
            existing = AskIdempotency.objects.filter(
                user_id=user.id, idempotency_key=idempotency_key,
            ).select_related('conversation', 'user_message', 'assistant_message').first()
            if existing is None:
                raise
            self._release_reserved_slot(conversation, user_message, conversation_is_new)
            return Response(self._build_ask_response(existing).data)

        state = quota_state(user.id)

        data = {
            'conversationId': str(conversation.id),
            'messageId': str(assistant_message.id),
            'answer': answer_text,
            'suggestedFollowUps': follow_ups,
            'remainingQuota': state['remainingQuota'],
            'credits': state['credits'],
            'diagram': assistant_message.diagram,
            'usedChunks': [
                {'title': c.get('title', ''), 'summary': c.get('summary', '')}
                for c in used_chunks
            ],
        }
        # فقط برای بازبینیِ داخلیِ خودمان (نه کاربر): از کدام collection/فصل
        # چانک‌ها برداشته شد. فقط وقتی DEBUG روشن است در پاسخ می‌آید.
        if settings.DEBUG and retrieval_meta:
            data['retrievalDebug'] = {
                'collection': retrieval_meta.get('collection'),
                'filterKey': retrieval_meta.get('filterKey'),
                'chapterOrLesson': retrieval_meta.get('filterValue'),
                'gradeTitle': retrieval_meta.get('gradeTitle'),
                'chunkCount': retrieval_meta.get('chunkCount'),
                'chunkTitles': [c.get('title', '') for c in used_chunks],
            }
        return Response(AskResponseSerializer(data).data)

    def _streaming_response(self, *, request, v, user, conversation, user_message,
                            conversation_is_new, idempotency_key, used_today,
                            subject, grade_id, grade_title, lesson_number, screened=None, quota=None,
                            age=None):
        """
        نسخه‌ی SSE از post(): همون منطقِ retrieval + کوتا + idempotency، ولی
        پاسخ توکن‌به‌توکن استریم می‌شه. رویدادها:
          event: chunk  data: {"delta": "..."}
          event: done   data: {conversationId, messageId, remainingQuota, diagram, usedChunks[, retrievalDebug]}
          event: error  data: {"detail": "..."}
        فاز ۱ (رزرو کوتا + conversation + user_message) قبلاً در post() انجام شده.
        """
        from rag.services import answer_cache, guardrails
        from rag.services.generation import (
            PROMPT_VERSION, chat_model_for, default_follow_ups, stream_answer,
        )
        from rag.services.retrieval import resolve_target, retrieve_context

        def _sse(event: str, payload: dict) -> str:
            return f'event: {event}\ndata: {json.dumps(payload, ensure_ascii=False)}\n\n'

        def _gen():
            nonlocal used_today
            generation_meta: dict = {}
            answer_from_cache = False
            _col, _fk, _fv = resolve_target(subject, grade_title, grade_id, lesson_number)
            cache_key = answer_cache.make_key(_col, _fv, v['userQuestion'], _cache_grade(grade_title, age))
            try:
                history = _conversation_history(conversation, user_message)
                focus = _question_focus(v, user)
                cacheable = answer_cache.cacheable(history) and not focus
                cached = answer_cache.get(cache_key) if cacheable and not screened else None
                if screened:
                    answer_text, follow_ups = screened[1], []
                    used_chunks, retrieval_meta = [], {}
                    generation_meta = {'promptVersion': f'guardrail-{screened[0]}'}
                    yield _sse('chunk', {'delta': answer_text})
                elif cached is not None:
                    answer_text = cached['answer']
                    follow_ups = list(cached['followUps'])
                    used_chunks = cached['usedChunks']
                    retrieval_meta = cached.get('retrievalMeta') or {}
                    generation_meta = cached.get('generationMeta') or {}
                    answer_from_cache = True
                    _uncharge(user_message)
                    yield _sse('chunk', {'delta': answer_text})
                else:
                    used_chunks, retrieval_meta = retrieve_context(
                        _retrieval_query(v['userQuestion'], history, focus),
                        subject=subject, grade_id=grade_id,
                        grade_title=grade_title, lesson_number=lesson_number,
                    )
                    used_chunks = [
                        {'title': c.get('title', ''), 'summary': c.get('summary', '')}
                        for c in used_chunks
                    ]
                    parts: list[str] = []
                    stream_result: dict = {}
                    started = time.monotonic()
                    effective_grade = retrieval_meta.get('gradeTitle') or grade_title or ''
                    try:
                        for delta in stream_answer(
                            v['userQuestion'], used_chunks or None, effective_grade,
                            history=history, focus=focus, result=stream_result, age=age,
                        ):
                            parts.append(delta)
                            yield _sse('chunk', {'delta': delta})
                    except GeneratorExit:
                        # کاربر صفحه را وسطِ جواب بست. سوال خرج می‌ماند (مدل هزینه‌اش را گرفته)، ولی
                        # همان بخشی که ساخته شد ثبت می‌شود تا گفتگو سوالِ بی‌جواب نداشته باشد.
                        self._save_partial_answer(conversation, user, v, ''.join(parts))
                        raise
                    provider_status.record_success()
                    answer_text = ''.join(parts).strip() or 'متأسفم، الان نتونستم پاسخِ مناسبی بسازم.'
                    follow_ups = default_follow_ups()
                    # توکن‌ها از آخرین تکه‌ی استریم می‌آیند (stream_options.include_usage).
                    # تا ۲۰۲۶-۱۰-۰۱ این‌جا صفر ثبت می‌شد، یعنی هزینه‌ی واقعیِ اپ هیچ‌جا
                    # اندازه‌گیری نمی‌شد؛ providerی که usage نفرستد هنوز صفر می‌دهد.
                    usage = stream_result.get('usage') or {}
                    generation_meta = {
                        'model': chat_model_for(effective_grade),
                        'promptVersion': PROMPT_VERSION,
                        'latencyMs': int((time.monotonic() - started) * 1000),
                        'hadContext': bool(used_chunks),
                        'diagram': stream_result.get('diagram'),
                        'promptTokens': usage.get('promptTokens', 0),
                        'completionTokens': usage.get('completionTokens', 0),
                    }
                    if stream_result.get('refused'):
                        used_chunks = []
                        answer_text = guardrails.REFUSAL
                    elif cacheable:
                        answer_cache.set(cache_key, {
                            'answer': answer_text, 'followUps': follow_ups,
                            'usedChunks': used_chunks, 'retrievalMeta': retrieval_meta,
                            'generationMeta': generation_meta,
                        })
            except Exception as exc:
                logger.exception('stream generate failed for user_id=%s', user.id)
                provider_status.record_failure(exc)
                self._release_reserved_slot(conversation, user_message, conversation_is_new)
                yield _sse('error', {'detail': AIProviderUnavailable.default_detail})
                return

            # فاز ۳: ثبتِ پیامِ دستیار + idempotency (همون منطقِ post()).
            try:
                with transaction.atomic():
                    assistant_message = Message.objects.create(
                        conversation=conversation, role='assistant', text=answer_text,
                        question_id=_db_ref_or_none(v.get('questionId')),
                        **_annotation(retrieval_meta, generation_meta, from_cache=answer_from_cache),
                    )
                    if idempotency_key:
                        AskIdempotency.objects.create(
                            user_id=user.id, idempotency_key=idempotency_key,
                            conversation=conversation, user_message=user_message,
                            assistant_message=assistant_message,
                            suggested_follow_ups=follow_ups,
                        )
            except IntegrityError:
                existing = AskIdempotency.objects.filter(
                    user_id=user.id, idempotency_key=idempotency_key,
                ).select_related('assistant_message').first() if idempotency_key else None
                if existing is None:
                    self._release_reserved_slot(conversation, user_message, conversation_is_new)
                    yield _sse('error', {'detail': AIProviderUnavailable.default_detail})
                    return
                self._release_reserved_slot(conversation, user_message, conversation_is_new)
                yield _sse('done', {
                    'conversationId': str(existing.conversation_id),
                    'messageId': str(existing.assistant_message_id),
                    'suggestedFollowUps': existing.suggested_follow_ups,
                    **quota_state(user.id),
                    'diagram': existing.assistant_message.diagram,
                    'usedChunks': [],
                })
                return

            done = {
                'conversationId': str(conversation.id),
                'messageId': str(assistant_message.id),
                'suggestedFollowUps': follow_ups,
                **quota_state(user.id),
                'diagram': assistant_message.diagram,
                'usedChunks': [
                    {'title': c.get('title', ''), 'summary': c.get('summary', '')}
                    for c in used_chunks
                ],
            }
            if settings.DEBUG and retrieval_meta:
                done['retrievalDebug'] = {
                    'collection': retrieval_meta.get('collection'),
                    'filterKey': retrieval_meta.get('filterKey'),
                    'chapterOrLesson': retrieval_meta.get('filterValue'),
                    'gradeTitle': retrieval_meta.get('gradeTitle'),
                    'chunkCount': retrieval_meta.get('chunkCount'),
                    'chunkTitles': [c.get('title', '') for c in used_chunks],
                }
            yield _sse('done', done)

        response = StreamingHttpResponse(_gen(), content_type='text/event-stream')
        response['Cache-Control'] = 'no-cache'
        response['X-Accel-Buffering'] = 'no'
        return response

    @staticmethod
    def _save_partial_answer(conversation, user, v, text):
        logger.info('stream closed by client mid-answer: user_id=%s conversation_id=%s chars=%s',
                    user.id, conversation.id, len(text))
        if not text.strip():
            return
        try:
            Message.objects.create(
                conversation=conversation, role='assistant', text=text.strip() + ' …',
                question_id=_db_ref_or_none(v.get('questionId')),
            )
        except Exception:
            logger.exception('saving partial answer failed: conversation_id=%s', conversation.id)

    def _release_reserved_slot(self, conversation, user_message, conversation_is_new):
        """پاک‌کردن رزرو فاز ۱ وقتی فاز بعدی شکست می‌خورد، تا نه پیام یتیم بماند و نه کوتا مصرف‌شده حساب شود."""
        from subscription.services import refund_credit

        with transaction.atomic():
            refund_credit(user_message.id)
            if conversation_is_new:
                conversation.delete()
            else:
                user_message.delete()

    def _build_ask_response(self, existing: AskIdempotency) -> AskResponseSerializer:
        """
        ساختِ همون پاسخِ AskResponseSerializer برای یک AskIdempotency
        از قبل ثبت‌شده (مسیرِ idempotent replay)، بدونِ ساختِ پیامِ جدید یا
        مصرفِ کوتا.

        ⚠️ اصلاح: قبلاً suggestedFollowUps ذخیره نمی‌شد و اینجا با
        فراخوانیِ دوباره‌ی generate_tutor_answer بازسازی می‌شد. تا وقتی
        generate_tutor_answer یک stub خالی از اثر جانبیه بی‌ضرر بود، اما
        وقتی به یک AI provider واقعی وصل بشه، هر replay یک فراخوانیِ
        واقعیِ دیگه به provider می‌زد (هزینه/تأخیرِ تکراری، و
        suggestedFollowUps متفاوت در هر retry — دقیقاً برخلافِ هدفِ
        idempotency). حالا suggested_follow_ups مستقیماً از خودِ رکوردِ
        AskIdempotency (ذخیره‌شده در همون درخواستِ اولِ موفق) خونده می‌شه؛
        هیچ فراخوانیِ دیگری به generate_tutor_answer در مسیرِ replay انجام
        نمی‌شه. answer هم مثلِ قبل از Message.text (بدونِ تغییر) خونده
        می‌شه؛ remainingQuota هم مثلِ قبل از روی مصرفِ *فعلیِ* امروز
        محاسبه می‌شه، نه لحظه‌ی درخواستِ اولیه، چون این فراخوانیِ
        idempotent به‌خودی‌خود چیزی مصرف نکرده.
        """
        conversation = existing.conversation
        state = quota_state(existing.user_id)

        data = {
            'conversationId': str(conversation.id),
            'messageId': str(existing.assistant_message.id),
            'answer': existing.assistant_message.text,
            'suggestedFollowUps': existing.suggested_follow_ups,
            'remainingQuota': state['remainingQuota'],
            'credits': state['credits'],
            'diagram': existing.assistant_message.diagram,
        }
        return AskResponseSerializer(data)


class ConversationMessagesView(APIView):
    """GET /api/ai/conversations/{id}/messages — تاریخچه‌ی مکالمه"""

    # سقفِ نرخِ این مسیر تا ۲۰۲۶-۱۰-۰۷ نبود؛ `read` همان نرخِ مسیرهای هم‌خانواده است.
    throttle_scope = 'read'
    throttle_classes = [ScopedRateThrottle]

    permission_classes = [IsAuthenticated]

    @extend_schema(responses=ConversationMessagesResponseSerializer, tags=['AI Tutor (ai_tutor)'])
    def get(self, request, conversation_id):
        conversation = get_object_or_404(Conversation, pk=conversation_id, user_id=request.user.id)
        # آخرین ۲۰۰ پیام، به ترتیبِ زمانی (قدیمی→جدید). جلوی رشدِ نامحدودِ پاسخ
        # برای یک مکالمه‌ی خیلی طولانی رو می‌گیره.
        messages = reversed(list(conversation.messages.order_by('-id')[:200]))

        data = {
            'conversationId': str(conversation.id),
            'lessonId': str(conversation.lesson_id) if conversation.lesson_id else None,
            'messages': [
                {'messageId': str(m.id), 'role': m.role, 'text': m.text, 'diagram': m.diagram}
                for m in messages
            ],
        }
        return Response(ConversationMessagesResponseSerializer(data).data)


class MessageFeedbackView(APIView):
    """POST /api/ai/messages/{id}/feedback — فیدبک 👍/👎"""

    # سقفِ نرخِ این مسیر تا ۲۰۲۶-۱۰-۰۷ نبود؛ `user_profile_write` همان نرخِ مسیرهای هم‌خانواده است.
    throttle_scope = 'user_profile_write'
    throttle_classes = [ScopedRateThrottle]

    permission_classes = [IsAuthenticated]

    @extend_schema(
        request=FeedbackRequestSerializer,
        responses=FeedbackResponseSerializer,
        tags=['AI Tutor (ai_tutor)'],
    )
    def post(self, request, message_id):
        message = get_object_or_404(Message, pk=message_id, conversation__user_id=request.user.id)
        serializer = FeedbackRequestSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)

        # `reason` فقط وقتی نوشته می‌شود که در همین درخواست آمده باشد: کاربری
        # که رأیش را عوض می‌کند معمولاً دلیل نمی‌نویسد، و پیش‌فرضِ '' متنِ
        # قبلی‌اش را پاک می‌کرد — یعنی مفیدترین بخشِ فیدبک از دست می‌رفت.
        defaults = {'is_helpful': serializer.validated_data['isHelpful']}
        if 'reason' in serializer.validated_data:
            defaults['reason'] = serializer.validated_data['reason']
        MessageFeedback.objects.update_or_create(message=message, defaults=defaults)
        return Response(FeedbackResponseSerializer({'success': True}).data)


class QuotaView(APIView):
    """
    GET /api/ai/quota — «امروز X سوالِ رایگان + Y سوالِ خریده‌شده». اپ پیش از اولین سوال این را
    نشان می‌دهد؛ بعد از هر جواب همین عددها در خودِ پاسخ هم می‌آیند (remainingQuota، credits).
    """

    permission_classes = [IsAuthenticated]
    throttle_scope = 'read'
    throttle_classes = [ScopedRateThrottle]

    @extend_schema(responses=QuotaResponseSerializer, tags=['AI Tutor (ai_tutor)'])
    def get(self, request):
        data = {'freeDaily': daily_quota(request.user.id), **quota_state(request.user.id)}
        return Response(QuotaResponseSerializer(data).data)