Se agregaron estados y update a cada una de las tareas

api/api_v2/modules/coves/routers.py

@@ -2,10 +2,13 @@ from fastapi import APIRouter, HTTPException
 from .schemas import CoveListSchema, CoveRequestSchema
 from typing import List, Dict, Any
 from uuid import UUID
+import logging
 
 from .tasks import process_cove_request, process_acuse_cove_request
 from ..tasks.services import register_task
 
+logger = logging.getLogger(__name__)
+
 router = APIRouter()
 # Aquí puedes definir tus endpoints relacionados con COVES usando el esquema CoveBaseSchema
 
@@ -17,16 +20,6 @@ async def get_cove(cove: CoveRequestSchema):
     cove_dict = cove.model_dump()
     task = process_cove_request.delay(cove_dict)
     
-    # Registrar la tarea en el servicio de seguimiento
-    await register_task(
-        task_id=task.id,
-        message=f"Procesando COVE para pedimento {cove_dict.get('pedimento', {}).get('pedimento_app', 'N/A')}",
-        status="submitted",
-        pedimento_id=cove_dict.get('pedimento', {}).get('id'),
-        organizacion_id=cove_dict.get('pedimento', {}).get('organizacion'),
-        servicio=8  # 8 corresponde a "Cove"
-    )
-    
     return {"task_id": task.id, "status": "submitted"}
 
 
@@ -46,16 +39,6 @@ async def get_coves(coves_request: CoveListSchema):
         task = process_cove_request.delay(cove_dict)
         task_ids.append(task.id)
         
-        # Registrar cada tarea en el servicio de seguimiento
-        await register_task(
-            task_id=task.id,
-            message=f"Procesando COVE masivo para pedimento {pedimento.get('pedimento_app', 'N/A')}",
-            status="submitted",
-            pedimento_id=pedimento.get('id'),
-            organizacion_id=pedimento.get('organizacion'),
-            servicio=8  # 8 corresponde a "Cove"
-        )
-
     return {"task_ids": task_ids, "status": "submitted", "total": len(task_ids)}
 
 
@@ -65,16 +48,6 @@ async def get_acuse_cove(cove: CoveRequestSchema):
     cove_dict = cove.model_dump()
     task = process_acuse_cove_request.delay(cove_dict)
     
-    # Registrar la tarea en el servicio de seguimiento
-    await register_task(
-        task_id=task.id,
-        message=f"Procesando acuse de COVE para pedimento {cove_dict.get('pedimento', {}).get('pedimento_app', 'N/A')}",
-        status="submitted",
-        pedimento_id=cove_dict.get('pedimento', {}).get('id'),
-        organizacion_id=cove_dict.get('pedimento', {}).get('organizacion'),
-        servicio=9  # 9 corresponde a "Acuse Cove"
-    )
-    
     return {"task_id": task.id, "status": "submitted"}
 
 @router.post("/services/all/acuse/cove/", response_model=Dict[str, Any])
@@ -93,14 +66,4 @@ async def get_acuses_cove(coves_request: CoveListSchema):
         task = process_acuse_cove_request.delay(acuse_dict)
         task_ids.append(task.id)
         
-        # Registrar cada tarea en el servicio de seguimiento
-        await register_task(
-            task_id=task.id,
-            message=f"Procesando acuse masivo de COVE para pedimento {pedimento.get('pedimento_app', 'N/A')}",
-            status="submitted",
-            pedimento_id=pedimento.get('id'),
-            organizacion_id=pedimento.get('organizacion'),
-            servicio=9  # 9 corresponde a "Acuse Cove"
-        )
-
     return {"task_ids": task_ids, "status": "submitted", "total": len(task_ids)}

api/api_v2/modules/coves/tasks.py

@@ -1,11 +1,12 @@
 import asyncio
 import logging
+import time
 from celery import Celery
 from celery_app import celery_app
 from typing import Dict, Any
 
 from .services import consume_ws_get_cove, consume_ws_get_acuse_cove
-from api.api_v2.modules.tasks.tasks import run_async_task
+from api.api_v2.modules.tasks.services import update_task, register_task
 
 # Logger para el módulo
 logger = logging.getLogger(__name__)
@@ -13,16 +14,156 @@ logger = logging.getLogger(__name__)
 
 @celery_app.task(bind=True)
 def process_cove_request(self, cove_request: Dict[str, Any]) -> Dict[str, Any]:
-    loop = asyncio.get_event_loop()
-    cove_response = loop.run_until_complete(consume_ws_get_cove(**cove_request))
+    """
+    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', {})
+    cove_number = cove_info.get('numero_cove', 'N/A')
+    
+    try:
+        # Registrar el inicio de la tarea
+        loop = asyncio.get_event_loop()
+        logger.info(f"[COVE] Registrando inicio de tarea {task_id}")
+        loop.run_until_complete(
+            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"
+            )
+        )
 
-    return {"status": "processed", "data": cove_response}
+        # Esperar un momento breve para asegurar que el registro se complete
+        time.sleep(1)
+        
+        # Actualizar estado: procesando
+        logger.info(f"[COVE] Actualizando estado a processing para tarea {task_id}")
+        loop.run_until_complete(
+            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
+            )
+        )
+        
+        # Obtener el COVE
+        cove_response = loop.run_until_complete(consume_ws_get_cove(**cove_request))
+        
+        # Actualizar estado: completado
+        loop.run_until_complete(
+            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:
+        # En caso de error, actualizar estado
+        error_message = f"Error al procesar COVE {cove_number} para pedimento {pedimento_app}: {str(e)}"
+        logger.error(error_message)
+        loop.run_until_complete(
+            update_task(
+                task_id=task_id,
+                status="failed",
+                message=error_message,
+                pedimento_id=pedimento_id,
+                organizacion_id=organizacion_id,
+                servicio=8
+            )
+        )
+        raise
 
 
 @celery_app.task(bind=True)
 def process_acuse_cove_request(self, cove_request: Dict[str, Any]) -> Dict[str, Any]:
-    loop = asyncio.get_event_loop()
-    acuse_response = loop.run_until_complete(consume_ws_get_acuse_cove(**cove_request))
+    """
+    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', {})
+    cove_number = cove_info.get('numero_cove', 'N/A')
+    
+    try:
+        # Registrar el inicio de la tarea
+        loop = asyncio.get_event_loop()
+        logger.info(f"[COVE] Registrando inicio de tarea de acuse {task_id}")
+        loop.run_until_complete(
+            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"
+            )
+        )
 
-    return {"status": "processed", "data": acuse_response}
+        # Esperar un momento breve para asegurar que el registro se complete
+        time.sleep(1)
+        
+        # Actualizar estado: procesando
+        logger.info(f"[COVE] Actualizando estado a processing para tarea de acuse {task_id}")
+        loop.run_until_complete(
+            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
+            )
+        )
+        
+        # Obtener el acuse del COVE
+        acuse_response = loop.run_until_complete(consume_ws_get_acuse_cove(**cove_request))
+        
+        # Actualizar estado: completado
+        loop.run_until_complete(
+            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:
+        # En caso de error, actualizar estado
+        error_message = f"Error al procesar acuse de COVE {cove_number} para pedimento {pedimento_app}: {str(e)}"
+        logger.error(error_message)
+        loop.run_until_complete(
+            update_task(
+                task_id=task_id,
+                status="failed",
+                message=error_message,
+                pedimento_id=pedimento_id,
+                organizacion_id=organizacion_id,
+                servicio=9
+            )
+        )
+        raise