Version con fallas
This commit is contained in:
@@ -1,262 +0,0 @@
|
||||
#!/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())
|
||||
Reference in New Issue
Block a user