#!/usr/bin/env python3 """ Database Utilities Script Herramientas administrativas para gestión de base de datos Uso: python scripts/db_utils.py list-users [--tenant-id UUID] python scripts/db_utils.py check-user EMAIL python scripts/db_utils.py reset-password EMAIL [--password PASSWORD] python scripts/db_utils.py list-tickets [--tenant-id UUID] [--limit N] python scripts/db_utils.py check-ticket TICKET_ID Ejemplos: python scripts/db_utils.py list-users python scripts/db_utils.py check-user admin@example.com python scripts/db_utils.py reset-password admin@example.com --password admin123 python scripts/db_utils.py list-tickets --limit 10 """ import asyncio import sys import os from typing import Optional import argparse from pathlib import Path # Agregar backend al path para imports backend_path = Path(__file__).parent.parent / "backend" sys.path.insert(0, str(backend_path)) from sqlalchemy import select from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession from sqlalchemy.orm import sessionmaker from app.models.user import User from app.models.ticket import Ticket from app.models.tenant import Tenant from app.core.security import SecurityUtils class DBUtils: """Utilidades de gestión de base de datos""" def __init__(self, database_url: Optional[str] = None): self.database_url = database_url or os.getenv( 'DATABASE_URL', 'postgresql+asyncpg://postgres:postgres@localhost:5432/servicemanager' ) self.engine = create_async_engine(self.database_url, echo=False) self.async_session = sessionmaker( self.engine, class_=AsyncSession, expire_on_commit=False ) async def list_users(self, tenant_id: Optional[str] = None): """Listar todos los usuarios""" async with self.async_session() as session: query = select(User) if tenant_id: query = query.where(User.tenant_id == tenant_id) result = await session.execute(query) users = result.scalars().all() if not users: print("❌ No se encontraron usuarios") return print(f"\n{'='*80}") print(f"📋 USUARIOS ({len(users)} encontrados)") print(f"{'='*80}\n") for user in users: print(f" Email: {user.email}") print(f" Role: {user.role}") print(f" ID: {user.id}") print(f" Tenant ID: {user.tenant_id}") print(f" Activo: {'✅' if user.is_active else '❌'}") print(f" {'-'*76}") async def check_user(self, email: str): """Verificar información de un usuario específico""" async with self.async_session() as session: result = await session.execute( select(User).where(User.email == email) ) user = result.scalar_one_or_none() if not user: print(f"❌ Usuario '{email}' no encontrado") return print(f"\n{'='*80}") print(f"👤 INFORMACIÓN DEL USUARIO") print(f"{'='*80}\n") print(f" Email: {user.email}") print(f" Nombre: {user.first_name} {user.last_name}") print(f" Role: {user.role}") print(f" ID: {user.id}") print(f" Tenant ID: {user.tenant_id}") print(f" Activo: {'✅' if user.is_active else '❌'}") print(f" 2FA: {'✅ Habilitado' if user.totp_secret else '❌ Deshabilitado'}") print(f" Creado: {user.created_at}") print(f"\n{'='*80}") async def reset_password(self, email: str, new_password: str = "admin123"): """Resetear contraseña de un usuario""" async with self.async_session() as session: result = await session.execute( select(User).where(User.email == email) ) user = result.scalar_one_or_none() if not user: print(f"❌ Usuario '{email}' no encontrado") return # Hash nueva contraseña password_hash = SecurityUtils.hash_password(new_password) user.password_hash = password_hash try: await session.commit() print(f"\n✅ Contraseña actualizada exitosamente") print(f" Usuario: {email}") print(f" Nueva contraseña: {new_password}") print(f"\n⚠️ IMPORTANTE: Cambia esta contraseña después del primer login") except Exception as e: await session.rollback() print(f"❌ Error al actualizar contraseña: {e}") async def list_tickets(self, tenant_id: Optional[str] = None, limit: int = 20): """Listar tickets""" async with self.async_session() as session: query = select(Ticket).order_by(Ticket.created_at.desc()).limit(limit) if tenant_id: query = query.where(Ticket.tenant_id == tenant_id) result = await session.execute(query) tickets = result.scalars().all() if not tickets: print("❌ No se encontraron tickets") return print(f"\n{'='*80}") print(f"🎫 TICKETS ({len(tickets)} encontrados, límite: {limit})") print(f"{'='*80}\n") for ticket in tickets: print(f" {ticket.ticket_number} | {ticket.status} | {ticket.priority}") print(f" Asunto: {ticket.subject}") print(f" ID: {ticket.id}") print(f" Tenant: {ticket.tenant_id}") print(f" Creado: {ticket.created_at}") print(f" {'-'*76}") async def check_ticket(self, ticket_id: str): """Verificar información de un ticket específico""" async with self.async_session() as session: result = await session.execute( select(Ticket).where(Ticket.id == ticket_id) ) ticket = result.scalar_one_or_none() if not ticket: print(f"❌ Ticket '{ticket_id}' no encontrado") return # Obtener creador creator_result = await session.execute( select(User).where(User.id == ticket.created_by) ) creator = creator_result.scalar_one_or_none() # Obtener asignado assigned = None if ticket.assigned_to: assigned_result = await session.execute( select(User).where(User.id == ticket.assigned_to) ) assigned = assigned_result.scalar_one_or_none() print(f"\n{'='*80}") print(f"🎫 INFORMACIÓN DEL TICKET") print(f"{'='*80}\n") print(f" Número: {ticket.ticket_number}") print(f" Asunto: {ticket.subject}") print(f" Estado: {ticket.status}") print(f" Prioridad: {ticket.priority}") print(f" ID: {ticket.id}") print(f" Tenant ID: {ticket.tenant_id}") if creator: print(f" Creado por: {creator.email} ({creator.role})") if assigned: print(f" Asignado a: {assigned.email} ({assigned.role})") print(f" Creado: {ticket.created_at}") print(f" Actualizado: {ticket.updated_at}") print(f"\n{'='*80}") async def close(self): """Cerrar conexión""" await self.engine.dispose() async def main(): parser = argparse.ArgumentParser( description='Utilidades de gestión de base de datos', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=__doc__ ) subparsers = parser.add_subparsers(dest='command', help='Comando a ejecutar') # list-users list_users_parser = subparsers.add_parser('list-users', help='Listar usuarios') list_users_parser.add_argument('--tenant-id', help='Filtrar por tenant ID') # check-user check_user_parser = subparsers.add_parser('check-user', help='Verificar usuario') check_user_parser.add_argument('email', help='Email del usuario') # reset-password reset_password_parser = subparsers.add_parser('reset-password', help='Resetear contraseña') reset_password_parser.add_argument('email', help='Email del usuario') reset_password_parser.add_argument('--password', default='admin123', help='Nueva contraseña') # list-tickets list_tickets_parser = subparsers.add_parser('list-tickets', help='Listar tickets') list_tickets_parser.add_argument('--tenant-id', help='Filtrar por tenant ID') list_tickets_parser.add_argument('--limit', type=int, default=20, help='Límite de resultados') # check-ticket check_ticket_parser = subparsers.add_parser('check-ticket', help='Verificar ticket') check_ticket_parser.add_argument('ticket_id', help='ID del ticket') args = parser.parse_args() if not args.command: parser.print_help() return utils = DBUtils() try: if args.command == 'list-users': await utils.list_users(args.tenant_id) elif args.command == 'check-user': await utils.check_user(args.email) elif args.command == 'reset-password': await utils.reset_password(args.email, args.password) elif args.command == 'list-tickets': await utils.list_tickets(args.tenant_id, args.limit) elif args.command == 'check-ticket': await utils.check_ticket(args.ticket_id) finally: await utils.close() if __name__ == "__main__": asyncio.run(main())