ADUANASOFT/plantillas-proyectos
13
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feature/digitalizacion-api

Browse Source
This commit is contained in:
hreyes
2026-04-17 15:59:47 -06:00
parent fa3cbb4d0a
commit 129608bb31
21 changed files with 2573 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
backend/api/v1/modules/a76/expediente_archivos/__init__.py Normal file
View File

133
backend/api/v1/modules/a76/expediente_archivos/dto.py Normal file
View File

@@ -0,0 +1,133 @@
from __future__ import annotations
from datetime import date
from typing import List, Optional
from pydantic import BaseModel, Field
# ---------------------------------------------------------------------------
# CRUD DTOs
# ---------------------------------------------------------------------------
class ExpedienteArchivoCreateDTO(BaseModel):
e_document: Optional[str] = Field(None, max_length=50)
num_operacion: Optional[str] = Field(None, max_length=50)
tipo_documento: Optional[str] = Field(None, max_length=10)
archivo_digitalizado_en: Optional[str] = Field(None, max_length=500)
fecha_digitalizacion: Optional[date] = None
agente_aduanal: Optional[str] = Field(None, max_length=50)
pedimento: Optional[str] = Field(None, max_length=21)
rfc_consulta: Optional[str] = Field(None, max_length=13)
nombre_archivo: Optional[str] = Field(None, max_length=255)
class ExpedienteArchivoUpdateDTO(BaseModel):
e_document: Optional[str] = Field(None, max_length=50)
num_operacion: Optional[str] = Field(None, max_length=50)
tipo_documento: Optional[str] = Field(None, max_length=10)
archivo_digitalizado_en: Optional[str] = Field(None, max_length=500)
fecha_digitalizacion: Optional[date] = None
agente_aduanal: Optional[str] = Field(None, max_length=50)
pedimento: Optional[str] = Field(None, max_length=21)
rfc_consulta: Optional[str] = Field(None, max_length=13)
nombre_archivo: Optional[str] = Field(None, max_length=255)
class ExpedienteArchivoResponseDTO(BaseModel):
id: int
e_document: Optional[str] = None
num_operacion: Optional[str] = None
tipo_documento: Optional[str] = None
archivo_digitalizado_en: Optional[str] = None
fecha_digitalizacion: Optional[date] = None
agente_aduanal: Optional[str] = None
pedimento: Optional[str] = None
rfc_consulta: Optional[str] = None
nombre_archivo: Optional[str] = None
status: Optional[str] = None
task_id: Optional[str] = None
external_task_id: Optional[str] = None
acuse_pdf_path: Optional[str] = None
company_id: int
tenant_id: int
model_config = {"from_attributes": True}
class ExpedienteArchivoListResponse(BaseModel):
items: List[ExpedienteArchivoResponseDTO]
total: int
page: int
page_size: int
# ---------------------------------------------------------------------------
# Digitalización DTOs
# ---------------------------------------------------------------------------
class DigitalizarRequest(BaseModel):
"""
Solicitud de digitalización enviada por el frontend.
La configuracion_vu se ensambla server-side desde CustomsBrokerVU / company.
"""
rfc_consulta: str = Field(..., max_length=13)
clave_documento: str = Field(..., max_length=10)
nombre_archivo: str = Field(..., max_length=255)
archivo_base64: str # contenido del archivo en base64
class RegistrarDigitalizacionRequest(BaseModel):
"""Digitalizar múltiples expedientes existentes por ID."""
rfc_consulta: str = Field(..., max_length=13)
ids_archivos: List[int]
class DigitalizarResponse(BaseModel):
task_id: str
message: str
status: str
# ---------------------------------------------------------------------------
# Task status DTOs (espejo del schema del servicio externo)
# ---------------------------------------------------------------------------
class DigitalizacionResult(BaseModel):
status: Optional[str] = None
message: Optional[str] = None
request_id: Optional[str] = None
response_code: Optional[int] = None
xml_path: Optional[str] = None
response_path: Optional[str] = None
acuese_digitalizacion_pdf_base64: Optional[str] = None
nombre_archivo: Optional[str] = None
timestamp: Optional[str] = None
numero_operacion: Optional[str] = None
e_document: Optional[str] = None
envio_xml_base64: Optional[str] = None
respuesta_xml_base64: Optional[str] = None
consulta_envio_xml_base64: Optional[str] = None
consulta_respuesta_xml_base64: Optional[str] = None
class DigitalizacionErrorDetail(BaseModel):
codigo: Optional[str] = None
descripcion: Optional[str] = None
paso: Optional[str] = None
sugerencias: Optional[List[str]] = None
class DigitalizacionTaskDetailResponse(BaseModel):
task_id: str
state: str
status: Optional[str] = None
current_step: Optional[str] = None
progress: Optional[int] = None
total_steps: Optional[int] = None
request_id: Optional[str] = None
result: Optional[DigitalizacionResult] = None
error: Optional[str] = None
error_type: Optional[str] = None
error_detail: Optional[DigitalizacionErrorDetail] = None
info: Optional[dict] = None

