proyecto-opendata/internal/scripts/seed_admin.py

#!/usr/bin/env python3
"""
Crea el primer usuario superadmin en la base de datos interna.

Uso (desde internal/):
    python scripts/seed_admin.py

Requiere que las variables de entorno estén configuradas (lee .env automáticamente).
La base de datos PostgreSQL debe estar corriendo.
"""

import sys
import os

# Permite ejecutar el script desde internal/ o desde la raíz
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

# Lee el .env manualmente para conexión local (sin depender de python-dotenv)
def _read_env(path: str) -> dict[str, str]:
    result: dict[str, str] = {}
    if os.path.exists(path):
        with open(path) as f:
            for line in f:
                line = line.strip()
                if line and not line.startswith("#") and "=" in line:
                    key, _, val = line.partition("=")
                    result[key.strip()] = val.strip()
    return result

# Lee el .env raíz del proyecto
_here = os.path.dirname(os.path.abspath(__file__))
env = _read_env(os.path.join(_here, "..", "..", ".env"))

pg_user = env.get("INTERNAL_DB_USER", "internal-database-user")
pg_pass = env.get("INTERNAL_DB_PASSWORD", "")
pg_db   = env.get("INTERNAL_DB_NAME", "internal-database")
pg_port = env.get("INTERNAL_DB_PORT", "5555")
jwt_key = env.get("JWT_SECRET_KEY", "seed-only")
plat_user = env.get("PLATFORM_DB_USER", "x")
plat_pass = env.get("PLATFORM_DB_PASSWORD", "x")
plat_db   = env.get("PLATFORM_DB_NAME", "x")
plat_port = env.get("PLATFORM_DB_PORT", "55557")

os.environ.setdefault("INTERNAL_DB_URL", f"postgresql://{pg_user}:{pg_pass}@localhost:{pg_port}/{pg_db}")
os.environ.setdefault("PLATFORM_DB_URL", f"postgresql://{plat_user}:{plat_pass}@localhost:{plat_port}/{plat_db}")
os.environ.setdefault("JWT_SECRET_KEY", jwt_key)

from src.backend.core.database import internal_engine, InternalSession  # noqa: E402
from src.backend.core.security import hash_password  # noqa: E402
from src.backend.db_models.internal import InternalBase, LogAdmin, RolAdmin, UsuarioAdmin  # noqa: E402


def seed():
    print("\n=== Seed: Administrador Interno ===\n")

    # Crea las tablas si no existen
    InternalBase.metadata.create_all(bind=internal_engine)
    print("✓ Esquema verificado")

    db = InternalSession()
    try:
        # Roles
        roles_existentes = db.query(RolAdmin).count()
        if roles_existentes == 0:
            for nombre in ("superadmin", "admin", "moderador", "viewer"):
                db.add(RolAdmin(nombre=nombre))
            db.commit()
            print("✓ Roles creados: superadmin, admin, moderador, viewer")
        else:
            print(f"  Roles ya existentes ({roles_existentes})")

        # Superadmin
        username = input("\nUsername [admin]: ").strip() or "admin"
        existing = db.query(UsuarioAdmin).filter(UsuarioAdmin.username == username).first()
        if existing:
            print(f"\n⚠ El usuario '{username}' ya existe. Abortando.")
            return

        email = input("Email: ").strip()
        while not email:
            email = input("Email (requerido): ").strip()

        password = input("Contraseña: ").strip()
        while len(password) < 8:
            password = input("Contraseña (mínimo 8 caracteres): ").strip()

        nombre = input("Nombre completo [Administrador]: ").strip() or "Administrador"

        rol = db.query(RolAdmin).filter(RolAdmin.nombre == "superadmin").first()

        user = UsuarioAdmin(
            username=username,
            email=email,
            password_hash=hash_password(password),
            nombre=nombre,
            id_rol=rol.id_rol,
        )
        db.add(user)
        db.commit()
        db.refresh(user)

        db.add(LogAdmin(
            id_usuario=user.id_usuario,
            accion="SEED",
            descripcion="Usuario superadmin creado via seed script",
        ))
        db.commit()

        print(f"\n✓ Usuario '{username}' ({email}) creado con rol superadmin")
        print("  Ya puedes hacer login en POST /auth/login\n")

    except Exception as e:
        db.rollback()
        print(f"\n✗ Error: {e}")
        raise
    finally:
        db.close()


if __name__ == "__main__":
    seed()