import asyncio import logging from celery import Celery from celery_app import celery_app from typing import Dict, Any from fastapi import HTTPException as _HTTPException from .services import consume_ws_get_cove, consume_ws_get_acuse_cove, marcar_error_cove from api.api_v2.modules.tasks.services import update_task, register_task # Logger para el módulo logger = logging.getLogger(__name__) # Reintentos internos del worker: pertenecen al MISMO intento orquestado y no # incrementan el contador de intentos del backend (T2026-05-027) WORKER_MAX_RETRIES = 2 @celery_app.task(bind=True) def process_cove_request(self, cove_request: Dict[str, Any]) -> Dict[str, Any]: """ Tarea para procesar solicitudes de COVE. """ task_id = self.request.id pedimento_info = cove_request.get('pedimento', {}) pedimento_id = pedimento_info.get('id') organizacion_id = pedimento_info.get('organizacion') pedimento_app = pedimento_info.get('pedimento_app', 'N/A') cove_info = cove_request.get('cove', {}) # aqui se cambio de cove_number a 'cove' porque no esta asi en el mapeo cove_number = cove_info.get('cove', 'N/A') try: # Registrar el inicio de la tarea (solo en la primera ejecución, no en reintentos) if self.request.retries == 0: logger.info(f"[COVE] Registrando inicio de tarea {task_id}") asyncio.run( register_task( task_id=task_id, status="submitted", message=f"Iniciando proceso de COVE {cove_number} para pedimento {pedimento_app}", pedimento_id=pedimento_id, organizacion_id=organizacion_id, servicio=8 # 8 corresponde a "Cove" ) ) logger.info(f"[COVE] Actualizando estado a processing para tarea {task_id}") asyncio.run( update_task( task_id=task_id, status="processing", message=f"Procesando COVE {cove_number} para pedimento {pedimento_app}", pedimento_id=pedimento_id, organizacion_id=organizacion_id, servicio=8 ) ) cove_response = asyncio.run(consume_ws_get_cove(**cove_request)) asyncio.run( update_task( task_id=task_id, status="completed", message=f"COVE {cove_number} obtenido exitosamente para pedimento {pedimento_app}", pedimento_id=pedimento_id, organizacion_id=organizacion_id, servicio=8 ) ) return {"status": "processed", "data": cove_response} except Exception as e: error_message = f"Error al procesar COVE {cove_number} para pedimento {pedimento_app}: {str(e)}" logger.error(error_message) # Reintento interno del worker para fallos transitorios (red, timeout): # mismo intento orquestado, NO incrementa el contador del backend if not isinstance(e, _HTTPException) and self.request.retries < WORKER_MAX_RETRIES: raise self.retry(exc=e, countdown=60 * (self.request.retries + 1)) # Fallo definitivo de este intento: registrar el detalle en el registro de # negocio. Permanece 'pendiente'; el tope de intentos automáticos del # backend (MAX_INTENTOS_AUTO) gobierna la transición a 'error'. try: asyncio.run( marcar_error_cove(cove_info, pedimento_info, error_message, acuse=False, definitivo=False) ) except Exception as report_error: logger.error(f"No se pudo registrar el error en el registro de negocio: {report_error}") try: asyncio.run( update_task( task_id=task_id, status="failed", message=error_message, pedimento_id=pedimento_id, organizacion_id=organizacion_id, servicio=8 ) ) except Exception as update_error: logger.error(f"No se pudo actualizar el estado de la tarea: {update_error}") if isinstance(e, _HTTPException): raise RuntimeError(f"HTTP {e.status_code}: {e.detail}") from None raise @celery_app.task(bind=True) def process_acuse_cove_request(self, cove_request: Dict[str, Any]) -> Dict[str, Any]: """ Tarea para procesar solicitudes de acuse de COVE. """ task_id = self.request.id pedimento_info = cove_request.get('pedimento', {}) pedimento_id = pedimento_info.get('id') organizacion_id = pedimento_info.get('organizacion') pedimento_app = pedimento_info.get('pedimento_app', 'N/A') cove_info = cove_request.get('cove', {}) # se cambio el cove_number, ya que no existe directamente en el mapeo, se conoce solo como 'cove' cove_number = cove_info.get('cove', 'N/A') try: # Registrar el inicio de la tarea (solo en la primera ejecución, no en reintentos) if self.request.retries == 0: logger.info(f"[COVE] Registrando inicio de tarea de acuse {task_id}") asyncio.run( register_task( task_id=task_id, status="submitted", message=f"Iniciando proceso de acuse de COVE {cove_number} para pedimento {pedimento_app}", pedimento_id=pedimento_id, organizacion_id=organizacion_id, servicio=9 # 9 corresponde a "Acuse Cove" ) ) logger.info(f"[COVE] Actualizando estado a processing para tarea de acuse {task_id}") asyncio.run( update_task( task_id=task_id, status="processing", message=f"Procesando acuse de COVE {cove_number} para pedimento {pedimento_app}", pedimento_id=pedimento_id, organizacion_id=organizacion_id, servicio=9 ) ) acuse_response = asyncio.run(consume_ws_get_acuse_cove(**cove_request)) asyncio.run( update_task( task_id=task_id, status="completed", message=f"Acuse de COVE {cove_number} obtenido exitosamente para pedimento {pedimento_app}", pedimento_id=pedimento_id, organizacion_id=organizacion_id, servicio=9 ) ) return {"status": "processed", "data": acuse_response} except Exception as e: error_message = f"Error al procesar acuse de COVE {cove_number} para pedimento {pedimento_app}: {str(e)}" logger.error(error_message) # Reintento interno del worker para fallos transitorios (red, timeout): # mismo intento orquestado, NO incrementa el contador del backend if not isinstance(e, _HTTPException) and self.request.retries < WORKER_MAX_RETRIES: raise self.retry(exc=e, countdown=60 * (self.request.retries + 1)) # Fallo definitivo de este intento: registrar el detalle en el registro de # negocio. Permanece 'pendiente'; el tope de intentos automáticos del # backend (MAX_INTENTOS_AUTO) gobierna la transición a 'error'. try: asyncio.run( marcar_error_cove(cove_info, pedimento_info, error_message, acuse=True, definitivo=False) ) except Exception as report_error: logger.error(f"No se pudo registrar el error en el registro de negocio: {report_error}") try: asyncio.run( update_task( task_id=task_id, status="failed", message=error_message, pedimento_id=pedimento_id, organizacion_id=organizacion_id, servicio=9 ) ) except Exception as update_error: logger.error(f"No se pudo actualizar el estado de la tarea: {update_error}") if isinstance(e, _HTTPException): raise RuntimeError(f"HTTP {e.status_code}: {e.detail}") from None raise