113 lines
4.4 KiB
Python
113 lines
4.4 KiB
Python
import asyncio
|
|
import os
|
|
import uuid
|
|
from sqlmodel import Session, select
|
|
from datetime import datetime
|
|
|
|
from database import engine, init_db
|
|
import models
|
|
import storage
|
|
from pdf_parser import parse_pdf_to_questions
|
|
|
|
async def seed_pdfs():
|
|
# Admin ID fallback
|
|
admin_id_str = "b0e17378-864d-439e-8806-708dc48a891b" # Or fetch first user
|
|
|
|
with Session(engine) as session:
|
|
admin = session.exec(select(models.User).limit(1)).first()
|
|
if admin:
|
|
admin_id = admin.id
|
|
else:
|
|
admin_id = uuid.UUID(admin_id_str)
|
|
|
|
# Paths to the 4 PDFs
|
|
base_dir = r"/app/pdfs"
|
|
pdf_files = [
|
|
"media_e3dd3511-9fc4-402b-8054-6a56f7ed8202_1782971559253.pdf",
|
|
"media_e3dd3511-9fc4-402b-8054-6a56f7ed8202_1782979477911.pdf",
|
|
"media_e3dd3511-9fc4-402b-8054-6a56f7ed8202_1783273583537.pdf",
|
|
"media_e3dd3511-9fc4-402b-8054-6a56f7ed8202_1783410386530.pdf"
|
|
]
|
|
|
|
subjects = ["Physics", "Chemistry", "Biology", "Mathematics"]
|
|
topics = ["Rotational Dynamics", "Chemical Kinetics", "Genetics", "Calculus"]
|
|
|
|
for idx, filename in enumerate(pdf_files):
|
|
file_path = os.path.join(base_dir, filename)
|
|
if not os.path.exists(file_path):
|
|
print(f"File not found: {file_path}")
|
|
continue
|
|
|
|
print(f"Processing {filename}...")
|
|
|
|
with open(file_path, "rb") as f:
|
|
pdf_bytes = f.read()
|
|
|
|
# 1. Upload to Minio
|
|
object_name = f"dpps/{uuid.uuid4()}_{filename}"
|
|
storage.upload_file(pdf_bytes, object_name, content_type="application/pdf")
|
|
|
|
# 2. Extract
|
|
questions_data = await parse_pdf_to_questions(pdf_bytes, "JEE_Main")
|
|
|
|
if not questions_data:
|
|
print(f"Warning: No questions extracted for {filename}. Using mock data fallback.")
|
|
questions_data = [
|
|
{
|
|
"subject": subjects[idx],
|
|
"topic": topics[idx],
|
|
"subtopic": "General",
|
|
"difficulty": "Hard",
|
|
"type": "MCQ_Single",
|
|
"question_text": f"What is the most fundamental principle of {topics[idx]}?",
|
|
"options": {"A": "Conservation of Energy", "B": "Newton's Second Law", "C": "Kinetic Theory", "D": "Thermodynamics"},
|
|
"correct_answer": {"value": "A"},
|
|
"solution_text": "By fundamental principles."
|
|
}
|
|
]
|
|
|
|
question_ids = []
|
|
for q_data in questions_data:
|
|
new_q = models.Question(
|
|
subject=q_data.get("subject", subjects[idx]),
|
|
topic=q_data.get("topic", topics[idx]),
|
|
subtopic=q_data.get("subtopic", "General"),
|
|
difficulty=q_data.get("difficulty", "Medium"),
|
|
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.JEE_MAIN,
|
|
created_by=admin_id
|
|
)
|
|
session.add(new_q)
|
|
session.flush()
|
|
question_ids.append(str(new_q.id))
|
|
|
|
# Create Test record
|
|
test_obj = models.Test(
|
|
title=f"AI DPP: {topics[idx]}",
|
|
type="DPS",
|
|
exam_pattern=models.ExamEnum.JEE_MAIN,
|
|
sections={
|
|
"Section 1": {
|
|
"questions": question_ids,
|
|
"time_limit": 60
|
|
}
|
|
},
|
|
total_time=60,
|
|
difficulty="Hard" if idx % 2 == 0 else "Medium",
|
|
scheduled_date=datetime.utcnow(),
|
|
is_active=True,
|
|
created_by=admin_id
|
|
)
|
|
session.add(test_obj)
|
|
session.commit()
|
|
print(f"Successfully created DPS Test: {test_obj.title} with {len(question_ids)} questions.")
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(seed_pdfs())
|
|
|