from __future__ import annotations from typing import Any from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy import select from sqlalchemy.orm import Session from app.core.auth import require_user from app.core.database import get_db from app.models.document import Document from app.models.previous_paper import PreviousPaper from app.models.previous_question import PreviousQuestion from app.models.study_profile import StudyProfile from app.models.user import User from app.schemas.study_path import StudyPathRequest, StudyPathResult from app.services.retrieval import chunks_to_context, retrieve_relevant_chunks from app.services.evidence_contract import build_evidence_context, _syllabus_matches from app.services.source_guard import assert_sources_eligible_for_exam from app.services.study_path_engine import build_study_path from app.services.study_request_analyzer import analyze_study_request router = APIRouter() def _profile_dict(profile: StudyProfile | None) -> dict[str, Any]: if profile is None or (profile.extra or {}).get("onboardingSkipped"): return {} return { "exam": profile.exam, "board": profile.board, "grade": profile.grade, "subject": profile.subject, "chapter": profile.chapter, "topic": profile.topic, "level": profile.level, "goal": profile.goal, "time_left": profile.time_left, "language_preference": profile.language_preference, "primary_need": profile.primary_need, "weak_areas": list(profile.weak_areas or []), } def _pyq_summary( db: Session, user_id: str, subject: str | None, *, board: str | None = None, class_level: str | None = None, ) -> dict[str, Any]: query = ( select(PreviousQuestion) .join(PreviousPaper, PreviousPaper.id == PreviousQuestion.previous_paper_id) .where(PreviousPaper.user_id == user_id) .where(PreviousPaper.verification_status == "verified") ) if subject: query = query.where(PreviousQuestion.subject == subject) questions = [ question for question in db.scalars(query).all() if _syllabus_matches(question.previous_paper.syllabus, board, class_level) ] count = len(questions) return {"available": count > 0, "question_count": count} @router.post("/generate", response_model=StudyPathResult, summary="Generate a personalised study path", description="Generate a step-by-step study timeline with readiness score, actions, and trust notes. Supports single or multiple source grounding.") def generate_study_path( payload: StudyPathRequest, db: Session = Depends(get_db), current_user: User = Depends(require_user), ) -> StudyPathResult: profile: StudyProfile | None = None if payload.use_my_profile: profile = db.scalar( select(StudyProfile).where(StudyProfile.user_id == current_user.id), ) profile_data = _profile_dict(profile) if payload.raw_text: analysis = analyze_study_request(payload.raw_text, payload.syllabus_text) else: analysis = { "raw_text": "", "exam": payload.exam or profile_data.get("exam"), "board": None, "grade": None, "subject": payload.subject or profile_data.get("subject"), "chapter": None, "topic": payload.topic or profile_data.get("topic"), "goal": payload.goal or profile_data.get("goal"), "time_left": payload.time_left or profile_data.get("time_left"), "level": payload.level or profile_data.get("level"), "language_preference": payload.language_preference or profile_data.get("language_preference"), "primary_need": profile_data.get("primary_need"), "confidence": 0.6 if (payload.topic or profile_data.get("topic")) else 0.2, "missing_fields": [], "warnings": [], "syllabus_status": "not_provided", "pyq_status": "unavailable", } source_context: str | None = None retrieval_query = ( " ".join(filter(None, [analysis.get("topic"), analysis.get("subject"), "exam keywords"])) or "exam keywords" ) # Build combined source_ids list (merge source_id + source_ids) all_source_ids: list[str] = [] if payload.source_id: all_source_ids.append(payload.source_id) if payload.source_ids: for sid in payload.source_ids: if sid not in all_source_ids: all_source_ids.append(sid) if all_source_ids: # Source Reality Guard — block resumes/unclassified before AI call. assert_sources_eligible_for_exam(db, current_user.id, all_source_ids) context_parts: list[str] = [] source_titles: list[str] = [] skipped_ids: list[str] = [] for sid in all_source_ids: document = db.get(Document, sid) if document is None or document.user_id != current_user.id: # Security: silently skip foreign/missing sources (safe 404 behavior) skipped_ids.append(sid) continue if document.status not in {"ready"}: # Source is still processing — skip with warning skipped_ids.append(sid) continue chunks = retrieve_relevant_chunks( db=db, document_id=document.id, query=retrieval_query, limit=4, # fewer chunks per source when multi-source user_id=current_user.id, ) ctx = chunks_to_context(chunks, fallback_text=document.extracted_text, max_chars=3600) if ctx and ctx.strip(): context_parts.append(ctx) source_titles.append(document.title) if context_parts: # Merge and lightly limit total context size (~4000 chars) merged = "\n\n---\n\n".join(context_parts) source_context = merged[:4000] if skipped_ids and not context_parts: # All requested sources were unavailable source_context = None pyq_summary = _pyq_summary( db, current_user.id, analysis.get("subject"), board=profile_data.get("board"), class_level=profile_data.get("grade"), ) evidence_ctx = build_evidence_context( db, current_user.id, subject=analysis.get("subject"), board=profile_data.get("board"), class_level=profile_data.get("grade"), explicit_source_ids=all_source_ids, ) result = build_study_path( study_profile=profile_data, analysis=analysis, source_context=source_context, pyq_summary=pyq_summary, ) result["analysis"] = analysis result["evidence_label"] = evidence_ctx.evidence_label return StudyPathResult(**result)