from __future__ import annotations import base64 import logging import os import time from typing import Any, Dict from celery import Task from celery.exceptions import Ignore import httpx from core.celery_app import celery_app from core.database import CoreSessionLocal from core.exceptions import ErrorCollector, ValidationException from core.storage_s3 import delete_object_if_exists, get_object_bytes, object_exists, put_object_bytes from core.s3_keys import expediente_archivo_artifact_key from .models import ExpedienteArchivo from .service import ExpedienteArchivoService, build_configuracion_vu, resolve_rfc_consulta_value from .external_service import ExpedienteExternalService logger = logging.getLogger(__name__) TOTAL_STEPS = 4 def _fail_task( task: Task, *, error: str, error_type: str, codigo: str, descripcion: str, paso: str, sugerencias: list[str] | None = None, ) -> None: task.update_state( state="FAILED", meta={ "error": error, "error_type": error_type, "error_detail": { "codigo": codigo, "descripcion": descripcion, "paso": paso, "sugerencias": sugerencias or [], }, }, ) raise Ignore() def _load_record_file_base64(record: ExpedienteArchivo) -> str: stored_path = (record.archivo_digitalizado_en or "").strip() if not stored_path: raise ValidationException( "El expediente no tiene archivo cargado", errors=[ { "field": "archivo_digitalizado_en", "message": "El expediente no tiene archivo almacenado para digitalizar.", "code": "MISSING_FILE", "solution": ["Edita el expediente y vuelve a seleccionar el archivo antes de digitalizar."], } ], ) load_started_at = time.perf_counter() source = "unknown" try: if os.path.exists(stored_path): source = "local" with open(stored_path, "rb") as file_handle: raw = file_handle.read() elif object_exists(stored_path): source = "s3" raw = get_object_bytes(stored_path) else: raise ValidationException( "Archivo del expediente no encontrado", errors=[ { "field": "archivo_digitalizado_en", "message": "No se encontró el archivo almacenado del expediente.", "code": "FILE_NOT_FOUND", "solution": ["Edita el expediente y vuelve a cargar el documento."], } ], ) except ValidationException: raise except Exception as exc: raise ValidationException( "No se pudo leer el archivo del expediente", errors=[ { "field": "archivo_digitalizado_en", "message": "Ocurrió un error leyendo el archivo almacenado del expediente.", "code": "FILE_READ_ERROR", "solution": ["Vuelve a cargar el archivo del expediente e inténtalo nuevamente."], } ], ) from exc logger.info( "Expediente file loaded expediente_id=%s source=%s bytes=%s elapsed_ms=%.1f", record.id, source, len(raw), (time.perf_counter() - load_started_at) * 1000, ) return base64.b64encode(raw).decode("ascii") def _progress(task: Task, current: int, status: str) -> None: task.update_state( state="PROGRESS", meta={"current": current, "status": status, "current_step": status, "progress": current, "total_steps": TOTAL_STEPS}, ) def _poll_external( task: Task, external: ExpedienteExternalService, external_task_id: str, timeout_seconds: int = 300, ) -> Dict[str, Any]: """ Hace polling al API externo hasta obtener un estado final o agotar el timeout. Retorna el payload final tal como lo devuelve el API externo. """ start = time.perf_counter() last_payload: Dict[str, Any] = {} attempts = 0 while True: attempts += 1 elapsed = time.perf_counter() - start if elapsed > timeout_seconds: logger.error( "Timeout en polling externo de digitalización: task_id=%s attempts=%s elapsed_s=%.2f", external_task_id, attempts, elapsed, ) raise TimeoutError(f"Timeout consultando estado de digitalización (task_id={external_task_id}).") try: status_payload = external.get_status(external_task_id) or {} except httpx.ReadTimeout: # VU mantiene la conexión abierta mientras procesa; si httpx corta antes, # lo tratamos como "sigue en proceso" y reintentamos. logger.warning( "ReadTimeout consultando estado externo, reintentando task_id=%s attempts=%s elapsed_s=%.2f", external_task_id, attempts, elapsed, ) time.sleep(5) continue except Exception: logger.exception( "Error consultando estado externo de digitalización task_id=%s attempts=%s elapsed_s=%.2f", external_task_id, attempts, elapsed, ) raise last_payload = status_payload state = str(status_payload.get("state") or "").upper() progress_info = status_payload.get("progress") percent = 0.0 current_step = str( status_payload.get("current_step") or status_payload.get("status") or "Consultando estado en Ventanilla Única..." ) if isinstance(progress_info, dict): raw_percent = progress_info.get("progress", progress_info.get("current", 0.0)) try: percent = float(raw_percent) except (TypeError, ValueError): percent = 0.0 current_step = str( progress_info.get("current_step") or progress_info.get("status") or current_step ) elif isinstance(progress_info, (int, float, str)): try: percent = float(progress_info) except (TypeError, ValueError): percent = 0.0 _progress(task, int(percent), str(current_step)) if state in {"PENDING", "STARTED", "PROGRESS"} or not state: time.sleep(5) continue logger.info( "Digitalization external polling finished task_id=%s final_state=%s attempts=%s elapsed_s=%.2f", external_task_id, state or "UNKNOWN", attempts, elapsed, ) return last_payload @celery_app.task(bind=True, name="expediente_archivos_digitalizar") def digitalizar_task( self: Task, expediente_id: int, request_data: Dict[str, Any], company_id: int, tenant_id: int, ) -> dict: """ Celery task: digitaliza un ExpedienteArchivo en Ventanilla Única. Pasos: 1. Cargar el registro y construir configuracion_vu (server-side). 2. Construir y enviar el payload al API externo. 3. Hacer polling del estado del task externo. 4. Persistir resultado (e_document, num_operacion, acuse_pdf_path) en DB. """ db = CoreSessionLocal() task_started_at = time.perf_counter() try: logger.info( "Digitalization task started task_id=%s expediente_id=%s company_id=%s tenant_id=%s", self.request.id, expediente_id, company_id, tenant_id, ) # ------------------------------------------------------------------ # # Paso 1 – cargar registro y construir configuracion_vu # # ------------------------------------------------------------------ # _progress(self, 5, "Construyendo configuración VU...") record: ExpedienteArchivo | None = db.get(ExpedienteArchivo, expediente_id) if not record: raise ValidationException( "Expediente no encontrado", errors=[{"field": "expediente_id", "message": f"No existe expediente {expediente_id}"}], ) errors = ErrorCollector() agente_key = request_data.get("agente_aduanal") or record.agente_aduanal config_started_at = time.perf_counter() config_vu = build_configuracion_vu(db, company_id, tenant_id, agente_key, errors) logger.info( "Digitalization VU config resolved task_id=%s expediente_id=%s agente_aduanal=%s elapsed_ms=%.1f", self.request.id, expediente_id, agente_key, (time.perf_counter() - config_started_at) * 1000, ) if errors.has_errors(): error_list = errors._errors # type: ignore[attr-defined] first = error_list[0] if error_list else {} _fail_task( self, error=first.get("message", "Error de configuración VU"), error_type="VALIDATION_ERROR", codigo=first.get("code", "VALIDATION_ERROR"), descripcion=first.get("message", ""), paso="Construcción de configuración VU", sugerencias=first.get("solution") or [], ) # Actualizar status en DB resolved_rfc_consulta = resolve_rfc_consulta_value( db, company_id, tenant_id, agente_key, request_data.get("rfc_consulta"), record.rfc_consulta, config_vu.get("rfc_usuario_vu"), ) current_record_rfc = (record.rfc_consulta or "").strip().upper() config_vu_rfc = (config_vu.get("rfc_usuario_vu") or "").strip().upper() if not current_record_rfc or current_record_rfc == config_vu_rfc: record.rfc_consulta = resolved_rfc_consulta # Limpiar artefactos previos de S3 antes de iniciar nueva digitalización _ARTIFACT_PATH_FIELDS = [ "acuse_pdf_path", "envio_xml_path", "respuesta_xml_path", "consulta_envio_xml_path", "consulta_respuesta_xml_path", ] for _field in _ARTIFACT_PATH_FIELDS: _old_key = (getattr(record, _field, None) or "").strip() if _old_key and _old_key != "inline": try: delete_object_if_exists(_old_key) logger.info( "Digitalization old artifact deleted task_id=%s expediente_id=%s field=%s key=%s", self.request.id, expediente_id, _field, _old_key, ) except Exception: logger.warning( "Could not delete old artifact task_id=%s expediente_id=%s field=%s key=%s", self.request.id, expediente_id, _field, _old_key, exc_info=True, ) record.status = "processing" record.task_id = self.request.id record.external_task_id = None record.e_document = None record.num_operacion = None record.acuse_pdf_path = None record.envio_xml_path = None record.respuesta_xml_path = None record.consulta_envio_xml_path = None record.consulta_respuesta_xml_path = None db.commit() # ------------------------------------------------------------------ # # Paso 2 – construir payload y enviar al API externo # # ------------------------------------------------------------------ # _progress(self, 30, "Enviando documento a Ventanilla Única...") file_started_at = time.perf_counter() archivo_base64 = request_data.get("archivo_base64") or _load_record_file_base64(record) logger.info( "Digitalization payload document ready task_id=%s expediente_id=%s provided_inline=%s base64_len=%s elapsed_ms=%.1f", self.request.id, expediente_id, bool(request_data.get("archivo_base64")), len(archivo_base64), (time.perf_counter() - file_started_at) * 1000, ) payload = { "rfc_consulta": resolved_rfc_consulta, "clave_documento": request_data.get("clave_documento") or record.tipo_documento or "", "nombre_archivo": request_data.get("nombre_archivo") or record.nombre_archivo or "", "archivo_base64": archivo_base64, "configuracion_vu": { key: value for key, value in config_vu.items() if not key.startswith("_") }, } external = ExpedienteExternalService() external_submit_started_at = time.perf_counter() response = external.digitalizar_archivo_json(payload) logger.info( "Digitalization external submission finished task_id=%s expediente_id=%s external_task_id=%s response_state=%s elapsed_ms=%.1f", self.request.id, expediente_id, response.get("task_id") or response.get("id"), response.get("state") or response.get("status"), (time.perf_counter() - external_submit_started_at) * 1000, ) # Chequear si el API externo devolvió un error inmediato resp_state = str(response.get("state") or response.get("status") or "").upper() if resp_state in {"ERROR", "FAILURE", "FAILED"}: error_msg = response.get("message") or response.get("error") or "Error en API externo" record.status = "failed" db.commit() _fail_task( self, error=error_msg, error_type="EXTERNAL_API_ERROR", codigo="EXTERNAL_API_ERROR", descripcion=error_msg, paso="Envío a Ventanilla Única", sugerencias=["Verifica las credenciales VU y vuelve a intentarlo."], ) # Extraer task_id externo si el API lo devolvió de inmediato en PENDING/PROCESSING external_task_id = response.get("task_id") or response.get("id") # ------------------------------------------------------------------ # # Paso 3 – polling del estado externo # # ------------------------------------------------------------------ # if external_task_id: _progress(self, 50, "Esperando respuesta de Ventanilla Única...") record.external_task_id = str(external_task_id) db.commit() polling_started_at = time.perf_counter() try: final_response = _poll_external(self, external, str(external_task_id)) except TimeoutError as exc: record.status = "failed" db.commit() _fail_task( self, error=str(exc), error_type="TIMEOUT", codigo="TIMEOUT", descripcion=str(exc), paso="Polling Ventanilla Única", sugerencias=["Vuelve a intentarlo o consulta el estado manualmente."], ) logger.info( "Digitalization external wait completed task_id=%s expediente_id=%s external_task_id=%s elapsed_s=%.2f", self.request.id, expediente_id, external_task_id, time.perf_counter() - polling_started_at, ) else: # El API devolvió resultado directo final_response = response final_state = str(final_response.get("state") or final_response.get("status") or "").upper() if final_state in {"ERROR", "FAILURE", "FAILED"}: error_detail = final_response.get("error_detail") or {} suggestions = error_detail.get("sugerencias") or [] if config_vu.get("_ws_key_source") == "fallback": suggestions = [ "No hay clave real de web service configurada en VU ni en la empresa; se usó la clave fallback del sistema.", *suggestions, ] error_msg = ( final_response.get("error") or final_response.get("message") or error_detail.get("descripcion") or final_response.get("status") or "Error en Ventanilla Única" ) record.status = "failed" db.commit() _fail_task( self, error=str(error_msg), error_type=str(final_response.get("error_type") or "EXTERNAL_API_ERROR"), codigo=str(error_detail.get("codigo") or "EXTERNAL_TASK_FAILURE"), descripcion=str(error_detail.get("descripcion") or error_msg), paso=str(error_detail.get("paso") or "Respuesta final de Ventanilla Única"), sugerencias=suggestions or ["Revisa el detalle devuelto por Ventanilla Única y vuelve a intentarlo."], ) # ------------------------------------------------------------------ # # Paso 4 – persistir resultado # # ------------------------------------------------------------------ # _progress(self, 90, "Guardando resultado...") result_payload = final_response.get("result") or final_response e_doc = result_payload.get("e_document") or result_payload.get("eDocument") num_op = result_payload.get("numero_operacion") or result_payload.get("numeroOperacion") record.status = "success" if e_doc: record.e_document = str(e_doc) if num_op: record.num_operacion = str(num_op) # Guardar todos los artefactos base64 en S3 _ARTIFACT_FIELDS = { "acuse": ("acuese_digitalizacion_pdf_base64", "application/pdf", "acuse_pdf_path"), "envio_xml": ("envio_xml_base64", "application/xml", "envio_xml_path"), "respuesta_xml": ("respuesta_xml_base64", "application/xml", "respuesta_xml_path"), "consulta_envio_xml": ("consulta_envio_xml_base64", "application/xml", "consulta_envio_xml_path"), "consulta_respuesta_xml": ("consulta_respuesta_xml_base64", "application/xml", "consulta_respuesta_xml_path"), } artifact_ts = time.strftime("%Y%m%d_%H%M%S", time.gmtime()) for artifact_type, (result_field, content_type, record_field) in _ARTIFACT_FIELDS.items(): b64 = result_payload.get(result_field) if not b64: continue try: key = expediente_archivo_artifact_key( tenant_id, company_id, expediente_id, artifact_type, artifact_ts ) put_object_bytes(key, base64.b64decode(b64), content_type=content_type) setattr(record, record_field, key) logger.info( "Digitalization artifact saved task_id=%s expediente_id=%s type=%s key=%s", self.request.id, expediente_id, artifact_type, key, ) except Exception: logger.exception( "Failed to save artifact %s to S3 task_id=%s expediente_id=%s", artifact_type, self.request.id, expediente_id, ) setattr(record, record_field, None) db.commit() logger.info( "Digitalization task finished task_id=%s expediente_id=%s status=success total_elapsed_s=%.2f", self.request.id, expediente_id, time.perf_counter() - task_started_at, ) return { "status": "success", "message": "Digitalización completada exitosamente.", "e_document": e_doc, "numero_operacion": num_op, "nombre_archivo": payload["nombre_archivo"], "timestamp": result_payload.get("timestamp"), "request_id": result_payload.get("request_id"), "response_code": result_payload.get("response_code"), } except Ignore: raise except ValidationException as exc: if db: try: record = db.get(ExpedienteArchivo, expediente_id) # type: ignore if record: record.status = "failed" db.commit() except Exception: pass first_error = (exc.errors or [{}])[0] self.update_state( state="PROGRESS", meta={"current": 0, "status": "Preparando error de validación..."}, ) logger.info( "Digitalization task failed by validation task_id=%s expediente_id=%s elapsed_s=%.2f", self.request.id, expediente_id, time.perf_counter() - task_started_at, ) _fail_task( self, error=first_error.get("message", str(exc)), error_type="VALIDATION_ERROR", codigo=first_error.get("code", "VALIDATION_ERROR"), descripcion=first_error.get("message", ""), paso="Validación", sugerencias=first_error.get("solution") or [], ) except Exception as exc: logger.exception( "Error inesperado en digitalizar_task task_id=%s expediente_id=%s elapsed_s=%.2f", self.request.id, expediente_id, time.perf_counter() - task_started_at, ) if db: try: record = db.get(ExpedienteArchivo, expediente_id) # type: ignore if record: record.status = "failed" db.commit() except Exception: pass _fail_task( self, error=str(exc), error_type=type(exc).__name__, codigo="UNEXPECTED_ERROR", descripcion=str(exc), paso="Proceso de digitalización", sugerencias=["Contacta al soporte técnico."], ) finally: db.close()