leocrm/scripts/seed_admin.py

#!/usr/bin/env python3
"""Seed a default tenant and admin user for LeoCRM.

Usage: python scripts/seed_admin.py

Creates:
  - Tenant: "Default Org" (slug: default)
  - Admin user: admin@media-on.de / Admin123!

If tenant or user already exists, skips creation.
"""

import asyncio
import sys
import os

# Ensure app is importable
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from app.core.db import get_engine, close_engine
from app.core.auth import hash_password
from app.models.tenant import Tenant
from app.models.user import User, UserTenant
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.ext.asyncio import async_sessionmaker


async def seed():
    engine = get_engine()
    async_session = async_sessionmaker(engine, expire_on_commit=False)

    async with async_session() as db:  # type: AsyncSession
        # Check if default tenant exists
        result = await db.execute(select(Tenant).where(Tenant.slug == "default"))
        tenant = result.scalar_one_or_none()

        if tenant is None:
            tenant = Tenant(name="Default Org", slug="default")
            db.add(tenant)
            await db.flush()
            print(f"Created tenant: {tenant.name} (slug: {tenant.slug}, id: {tenant.id})")
        else:
            print(f"Tenant exists: {tenant.name} (id: {tenant.id})")

        # Check if admin user exists
        result = await db.execute(select(User).where(User.email == "admin@media-on.de"))
        user = result.scalar_one_or_none()

        if user is None:
            user = User(
                tenant_id=tenant.id,
                email="admin@media-on.de",
                name="Administrator",
                password_hash=hash_password("Admin123!"),
                role="admin",
                is_active=True,
                preferences={},
            )
            db.add(user)
            await db.flush()

            # Link user to tenant
            ut = UserTenant(user_id=user.id, tenant_id=tenant.id, is_default=True)
            db.add(ut)
            await db.flush()
            print(f"Created admin user: {user.email} (id: {user.id})")
        else:
            print(f"Admin user exists: {user.email} (id: {user.id})")

        await db.commit()
        print("\nSeed complete!")
        print(f"  Login URL: https://crm.media-on.de/login")
        print(f"  Email:     admin@media-on.de")
        print(f"  Password:   Admin123!")

    await close_engine()


if __name__ == "__main__":
    asyncio.run(seed())