97 lines
3.6 KiB
Python
97 lines
3.6 KiB
Python
from django.core.management.base import BaseCommand
|
|
from api.customs.tasks.auditoria import (
|
|
auditar_procesamiento_remesas,
|
|
auditar_coves,
|
|
auditar_acuse_cove,
|
|
auditar_edocuments,
|
|
auditar_acuse,
|
|
crear_partidas
|
|
)
|
|
from api.customs.models import Pedimento
|
|
from datetime import datetime
|
|
|
|
class Command(BaseCommand):
|
|
help = 'Ejecuta tareas de auditoría para una organización'
|
|
|
|
# Definir las tareas disponibles
|
|
TAREAS_DISPONIBLES = {
|
|
'remesas': (auditar_procesamiento_remesas, "Auditoría de remesas"),
|
|
'partidas': (crear_partidas, "Creación de partidas"),
|
|
'coves': (auditar_coves, "Auditoría de COVEs"),
|
|
'acuse-cove': (auditar_acuse_cove, "Auditoría de acuses de COVEs"),
|
|
'edocs': (auditar_edocuments, "Auditoría de E-documents"),
|
|
'acuses': (auditar_acuse, "Auditoría de acuses")
|
|
}
|
|
|
|
def add_arguments(self, parser):
|
|
parser.add_argument(
|
|
'organizacion_id',
|
|
type=str,
|
|
help='ID de la organización a auditar'
|
|
)
|
|
parser.add_argument(
|
|
'--tareas',
|
|
nargs='+',
|
|
choices=['todas'] + list(self.TAREAS_DISPONIBLES.keys()),
|
|
default=['todas'],
|
|
help='Lista de tareas a ejecutar. Opciones: todas, ' +
|
|
', '.join(self.TAREAS_DISPONIBLES.keys())
|
|
)
|
|
parser.add_argument(
|
|
'--list',
|
|
action='store_true',
|
|
help='Muestra la lista de tareas disponibles'
|
|
)
|
|
|
|
def handle(self, *args, **options):
|
|
organizacion_id = options['organizacion_id']
|
|
start_time = datetime.now()
|
|
|
|
# Verificar si la organización tiene pedimentos
|
|
pedimentos_count = Pedimento.objects.filter(organizacion_id=organizacion_id).count()
|
|
if not pedimentos_count:
|
|
self.stdout.write(
|
|
self.style.ERROR(f'No se encontraron pedimentos para la organización {organizacion_id}')
|
|
)
|
|
return
|
|
|
|
self.stdout.write(f'Iniciando auditoría para organización {organizacion_id}')
|
|
self.stdout.write(f'Total de pedimentos a procesar: {pedimentos_count}')
|
|
|
|
# Si se solicita listar las tareas
|
|
if options['list']:
|
|
self.stdout.write('Tareas disponibles:')
|
|
for key, (_, desc) in self.TAREAS_DISPONIBLES.items():
|
|
self.stdout.write(f' - {key}: {desc}')
|
|
return
|
|
|
|
# Determinar qué tareas ejecutar
|
|
tareas_seleccionadas = []
|
|
if 'todas' in options['tareas']:
|
|
tareas_seleccionadas = list(self.TAREAS_DISPONIBLES.values())
|
|
else:
|
|
for tarea in options['tareas']:
|
|
if tarea in self.TAREAS_DISPONIBLES:
|
|
tareas_seleccionadas.append(self.TAREAS_DISPONIBLES[tarea])
|
|
|
|
self.stdout.write(f'Tareas seleccionadas: {len(tareas_seleccionadas)}')
|
|
for _, desc in tareas_seleccionadas:
|
|
self.stdout.write(f' - {desc}')
|
|
|
|
# Ejecutar cada tarea
|
|
for tarea, descripcion in tareas_seleccionadas:
|
|
self.stdout.write(f'\nIniciando {descripcion}...')
|
|
try:
|
|
resultado = tarea.delay(organizacion_id)
|
|
self.stdout.write(
|
|
self.style.SUCCESS(f'✓ {descripcion} iniciada (Task ID: {resultado.id})')
|
|
)
|
|
except Exception as e:
|
|
self.stdout.write(
|
|
self.style.ERROR(f'✗ Error en {descripcion}: {str(e)}')
|
|
)
|
|
|
|
tiempo_total = datetime.now() - start_time
|
|
self.stdout.write(self.style.SUCCESS(
|
|
f'\nProceso de auditoría completado en {tiempo_total}'
|
|
)) |