Feat: Complete CBT PDF upload, AI parsing, and MinIO storage integration

backend/main.py

@@ -1,4 +1,4 @@
-from fastapi import FastAPI, Depends, HTTPException, status, Request
+from fastapi import FastAPI, Depends, HTTPException, status, Request, UploadFile, File, Form, BackgroundTasks
 from fastapi.responses import StreamingResponse
 from fastapi.middleware.cors import CORSMiddleware
 from sqlmodel import Session, select
@@ -75,6 +75,13 @@ def verify_otp(phone_or_email: str, code: str, session: Session = Depends(get_se
     return {"status": "authenticated", "token": "mock_jwt_session_token_string"}
 
 # --- CBT TEST ENVIRONMENT CONTENT DELIVERY ENDPOINTS ---
+@app.get("/api/tests", response_model=List[models.Test], tags=["CBT Engine"])
+def list_active_tests(exam_pattern: Optional[str] = None, session: Session = Depends(get_session)):
+    query = select(models.Test).where(models.Test.is_active == True)
+    if exam_pattern:
+        query = query.where(models.Test.exam_pattern == exam_pattern)
+    return session.exec(query).all()
+
 
 @app.get("/api/tests/{test_id}", response_model=models.Test, tags=["CBT Engine"])
 def get_test_by_id(test_id: UUID, user_id: UUID, session: Session = Depends(get_session), current_user = Depends(require_tier([models.TierEnum.PRO, models.TierEnum.AI_COACH]))):
@@ -512,3 +519,69 @@ def get_user_attempts(user_id: UUID, session: Session = Depends(get_session)):
     query = select(models.Attempt).where(models.Attempt.user_id == user_id)
     return session.exec(query).all()
 
+
+
+# --- ADMIN ROUTES ---
+
+@app.get("/api/admin/tests", tags=["Admin"])
+def get_admin_tests(session: Session = Depends(get_session)):
+    tests = session.exec(select(models.Test)).all()
+    return tests
+
+@app.post("/api/admin/tests/upload", tags=["Admin"])
+async def upload_admin_test(
+    file: UploadFile = File(...),
+    title: str = Form(...),
+    exam_pattern: str = Form(...),
+    time_limit: int = Form(...),
+    scheduled_date: Optional[datetime] = Form(None),
+    admin_id: UUID = Form(...),
+    session: Session = Depends(get_session)
+):
+    import storage
+    import uuid
+    from worker import process_pdf_test
+    
+    pdf_bytes = await file.read()
+    object_name = f"tests/{uuid.uuid4()}_{file.filename}"
+    url = storage.upload_file(pdf_bytes, object_name, content_type="application/pdf")
+    if not url:
+        raise HTTPException(status_code=500, detail="Failed to upload PDF to MinIO")
+
+    # Create inactive test
+    test_obj = models.Test(
+        title=title,
+        type="Mock",
+        exam_pattern=models.ExamEnum(exam_pattern),
+        sections={},
+        total_time=time_limit,
+        difficulty="Medium",
+        scheduled_date=scheduled_date,
+        is_active=False,  # Inactive until processing finishes
+        created_by=admin_id
+    )
+    session.add(test_obj)
+    session.commit()
+    session.refresh(test_obj)
+
+    # Trigger async background worker
+    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": "processing", "test_id": test_obj.id, "message": "Test is being processed by AI"}
+
+@app.put("/api/admin/tests/{test_id}/toggle", tags=["Admin"])
+def toggle_test_status(test_id: UUID, session: Session = Depends(get_session)):
+    test_obj = session.get(models.Test, test_id)
+    if not test_obj:
+        raise HTTPException(status_code=404, detail="Test not found")
+        
+    test_obj.is_active = not test_obj.is_active
+    session.add(test_obj)
+    session.commit()
+    
+    return {"status": "success", "is_active": test_obj.is_active}

backend/pdf_parser.py

@@ -0,0 +1,85 @@
+import fitz  # PyMuPDF
+import json
+import logging
+import asyncio
+from typing import List, Dict, Any
+import os
+from openai import AsyncOpenAI
+
+logger = logging.getLogger("pdf_parser")
+
+client = AsyncOpenAI(
+    base_url=os.getenv("LLM_BASE_URL", "http://host.docker.internal:11434/v1"),
+    api_key=os.getenv("OPENAI_API_KEY", "ollama"),
+)
+
+async def parse_pdf_to_questions(pdf_bytes: bytes, exam_pattern: str) -> List[Dict[str, Any]]:
+    """
+    Extracts text from a PDF and uses the LLM to structure it into a JSON array of questions.
+    """
+    text = ""
+    try:
+        doc = fitz.open(stream=pdf_bytes, filetype="pdf")
+        for page in doc:
+            text += page.get_text("text") + "\n\n"
+    except Exception as e:
+        logger.error(f"Failed to parse PDF with PyMuPDF: {e}")
+        return []
+
+    system_prompt = f"""
+    You are an AI expert at extracting {exam_pattern} test questions from messy PDF text.
+    The text contains complex Chemistry and Math formulas. 
+    CRITICAL: You MUST format all formulas perfectly in LaTeX wrapping inline math in $ and block math in $$.
+    
+    Extract the questions and return them STRICTLY as a JSON array of objects. 
+    If your API requires a JSON object root, return {{"questions": [...]}}.
+    
+    Schema per question:
+    {{
+      "subject": "Physics" | "Chemistry" | "Maths" | "Biology",
+      "topic": "String",
+      "subtopic": "String",
+      "difficulty": "Easy" | "Medium" | "Hard",
+      "type": "MCQ_Single" | "MCQ_Multiple" | "Numerical",
+      "question_text": "Full text of the question with LaTeX",
+      "options": {{"A": "Option A text", "B": "Option B text", "C": "Option C text", "D": "Option D text"}},
+      "correct_answer": {{"value": "A"}},
+      "solution_text": "Detailed step-by-step solution if available, else 'Solution not provided'",
+      "language": "English"
+    }}
+    """
+    
+    try:
+        truncated_text = text[:15000] 
+        
+        response = await client.chat.completions.create(
+            model=os.getenv("LLM_MODEL_NAME", "gemma"),
+            messages=[
+                {"role": "system", "content": system_prompt},
+                {"role": "user", "content": truncated_text}
+            ],
+            response_format={ "type": "json_object" }
+        )
+        
+        output = response.choices[0].message.content
+        if not output:
+            return []
+            
+        if "```json" in output:
+            output = output.split("```json")[1].split("```")[0].strip()
+        elif "```" in output:
+            output = output.split("```")[1].split("```")[0].strip()
+            
+        data = json.loads(output)
+        
+        if isinstance(data, list):
+            return data
+        elif isinstance(data, dict) and "questions" in data:
+            return data["questions"]
+        elif isinstance(data, dict):
+            return [data]
+        else:
+            return []
+    except Exception as e:
+        logger.error(f"LLM PDF parsing failed: {e}")
+        return []

backend/requirements.txt

@@ -12,3 +12,4 @@ openai==1.14.0
 httpx==0.27.2
 boto3==1.34.0
 python-multipart==0.0.9
+PyMuPDF==1.23.26

backend/worker.py

@@ -4,9 +4,12 @@ from celery import Celery
 from celery.schedules import crontab
 from sqlmodel import Session, select
 import json
+import asyncio
 
 from database import engine
 import models
+import storage
+from pdf_parser import parse_pdf_to_questions
 
 REDIS_URL = os.getenv("REDIS_URL", "redis://redis:6379/0")
 
@@ -75,6 +78,76 @@ def generate_daily_study_plans(self):
         print(f"Database error during study plan generation: {exc}")
         raise self.retry(exc=exc)
 
+@celery_app.task(name="process_pdf_test", bind=True, max_retries=1)
+def process_pdf_test(self, test_id_str: str, object_name: str, admin_id_str: str, exam_pattern: str):
+    """
+    Background task to download PDF from MinIO, extract text, 
+    use LLM to generate questions, and attach them to a Test record.
+    """
+    try:
+        print(f"Starting PDF processing for test {test_id_str}")
+        
+        # 1. Download PDF from MinIO
+        pdf_bytes = storage.get_file(object_name)
+        if not pdf_bytes:
+            print("Failed to download PDF from MinIO")
+            return
+            
+        # 2. Extract Questions using LLM (Async)
+        questions_data = asyncio.run(parse_pdf_to_questions(pdf_bytes, exam_pattern))
+        if not questions_data:
+            print("LLM parsing returned empty data")
+            return
+            
+        import uuid
+        test_id = uuid.UUID(test_id_str)
+        admin_id = uuid.UUID(admin_id_str)
+        
+        with Session(engine) as session:
+            test_obj = session.get(models.Test, test_id)
+            if not test_obj:
+                print("Test record not found in DB")
+                return
+
+            question_ids = []
+            for q_data in questions_data:
+                # Create a Question DB object
+                new_q = models.Question(
+                    subject=q_data.get("subject", "Physics"),
+                    topic=q_data.get("topic", "General"),
+                    subtopic=q_data.get("subtopic", "General"),
+                    difficulty=q_data.get("difficulty", "Medium"),
+                    type=q_data.get("type", models.QuestionTypeEnum.MCQ_SINGLE),
+                    question_text=q_data.get("question_text", ""),
+                    options=q_data.get("options", {}),
+                    correct_answer=q_data.get("correct_answer", {}),
+                    solution_text=q_data.get("solution_text", ""),
+                    language=models.MediumEnum.ENGLISH,
+                    exam=models.ExamEnum(exam_pattern),
+                    created_by=admin_id
+                )
+                session.add(new_q)
+                session.flush() # To get the ID
+                question_ids.append(str(new_q.id))
+                
+            # Update Test record with the new sections
+            test_obj.sections = {
+                "Section 1": {
+                    "questions": question_ids,
+                    "time_limit": test_obj.total_time
+                }
+            }
+            # Automatically activate it when done processing
+            test_obj.is_active = True
+            
+            session.commit()
+            print(f"Successfully processed PDF and created {len(question_ids)} questions for test {test_id_str}")
+            return f"Processed {len(question_ids)} questions"
+    except Exception as exc:
+        print(f"Error processing PDF test: {exc}")
+        # Mark test as failed if we had a status field, for now just log it
+        raise self.retry(exc=exc)
+
 # Schedule the tasks using Celery Beat
 celery_app.conf.beat_schedule = {
     'generate-study-plans-every-morning': {