| import sys |
| import os |
| from datetime import datetime |
|
|
| |
| sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) |
|
|
| from app.db.session import SessionLocal, engine |
| from app.models.base import Base |
| from app.models.assessment import AssessmentItem, AssessmentDimension |
| from app.models.auth import Organization, Role, Permission, User |
| from app.models.child import Child, GenderEnum |
| from app.models.tenant import Tenant |
| from app.core.security import get_password_hash |
|
|
|
|
| def seed_data(): |
| """Seed demo data: organization, admin user, demo child, and 60 assessment items.""" |
| Base.metadata.create_all(bind=engine) |
| db = SessionLocal() |
|
|
| try: |
| |
| org = db.query(Organization).filter(Organization.code == "HEADQUARTERS").first() |
| if not org: |
| org = Organization(name="LearningOS HQ", code="HEADQUARTERS", enabled=True) |
| db.add(org) |
| db.flush() |
|
|
| |
| permissions_data = [ |
| ("child:view", "View Child Profile", "Child"), |
| ("child:create", "Create Child Profile", "Child"), |
| ("assessment:start", "Start Assessment", "Assessment"), |
| ("report:view", "View AI Report", "Report"), |
| ] |
| perms = [] |
| for code, name, module in permissions_data: |
| p = db.query(Permission).filter(Permission.code == code).first() |
| if not p: |
| p = Permission(code=code, name=name, module=module) |
| db.add(p) |
| db.flush() |
| perms.append(p) |
|
|
| |
| role = db.query(Role).filter(Role.name == "SUPER_ADMIN").first() |
| if not role: |
| role = Role(name="SUPER_ADMIN", description="System Super Admin", permissions=perms) |
| db.add(role) |
| db.flush() |
|
|
| user = db.query(User).filter(User.username == "admin").first() |
| if not user: |
| user = User( |
| username="admin", |
| email="admin@learningos.org", |
| hashed_password=get_password_hash("Admin123456!"), |
| is_active=True, |
| is_superuser=True, |
| organization_id=org.id, |
| roles=[role], |
| ) |
| db.add(user) |
|
|
| |
| tenant = db.query(Tenant).filter(Tenant.code == "DEMO_HOSP").first() |
| if not tenant: |
| tenant = Tenant(name="Demo Hospital", code="DEMO_HOSP", enabled=True, quota_credits=1000) |
| db.add(tenant) |
| db.flush() |
|
|
| |
| child = db.query(Child).filter(Child.name == "Alex").first() |
| if not child: |
| child = Child( |
| archive_no="ARCH-001", |
| name="Alex", |
| gender=GenderEnum.MALE, |
| dob=datetime(2020, 5, 1).date(), |
| organization_id=org.id, |
| ) |
| db.add(child) |
| db.flush() |
|
|
| |
| item_count = db.query(AssessmentItem).count() |
| if item_count == 0: |
| sample_items = [] |
| for dim in AssessmentDimension: |
| difficulties = [-1.5, -0.8, 0.0, 0.8, 1.5] |
| for idx, diff in enumerate(difficulties): |
| item = AssessmentItem( |
| dimension=dim, |
| content={ |
| "title": f"{dim.value.capitalize()} Assessment Item #{idx+1}", |
| "question": f"Sample task evaluating {dim.value} at level {diff}", |
| "options": ["Option A", "Option B", "Option C", "Option D"], |
| "correct_answer": 0 |
| }, |
| a_parameter=1.2 + (idx * 0.1), |
| b_parameter=diff, |
| c_parameter=0.0 |
| ) |
| sample_items.append(item) |
|
|
| db.bulk_save_objects(sample_items) |
| db.commit() |
| print(f"Successfully seeded {len(sample_items)} assessment items across 12 dimensions.") |
| else: |
| print("Assessment items already present. Skipping item seeding.") |
|
|
| db.commit() |
| print("Seed data initialized successfully.") |
|
|
| except Exception as e: |
| db.rollback() |
| print(f"Error during database seeding: {e}") |
| raise e |
| finally: |
| db.close() |
|
|
|
|
| if __name__ == "__main__": |
| seed_data() |
|
|
