Files
pragativeer-v2/backend/seed.py

318 lines
13 KiB
Python

from sqlmodel import Session, create_engine, select
from uuid import uuid4, UUID
from datetime import datetime, timedelta, date
import os
import json
import glob
from database import engine, init_db
import models
import bcrypt
DEFAULT_SEED_HASH = bcrypt.hashpw("Password123!".encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
def seed_local_database():
# Ensure tables are created
init_db()
has_data = False
with Session(engine) as session:
# Prevent double seeding in local testing environment
if session.exec(select(models.Question)).first() is not None:
has_data = True
if has_data:
print("Local environment database already contains data components. Dropping all and reseeding...")
import sqlmodel
sqlmodel.SQLModel.metadata.drop_all(engine)
init_db()
print("Seeding baseline content entities...")
with Session(engine) as session:
# 1. Super Admin User
admin_id = UUID("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa")
admin_user = models.User(
id=admin_id,
phone="9999999998",
email="admin@pragativeer.com",
password_hash=bcrypt.hashpw("admin123".encode('utf-8'), bcrypt.gensalt()).decode('utf-8'),
name="Super Admin",
student_class=models.ClassEnum.TWELFTH,
medium=models.MediumEnum.ENGLISH,
target_exam=models.ExamEnum.JEE_MAIN,
city="New Delhi",
state="Delhi",
tier=models.TierEnum.AI_COACH,
streak_count=0,
last_active=datetime.utcnow()
)
session.add(admin_user)
# 2. AI Coach Demo Student User
student_id = UUID("b0e17378-864d-439e-8806-708dc48a891b")
student_user = models.User(
id=student_id,
phone="9999999999",
email="demo@pragativeer.com",
password_hash=bcrypt.hashpw("Pv9#kL2$mQ8!vX4@".encode('utf-8'), bcrypt.gensalt()).decode('utf-8'),
name="Rahul Sharma (Demo AI Coach)",
student_class=models.ClassEnum.TWELFTH,
medium=models.MediumEnum.ENGLISH,
target_exam=models.ExamEnum.JEE_MAIN,
school="Delhi Public School, R.K. Puram",
city="New Delhi",
state="Delhi",
tier=models.TierEnum.AI_COACH,
trial_end=None,
subscription_end=datetime.utcnow() + timedelta(days=345),
streak_count=6,
longest_streak=6,
last_active=datetime.utcnow()
)
session.add(student_user)
# 3. Neha Verma (AI Coach demo student)
neha_id = UUID("33333333-3333-3333-3333-333333333333")
neha_user = models.User(
id=neha_id,
phone="9999999997",
email="neha.demo@pragativeer.com",
password_hash=bcrypt.hashpw("Pv9#kL2$mQ8!vX4@".encode('utf-8'), bcrypt.gensalt()).decode('utf-8'),
name="Neha Verma",
student_class=models.ClassEnum.TWELFTH,
medium=models.MediumEnum.ENGLISH,
target_exam=models.ExamEnum.JEE_MAIN,
school="Modern School, Barakhamba Road",
city="New Delhi",
state="Delhi",
tier=models.TierEnum.AI_COACH,
trial_end=None,
subscription_end=datetime.utcnow() + timedelta(days=345),
streak_count=15,
longest_streak=25,
last_active=datetime.utcnow()
)
session.add(neha_user)
# 4. Priya Singh (Pro Tier demo student)
priya_id = UUID("22222222-2222-2222-2222-222222222222")
priya_user = models.User(
id=priya_id,
phone="9999999996",
email="priya.demo@pragativeer.com",
password_hash=bcrypt.hashpw("Pv9#kL2$mQ8!vX4@".encode('utf-8'), bcrypt.gensalt()).decode('utf-8'),
name="Priya Singh",
student_class=models.ClassEnum.TWELFTH,
medium=models.MediumEnum.ENGLISH,
target_exam=models.ExamEnum.NEET,
school="City Montessori School",
city="Lucknow",
state="Uttar Pradesh",
tier=models.TierEnum.PRO,
trial_end=None,
subscription_end=datetime.utcnow() + timedelta(days=345),
streak_count=42,
longest_streak=42,
last_active=datetime.utcnow()
)
session.add(priya_user)
# 5. Amit Kumar (Free Tier demo student)
amit_id = UUID("11111111-1111-1111-1111-111111111111")
amit_user = models.User(
id=amit_id,
phone="9999999995",
email="amit.demo@pragativeer.com",
password_hash=bcrypt.hashpw("Pv9#kL2$mQ8!vX4@".encode('utf-8'), bcrypt.gensalt()).decode('utf-8'),
name="Amit Kumar",
student_class=models.ClassEnum.ELEVENTH,
medium=models.MediumEnum.ENGLISH,
target_exam=models.ExamEnum.JEE_MAIN,
school="Loyola High School",
city="Patna",
state="Bihar",
tier=models.TierEnum.FREE,
trial_end=None,
streak_count=38,
longest_streak=38,
last_active=datetime.utcnow()
)
session.add(amit_user)
# Flush users first so FK constraints for subscription/transaction are satisfied
session.flush()
# Seed Demo Student Subscription records
for uid, tier in [(student_id, models.TierEnum.AI_COACH), (neha_id, models.TierEnum.AI_COACH), (priya_id, models.TierEnum.PRO)]:
demo_subscription = models.Subscription(
id=uuid4(),
user_id=uid,
tier=tier,
billing_cycle="Annual",
start_date=datetime.utcnow() - timedelta(days=20),
end_date=datetime.utcnow() + timedelta(days=345),
amount=1199900,
payment_id=f"PAY_DEMO_{str(uid)[:6]}",
status="Active",
auto_renew=True,
)
session.add(demo_subscription)
# Seed Demo Student Transaction record
demo_txn = models.Transaction(
id=uuid4(),
user_id=student_id,
amount=11999.0,
tier="Al_Coach",
status="Success",
created_at=datetime.utcnow() - timedelta(days=20),
)
session.add(demo_txn)
# admin_id is already defined above as the fixed UUID for the admin user
jee_test_id = UUID("b6faa295-3bf5-4de2-a587-d72b889629c4")
neet_test_id = UUID("e7c80521-825b-4c07-889a-ee646c05d762")
# --- MOCK TESTS from admin_uploads JSON files ---
print("Seeding Mock tests from admin_uploads...")
_script_dir = os.path.dirname(os.path.abspath(__file__))
_docker_path = os.path.join(_script_dir, "admin_uploads")
_local_path = os.path.join(_script_dir, "..", "admin_uploads")
mock_json_dir = _docker_path if os.path.isdir(_docker_path) else _local_path
mock_json_files = sorted(glob.glob(os.path.join(mock_json_dir, "mock_*.json")))
for json_path in mock_json_files:
with open(json_path, "r", encoding="utf-8") as f:
raw_questions = json.load(f)
is_neet = "neet" in json_path.lower()
pattern = models.ExamEnum.NEET if is_neet else models.ExamEnum.JEE_MAIN
title = "NEET UG Full Mock Test (2025)" if is_neet else "JEE Advanced Paper 1 (2025)"
sections = {}
for raw_q in raw_questions:
subject = raw_q.get("subject", "General")
if subject not in sections:
sections[subject] = {"name": subject, "questions": [], "time_limit": 45 if is_neet else 60}
q = models.Question(
id=uuid4(),
subject=subject,
topic=raw_q.get("topic", "Mixed"),
subtopic=raw_q.get("subtopic", "Mixed"),
difficulty=raw_q.get("difficulty", "Medium"),
type=models.QuestionTypeEnum(raw_q.get("type", "MCQ_Single")),
question_text=raw_q.get("question_text", ""),
options=raw_q.get("options"),
correct_answer=raw_q.get("correct_answer"),
solution_text=raw_q.get("solution_text", ""),
language=models.MediumEnum.ENGLISH,
exam=pattern,
created_by=admin_id
)
session.add(q)
sections[subject]["questions"].append(str(q.id))
mock_test = models.Test(
id=neet_test_id if is_neet else jee_test_id,
title=title,
type="Mock",
exam_pattern=pattern,
sections=sections,
total_time=180,
difficulty="Mixed",
scheduled_date=datetime.utcnow(),
is_active=True,
created_by=admin_id
)
session.add(mock_test)
session.commit()
# --- DPS (Daily Practice Sheets) from admin_uploads JSON files ---
print("Seeding DPS tests from admin_uploads...")
# Resolve path: works inside Docker (/app/admin_uploads) and locally (../../admin_uploads)
_script_dir = os.path.dirname(os.path.abspath(__file__))
_docker_path = os.path.join(_script_dir, "admin_uploads")
_local_path = os.path.join(_script_dir, "..", "admin_uploads")
dps_json_dir = _docker_path if os.path.isdir(_docker_path) else _local_path
dps_json_files = sorted(glob.glob(os.path.join(dps_json_dir, "dps_*.json")))
with Session(engine) as dps_session:
for i, json_path in enumerate(dps_json_files):
with open(json_path, "r", encoding="utf-8") as f:
raw_questions = json.load(f)
subject = raw_questions[0].get("subject", "General") if raw_questions else "General"
topic = raw_questions[0].get("topic", "Chapter 4") if raw_questions else "Chapter 4"
sheet_title = f"{subject} DPS"
sheet_date = date.today() - timedelta(days=i)
is_neet = any(s in subject.lower() for s in ["botany", "zoology", "bio", "biology"])
dps_pattern = models.ExamEnum.NEET if is_neet else models.ExamEnum.JEE_MAIN
# Create Question records from the JSON
q_ids = []
for raw_q in raw_questions:
q = models.Question(
id=uuid4(),
subject=raw_q.get("subject", subject),
topic=raw_q.get("topic", topic),
subtopic=raw_q.get("subtopic", "Mixed"),
difficulty=raw_q.get("difficulty", "Medium"),
type=models.QuestionTypeEnum(raw_q.get("type", "MCQ_Single")),
question_text=raw_q.get("question_text", "Sample question"),
options=raw_q.get("options"),
correct_answer=raw_q.get("correct_answer", {"value": "A"}),
solution_text=raw_q.get("solution_text", "Refer to standard solution."),
language=models.MediumEnum.ENGLISH,
exam=dps_pattern,
created_by=admin_id,
)
dps_session.add(q)
q_ids.append(str(q.id))
dps_id = uuid4()
dps = models.DPS(
id=dps_id,
date=sheet_date,
title=sheet_title,
questions=q_ids,
difficulty="Medium",
time_limit=30,
is_active=True,
created_by=admin_id,
)
dps_session.add(dps)
# Also create as Test(type='DPS') so /api/tests?type=DPS and CBT engine can run it
dps_test = models.Test(
id=dps_id,
title=sheet_title,
type="DPS",
exam_pattern=dps_pattern,
sections={"Practice": {"questions": q_ids, "time_limit": 30}},
total_time=30,
difficulty="Medium",
scheduled_date=datetime.combine(sheet_date, datetime.min.time()),
is_active=True,
created_by=admin_id,
)
dps_session.add(dps_test)
print(f" ✓ DPS & CBT Test seeded: {sheet_title} ({len(q_ids)} questions)")
dps_session.commit()
print("\n=== LOCAL SEED COMPLETED SUCCESSFULLY ===")
print(f"JEE Test ID: {jee_test_id}")
print(f"NEET Test ID: {neet_test_id}")
print("DPS Sheets: seeded from admin_uploads/\n")
if __name__ == "__main__":
seed_local_database()