561 lines
21 KiB
Python
561 lines
21 KiB
Python
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 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
|
|
|
|
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 _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,
|
|
company_id: int,
|
|
tenant_id: int,
|
|
page: int = 1,
|
|
page_size: int = 50,
|
|
search: Optional[str] = None,
|
|
status: Optional[str] = None,
|
|
rfc_consulta: Optional[str] = None,
|
|
e_document: 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 status:
|
|
query = query.filter(ExpedienteArchivo.status == status)
|
|
if rfc_consulta:
|
|
query = query.filter(ExpedienteArchivo.rfc_consulta.ilike(f"%{rfc_consulta}%"))
|
|
if e_document:
|
|
query = query.filter(ExpedienteArchivo.e_document.ilike(f"%{e_document}%"))
|
|
if search:
|
|
like = f"%{search}%"
|
|
query = query.filter(
|
|
or_(
|
|
ExpedienteArchivo.e_document.ilike(like),
|
|
ExpedienteArchivo.tipo_documento.ilike(like),
|
|
ExpedienteArchivo.rfc_consulta.ilike(like),
|
|
ExpedienteArchivo.num_operacion.ilike(like),
|
|
ExpedienteArchivo.nombre_archivo.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:
|
|
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,
|
|
)
|
|
|
|
# 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",
|
|
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,
|
|
)
|
|
|
|
# PROGRESS / PENDING / STARTED
|
|
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,
|
|
external_task_id=task_metadata.get("external_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 _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,
|
|
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 = _resolve_broker_for_vu(db, company_id, tenant_id, agente_aduanal_key)
|
|
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
|
|
|
|
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))
|
|
|
|
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.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:
|
|
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 ""
|
|
)
|
|
|
|
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 ""
|
|
)
|
|
|
|
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,
|
|
"clave_webservice": clave_webservice,
|
|
"archivo_cer_base64": cer_b64 or "",
|
|
"archivo_key_base64": key_b64 or "",
|
|
"clave_fiel": clave_fiel_value,
|
|
"email": email,
|
|
"_ws_key_source": ws_key_source,
|
|
}
|