Feat: Add script to parse 4 real PDF DPPs into Database

backend/seed_from_dpps.py

@@ -0,0 +1,138 @@
+import asyncio
+import os
+import uuid
+import re
+import glob
+from sqlmodel import Session, select, delete
+from datetime import datetime
+
+import fitz  # PyMuPDF
+from database import engine, init_db
+import models
+
+def parse_pdf(filepath):
+    doc = fitz.open(filepath)
+    text = ""
+    for page in doc:
+        text += page.get_text() + "\n"
+        
+    q_block = text.split("Questions\n")[-1].split("Answer Key\n")[0]
+    q_pattern = re.compile(r'^(\d+)\.\s+(.*?)(?=^\d+\.\s+|\Z)', re.DOTALL | re.MULTILINE)
+    
+    questions_dict = {}
+    for match in q_pattern.finditer(q_block):
+        q_num = int(match.group(1))
+        if 1 <= q_num <= 15:
+            questions_dict[q_num] = match.group(2).strip()
+
+    answers_dict = {}
+    if "Answer Key\n" in text:
+        a_block = text.split("Answer Key\n")[-1]
+        a_pattern = re.compile(r'^(\d+)\s+(.*?)(?=^\d+\s+|\Z)', re.DOTALL | re.MULTILINE)
+        for match in a_pattern.finditer(a_block):
+            q_num = int(match.group(1))
+            if 1 <= q_num <= 15:
+                # Some answers might get overwritten if footer matches the same number, but 
+                # answers are strictly linear 1 to 15, so we only take the first match.
+                if q_num not in answers_dict:
+                    answers_dict[q_num] = match.group(2).strip()
+            
+    return questions_dict, answers_dict
+
+async def seed_pdfs():
+    admin_id_str = "b0e17378-864d-439e-8806-708dc48a891b"
+    pdf_files = glob.glob('/app/pdfs_to_parse/*.pdf')
+    
+    if not pdf_files:
+        print("No PDFs found in /app/pdfs_to_parse/")
+        return
+
+    with Session(engine) as session:
+        admin = session.exec(select(models.User).limit(1)).first()
+        admin_id = admin.id if admin else uuid.UUID(admin_id_str)
+
+        statement = select(models.Test).where(models.Test.type == "DPS")
+        old_tests = session.exec(statement).all()
+        for t in old_tests:
+            session.delete(t)
+        session.commit()
+
+        for pdf_path in pdf_files:
+            filename = os.path.basename(pdf_path)
+            subject = "Physics"
+            topic = "General"
+            if "phy" in filename.lower():
+                subject = "Physics"
+                topic = "Laws of Motion"
+            elif "che" in filename.lower():
+                subject = "Chemistry"
+                topic = "Chemical Bonding"
+            elif "mth" in filename.lower():
+                subject = "Mathematics"
+                topic = "Calculus"
+            elif "bio" in filename.lower():
+                subject = "Biology"
+                topic = "Genetics"
+
+            print(f"Parsing {filename}...")
+            q_dict, a_dict = parse_pdf(pdf_path)
+            print(f"Found {len(q_dict)} questions and {len(a_dict)} answers.")
+
+            if not q_dict:
+                continue
+
+            question_ids = []
+            for i in range(1, 16):
+                if i not in q_dict:
+                    continue
+                q_text = q_dict[i].replace('\x00', '')
+                ans_text = a_dict.get(i, "Solution not provided in PDF.").replace('\x00', '')
+                
+                options = {
+                    "A": "Correct Answer - Refer to Solution Key below",
+                    "B": "Incorrect Placeholder",
+                    "C": "Incorrect Placeholder",
+                    "D": "Incorrect Placeholder"
+                }
+
+                new_q = models.Question(
+                    subject=subject,
+                    topic=topic,
+                    subtopic="General",
+                    difficulty="Medium",
+                    type=models.QuestionTypeEnum.MCQ_SINGLE,
+                    question_text=q_text,
+                    options=options,
+                    correct_answer={"value": "A"},
+                    solution_text=ans_text,
+                    language=models.MediumEnum.ENGLISH,
+                    exam=models.ExamEnum.JEE_MAIN,
+                    created_by=admin_id
+                )
+                session.add(new_q)
+                session.flush()
+                question_ids.append(str(new_q.id))
+                
+            if question_ids:
+                test_obj = models.Test(
+                    title=f"DPP: {subject} - {topic}",
+                    type="DPS",
+                    exam_pattern=models.ExamEnum.JEE_MAIN,
+                    sections={
+                        "Section 1": {
+                            "questions": question_ids,
+                            "time_limit": 45
+                        }
+                    },
+                    total_time=45,
+                    difficulty="Medium",
+                    scheduled_date=datetime.utcnow(),
+                    is_active=True,
+                    created_by=admin_id
+                )
+                session.add(test_obj)
+                session.commit()
+                print(f"Successfully created DPP Test: {test_obj.title} with {len(question_ids)} questions.")
+
+if __name__ == "__main__":
+    asyncio.run(seed_pdfs())