60
backend/api/v1/modules/a76/expediente_archivos/external_service.py Normal file
View File

@@ -0,0 +1,60 @@
from __future__ import annotations
import logging
from typing import Any, Dict, Optional
import httpx
from core.config import settings
logger = logging.getLogger(__name__)
class ExpedienteExternalService:
"""
Cliente HTTP real para el API externo de digitalización de archivos en Ventanilla Única.
Endpoints utilizados:
POST {base_url}/api/v1/expediente-archivos/digitalizar-archivo-json
GET {base_url}/api/v1/expediente-archivos/status-digitalizacion-task/{task_id}
"""
def __init__(self) -> None:
self.base_url = (settings.COVE_API_URL or "").strip() or "https://api.vu.aduanasoft.com"
def digitalizar_archivo_json(self, payload: Dict[str, Any]) -> Dict[str, Any]:
"""
Envía un JSON con el documento + configuracion_vu al endpoint de digitalización.
Retorna la respuesta JSON tal como viene del API externo.
"""
url = f"{self.base_url.rstrip('/')}/api/v1/expediente-archivos/digitalizar-archivo-json"
configuracion_vu = payload.get("configuracion_vu") or {}
logger.info(
"Sending digitalizar-archivo-json: nombre_archivo=%s rfc_vu=%s clave_fiel_len=%s clave_ws_len=%s cer_len=%s key_len=%s",
payload.get("nombre_archivo"),
configuracion_vu.get("rfc_usuario_vu"),
len(configuracion_vu.get("clave_fiel") or ""),
len(configuracion_vu.get("clave_webservice") or ""),
len(configuracion_vu.get("archivo_cer_base64") or ""),
len(configuracion_vu.get("archivo_key_base64") or ""),
)
# verify=False: el entorno externo puede usar certificados auto-firmados,
# igual que en factura_cove.
with httpx.Client(timeout=60.0, verify=False) as client:
response = client.post(url, json=payload)
response.raise_for_status()
return response.json()
def get_status(self, task_id: str) -> Optional[Dict[str, Any]]:
"""
Consulta el estado de una tarea de digitalización en el API externo.
"""
url = f"{self.base_url.rstrip('/')}/api/v1/expediente-archivos/status-digitalizacion-task/{task_id}"
logger.debug("Consulting expediente task status: task_id=%s url=%s", task_id, url)
with httpx.Client(timeout=30.0, verify=False) as client:
response = client.get(url)
response.raise_for_status()
return response.json()

35
backend/api/v1/modules/a76/expediente_archivos/models.py Normal file
View File

