139 lines
4.9 KiB
Python
139 lines
4.9 KiB
Python
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.NUMERICAL,
|
|
question_text=q_text,
|
|
options=None,
|
|
correct_answer={"value": "Subjective input"},
|
|
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())
|