#!/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()