104 lines
3.7 KiB
Python
104 lines
3.7 KiB
Python
from celery_app import celery_app
|
|
from .services import post_remesa_data
|
|
import asyncio
|
|
import logging
|
|
from ..tasks.services import register_task, update_task
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
@celery_app.task(bind=True)
|
|
def process_remesa_request(self, remesa_request: dict) -> dict:
|
|
"""
|
|
Tarea de Celery para procesar la solicitud de remesa.
|
|
|
|
Args:
|
|
remesa_request (dict): Datos de la remesa a procesar
|
|
|
|
Returns:
|
|
dict: Resultado del procesamiento con estado y detalles
|
|
"""
|
|
task_id = self.request.id
|
|
servicio = 5 # Código para Pedimento Remesas
|
|
pedimento_id = remesa_request.get('pedimento', {}).get('id')
|
|
organizacion_id = remesa_request.get('pedimento', {}).get('organizacion')
|
|
remesa_num = remesa_request.get('remesa', 'N/A')
|
|
|
|
# Crear un NUEVO event loop para esta tarea (evita problemas de loop cerrado)
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
|
|
try:
|
|
# Actualizar estado a processing
|
|
loop.run_until_complete(
|
|
update_task(
|
|
task_id=task_id,
|
|
message=f"Iniciando procesamiento de la remesa {remesa_num}",
|
|
status="processing",
|
|
pedimento_id=pedimento_id,
|
|
organizacion_id=organizacion_id,
|
|
servicio=servicio
|
|
)
|
|
)
|
|
|
|
# Procesar remesa
|
|
result = loop.run_until_complete(post_remesa_data(**remesa_request))
|
|
|
|
# Actualizar estado a completed
|
|
loop.run_until_complete(
|
|
update_task(
|
|
task_id=task_id,
|
|
message=f"Remesa {remesa_num} procesada exitosamente",
|
|
status="completed",
|
|
pedimento_id=pedimento_id,
|
|
organizacion_id=organizacion_id,
|
|
servicio=servicio,
|
|
)
|
|
)
|
|
|
|
return {"status": "success", "result": result}
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error procesando remesa {remesa_num}: {str(e)}", exc_info=True)
|
|
|
|
# Actualizar estado a failed
|
|
try:
|
|
# Verificar si el loop aún está abierto
|
|
if not loop.is_closed():
|
|
loop.run_until_complete(
|
|
update_task(
|
|
task_id=task_id,
|
|
message=f"Error al procesar remesa {remesa_num}: {str(e)}",
|
|
status="failed",
|
|
pedimento_id=pedimento_id,
|
|
organizacion_id=organizacion_id,
|
|
servicio=servicio,
|
|
)
|
|
)
|
|
else:
|
|
# Si el loop está cerrado, crear uno nuevo temporal
|
|
logger.warning(f"Loop cerrado, creando loop temporal para actualizar error")
|
|
temp_loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(temp_loop)
|
|
try:
|
|
temp_loop.run_until_complete(
|
|
update_task(
|
|
task_id=task_id,
|
|
message=f"Error al procesar remesa {remesa_num}: {str(e)}",
|
|
status="failed",
|
|
pedimento_id=pedimento_id,
|
|
organizacion_id=organizacion_id,
|
|
servicio=servicio
|
|
)
|
|
)
|
|
finally:
|
|
temp_loop.close()
|
|
except Exception as update_error:
|
|
logger.error(f"Error actualizando estado de tarea: {update_error}")
|
|
|
|
# Re-lanzar la excepción para que Celery la marque como fallida
|
|
raise
|
|
|
|
finally:
|
|
# Limpiar el event loop
|
|
if not loop.is_closed():
|
|
loop.close() |