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()