@@ -0,0 +1,35 @@
from __future__ import annotations
from datetime import date
from sqlalchemy import Date, Integer, String
from sqlalchemy.orm import Mapped, mapped_column
from api.v1.common.base_models import TenantScopedMixin, TimestampMixin
from core.database import Base
class ExpedienteArchivo(Base, TenantScopedMixin, TimestampMixin):
"""Registro de documentos digitalizados en Ventanilla Única."""
__tablename__ = "expediente_archivo"
__table_args__ = ({"schema": "a76"},)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
# Datos del documento (campos del catálogo legacy)
e_document: Mapped[str | None] = mapped_column(String(50), nullable=True)
num_operacion: Mapped[str | None] = mapped_column(String(50), nullable=True)
tipo_documento: Mapped[str | None] = mapped_column(String(10), nullable=True)
archivo_digitalizado_en: Mapped[str | None] = mapped_column(String(500), nullable=True)
fecha_digitalizacion: Mapped[date | None] = mapped_column(Date, nullable=True)
agente_aduanal: Mapped[str | None] = mapped_column(String(50), nullable=True)
pedimento: Mapped[str | None] = mapped_column(String(21), nullable=True)
rfc_consulta: Mapped[str | None] = mapped_column(String(13), nullable=True)
nombre_archivo: Mapped[str | None] = mapped_column(String(255), nullable=True)
# Estado de la tarea de digitalización
status: Mapped[str | None] = mapped_column(String(20), nullable=True) # pending/processing/success/failed
task_id: Mapped[str | None] = mapped_column(String(255), nullable=True, index=True)
external_task_id: Mapped[str | None] = mapped_column(String(255), nullable=True, index=True)
acuse_pdf_path: Mapped[str | None] = mapped_column(String(500), nullable=True)

182
backend/api/v1/modules/a76/expediente_archivos/routes.py Normal file
View File

