570 lines
22 KiB
Python
570 lines
22 KiB
Python
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()
|