icamarillo
96084b89c0
🗑️ Eliminados: - Scripts temporales de debugging (9 archivos en backend/) - Archivos temporales en raíz (4 archivos) - Reportes de cobertura htmlcov/ (~543 KB) - Directorio backups/ - Script de migración update_password_hashes.py ✨ Optimizaciones: - Creado scripts/db_utils.py - Herramienta consolidada para administración - Actualizado README.md con sección de Utilidades Administrativas - Mejorado .gitignore para prevenir archivos temporales futuros 📦 Espacio liberado: ~600-800 KB Las funcionalidades de debugging ahora están consolidadas en: - scripts/db_utils.py (herramienta CLI profesional) - Documentación en README.md - Alternativas sugeridas (psql, tests, API docs)
263 lines
9.9 KiB
Python
263 lines
9.9 KiB
Python
#!/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())
|