File size: 7,003 Bytes
7c6ffa6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 | 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)
|