242 lines
7.5 KiB
Python
242 lines
7.5 KiB
Python
"""
|
|
Script de datos iniciales (seed) para ServiceManagerWeb.
|
|
Crea categorías, sistemas, y usuarios de prueba en el tenant aduanasoft-demo.
|
|
Ejecutar desde dentro del contenedor: python /scripts/seed_data.py
|
|
"""
|
|
import asyncio
|
|
import sys
|
|
|
|
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
|
|
from sqlalchemy.orm import sessionmaker
|
|
from sqlalchemy import select
|
|
|
|
# Configurar path para importar el app
|
|
sys.path.insert(0, '/app')
|
|
|
|
from app.core.config import get_settings
|
|
|
|
settings = get_settings()
|
|
from app.core.security import security
|
|
from app.models.tenant import Tenant
|
|
from app.models.category import Category
|
|
from app.models.system import System
|
|
from app.models.user import User, UserRole
|
|
|
|
TENANT_SLUG = "aduanasoft-demo"
|
|
|
|
CATEGORIES = [
|
|
{
|
|
"name": "Soporte Técnico",
|
|
"description": "Problemas técnicos con sistemas y aplicaciones empresariales",
|
|
"color": "#EF4444",
|
|
"sla_response_hours": 2,
|
|
"sla_resolution_hours": 24,
|
|
"is_active": True,
|
|
},
|
|
{
|
|
"name": "Facturación",
|
|
"description": "Consultas y problemas relacionados con facturación y pagos",
|
|
"color": "#F59E0B",
|
|
"sla_response_hours": 4,
|
|
"sla_resolution_hours": 48,
|
|
"is_active": True,
|
|
},
|
|
{
|
|
"name": "Incidentes Críticos",
|
|
"description": "Fallas graves que afectan la operación del negocio",
|
|
"color": "#DC2626",
|
|
"sla_response_hours": 1,
|
|
"sla_resolution_hours": 8,
|
|
"is_active": True,
|
|
},
|
|
{
|
|
"name": "Consultas Generales",
|
|
"description": "Preguntas generales sobre productos y servicios",
|
|
"color": "#3B82F6",
|
|
"sla_response_hours": 8,
|
|
"sla_resolution_hours": 72,
|
|
"is_active": True,
|
|
},
|
|
{
|
|
"name": "Capacitación",
|
|
"description": "Solicitudes de entrenamiento y capacitación en sistemas",
|
|
"color": "#8B5CF6",
|
|
"sla_response_hours": 24,
|
|
"sla_resolution_hours": 96,
|
|
"is_active": True,
|
|
},
|
|
{
|
|
"name": "Infraestructura",
|
|
"description": "Problemas de red, servidores y componentes de infraestructura",
|
|
"color": "#06B6D4",
|
|
"sla_response_hours": 2,
|
|
"sla_resolution_hours": 16,
|
|
"is_active": True,
|
|
},
|
|
]
|
|
|
|
SYSTEMS = [
|
|
{
|
|
"name": "ERP Aduanero",
|
|
"description": "Sistema principal de gestión aduanera y comercio exterior",
|
|
"is_active": True,
|
|
},
|
|
{
|
|
"name": "Portal Web",
|
|
"description": "Portal de autogestión y consultas en línea para clientes",
|
|
"is_active": True,
|
|
},
|
|
{
|
|
"name": "Gestión Documental",
|
|
"description": "Sistema de administración y archivo de documentos aduaneros",
|
|
"is_active": True,
|
|
},
|
|
{
|
|
"name": "App Móvil",
|
|
"description": "Aplicación móvil para seguimiento de trámites en tiempo real",
|
|
"is_active": True,
|
|
},
|
|
{
|
|
"name": "Reportes y BI",
|
|
"description": "Plataforma de inteligencia de negocio y generación de reportes",
|
|
"is_active": True,
|
|
},
|
|
]
|
|
|
|
USERS = [
|
|
{
|
|
"email": "agente@aduanasoft.com",
|
|
"password": "agente123",
|
|
"first_name": "Carlos",
|
|
"last_name": "Agente",
|
|
"role": UserRole.AGENT,
|
|
"is_active": True,
|
|
},
|
|
{
|
|
"email": "manager@aduanasoft.com",
|
|
"password": "manager123",
|
|
"first_name": "Laura",
|
|
"last_name": "Gerente",
|
|
"role": UserRole.SUPPORT_MANAGER,
|
|
"is_active": True,
|
|
},
|
|
{
|
|
"email": "cliente@empresa-demo.com",
|
|
"password": "cliente123",
|
|
"first_name": "Roberto",
|
|
"last_name": "Cliente",
|
|
"role": UserRole.CLIENT_USER,
|
|
"is_active": True,
|
|
},
|
|
{
|
|
"email": "admin-cliente@empresa-demo.com",
|
|
"password": "clienteadmin123",
|
|
"first_name": "Ana",
|
|
"last_name": "Admin",
|
|
"role": UserRole.CLIENT_ADMIN,
|
|
"is_active": True,
|
|
},
|
|
]
|
|
|
|
|
|
async def main():
|
|
engine = create_async_engine(settings.DATABASE_URL, echo=False)
|
|
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
|
|
|
|
async with async_session() as session:
|
|
# 1. Obtener tenant
|
|
result = await session.execute(
|
|
select(Tenant).where(Tenant.slug == TENANT_SLUG)
|
|
)
|
|
tenant = result.scalar_one_or_none()
|
|
|
|
if not tenant:
|
|
print(f"ERROR: Tenant '{TENANT_SLUG}' no encontrado. Ejecuta el seed de tenants primero.")
|
|
return
|
|
|
|
tenant_id = tenant.id
|
|
print(f"✅ Tenant: {tenant.name} (id={str(tenant_id)[:8]}...)")
|
|
|
|
# 2. Categorías
|
|
print("\n📁 Creando categorías:")
|
|
# Eliminar la categoría "Infraestructura" duplicada creada previamente si existe
|
|
existing_cats = (await session.execute(
|
|
select(Category).where(Category.tenant_id == tenant_id)
|
|
)).scalars().all()
|
|
existing_names = {c.name for c in existing_cats}
|
|
|
|
created = 0
|
|
for cat_data in CATEGORIES:
|
|
if cat_data["name"] in existing_names:
|
|
print(f" ⏭ Ya existe: {cat_data['name']}")
|
|
continue
|
|
cat = Category(
|
|
tenant_id=tenant_id,
|
|
**cat_data,
|
|
)
|
|
session.add(cat)
|
|
print(f" ✓ {cat_data['name']} (resp={cat_data['sla_response_hours']}h, resol={cat_data['sla_resolution_hours']}h)")
|
|
created += 1
|
|
|
|
await session.flush()
|
|
print(f" → {created} categorías creadas, {len(existing_names)} ya existían")
|
|
|
|
# 3. Sistemas
|
|
print("\n🖥 Creando sistemas:")
|
|
existing_sys = (await session.execute(
|
|
select(System).where(System.tenant_id == tenant_id)
|
|
)).scalars().all()
|
|
existing_sys_names = {s.name for s in existing_sys}
|
|
|
|
created_sys = 0
|
|
for sys_data in SYSTEMS:
|
|
if sys_data["name"] in existing_sys_names:
|
|
print(f" ⏭ Ya existe: {sys_data['name']}")
|
|
continue
|
|
sys_obj = System(
|
|
tenant_id=tenant_id,
|
|
**sys_data,
|
|
)
|
|
session.add(sys_obj)
|
|
print(f" ✓ {sys_data['name']}")
|
|
created_sys += 1
|
|
|
|
await session.flush()
|
|
print(f" → {created_sys} sistemas creados")
|
|
|
|
# 4. Usuarios
|
|
print("\n👤 Creando usuarios:")
|
|
existing_users = (await session.execute(
|
|
select(User).where(User.tenant_id == tenant_id)
|
|
)).scalars().all()
|
|
existing_emails = {u.email for u in existing_users}
|
|
|
|
created_users = 0
|
|
for user_data in USERS:
|
|
if user_data["email"] in existing_emails:
|
|
print(f" ⏭ Ya existe: {user_data['email']}")
|
|
continue
|
|
# Usar copia para no mutar el dict original (permite re-ejecutar el script)
|
|
ud = user_data.copy()
|
|
pwd = ud.pop("password")
|
|
hashed_pwd = security.hash_password(pwd)
|
|
user = User(
|
|
tenant_id=tenant_id,
|
|
password_hash=hashed_pwd,
|
|
email_verified=True, # Marcar como verificado para permitir login
|
|
**ud,
|
|
)
|
|
session.add(user)
|
|
print(f" ✓ {ud['email']} [{ud['role'].value}] pwd={pwd}")
|
|
created_users += 1
|
|
|
|
await session.commit()
|
|
print(f" → {created_users} usuarios creados")
|
|
|
|
await engine.dispose()
|
|
print("\n🎉 Seed completado exitosamente.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|