fix(cbt): comprehensive overhaul of CBT & DPP test pipelines, JSON schemas, active states, and exam pattern filters
This commit is contained in:
180
backend/main.py
180
backend/main.py
@@ -443,21 +443,28 @@ def login(payload: LoginPayload, session: Session = Depends(get_session)):
|
||||
def list_active_tests(exam_pattern: Optional[str] = None, type: Optional[str] = None, session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)):
|
||||
query = select(models.Test).where(models.Test.is_active == True)
|
||||
if exam_pattern:
|
||||
if exam_pattern == 'JEE_NEET':
|
||||
pattern_str = str(exam_pattern).strip()
|
||||
if pattern_str in ['JEE_NEET', 'ALL', 'All']:
|
||||
pass
|
||||
elif exam_pattern == 'JEE':
|
||||
query = query.where(models.Test.exam_pattern.in_([models.ExamEnum.JEE_MAIN, models.ExamEnum.JEE_ADVANCED]))
|
||||
elif pattern_str in ['JEE', 'JEE_Main', 'JEE_MAIN', 'JEE Main', 'JEE_Advanced', 'JEE_ADVANCED', 'JEE Advanced']:
|
||||
query = query.where(models.Test.exam_pattern.in_([models.ExamEnum.JEE_MAIN, models.ExamEnum.JEE_ADVANCED, models.ExamEnum.JEE_NEET]))
|
||||
elif pattern_str in ['NEET']:
|
||||
query = query.where(models.Test.exam_pattern.in_([models.ExamEnum.NEET, models.ExamEnum.JEE_NEET]))
|
||||
else:
|
||||
try:
|
||||
# Find matching enum by value string (e.g. "JEE_Main")
|
||||
enum_val = models.ExamEnum(exam_pattern)
|
||||
query = query.where(models.Test.exam_pattern == enum_val)
|
||||
enum_val = models.ExamEnum(pattern_str)
|
||||
query = query.where(models.Test.exam_pattern.in_([enum_val, models.ExamEnum.JEE_NEET]))
|
||||
except ValueError:
|
||||
pass
|
||||
if type:
|
||||
query = query.where(models.Test.type == type)
|
||||
if type in ["Mock", "CBT"]:
|
||||
query = query.where(models.Test.type.in_(["Mock", "CBT"]))
|
||||
elif type in ["DPS", "DPP"]:
|
||||
query = query.where(models.Test.type.in_(["DPS", "DPP"]))
|
||||
else:
|
||||
query = query.where(models.Test.type == type)
|
||||
else:
|
||||
query = query.where(models.Test.type != "DPS")
|
||||
query = query.where(models.Test.type.not_in(["DPS", "DPP"]))
|
||||
return session.exec(query).all()
|
||||
|
||||
@app.get("/api/tests/{test_id}", response_model=models.Test, tags=["CBT Engine"])
|
||||
@@ -1303,26 +1310,30 @@ def get_admin_tests(session: Session = Depends(get_session), current_user: model
|
||||
return tests
|
||||
|
||||
class ManualQuestionInput(BaseModel):
|
||||
section: str = "General"
|
||||
subject: str
|
||||
topic: str
|
||||
subtopic: str
|
||||
difficulty: str
|
||||
type: str
|
||||
question_text: str
|
||||
options: Optional[Dict[str, Any]] = None
|
||||
correct_answer: Dict[str, Any]
|
||||
solution_text: str
|
||||
language: str
|
||||
section: Optional[str] = "General"
|
||||
subject: Optional[str] = "General"
|
||||
topic: Optional[str] = "General"
|
||||
subtopic: Optional[str] = "General"
|
||||
difficulty: Optional[str] = "Medium"
|
||||
type: Optional[str] = "MCQ_Single"
|
||||
question_text: Optional[str] = None
|
||||
question: Optional[str] = None
|
||||
options: Optional[Any] = None
|
||||
options_hi: Optional[Any] = None
|
||||
correct_answer: Optional[Any] = None
|
||||
solution_text: Optional[str] = None
|
||||
solution: Optional[str] = None
|
||||
explanation: Optional[str] = None
|
||||
language: Optional[str] = "English"
|
||||
|
||||
class ManualTestCreate(BaseModel):
|
||||
title: str
|
||||
type: str = "CBT"
|
||||
type: str = "Mock"
|
||||
exam_pattern: str
|
||||
time_limit: int
|
||||
scheduled_date: Optional[datetime] = None
|
||||
end_date: Optional[datetime] = None
|
||||
admin_id: UUID
|
||||
admin_id: Optional[UUID] = None
|
||||
questions_json: List[ManualQuestionInput]
|
||||
|
||||
@app.post("/api/admin/tests/upload", tags=["Admin"])
|
||||
@@ -1332,7 +1343,7 @@ async def upload_admin_test(
|
||||
exam_pattern: str = Form(...),
|
||||
time_limit: int = Form(...),
|
||||
scheduled_date: Optional[datetime] = Form(None),
|
||||
admin_id: UUID = Form(...),
|
||||
admin_id: Optional[UUID] = Form(None),
|
||||
session: Session = Depends(get_session),
|
||||
current_user: models.User = Depends(get_current_user),
|
||||
):
|
||||
@@ -1344,6 +1355,24 @@ async def upload_admin_test(
|
||||
import storage
|
||||
import uuid
|
||||
|
||||
# Safe pattern normalization
|
||||
pat_str = str(exam_pattern).strip()
|
||||
if pat_str in ["JEE", "JEE_MAIN", "JEE Main", "JEE_Main"]:
|
||||
pat_enum = models.ExamEnum.JEE_MAIN
|
||||
elif pat_str in ["JEE_ADVANCED", "JEE Advanced", "JEE_Advanced"]:
|
||||
pat_enum = models.ExamEnum.JEE_ADVANCED
|
||||
elif pat_str in ["NEET"]:
|
||||
pat_enum = models.ExamEnum.NEET
|
||||
elif pat_str in ["JEE_NEET", "JEE & NEET"]:
|
||||
pat_enum = models.ExamEnum.JEE_NEET
|
||||
else:
|
||||
try:
|
||||
pat_enum = models.ExamEnum(pat_str)
|
||||
except ValueError:
|
||||
pat_enum = models.ExamEnum.JEE_MAIN
|
||||
|
||||
effective_admin_id = admin_id or uuid.UUID("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa")
|
||||
|
||||
pdf_bytes = await file.read()
|
||||
current_year = datetime.utcnow().year
|
||||
object_name = f"tests/{current_year}/{uuid.uuid4()}_{file.filename}"
|
||||
@@ -1354,28 +1383,19 @@ async def upload_admin_test(
|
||||
test_obj = models.Test(
|
||||
title=title,
|
||||
type="DPS",
|
||||
exam_pattern=models.ExamEnum(exam_pattern),
|
||||
exam_pattern=pat_enum,
|
||||
sections={},
|
||||
total_time=time_limit,
|
||||
difficulty="Medium",
|
||||
scheduled_date=scheduled_date,
|
||||
is_active=False,
|
||||
created_by=admin_id,
|
||||
is_active=True,
|
||||
created_by=effective_admin_id,
|
||||
pdf_url=url
|
||||
)
|
||||
session.add(test_obj)
|
||||
session.commit()
|
||||
session.refresh(test_obj)
|
||||
|
||||
# Phase 2: Uncomment below to trigger AI question extraction from PDF
|
||||
# from worker import process_pdf_test
|
||||
# process_pdf_test.delay(
|
||||
# test_id_str=str(test_obj.id),
|
||||
# object_name=object_name,
|
||||
# admin_id_str=str(admin_id),
|
||||
# exam_pattern=exam_pattern
|
||||
# )
|
||||
|
||||
return {
|
||||
"status": "uploaded",
|
||||
"test_id": test_obj.id,
|
||||
@@ -1388,18 +1408,37 @@ def create_manual_test(
|
||||
session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)
|
||||
):
|
||||
import uuid
|
||||
|
||||
# Safe pattern normalization
|
||||
pat_str = str(data.exam_pattern).strip()
|
||||
if pat_str in ["JEE", "JEE_MAIN", "JEE Main", "JEE_Main"]:
|
||||
pat_enum = models.ExamEnum.JEE_MAIN
|
||||
elif pat_str in ["JEE_ADVANCED", "JEE Advanced", "JEE_Advanced"]:
|
||||
pat_enum = models.ExamEnum.JEE_ADVANCED
|
||||
elif pat_str in ["NEET"]:
|
||||
pat_enum = models.ExamEnum.NEET
|
||||
elif pat_str in ["JEE_NEET", "JEE & NEET"]:
|
||||
pat_enum = models.ExamEnum.JEE_NEET
|
||||
else:
|
||||
try:
|
||||
pat_enum = models.ExamEnum(pat_str)
|
||||
except ValueError:
|
||||
pat_enum = models.ExamEnum.JEE_MAIN
|
||||
|
||||
effective_admin_id = data.admin_id or uuid.UUID("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa")
|
||||
|
||||
# 1. Create the test object
|
||||
test_obj = models.Test(
|
||||
title=data.title,
|
||||
type=data.type,
|
||||
exam_pattern=models.ExamEnum(data.exam_pattern),
|
||||
exam_pattern=pat_enum,
|
||||
sections={},
|
||||
total_time=data.time_limit,
|
||||
difficulty="Medium",
|
||||
scheduled_date=data.scheduled_date,
|
||||
end_date=data.end_date,
|
||||
is_active=True,
|
||||
created_by=data.admin_id
|
||||
created_by=effective_admin_id
|
||||
)
|
||||
session.add(test_obj)
|
||||
|
||||
@@ -1407,22 +1446,67 @@ def create_manual_test(
|
||||
sections_dict = {}
|
||||
|
||||
for q_data in data.questions_json:
|
||||
# Extract text with fallbacks
|
||||
q_text = q_data.question_text or q_data.question or "Question"
|
||||
sol_text = q_data.solution_text or q_data.solution or q_data.explanation or "Detailed solution will be available after review."
|
||||
subj = q_data.subject or "General"
|
||||
top = q_data.topic or subj
|
||||
subtop = q_data.subtopic or top
|
||||
diff = q_data.difficulty or "Medium"
|
||||
|
||||
# Normalize question type safely
|
||||
t_raw = str(q_data.type or "MCQ_Single").strip()
|
||||
if t_raw in ["Single_Choice", "MCQ", "Single", "MCQ_Single"]:
|
||||
q_type = models.QuestionTypeEnum.MCQ_SINGLE
|
||||
elif t_raw in ["Multiple_Choice", "Multiple", "MCQ_Multiple"]:
|
||||
q_type = models.QuestionTypeEnum.MCQ_MULTIPLE
|
||||
elif t_raw in ["Numerical", "Subjective", "Integer"]:
|
||||
q_type = models.QuestionTypeEnum.NUMERICAL
|
||||
else:
|
||||
try:
|
||||
q_type = models.QuestionTypeEnum(t_raw)
|
||||
except ValueError:
|
||||
q_type = models.QuestionTypeEnum.MCQ_SINGLE
|
||||
|
||||
# Normalize language
|
||||
l_raw = str(q_data.language or "English").strip().capitalize()
|
||||
if l_raw not in ["English", "Hindi", "Bilingual"]:
|
||||
l_raw = "English"
|
||||
lang_enum = models.MediumEnum(l_raw)
|
||||
|
||||
# Normalize options
|
||||
opts = q_data.options
|
||||
if isinstance(opts, list):
|
||||
opts = {chr(65 + i): str(v) for i, v in enumerate(opts)}
|
||||
|
||||
opts_hi = q_data.options_hi
|
||||
if isinstance(opts_hi, list):
|
||||
opts_hi = {chr(65 + i): str(v) for i, v in enumerate(opts_hi)}
|
||||
|
||||
# Normalize correct_answer
|
||||
ca = q_data.correct_answer
|
||||
if not isinstance(ca, dict):
|
||||
ca = {"value": ca if ca is not None else "A"}
|
||||
elif "value" not in ca:
|
||||
ca = {"value": ca.get("correct") or ca.get("answer") or "A"}
|
||||
|
||||
q_obj = models.Question(
|
||||
subject=q_data.subject,
|
||||
topic=q_data.topic,
|
||||
subtopic=q_data.subtopic,
|
||||
difficulty=q_data.difficulty,
|
||||
type=models.QuestionTypeEnum(q_data.type),
|
||||
question_text=q_data.question_text,
|
||||
options=q_data.options,
|
||||
correct_answer=q_data.correct_answer,
|
||||
solution_text=q_data.solution_text,
|
||||
language=models.MediumEnum(q_data.language),
|
||||
created_by=data.admin_id
|
||||
subject=subj,
|
||||
topic=top,
|
||||
subtopic=subtop,
|
||||
difficulty=diff,
|
||||
type=q_type,
|
||||
question_text=q_text,
|
||||
options=opts,
|
||||
options_hi=opts_hi,
|
||||
correct_answer=ca,
|
||||
solution_text=sol_text,
|
||||
language=lang_enum,
|
||||
created_by=effective_admin_id
|
||||
)
|
||||
session.add(q_obj)
|
||||
|
||||
sec_name = q_data.section
|
||||
sec_name = q_data.section or subj or "General"
|
||||
if sec_name not in sections_dict:
|
||||
sections_dict[sec_name] = {
|
||||
"name": sec_name,
|
||||
|
||||
Reference in New Issue
Block a user