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-20 15:13:48 -06:00
parent 32f0dcc35f
commit 1cd81ebc03
29 changed files with 3275 additions and 1004 deletions
Download Patch File Download Diff File Expand all files Collapse all files

239
backend/api/v1/modules/a76/doc_types_dig/routes.py
View File

@@ -1,170 +1,97 @@
from typing import List
from typing import Any, Dict, Optional
from api.v1.modules.a76.doc_types_dig.dto import (
DocumentTypeDigitizationCreate,
DocumentTypeDigitizationResponse,
DocumentTypeDigitizationUpdate,
)
from api.v1.modules.a76.doc_types_dig.models import DocumentTypeDigitization
from core.database import get_core_db
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy import select
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session
router = APIRouter(prefix="/document-types-digitization", tags=["Document Types Digitization"])
from core.database import get_core_db
from core.security import get_current_user, validate_access_to_resource
from .dto import DocumentTypeDigitizationResponse
from .service import DocumentTypeDigitizationService
router = APIRouter(prefix='/document-types-digitization', tags=['Document Types Digitization'])
@router.get("", response_model=List[DocumentTypeDigitizationResponse])
def get_all_document_types(
active_only: bool = True,
db: Session = Depends(get_core_db),
@router.get('/', response_model=Dict[str, Any])
async def list_document_types(
company_id: int = Query(..., description='Company ID'),
page: int = Query(1, ge=1, description='Page number'),
page_size: int = Query(50, ge=1, le=2000, description='Page size'),
search: Optional[str] = Query(None, description='Search by code or description'),
active_only: bool = Query(False, description='Only active records'),
sort_by: Optional[str] = Query('code', description='Column to sort by'),
sort_order: str = Query('asc', pattern='^(asc|desc)$', description='Sort order'),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Obtener todos los tipos de documentos para digitalización
Args:
active_only: Si es True, solo devuelve los tipos activos
"""
query = select(DocumentTypeDigitization)
if active_only:
query = query.where(DocumentTypeDigitization.active == True)
query = query.order_by(DocumentTypeDigitization.code)
result = db.execute(query)
document_types = result.scalars().all()
return document_types
tenant_id = validate_access_to_resource(db, company_id, current_user)
skip = (page - 1) * page_size
filters: Dict[str, Any] = {}
if search:
filters['search'] = search
if active_only:
filters['active_only'] = True
items, total = DocumentTypeDigitizationService.get_all(
db,
tenant_id,
company_id,
skip,
page_size,
filters,
sort_by,
sort_order,
)
return {
'items': [DocumentTypeDigitizationResponse.model_validate(item) for item in items],
'total': total,
'page': page,
'page_size': page_size,
}
@router.get("/{document_type_id}", response_model=DocumentTypeDigitizationResponse)
def get_document_type(
document_type_id: int,
db: Session = Depends(get_core_db),
@router.get('/{document_type_id}/', response_model=DocumentTypeDigitizationResponse)
async def get_document_type(
document_type_id: int,
company_id: int = Query(..., description='Company ID'),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""Obtener un tipo de documento por ID"""
result = db.execute(
select(DocumentTypeDigitization).where(DocumentTypeDigitization.id == document_type_id)
)
document_type = result.scalar_one_or_none()
if not document_type:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Tipo de documento con ID {document_type_id} no encontrado"
)
return document_type
tenant_id = validate_access_to_resource(db, company_id, current_user)
document_type = DocumentTypeDigitizationService.get_by_id(
db,
document_type_id,
tenant_id,
company_id,
)
if not document_type:
raise HTTPException(
status_code=404,
detail=f'Tipo de documento con ID {document_type_id} no encontrado',
)
return document_type
@router.get("/by-code/{code}", response_model=DocumentTypeDigitizationResponse)
def get_document_type_by_code(
code: str,
db: Session = Depends(get_core_db),
@router.get('/by-code/{code}/', response_model=DocumentTypeDigitizationResponse)
async def get_document_type_by_code(
code: str,
company_id: int = Query(..., description='Company ID'),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""Obtener un tipo de documento por código"""
result = db.execute(
select(DocumentTypeDigitization).where(DocumentTypeDigitization.code == code)
)
document_type = result.scalar_one_or_none()
if not document_type:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Tipo de documento con código {code} no encontrado"
)
return document_type
tenant_id = validate_access_to_resource(db, company_id, current_user)
document_type = DocumentTypeDigitizationService.get_by_code(
db,
code,
tenant_id,
company_id,
)
if not document_type:
raise HTTPException(status_code=404, detail=f'Tipo de documento con codigo {code} no encontrado')
@router.post("", response_model=DocumentTypeDigitizationResponse, status_code=status.HTTP_201_CREATED)
def create_document_type(
document_type_data: DocumentTypeDigitizationCreate,
db: Session = Depends(get_core_db),
):
"""Crear un nuevo tipo de documento"""
# Verificar si el código ya existe
result = db.execute(
select(DocumentTypeDigitization).where(DocumentTypeDigitization.code == document_type_data.code)
)
existing = result.scalar_one_or_none()
if existing:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Ya existe un tipo de documento con el código {document_type_data.code}"
)
new_document_type = DocumentTypeDigitization(**document_type_data.model_dump())
db.add(new_document_type)
db.commit()
db.refresh(new_document_type)
return new_document_type
@router.put("/{document_type_id}", response_model=DocumentTypeDigitizationResponse)
def update_document_type(
document_type_id: int,
document_type_data: DocumentTypeDigitizationUpdate,
db: Session = Depends(get_core_db),
):
"""Actualizar un tipo de documento existente"""
result = db.execute(
select(DocumentTypeDigitization).where(DocumentTypeDigitization.id == document_type_id)
)
document_type = result.scalar_one_or_none()
if not document_type:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Tipo de documento con ID {document_type_id} no encontrado"
)
# Actualizar solo los campos proporcionados
update_data = document_type_data.model_dump(exclude_unset=True)
# Verificar si el nuevo código ya existe (si se está actualizando)
if "code" in update_data and update_data["code"] != document_type.code:
result = db.execute(
select(DocumentTypeDigitization).where(DocumentTypeDigitization.code == update_data["code"])
)
existing = result.scalar_one_or_none()
if existing:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Ya existe un tipo de documento con el código {update_data['code']}"
)
for field, value in update_data.items():
setattr(document_type, field, value)
db.commit()
db.refresh(document_type)
return document_type
@router.delete("/{document_type_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_document_type(
document_type_id: int,
db: Session = Depends(get_core_db),
):
"""Eliminar un tipo de documento (soft delete, marca como inactivo)"""
result = db.execute(
select(DocumentTypeDigitization).where(DocumentTypeDigitization.id == document_type_id)
)
document_type = result.scalar_one_or_none()
if not document_type:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Tipo de documento con ID {document_type_id} no encontrado"
)
# Soft delete - solo marcar como inactivo
document_type.active = False
db.commit()
return None
return document_type

211
backend/api/v1/modules/a76/doc_types_dig/service.py Normal file
View File

@@ -0,0 +1,211 @@
from __future__ import annotations
from typing import Any, Dict, Optional, Tuple
from sqlalchemy import or_
from sqlalchemy.orm import Session
from .dto import DocumentTypeDigitizationCreate, DocumentTypeDigitizationUpdate
from .models import DocumentTypeDigitization
class DocumentTypeDigitizationService:
@staticmethod
def _normalize_code(code: str) -> str:
return (code or '').strip().upper()
@staticmethod
def _normalize_description(description: str) -> str:
return (description or '').strip()
@staticmethod
def _is_truthy_filter(value: Any, default: bool = False) -> bool:
if value is None:
return default
if isinstance(value, bool):
return value
return str(value).strip().lower() in {'1', 'true', 'yes', 'si'}
@staticmethod
def get_all(
db: Session,
tenant_id: int,
company_id: Optional[int],
skip: int = 0,
limit: int = 100,
filters: Optional[Dict[str, Any]] = None,
sort_by: Optional[str] = None,
sort_order: str = 'asc',
) -> Tuple[list[DocumentTypeDigitization], int]:
query = db.query(DocumentTypeDigitization).filter(
DocumentTypeDigitization.tenant_id == tenant_id,
)
if company_id is not None:
query = query.filter(DocumentTypeDigitization.company_id == company_id)
filters = filters or {}
active_only = DocumentTypeDigitizationService._is_truthy_filter(
filters.get('active_only'),
default=False,
)
search = (filters.get('search') or '').strip()
if active_only:
query = query.filter(DocumentTypeDigitization.active.is_(True))
if search:
like = f'%{search}%'
query = query.filter(
or_(
DocumentTypeDigitization.code.ilike(like),
DocumentTypeDigitization.description.ilike(like),
)
)
total = query.count()
sort_column = {
'id': DocumentTypeDigitization.id,
'code': DocumentTypeDigitization.code,
'description': DocumentTypeDigitization.description,
'active': DocumentTypeDigitization.active,
}.get(sort_by or 'code', DocumentTypeDigitization.code)
if sort_order == 'desc':
query = query.order_by(sort_column.desc())
else:
query = query.order_by(sort_column.asc())
items = query.offset(skip).limit(limit).all()
return items, total
@staticmethod
def get_by_id(
db: Session,
id: int,
tenant_id: int,
company_id: int,
) -> Optional[DocumentTypeDigitization]:
return (
db.query(DocumentTypeDigitization)
.filter(
DocumentTypeDigitization.id == id,
DocumentTypeDigitization.tenant_id == tenant_id,
DocumentTypeDigitization.company_id == company_id,
)
.first()
)
@staticmethod
def get_by_code(
db: Session,
code: str,
tenant_id: int,
company_id: int,
) -> Optional[DocumentTypeDigitization]:
normalized_code = DocumentTypeDigitizationService._normalize_code(code)
return (
db.query(DocumentTypeDigitization)
.filter(
DocumentTypeDigitization.code == normalized_code,
DocumentTypeDigitization.tenant_id == tenant_id,
DocumentTypeDigitization.company_id == company_id,
)
.first()
)
@staticmethod
def create(
db: Session,
data: DocumentTypeDigitizationCreate,
tenant_id: int,
company_id: int,
) -> DocumentTypeDigitization:
payload = data.model_dump()
payload['code'] = DocumentTypeDigitizationService._normalize_code(payload['code'])
payload['description'] = DocumentTypeDigitizationService._normalize_description(
payload['description']
)
if not payload['code']:
raise ValueError('El codigo es obligatorio')
if not payload['description']:
raise ValueError('La descripcion es obligatoria')
existing = DocumentTypeDigitizationService.get_by_code(
db, payload['code'], tenant_id, company_id
)
if existing:
raise ValueError(
f'Ya existe un tipo de documento con el codigo {payload["code"]}'
)
db_obj = DocumentTypeDigitization(
**payload,
tenant_id=tenant_id,
company_id=company_id,
)
db.add(db_obj)
db.commit()
db.refresh(db_obj)
return db_obj
@staticmethod
def update(
db: Session,
id: int,
tenant_id: int,
data: DocumentTypeDigitizationUpdate,
company_id: int,
) -> Optional[DocumentTypeDigitization]:
db_obj = DocumentTypeDigitizationService.get_by_id(db, id, tenant_id, company_id)
if not db_obj:
return None
update_dict = data.model_dump(exclude_unset=True)
if 'code' in update_dict:
update_dict['code'] = DocumentTypeDigitizationService._normalize_code(update_dict['code'])
if not update_dict['code']:
raise ValueError('El codigo es obligatorio')
if update_dict['code'] != db_obj.code:
existing = DocumentTypeDigitizationService.get_by_code(
db,
update_dict['code'],
tenant_id,
company_id,
)
if existing:
raise ValueError(
f'Ya existe un tipo de documento con el codigo {update_dict["code"]}'
)
if 'description' in update_dict:
update_dict['description'] = DocumentTypeDigitizationService._normalize_description(
update_dict['description']
)
if not update_dict['description']:
raise ValueError('La descripcion es obligatoria')
for key, value in update_dict.items():
setattr(db_obj, key, value)
db.commit()
db.refresh(db_obj)
return db_obj
@staticmethod
def delete(
db: Session,
id: int,
tenant_id: int,
company_id: int,
) -> bool:
db_obj = DocumentTypeDigitizationService.get_by_id(db, id, tenant_id, company_id)
if not db_obj:
return False
db_obj.active = False
db.commit()
return True

15
backend/api/v1/modules/a76/expediente_archivos/dto.py
View File

@@ -49,6 +49,10 @@ class ExpedienteArchivoResponseDTO(BaseModel):
task_id: Optional[str] = None
external_task_id: Optional[str] = None
acuse_pdf_path: Optional[str] = None
envio_xml_path: Optional[str] = None
respuesta_xml_path: Optional[str] = None
consulta_envio_xml_path: Optional[str] = None
consulta_respuesta_xml_path: Optional[str] = None
company_id: int
tenant_id: int
@@ -71,15 +75,15 @@ 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
rfc_consulta: Optional[str] = Field(None, max_length=13)
clave_documento: Optional[str] = Field(None, max_length=10)
nombre_archivo: Optional[str] = Field(None, max_length=255)
archivo_base64: Optional[str] = None # contenido del archivo en base64; opcional si el expediente ya tiene archivo almacenado
class RegistrarDigitalizacionRequest(BaseModel):
"""Digitalizar múltiples expedientes existentes por ID."""
rfc_consulta: str = Field(..., max_length=13)
rfc_consulta: Optional[str] = Field(None, max_length=13)
ids_archivos: List[int]
@@ -120,6 +124,7 @@ class DigitalizacionErrorDetail(BaseModel):
class DigitalizacionTaskDetailResponse(BaseModel):
task_id: str
external_task_id: Optional[str] = None
state: str
status: Optional[str] = None
current_step: Optional[str] = None

9
backend/api/v1/modules/a76/expediente_archivos/external_service.py
View File

@@ -40,9 +40,8 @@ class ExpedienteExternalService:
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:
# connect=10s, read=120s: la subida del PDF puede tomar tiempo en el servidor VU
with httpx.Client(timeout=httpx.Timeout(120.0, connect=10.0), verify=False) as client:
response = client.post(url, json=payload)
response.raise_for_status()
return response.json()
@@ -54,7 +53,9 @@ class ExpedienteExternalService:
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:
# read=None: sin límite de lectura — VU mantiene la conexión abierta mientras procesa.
# El timeout global del polling loop (300 s) actúa como cota máxima real.
with httpx.Client(timeout=httpx.Timeout(None, connect=10.0), verify=False) as client:
response = client.get(url)
response.raise_for_status()
return response.json()

4
backend/api/v1/modules/a76/expediente_archivos/models.py
View File

@@ -33,3 +33,7 @@ class ExpedienteArchivo(Base, TenantScopedMixin, TimestampMixin):
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)
envio_xml_path: Mapped[str | None] = mapped_column(String(500), nullable=True)
respuesta_xml_path: Mapped[str | None] = mapped_column(String(500), nullable=True)
consulta_envio_xml_path: Mapped[str | None] = mapped_column(String(500), nullable=True)
consulta_respuesta_xml_path: Mapped[str | None] = mapped_column(String(500), nullable=True)

205
backend/api/v1/modules/a76/expediente_archivos/routes.py
View File

@@ -1,12 +1,20 @@
from __future__ import annotations
from datetime import datetime
import io
import mimetypes
import os
import zipfile
from typing import Any, Dict
from fastapi import APIRouter, Depends, HTTPException, Query, status
from fastapi import APIRouter, Depends, File, HTTPException, Query, Response, UploadFile, status
from sqlalchemy.orm import Session
from core.config import settings
from core.database import get_core_db
from core.security import get_current_user, validate_access_to_resource
from core.s3_keys import expediente_archivo_document_key
from core.storage_s3 import delete_object_if_exists, get_object_bytes, object_exists, put_object_bytes
from .dto import (
DigitalizacionTaskDetailResponse,
@@ -28,6 +36,20 @@ logger = logging.getLogger(__name__)
router = APIRouter(prefix="/expediente-archivos")
def _remove_stored_document_path(path: str | None) -> None:
raw = (path or "").strip()
if not raw:
return
try:
if settings.use_s3_object_storage and not os.path.isabs(raw):
delete_object_if_exists(raw)
return
if os.path.exists(raw):
os.remove(raw)
except Exception:
logger.warning("No se pudo eliminar archivo previo de expediente: %s", raw, exc_info=True)
# ──────────────────────────────────────────────────────────────────────────── #
# CRUD #
# ──────────────────────────────────────────────────────────────────────────── #
@@ -98,10 +120,191 @@ def delete_expediente_archivo(
record = ExpedienteArchivoService.get(db, record_id, company_id, tenant_id)
if not record:
raise HTTPException(status_code=404, detail="Expediente archivo no encontrado.")
_remove_stored_document_path(record.archivo_digitalizado_en)
ExpedienteArchivoService.delete(db, record)
return None
_ARTIFACT_TYPE_MAP = {
"acuse": ("acuse_pdf_path", "application/pdf"),
"envio-xml": ("envio_xml_path", "application/xml"),
"respuesta-xml": ("respuesta_xml_path", "application/xml"),
"consulta-envio-xml": ("consulta_envio_xml_path", "application/xml"),
"consulta-respuesta-xml": ("consulta_respuesta_xml_path", "application/xml"),
}
@router.get("/{record_id}/artifacts/{artifact_type}", response_class=Response)
def download_artifact(
record_id: int,
artifact_type: str,
company_id: int = Query(...),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
"""Descarga un artefacto de digitalización (acuse PDF o XMLs) desde S3."""
tenant_id = validate_access_to_resource(db, company_id, current_user)
if artifact_type not in _ARTIFACT_TYPE_MAP:
raise HTTPException(status_code=400, detail=f"Tipo de artefacto no válido: {artifact_type}")
record = ExpedienteArchivoService.get(db, record_id, company_id, tenant_id)
if not record:
raise HTTPException(status_code=404, detail="Expediente archivo no encontrado.")
field_name, content_type = _ARTIFACT_TYPE_MAP[artifact_type]
key = (getattr(record, field_name, None) or "").strip()
if not key or key == "inline":
raise HTTPException(status_code=404, detail="Artefacto no disponible para este expediente.")
if not object_exists(key):
raise HTTPException(status_code=404, detail="El archivo no se encontró en el almacenamiento.")
ext = ".pdf" if content_type == "application/pdf" else ".xml"
filename = f"{artifact_type}_{record.e_document or record_id}{ext}"
return Response(
content=get_object_bytes(key),
media_type=content_type,
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
)
@router.get("/{record_id}/artifacts-zip", response_class=Response)
def download_artifacts_zip(
record_id: int,
company_id: int = Query(...),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
"""Descarga todos los artefactos disponibles de un expediente en un archivo ZIP."""
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.")
base_name = record.e_document or str(record_id)
artifact_files = [
("acuse", "acuse_pdf_path", f"acuse_{base_name}.pdf"),
("envio-xml", "envio_xml_path", f"envio_{base_name}.xml"),
("respuesta-xml", "respuesta_xml_path", f"respuesta_{base_name}.xml"),
("consulta-envio-xml", "consulta_envio_xml_path", f"consulta_envio_{base_name}.xml"),
("consulta-respuesta-xml", "consulta_respuesta_xml_path", f"consulta_respuesta_{base_name}.xml"),
]
buf = io.BytesIO()
added = 0
with zipfile.ZipFile(buf, mode="w", compression=zipfile.ZIP_DEFLATED) as zf:
for _type, field_name, filename in artifact_files:
key = (getattr(record, field_name, None) or "").strip()
if not key or key == "inline":
continue
if not object_exists(key):
continue
zf.writestr(filename, get_object_bytes(key))
added += 1
if added == 0:
raise HTTPException(status_code=404, detail="No hay artefactos disponibles para este expediente.")
buf.seek(0)
zip_filename = f"expediente_{base_name}.zip"
return Response(
content=buf.getvalue(),
media_type="application/zip",
headers={"Content-Disposition": f'attachment; filename="{zip_filename}"'},
)
@router.post("/{record_id}/upload", response_model=Dict[str, Any])
async def upload_expediente_archivo_file(
record_id: int,
company_id: int = Query(...),
file: UploadFile = File(...),
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.")
content = await file.read()
if not content:
raise HTTPException(status_code=400, detail="El archivo está vacío.")
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
# Borrar artefactos de digitalización previa al subir documento nuevo
_ARTIFACT_FIELDS = [
"acuse_pdf_path",
"envio_xml_path",
"respuesta_xml_path",
"consulta_envio_xml_path",
"consulta_respuesta_xml_path",
]
for _field in _ARTIFACT_FIELDS:
_old_key = (getattr(record, _field, None) or "").strip()
if _old_key and _old_key != "inline":
try:
delete_object_if_exists(_old_key)
except Exception:
logger.warning("No se pudo eliminar artefacto previo %s=%s", _field, _old_key, exc_info=True)
setattr(record, _field, None)
# El documento fuente cambió — la digitalización anterior ya no aplica
record.status = "pending"
record.e_document = None
record.num_operacion = None
record.task_id = None
record.external_task_id = None
try:
if settings.use_s3_object_storage:
key = expediente_archivo_document_key(
tenant_id,
company_id,
record.id,
timestamp,
file.filename or "documento.pdf",
)
ct = file.content_type or mimetypes.guess_type(file.filename or "")[0] or "application/octet-stream"
_remove_stored_document_path(record.archivo_digitalizado_en)
put_object_bytes(key, content, content_type=ct)
stored = key
else:
key = expediente_archivo_document_key(
tenant_id,
company_id,
record.id,
timestamp,
file.filename or "documento.pdf",
)
base = os.path.join("uploads", "expediente_archivos", str(company_id), str(record.id))
os.makedirs(base, exist_ok=True)
filename = key.rsplit("/", 1)[-1]
stored = os.path.join(base, filename)
_remove_stored_document_path(record.archivo_digitalizado_en)
with open(stored, "wb") as destination:
destination.write(content)
record.archivo_digitalizado_en = stored
record.nombre_archivo = file.filename or record.nombre_archivo
db.add(record)
db.commit()
db.refresh(record)
except HTTPException:
db.rollback()
raise
except ValueError as exc:
db.rollback()
raise HTTPException(status_code=400, detail=str(exc)) from exc
except Exception as exc:
db.rollback()
raise HTTPException(status_code=500, detail=f"Error saving file: {str(exc)}") from exc
return {
"message": "Archivo cargado correctamente",
"record_id": record.id,
"path": stored,
"nombre_archivo": record.nombre_archivo,
}
# ──────────────────────────────────────────────────────────────────────────── #
# Digitalización #
# ──────────────────────────────────────────────────────────────────────────── #

275
backend/api/v1/modules/a76/expediente_archivos/service.py
View File

@@ -7,10 +7,12 @@ 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 import or_
from sqlalchemy.orm import Session
from core.celery_app import celery_app
from core.config import settings
from core.database import CoreSessionLocal
from core.exceptions import ErrorCollector, ValidationException
from core.storage_s3 import get_object_bytes, object_exists
@@ -37,6 +39,34 @@ logger = logging.getLogger(__name__)
class ExpedienteArchivoService:
@staticmethod
def _get_task_record_metadata(task_id: str) -> dict:
db = CoreSessionLocal()
try:
record = (
db.query(ExpedienteArchivo)
.filter(
ExpedienteArchivo.task_id == task_id,
ExpedienteArchivo.deleted_at.is_(None),
)
.order_by(ExpedienteArchivo.id.desc())
.first()
)
if not record:
return {"external_task_id": None}
return {
"external_task_id": record.external_task_id,
"db_status": record.status,
"e_document": record.e_document,
"num_operacion": record.num_operacion,
"nombre_archivo": record.nombre_archivo,
}
except Exception:
logger.exception("No se pudo obtener metadata del expediente para task_id=%s", task_id)
return {"external_task_id": None}
finally:
db.close()
@staticmethod
def list(
db: Session,
@@ -114,30 +144,90 @@ class ExpedienteArchivoService:
@staticmethod
def get_task_status(task_id: str) -> DigitalizacionTaskDetailResponse:
result = celery_app.AsyncResult(task_id)
state = result.state or "PENDING"
info = result.info or {}
task_metadata = ExpedienteArchivoService._get_task_record_metadata(task_id)
try:
result = celery_app.AsyncResult(task_id)
state = result.state or "PENDING"
info = result.info or {}
except Exception as exc:
logger.exception("No se pudo consultar el estado de la tarea de digitalización task_id=%s", task_id)
return DigitalizacionTaskDetailResponse(
task_id=task_id,
external_task_id=task_metadata.get("external_task_id"),
state="FAILURE",
status="failed",
error="No se pudo consultar el estado de la digitalización.",
error_type=type(exc).__name__,
error_detail=DigitalizacionErrorDetail(
codigo="TASK_STATUS_ERROR",
descripcion=str(exc),
paso="Consulta de estado",
sugerencias=["Cierra el diálogo y vuelve a intentar la digitalización."],
),
)
if state == "SUCCESS":
raw = result.result or {}
return DigitalizacionTaskDetailResponse(
task_id=task_id,
external_task_id=task_metadata.get("external_task_id"),
state="SUCCESS",
status="success",
request_id=raw.get("request_id"),
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
# Fallback: Celery puede tardar en propagar SUCCESS a Redis.
# Si la DB ya tiene status=success, retornamos SUCCESS inmediatamente.
if task_metadata.get("db_status") == "success":
logger.info(
"get_task_status: Celery state=%s but DB status=success — returning SUCCESS from DB task_id=%s",
state, task_id,
)
return DigitalizacionTaskDetailResponse(
task_id=task_id,
external_task_id=task_metadata.get("external_task_id"),
state="SUCCESS",
status="success",
result=DigitalizacionResult(
status="success",
message="Digitalización completada exitosamente.",
e_document=task_metadata.get("e_document"),
numero_operacion=task_metadata.get("num_operacion"),
nombre_archivo=task_metadata.get("nombre_archivo"),
),
progress=100,
total_steps=4,
)
if state in {"FAILURE", "FAILED"}:
err = info if not isinstance(info, dict) else None
error_detail_raw = info.get("error_detail") if isinstance(info, dict) else None
error_text = str(err or info.get("error", "")) if isinstance(info, dict) else str(err or "")
error_type = info.get("error_type") if isinstance(info, dict) else None
if not error_type and isinstance(info, BaseException):
error_type = type(info).__name__
if error_type == "Ignore" and not error_text:
error_text = "La digitalización no pudo completarse."
if not error_text and isinstance(info, BaseException):
error_text = "La digitalización no pudo completarse."
if isinstance(info, BaseException) and not error_detail_raw:
error_detail_raw = {
"codigo": "TASK_FAILED",
"descripcion": "La tarea terminó con error antes de completar la digitalización.",
"paso": "Proceso de digitalización",
"sugerencias": ["Revisa la configuración VU y vuelve a intentarlo."],
}
return DigitalizacionTaskDetailResponse(
task_id=task_id,
external_task_id=task_metadata.get("external_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,
request_id=info.get("request_id") if isinstance(info, dict) else None,
error=error_text,
error_type=error_type,
error_detail=DigitalizacionErrorDetail(**(error_detail_raw or {})) if error_detail_raw else None,
)
@@ -145,14 +235,21 @@ class ExpedienteArchivoService:
if isinstance(info, dict):
return DigitalizacionTaskDetailResponse(
task_id=task_id,
external_task_id=task_metadata.get("external_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,
request_id=info.get("request_id"),
)
return DigitalizacionTaskDetailResponse(task_id=task_id, state=state, status="pending")
return DigitalizacionTaskDetailResponse(
task_id=task_id,
external_task_id=task_metadata.get("external_task_id"),
state=state,
status="pending",
)
# ---------------------------------------------------------------------------
@@ -178,6 +275,94 @@ def _encrypt_fiel(raw_fiel: str) -> str:
return base64.b64encode(encrypted).decode("ascii")
def _resolve_broker_for_vu(
db: Session,
company_id: int,
tenant_id: int,
agente_aduanal_key: str,
) -> Optional[cb_models.CustomsBroker]:
normalized_key = (agente_aduanal_key or "").strip()
if not normalized_key:
return None
brokers = (
db.query(cb_models.CustomsBroker)
.filter(
or_(
cb_models.CustomsBroker.broker_key == normalized_key,
cb_models.CustomsBroker.license == normalized_key,
),
cb_models.CustomsBroker.company_id == company_id,
cb_models.CustomsBroker.tenant_id == tenant_id,
cb_models.CustomsBroker.deleted_at.is_(None),
)
.order_by(cb_models.CustomsBroker.id.desc())
.all()
)
if not brokers:
return None
exact_broker_key = next(
(broker for broker in brokers if (broker.broker_key or "").strip() == normalized_key),
None,
)
if exact_broker_key:
return exact_broker_key
if len(brokers) > 1:
logger.warning(
"Multiple customs brokers matched agente_aduanal=%s; falling back to first license match ids=%s broker_keys=%s",
normalized_key,
[broker.id for broker in brokers],
[broker.broker_key for broker in brokers],
)
return brokers[0]
def resolve_rfc_consulta_value(
db: Session,
company_id: int,
tenant_id: int,
agente_aduanal_key: Optional[str],
request_rfc_consulta: Optional[str],
record_rfc_consulta: Optional[str],
config_vu_rfc: Optional[str],
) -> str:
explicit_rfc = (request_rfc_consulta or "").strip().upper()
if explicit_rfc:
return explicit_rfc
broker_tax_id = ""
if agente_aduanal_key:
broker = _resolve_broker_for_vu(db, company_id, tenant_id, agente_aduanal_key)
broker_tax_id = (getattr(broker, "tax_id", None) or "").strip().upper() if broker else ""
if broker_tax_id:
return broker_tax_id
stored_rfc = (record_rfc_consulta or "").strip().upper()
if stored_rfc:
return stored_rfc
config_rfc = (config_vu_rfc or "").strip().upper()
if config_rfc:
return config_rfc
raise ValidationException(
"RFC Consulta no disponible",
errors=[
{
"field": "rfc_consulta",
"message": "No se pudo resolver el RFC Consulta desde el expediente ni desde el customs broker.",
"code": "MISSING_RFC_CONSULTA",
"solution": [
"Configura el RFC del agente aduanal en el customs broker o captura el RFC directamente en el expediente."
],
}
],
)
def build_configuracion_vu(
db: Session,
company_id: int,
@@ -191,16 +376,7 @@ def build_configuracion_vu(
"""
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()
)
broker = _resolve_broker_for_vu(db, company_id, tenant_id, agente_aduanal_key)
if broker:
vu = broker.vu
else:
@@ -235,17 +411,6 @@ def build_configuracion_vu(
)
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 "")
@@ -257,12 +422,29 @@ def build_configuracion_vu(
)
clave_fiel_value = _encrypt_fiel(str(secret))
if not effective_ws_user:
hardcoded_ws_key = (
"RZGd+CB4R6PfSrstOyN8Is9FXL9AK9NPFisyGGaEWa0vVHoVOl8v2SBcHBoGbt3T/4uHTGcsFQO3b7EonWVfugQjBooywbz74K+jM68j8/Y="
)
vu_ws_key = (getattr(vu, "web_service_access_key", None) or "").strip() if vu else ""
vu_access_key_raw = (getattr(vu, "access_key", None) or "").strip() if vu else ""
vu_access_key_encrypted = _encrypt_fiel(vu_access_key_raw) if vu_access_key_raw else ""
company_ws_key = (getattr(company_vu, "webservice_password", None) or "").strip() if company_vu else ""
if vu_ws_key:
ws_key_source = "web_service_access_key"
elif vu_access_key_encrypted:
ws_key_source = "access_key_encrypted"
elif company_ws_key:
ws_key_source = "company"
else:
ws_key_source = "fallback"
clave_webservice = vu_ws_key or vu_access_key_encrypted or company_ws_key or hardcoded_ws_key
if not clave_webservice:
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",
field="vu.clave_webservice",
message="La clave de web service no está configurada en VU ni en la empresa.",
solution=["Captura la clave de web service en la pestaña VU o completa la configuración VU de la empresa."],
code="MISSING_VU_WS_KEY",
)
if not clave_fiel_value:
@@ -337,17 +519,21 @@ def build_configuracion_vu(
(getattr(company_vu, "query_rfc", None) or "").strip() if company_vu else ""
)
hardcoded_ws_key = (
"RZGd+CB4R6PfSrstOyN8Is9FXL9AK9NPFisyGGaEWa0vVHoVOl8v2SBcHBoGbt3T/4uHTGcsFQO3b7EonWVfugQjBooywbz74K+jM68j8/Y="
email = (
(getattr(vu, "vu_email", None) or "").strip() if vu else ""
) or (
(getattr(company_vu, "email", None) or "").strip() if company_vu else ""
) or (
(getattr(getattr(company, "address", None), "email", None) or "").strip() if company else ""
)
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 ""
if ws_key_source == "fallback":
logger.warning(
"Expediente digitalization is using fallback web service key for agente_aduanal=%s company_id=%s tenant_id=%s",
agente_aduanal_key,
company_id,
tenant_id,
)
return {
"rfc_usuario_vu": rfc_usuario_vu,
@@ -356,4 +542,5 @@ def build_configuracion_vu(
"archivo_key_base64": key_b64 or "",
"clave_fiel": clave_fiel_value,
"email": email,
"_ws_key_source": ws_key_source,
}

461
backend/api/v1/modules/a76/expediente_archivos/tasks.py
View File

@@ -1,17 +1,23 @@
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
from .service import ExpedienteArchivoService, build_configuracion_vu, resolve_rfc_consulta_value
from .external_service import ExpedienteExternalService
logger = logging.getLogger(__name__)
@@ -19,6 +25,95 @@ 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",
@@ -36,40 +131,85 @@ def _poll_external(
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()
start = time.perf_counter()
last_payload: Dict[str, Any] = {}
attempts = 0
while True:
elapsed = time.time() - start
attempts += 1
elapsed = time.perf_counter() - start
if elapsed > timeout_seconds:
logger.error("Timeout en polling externo de digitalización: task_id=%s", external_task_id)
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 Exception as exc:
logger.exception("Error consultando estado externo de digitalización")
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") or {}
try:
percent = float(progress_info.get("progress", 0.0))
except (TypeError, ValueError):
percent = 0.0
current_step = (
progress_info.get("current_step")
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
@@ -91,7 +231,16 @@ def digitalizar_task(
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 #
# ------------------------------------------------------------------ #
@@ -106,29 +255,76 @@ def digitalizar_task(
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 {}
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 [],
},
},
_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 [],
)
return {}
# 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()
# ------------------------------------------------------------------ #
@@ -136,16 +332,38 @@ def digitalizar_task(
# ------------------------------------------------------------------ #
_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": request_data.get("rfc_consulta") or record.rfc_consulta or "",
"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": request_data.get("archivo_base64") or "",
"configuracion_vu": config_vu,
"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()
@@ -153,20 +371,15 @@ def digitalizar_task(
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."],
},
},
_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."],
)
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")
@@ -178,29 +391,60 @@ def digitalizar_task(
_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()
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."],
},
},
_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."],
)
return {}
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 #
# ------------------------------------------------------------------ #
@@ -209,31 +453,65 @@ def digitalizar_task(
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"
# 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,
"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 Ignore:
raise
except ValidationException as exc:
if db:
try:
@@ -245,21 +523,31 @@ def digitalizar_task(
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 [],
},
},
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 [],
)
return {}
except Exception as exc:
logger.exception("Error inesperado en digitalizar_task expediente_id=%s", expediente_id)
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
@@ -268,19 +556,14 @@ def digitalizar_task(
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."],
},
},
_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."],
)
return {}
finally:
db.close()

