EFC/microservice
5
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

feature/capturar errores, evitar duplicados, eliminar manejar nuevas flags para descargar datos de vucem

Browse Source
This commit is contained in:
marcos
2026-06-15 11:20:58 -06:00
parent 042d02e240
commit b9c6ab89c3
10 changed files with 326 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
api/api_v2/modules/coves/services.py
View File

@@ -146,6 +146,13 @@ async def consume_ws_get_cove(**kwargs):
logger.error(f"Error detectado en la respuesta SOAP: {str(e)}")
raise Exception(f"Error en la respuesta SOAP: {str(e)}")
# Validar que el documento quedó persistido en la API antes de marcar el
# COVE como descargado (T2026-05-027): post_or_update_document retorna
# None en error sin lanzar excepción
if not document_response or document_response.get("id") is None:
logger.error(f"No se pudo persistir el COVE {cove} en la API; el estatus no se actualiza")
raise Exception(f"No se pudo persistir el COVE {cove} en la API")
logger.info("Documento enviado, actualizando status de COVE")
# Actualizar status del COVE
@@ -329,6 +336,15 @@ async def consume_ws_get_acuse_cove(**kwargs):
identifier=cove_identifier,
)
# Validar la subida antes de marcar el acuse como descargado (T2026-05-027):
# post_or_update_document retorna None en error sin lanzar excepción
if not rest_response or rest_response.get("id") is None:
logger.error("Error al enviar el acuse de COVE a la API interna; el estatus no se actualiza")
raise HTTPException(
status_code=500,
detail="No se pudo persistir el acuse del COVE en la API; el estatus no se actualiza"
)
acuse_status = await change_acuse_status(
cove=kwargs.get('cove'),
status=True,
@@ -507,6 +523,7 @@ async def change_cove_status(cove: dict, status: bool, pedimento: dict):
data = {
"id": cove.get("id"),
"cove_descargado": status,
"cove_estado": "descargado" if status else "pendiente",
"numero_cove": cove.get("cove"),
"pedimento": pedimento.get("id"),
"organizacion": pedimento.get("organizacion"),
@@ -514,6 +531,11 @@ async def change_cove_status(cove: dict, status: bool, pedimento: dict):
response = await coves_rest_controller.put_cove_data(cove_id=cove.get("id"), data=data)
# Nunca reportar éxito si el estatus no quedó persistido (T2026-05-027)
if response is None:
logger.error(f"No se pudo actualizar el estatus del COVE {cove.get('cove')} en la API")
raise Exception(f"Fallo al actualizar el estatus del COVE {cove.get('cove')}")
return response
@@ -521,14 +543,46 @@ async def change_acuse_status(cove: dict, status: bool, pedimento: dict):
data = {
"id": cove.get("id"),
"acuse_cove_descargado": status,
"acuse_cove_estado": "descargado" if status else "pendiente",
"numero_cove": cove.get("cove"),
"pedimento": pedimento.get("id"),
"organizacion": pedimento.get("organizacion"),
}
print(data)
response = await coves_rest_controller.put_cove_data(cove_id=cove.get("id"), data=data)
# Nunca reportar éxito si el estatus no quedó persistido (T2026-05-027)
if response is None:
logger.error(f"No se pudo actualizar el estatus del acuse del COVE {cove.get('cove')} en la API")
raise Exception(f"Fallo al actualizar el estatus del acuse del COVE {cove.get('cove')}")
return response
async def marcar_error_cove(cove: dict, pedimento: dict, mensaje: str, acuse: bool = False, definitivo: bool = False):
"""
Reporta un fallo de descarga al registro de negocio (T2026-05-027).
- definitivo=False (fallo transitorio): solo registra ultimo_error; el registro
permanece 'pendiente' y el tope de intentos automáticos del backend gobierna
la transición a 'error'.
- definitivo=True (fallo permanente: VUCEM responde que no existe, credencial
rechazada): transiciona de inmediato a 'error'; queda fuera del ciclo
automático y solo el reproceso manual o el reset lo reincorporan.
"""
campo = "acuse_cove_estado" if acuse else "cove_estado"
data = {
"id": cove.get("id"),
"ultimo_error": (mensaje or "Error de descarga en VUCEM")[:2000],
"numero_cove": cove.get("cove"),
"pedimento": pedimento.get("id"),
"organizacion": pedimento.get("organizacion"),
}
if definitivo:
data[campo] = "error"
response = await coves_rest_controller.put_cove_data(cove_id=cove.get("id"), data=data)
if response is None:
logger.error(f"No se pudo registrar el error del COVE {cove.get('cove')} en la API")
return response

86
api/api_v2/modules/coves/tasks.py
View File

@@ -5,12 +5,16 @@ 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
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]:
@@ -27,18 +31,19 @@ def process_cove_request(self, cove_request: Dict[str, Any]) -> Dict[str, Any]:
cove_number = cove_info.get('cove', 'N/A')
try:
# Registrar el inicio de la tarea
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"
# 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(
@@ -70,7 +75,22 @@ def process_cove_request(self, cove_request: Dict[str, Any]) -> Dict[str, Any]:
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(
@@ -105,18 +125,19 @@ def process_acuse_cove_request(self, cove_request: Dict[str, Any]) -> Dict[str,
cove_number = cove_info.get('cove', 'N/A')
try:
# Registrar el inicio de la tarea
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"
# 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(
@@ -148,7 +169,22 @@ def process_acuse_cove_request(self, cove_request: Dict[str, Any]) -> Dict[str,
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(