@@ -0,0 +1,182 @@
from __future__ import annotations
from typing import Any, Dict
from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy.orm import Session
from core.database import get_core_db
from core.security import get_current_user, validate_access_to_resource
from .dto import (
DigitalizacionTaskDetailResponse,
DigitalizarRequest,
DigitalizarResponse,
ExpedienteArchivoCreateDTO,
ExpedienteArchivoListResponse,
ExpedienteArchivoResponseDTO,
ExpedienteArchivoUpdateDTO,
RegistrarDigitalizacionRequest,
)
from .models import ExpedienteArchivo
from .service import ExpedienteArchivoService
from .tasks import digitalizar_task
import logging
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/expediente-archivos")
# ──────────────────────────────────────────────────────────────────────────── #
# CRUD #
# ──────────────────────────────────────────────────────────────────────────── #
@router.get("/", response_model=ExpedienteArchivoListResponse)
def list_expediente_archivos(
company_id: int = Query(...),
page: int = Query(1, ge=1),
page_size: int = Query(50, ge=1, le=200),
search: str = Query(None),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
tenant_id = validate_access_to_resource(db, company_id, current_user)
return ExpedienteArchivoService.list(db, company_id, tenant_id, page, page_size, search)
@router.get("/{record_id}", response_model=ExpedienteArchivoResponseDTO)
def get_expediente_archivo(
record_id: int,
company_id: int = Query(...),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
tenant_id = validate_access_to_resource(db, company_id, current_user)
record = ExpedienteArchivoService.get(db, record_id, company_id, tenant_id)
if not record:
raise HTTPException(status_code=404, detail="Expediente archivo no encontrado.")
return ExpedienteArchivoResponseDTO.model_validate(record)
@router.post("/", response_model=ExpedienteArchivoResponseDTO, status_code=status.HTTP_201_CREATED)
def create_expediente_archivo(
dto: ExpedienteArchivoCreateDTO,
company_id: int = Query(...),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
tenant_id = validate_access_to_resource(db, company_id, current_user)
record = ExpedienteArchivoService.create(db, dto, company_id, tenant_id)
return ExpedienteArchivoResponseDTO.model_validate(record)
@router.put("/{record_id}", response_model=ExpedienteArchivoResponseDTO)
def update_expediente_archivo(
record_id: int,
dto: ExpedienteArchivoUpdateDTO,
company_id: int = Query(...),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
tenant_id = validate_access_to_resource(db, company_id, current_user)
record = ExpedienteArchivoService.get(db, record_id, company_id, tenant_id)
if not record:
raise HTTPException(status_code=404, detail="Expediente archivo no encontrado.")
record = ExpedienteArchivoService.update(db, record, dto)
return ExpedienteArchivoResponseDTO.model_validate(record)
@router.delete("/{record_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_expediente_archivo(
record_id: int,
company_id: int = Query(...),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
tenant_id = validate_access_to_resource(db, company_id, current_user)
record = ExpedienteArchivoService.get(db, record_id, company_id, tenant_id)
if not record:
raise HTTPException(status_code=404, detail="Expediente archivo no encontrado.")
ExpedienteArchivoService.delete(db, record)
return None
# ──────────────────────────────────────────────────────────────────────────── #
# Digitalización #
# ──────────────────────────────────────────────────────────────────────────── #
@router.post("/digitalizar/{record_id}", response_model=DigitalizarResponse)
def digitalizar_expediente_archivo(
record_id: int,
body: DigitalizarRequest,
company_id: int = Query(...),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
"""
Lanza la tarea Celery de digitalización para un expediente existente.
"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
record = ExpedienteArchivoService.get(db, record_id, company_id, tenant_id)
if not record:
raise HTTPException(status_code=404, detail="Expediente archivo no encontrado.")
task = digitalizar_task.apply_async(
kwargs={
"expediente_id": record_id,
"request_data": body.model_dump(),
"company_id": company_id,
"tenant_id": tenant_id,
}
)
return DigitalizarResponse(
task_id=task.id,
message="Tarea de digitalización iniciada.",
status="pending",
)
@router.post("/registrar-digitalizacion/", response_model=Dict[str, Any])
def registrar_digitalizacion(
body: RegistrarDigitalizacionRequest,
company_id: int = Query(...),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
"""
Lanza tareas de digitalización en batch para múltiples expedientes.
"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
launched = []
errors_list = []
for record_id in body.ids_archivos:
record = ExpedienteArchivoService.get(db, record_id, company_id, tenant_id)
if not record:
errors_list.append({"id": record_id, "error": "No encontrado"})
continue
task = digitalizar_task.apply_async(
kwargs={
"expediente_id": record_id,
"request_data": {"rfc_consulta": body.rfc_consulta},
"company_id": company_id,
"tenant_id": tenant_id,
}
)
launched.append({"id": record_id, "task_id": task.id})
return {"launched": launched, "errors": errors_list}
@router.get("/status-digitalizacion-task/{task_id}", response_model=DigitalizacionTaskDetailResponse)
def get_digitalizacion_task_status(
task_id: str,
current_user: Dict[str, Any] = Depends(get_current_user),
):
"""
Consulta el estado de una tarea Celery de digitalización.
"""
return ExpedienteArchivoService.get_task_status(task_id)

359
backend/api/v1/modules/a76/expediente_archivos/service.py Normal file
View File

@@ -0,0 +1,359 @@
from __future__ import annotations
import base64
import logging
from datetime import datetime
from typing import List, Optional
from cryptography.hazmat.primitives import padding as crypto_padding
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from sqlalchemy.orm import Session
from core.celery_app import celery_app
from core.config import settings
from core.exceptions import ErrorCollector, ValidationException
from core.storage_s3 import get_object_bytes, object_exists
from api.v1.modules.a76.customs_brokers import models as cb_models
from api.v1.modules.a76.general_catalogs.company.models import Company
from .dto import (
DigitalizacionErrorDetail,
DigitalizacionResult,
DigitalizacionTaskDetailResponse,
ExpedienteArchivoCreateDTO,
ExpedienteArchivoListResponse,
ExpedienteArchivoResponseDTO,
ExpedienteArchivoUpdateDTO,
)
from .models import ExpedienteArchivo
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# CRUD service
# ---------------------------------------------------------------------------
class ExpedienteArchivoService:
@staticmethod
def list(
db: Session,
company_id: int,
tenant_id: int,
page: int = 1,
page_size: int = 50,
search: Optional[str] = None,
) -> ExpedienteArchivoListResponse:
query = (
db.query(ExpedienteArchivo)
.filter(
ExpedienteArchivo.company_id == company_id,
ExpedienteArchivo.tenant_id == tenant_id,
ExpedienteArchivo.deleted_at.is_(None),
)
)
if search:
like = f"%{search}%"
query = query.filter(
ExpedienteArchivo.e_document.ilike(like)
| ExpedienteArchivo.tipo_documento.ilike(like)
)
total = query.count()
items = query.order_by(ExpedienteArchivo.id.desc()).offset((page - 1) * page_size).limit(page_size).all()
return ExpedienteArchivoListResponse(
items=[ExpedienteArchivoResponseDTO.model_validate(r) for r in items],
total=total,
page=page,
page_size=page_size,
)
@staticmethod
def get(db: Session, record_id: int, company_id: int, tenant_id: int) -> Optional[ExpedienteArchivo]:
return (
db.query(ExpedienteArchivo)
.filter(
ExpedienteArchivo.id == record_id,
ExpedienteArchivo.company_id == company_id,
ExpedienteArchivo.tenant_id == tenant_id,
ExpedienteArchivo.deleted_at.is_(None),
)
.first()
)
@staticmethod
def create(db: Session, dto: ExpedienteArchivoCreateDTO, company_id: int, tenant_id: int) -> ExpedienteArchivo:
record = ExpedienteArchivo(
company_id=company_id,
tenant_id=tenant_id,
**dto.model_dump(exclude_none=False),
)
db.add(record)
db.commit()
db.refresh(record)
return record
@staticmethod
def update(
db: Session,
record: ExpedienteArchivo,
dto: ExpedienteArchivoUpdateDTO,
) -> ExpedienteArchivo:
for field, value in dto.model_dump(exclude_unset=True).items():
setattr(record, field, value)
db.commit()
db.refresh(record)
return record
@staticmethod
def delete(db: Session, record: ExpedienteArchivo) -> None:
from datetime import datetime
record.deleted_at = datetime.utcnow()
db.commit()
@staticmethod
def get_task_status(task_id: str) -> DigitalizacionTaskDetailResponse:
result = celery_app.AsyncResult(task_id)
state = result.state or "PENDING"
info = result.info or {}
if state == "SUCCESS":
raw = result.result or {}
return DigitalizacionTaskDetailResponse(
task_id=task_id,
state="SUCCESS",
status="success",
result=DigitalizacionResult(**{k: raw.get(k) for k in DigitalizacionResult.model_fields}),
progress=100,
total_steps=4,
)
if state == "FAILURE":
err = info if not isinstance(info, dict) else None
error_detail_raw = info.get("error_detail") if isinstance(info, dict) else None
return DigitalizacionTaskDetailResponse(
task_id=task_id,
state="FAILURE",
status="failed",
error=str(err or info.get("error", "")),
error_type=info.get("error_type") if isinstance(info, dict) else None,
error_detail=DigitalizacionErrorDetail(**(error_detail_raw or {})) if error_detail_raw else None,
)
# PROGRESS / PENDING / STARTED
if isinstance(info, dict):
return DigitalizacionTaskDetailResponse(
task_id=task_id,
state=state,
status="processing",
current_step=info.get("current_step") or info.get("status"),
progress=info.get("progress") or info.get("current"),
total_steps=info.get("total_steps") or 4,
)
return DigitalizacionTaskDetailResponse(task_id=task_id, state=state, status="pending")
# ---------------------------------------------------------------------------
# VU config builder (mirror of factura_cove/service.py _build_configuracion_vu)
# ---------------------------------------------------------------------------
def _encrypt_fiel(raw_fiel: str) -> str:
"""AES-256-CBC + PKCS7 + base64 — mismo esquema que factura_cove."""
normalized = (raw_fiel or "").strip()
if not normalized:
return ""
key_bytes = (settings.COVE_FIEL_HASH_KEY or "").encode("utf-8")
iv_bytes = (settings.COVE_FIEL_HASH_IV or "").encode("utf-8")
if not key_bytes or not iv_bytes:
return ""
key32 = key_bytes[:32].ljust(32, b"\0")
iv16 = iv_bytes[:16].ljust(16, b"\0")
padder = crypto_padding.PKCS7(algorithms.AES.block_size).padder()
padded = padder.update(normalized.encode("utf-8")) + padder.finalize()
cipher = Cipher(algorithms.AES(key32), modes.CBC(iv16))
enc = cipher.encryptor()
encrypted = enc.update(padded) + enc.finalize()
return base64.b64encode(encrypted).decode("ascii")
def build_configuracion_vu(
db: Session,
company_id: int,
tenant_id: int,
agente_aduanal_key: Optional[str],
errors: ErrorCollector,
) -> Optional[dict]:
"""
Construye el dict de configuracion_vu para el servicio externo de digitalización.
Prioridad: CustomsBrokerVU (agente) → company.ventanilla_unica → company_fiel_certificate.
"""
vu: Optional[cb_models.CustomsBrokerVU] = None
if agente_aduanal_key:
broker = (
db.query(cb_models.CustomsBroker)
.filter(
cb_models.CustomsBroker.license == agente_aduanal_key,
cb_models.CustomsBroker.company_id == company_id,
cb_models.CustomsBroker.tenant_id == tenant_id,
cb_models.CustomsBroker.deleted_at.is_(None),
)
.first()
)
if broker:
vu = broker.vu
else:
broker = (
db.query(cb_models.CustomsBroker)
.filter(
cb_models.CustomsBroker.company_id == company_id,
cb_models.CustomsBroker.tenant_id == tenant_id,
cb_models.CustomsBroker.deleted_at.is_(None),
)
.first()
)
if broker:
vu = broker.vu
company = db.get(Company, company_id)
company_vu = company.ventanilla_unica if company else None
company_fiel_certificate = None
if company:
for cert in (company.digital_certificates or []):
if (cert.certificate_type or "").strip().lower() == "fiel":
company_fiel_certificate = cert
break
if not vu and not company_vu and not company_fiel_certificate:
errors.add_error(
field="vu",
message="La empresa no tiene configuración VU ni certificado FIEL.",
solution=["Configura los datos VU del agente aduanal o de la empresa y sube certificado (.cer) y llave (.key)."],
code="MISSING_VU_CONFIGURATION",
)
return None
effective_ws_user = (
(
vu.web_service_user
or vu.doda_web_service_user
or getattr(company_vu, "webservice_user", None)
or ""
).strip()
if (vu or company_vu)
else ""
)
clave_fiel_value = ""
if vu and getattr(vu, "fiel_access_key", None):
clave_fiel_value = _encrypt_fiel(vu.fiel_access_key or "")
elif company_fiel_certificate:
secret = (
getattr(company_fiel_certificate, "access_key", None)
or getattr(company_fiel_certificate, "password", None)
or ""
)
clave_fiel_value = _encrypt_fiel(str(secret))
if not effective_ws_user:
errors.add_error(
field="vu",
message="Faltan credenciales de web service en VU.",
solution=["Captura usuario y clave de web service en la pestaña VU del agente aduanal."],
code="MISSING_VU_CREDENTIALS",
)
if not clave_fiel_value:
errors.add_error(
field="vu.clave_fiel",
message="La clave FIEL no está configurada en VU ni en la empresa.",
solution=["Captura la clave FIEL en la configuración VU del agente aduanal o en los certificados de la empresa."],
code="MISSING_FIEL_PASSWORD",
)
certificate_path = (
(getattr(vu, "certificate_path", None) or "").strip() if vu else ""
) or (
(getattr(company_fiel_certificate, "cer_file_path", None) or "").strip()
if company_fiel_certificate
else ""
)
key_path = (
(getattr(vu, "key_path", None) or "").strip() if vu else ""
) or (
(getattr(company_fiel_certificate, "key_file_path", None) or "").strip()
if company_fiel_certificate
else ""
)
if not (certificate_path and key_path):
errors.add_error(
field="vu",
message="No hay rutas de certificado o llave en la configuración VU.",
solution=["Sube el certificado (.cer) y la llave (.key) en la configuración VU del agente aduanal."],
code="MISSING_VU_CERT_KEY",
)
return None
cer_b64 = None
key_b64 = None
try:
if not object_exists(certificate_path):
errors.add_error(
field="vu.certificate_path",
message="El certificado VU no existe en el almacenamiento.",
solution=["Vuelve a subir el certificado en la configuración VU."],
code="VU_CERT_NOT_FOUND",
)
else:
cer_b64 = base64.b64encode(get_object_bytes(certificate_path)).decode("ascii")
if not object_exists(key_path):
errors.add_error(
field="vu.key_path",
message="La llave VU no existe en el almacenamiento.",
solution=["Vuelve a subir la llave en la configuración VU."],
code="VU_KEY_NOT_FOUND",
)
else:
key_b64 = base64.b64encode(get_object_bytes(key_path)).decode("ascii")
except Exception:
logger.exception("Error leyendo certificados VU desde almacenamiento")
errors.add_error(
field="vu",
message="Error leyendo certificados VU.",
solution=["Verifica la configuración de MinIO/S3."],
code="VU_STORAGE_ERROR",
)
if errors.has_errors():
return None
rfc_usuario_vu = (
(getattr(vu, "query_tax_id", None) or "").strip() if vu else ""
) or (
(getattr(company_vu, "query_rfc", None) or "").strip() if company_vu else ""
)
hardcoded_ws_key = (
"RZGd+CB4R6PfSrstOyN8Is9FXL9AK9NPFisyGGaEWa0vVHoVOl8v2SBcHBoGbt3T/4uHTGcsFQO3b7EonWVfugQjBooywbz74K+jM68j8/Y="
)
clave_webservice = (
(getattr(vu, "web_service_access_key", None) or "").strip()
or (getattr(company_vu, "webservice_password", None) or "").strip()
or hardcoded_ws_key
)
email = (getattr(vu, "vu_email", None) or "").strip() if vu else ""
return {
"rfc_usuario_vu": rfc_usuario_vu,
"clave_webservice": clave_webservice,
"archivo_cer_base64": cer_b64 or "",
"archivo_key_base64": key_b64 or "",
"clave_fiel": clave_fiel_value,
"email": email,
}

286
backend/api/v1/modules/a76/expediente_archivos/tasks.py Normal file
View File

@@ -0,0 +1,286 @@
from __future__ import annotations
import logging
import time
from typing import Any, Dict
from celery import Task
from core.celery_app import celery_app
from core.database import CoreSessionLocal
from core.exceptions import ErrorCollector, ValidationException
from .models import ExpedienteArchivo
from .service import ExpedienteArchivoService, build_configuracion_vu
from .external_service import ExpedienteExternalService
logger = logging.getLogger(__name__)
TOTAL_STEPS = 4
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.time()
last_payload: Dict[str, Any] = {}
while True:
elapsed = time.time() - start
if elapsed > timeout_seconds:
logger.error("Timeout en polling externo de digitalización: task_id=%s", external_task_id)
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 Exception as exc:
logger.exception("Error consultando estado externo de digitalización")
raise
last_payload = status_payload
state = str(status_payload.get("state") or "").upper()
progress_info = status_payload.get("progress") or {}
try:
percent = float(progress_info.get("progress", 0.0))
except (TypeError, ValueError):
percent = 0.0
current_step = (
progress_info.get("current_step")
or "Consultando estado en Ventanilla Única..."
)
_progress(task, int(percent), str(current_step))
if state in {"PENDING", "STARTED", "PROGRESS"} or not state:
time.sleep(5)
continue
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()
try:
# ------------------------------------------------------------------ #
# 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_vu = build_configuracion_vu(db, company_id, tenant_id, agente_key, errors)
if errors.has_errors():
error_list = errors._errors # type: ignore[attr-defined]
first = error_list[0] if error_list else {}
self.update_state(
state="FAILURE",
meta={
"error": first.get("message", "Error de configuración VU"),
"error_type": "VALIDATION_ERROR",
"error_detail": {
"codigo": first.get("code", "VALIDATION_ERROR"),
"descripcion": first.get("message", ""),
"paso": "Construcción de configuración VU",
"sugerencias": first.get("solution") or [],
},
},
)
return {}
# Actualizar status en DB
record.status = "processing"
record.task_id = self.request.id
db.commit()
# ------------------------------------------------------------------ #
# Paso 2 – construir payload y enviar al API externo #
# ------------------------------------------------------------------ #
_progress(self, 30, "Enviando documento a Ventanilla Única...")
payload = {
"rfc_consulta": request_data.get("rfc_consulta") or record.rfc_consulta or "",
"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": request_data.get("archivo_base64") or "",
"configuracion_vu": config_vu,
}
external = ExpedienteExternalService()
response = external.digitalizar_archivo_json(payload)
# 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()
self.update_state(
state="FAILURE",
meta={
"error": error_msg,
"error_type": "EXTERNAL_API_ERROR",
"error_detail": {
"codigo": "EXTERNAL_API_ERROR",
"descripcion": error_msg,
"paso": "Envío a Ventanilla Única",
"sugerencias": ["Verifica las credenciales VU y vuelve a intentarlo."],
},
},
)
return {}
# 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()
try:
final_response = _poll_external(self, external, str(external_task_id))
except TimeoutError as exc:
record.status = "failed"
db.commit()
self.update_state(
state="FAILURE",
meta={
"error": str(exc),
"error_type": "TIMEOUT",
"error_detail": {
"codigo": "TIMEOUT",
"descripcion": str(exc),
"paso": "Polling Ventanilla Única",
"sugerencias": ["Vuelve a intentarlo o consulta el estado manualmente."],
},
},
)
return {}
else:
# El API devolvió resultado directo
final_response = response
# ------------------------------------------------------------------ #
# 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")
acuse_b64 = result_payload.get("acuese_digitalizacion_pdf_base64") or result_payload.get("acuse_pdf_base64")
record.status = "success"
if e_doc:
record.e_document = str(e_doc)
if num_op:
record.num_operacion = str(num_op)
# Guardamos el acuse en-línea (no en S3 por ahora) o en una ruta
# Los acuses se almacenan en el campo acuse_pdf_path como indicación
if acuse_b64:
record.acuse_pdf_path = "inline"
db.commit()
return {
"status": "success",
"message": "Digitalización completada exitosamente.",
"e_document": e_doc,
"numero_operacion": num_op,
"acuese_digitalizacion_pdf_base64": acuse_b64,
"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 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="FAILURE",
meta={
"error": first_error.get("message", str(exc)),
"error_type": "VALIDATION_ERROR",
"error_detail": {
"codigo": first_error.get("code", "VALIDATION_ERROR"),
"descripcion": first_error.get("message", ""),
"paso": "Validación",
"sugerencias": first_error.get("solution") or [],
},
},
)
return {}
except Exception as exc:
logger.exception("Error inesperado en digitalizar_task expediente_id=%s", expediente_id)
if db:
try:
record = db.get(ExpedienteArchivo, expediente_id) # type: ignore
if record:
record.status = "failed"
db.commit()
except Exception:
pass
self.update_state(
state="FAILURE",
meta={
"error": str(exc),
"error_type": type(exc).__name__,
"error_detail": {
"codigo": "UNEXPECTED_ERROR",
"descripcion": str(exc),
"paso": "Proceso de digitalización",
"sugerencias": ["Contacta al soporte técnico."],
},
},
)
return {}
finally:
db.close()