54
backend/api/v1/modules/a76/factura_cove/service.py
View File

@@ -145,21 +145,6 @@ class FacturaCoveDomainService:
)
return None
# Determinar usuario efectivo de WebService:
# - Preferimos el usuario configurado en VU (web_service_user)
# - Si no existe, usamos el de DODA-PITA (doda_web_service_user)
# - Si no existe, usamos la configuración VU de la empresa
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 ""
)
# Determinar clave FIEL efectiva desde la configuración persistida.
# Se envía cifrada con el mismo esquema AES-256-CBC del sistema legado.
clave_fiel_value = ""
@@ -175,17 +160,30 @@ class FacturaCoveDomainService:
)
clave_fiel_value = self._encrypt_fiel(str(company_fiel_secret))
# Validación básica de credenciales VU: para COVE necesitamos al menos
# un usuario de web service (VU o DODA) y una clave FIEL no vacía.
if not effective_ws_user:
# Validación básica de credenciales VU: usamos la clave/token efectiva
# del web service, que es lo que realmente viaja en configuracion_vu.
hardcoded_ws_key = (
"RZGd+CB4R6PfSrstOyN8Is9FXL9AK9NPFisyGGaEWa0vVHoVOl8v2SBcHBoGbt3T/4uHTGcsFQO3b7EonWVfugQjBooywbz74K+jM68j8/Y="
)
vu_ws_key = (getattr(vu, "web_service_access_key", None) or "").strip() if vu else ""
vu_access_key_raw = (getattr(vu, "access_key", None) or "").strip() if vu else ""
vu_access_key_encrypted = self._encrypt_fiel(vu_access_key_raw) if vu_access_key_raw else ""
clave_webservice = (
vu_ws_key
or vu_access_key_encrypted
or (getattr(company_vu, "webservice_password", None) or "").strip()
or hardcoded_ws_key
)
if not clave_webservice:
errors.add_error(
field="vu",
message="Faltan credenciales de web service o clave FIEL en VU",
field="vu.clave_webservice",
message="La clave de web service no está configurada en VU ni en la empresa.",
solution=[
"Captura usuario y clave de web service en la pestaña VU o DODA del agente, "
"o completa la configuración VU de la empresa y su certificado FIEL."
"Captura la clave de web service en la pestaña VU o DODA del agente, "
"o completa la configuración VU de la empresa."
],
code="MISSING_VU_CREDENTIALS",
code="MISSING_VU_WS_KEY",
)
if not clave_fiel_value:
@@ -269,17 +267,9 @@ class FacturaCoveDomainService:
# Clave/token del webservice: usar el valor de VU si existe, o una
# clave fija de pruebas mientras se termina la configuración real.
hardcoded_ws_key = (
"RZGd+CB4R6PfSrstOyN8Is9FXL9AK9NPFisyGGaEWa0vVHoVOl8v2SBcHBoGbt3T/4uHTGcsFQO3b7EonWVfugQjBooywbz74K+jM68j8/Y="
)
return ConfiguracionVU(
rfc_usuario_vu=rfc_usuario_vu,
clave_webservice=(
(getattr(vu, "web_service_access_key", None) or "").strip()
or (getattr(company_vu, "webservice_password", None) or "").strip()
or hardcoded_ws_key
),
clave_webservice=clave_webservice,
archivo_cer_base64=cer_b64 or "",
archivo_key_base64=key_b64 or "",
clave_fiel=clave_fiel_value,