774 lines
32 KiB
Python
774 lines
32 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
from dataclasses import dataclass
|
|
from decimal import Decimal
|
|
from typing import List, Tuple
|
|
|
|
from cryptography.hazmat.primitives import padding
|
|
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
|
|
from sqlalchemy.orm import Session
|
|
|
|
from core.config import settings
|
|
from core.database import CoreSessionLocal
|
|
from core.exceptions import ValidationException, ErrorCollector
|
|
from core.storage_s3 import get_object_bytes, object_exists
|
|
|
|
from api.v1.modules.a76.invoices.models import InvoiceHeader
|
|
from api.v1.modules.a76.customs_brokers import models as cb_models
|
|
from api.v1.modules.a76.clients_and_providers.models import ClientProvider
|
|
from api.v1.modules.a76.general_catalogs.company.models import Company
|
|
from api.v1.modules.a76.items.models import LineItem
|
|
|
|
from .schemas import (
|
|
ConfiguracionVU,
|
|
CoveEligibilityIssue,
|
|
CoveEligibilityResponse,
|
|
FacturaCoveRequest,
|
|
MercanciaCove,
|
|
PersonaCove,
|
|
)
|
|
|
|
@dataclass
|
|
class InvoiceContext:
|
|
invoice: InvoiceHeader
|
|
broker: cb_models.CustomsBroker | None
|
|
vu: cb_models.CustomsBrokerVU | None
|
|
|
|
|
|
class FacturaCoveDomainService:
|
|
"""
|
|
Servicio de dominio para validar y construir el payload de COVE a partir de una factura.
|
|
|
|
NOTA IMPORTANTE:
|
|
----------------
|
|
Este servicio prepara la estructura de datos y realiza validaciones de negocio,
|
|
pero **no** realiza todavía la llamada HTTP al webservice de COVE. Eso se puede
|
|
implementar posteriormente en un servicio dedicado (p. ej. CoveExternalService).
|
|
"""
|
|
|
|
def __init__(self, db: Session):
|
|
self.db = db
|
|
|
|
def _load_context(self, invoice_id: int, tenant_id: int, company_id: int) -> InvoiceContext:
|
|
invoice: InvoiceHeader | None = self.db.get(InvoiceHeader, invoice_id)
|
|
if not invoice:
|
|
raise ValidationException(
|
|
"Factura no encontrada",
|
|
errors=[{"field": "invoice_id", "message": f"Factura {invoice_id} no encontrada"}],
|
|
)
|
|
|
|
if invoice.company_id != company_id or invoice.tenant_id != tenant_id:
|
|
raise ValidationException(
|
|
"Factura no pertenece a la compañía/tenant actual",
|
|
errors=[
|
|
{
|
|
"field": "invoice_id",
|
|
"message": "La factura no pertenece a la compañía o tenant actuales",
|
|
}
|
|
],
|
|
)
|
|
|
|
compliance = invoice.compliance_mx
|
|
broker = None
|
|
vu = None
|
|
|
|
if compliance and compliance.customs_broker_id:
|
|
broker = self.db.get(cb_models.CustomsBroker, compliance.customs_broker_id)
|
|
if broker:
|
|
vu = broker.vu
|
|
|
|
return InvoiceContext(invoice=invoice, broker=broker, vu=vu)
|
|
|
|
def _get_company(self, ctx: InvoiceContext) -> Company | None:
|
|
company_id = getattr(ctx.invoice, "company_id", None)
|
|
if not company_id:
|
|
return None
|
|
return self.db.get(Company, company_id)
|
|
|
|
def _get_company_fiel_certificate(self, company: Company | None):
|
|
if not company:
|
|
return None
|
|
|
|
for certificate in company.digital_certificates or []:
|
|
if (certificate.certificate_type or "").strip().lower() == "fiel":
|
|
return certificate
|
|
|
|
return None
|
|
|
|
def _encrypt_fiel(self, raw_fiel: str) -> str:
|
|
"""
|
|
Cifra la clave FIEL con el mismo esquema del sistema legado PHP:
|
|
AES-256-CBC + PKCS7 + base64.
|
|
"""
|
|
normalized_fiel = (raw_fiel or "").strip()
|
|
if not normalized_fiel:
|
|
return ""
|
|
|
|
encryption_key = (settings.COVE_FIEL_HASH_KEY or "").encode("utf-8")
|
|
encryption_iv = (settings.COVE_FIEL_HASH_IV or "").encode("utf-8")
|
|
if not encryption_key or not encryption_iv:
|
|
return ""
|
|
|
|
key_bytes = encryption_key[:32].ljust(32, b"\0")
|
|
iv_bytes = encryption_iv[:16].ljust(16, b"\0")
|
|
|
|
padder = padding.PKCS7(algorithms.AES.block_size).padder()
|
|
padded_data = padder.update(normalized_fiel.encode("utf-8")) + padder.finalize()
|
|
|
|
cipher = Cipher(algorithms.AES(key_bytes), modes.CBC(iv_bytes))
|
|
encryptor = cipher.encryptor()
|
|
encrypted = encryptor.update(padded_data) + encryptor.finalize()
|
|
return base64.b64encode(encrypted).decode("ascii")
|
|
|
|
def _build_configuracion_vu(self, ctx: InvoiceContext, errors: ErrorCollector) -> ConfiguracionVU | None:
|
|
"""
|
|
Construye la sección configuracion_vu usando CustomsBrokerVU + S3.
|
|
|
|
Lee los archivos .cer y .key desde almacenamiento de objetos, los
|
|
convierte a base64 y construye un ConfiguracionVU listo para enviar
|
|
al API externo de COVE.
|
|
"""
|
|
vu = ctx.vu
|
|
company = self._get_company(ctx)
|
|
company_vu = company.ventanilla_unica if company else None
|
|
company_fiel_certificate = self._get_company_fiel_certificate(company)
|
|
|
|
if not vu and not company_vu and not company_fiel_certificate:
|
|
errors.add_error(
|
|
field="vu",
|
|
message="La factura no tiene configuración VU asociada ni configuración VU/certificado FIEL en la empresa",
|
|
solution=[
|
|
"Configura los datos VU del agente aduanal o de la empresa y sube certificado (.cer) y llave (.key) antes de generar COVE."
|
|
],
|
|
code="MISSING_VU_CONFIGURATION",
|
|
)
|
|
return None
|
|
|
|
# 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 = ""
|
|
|
|
if vu and getattr(vu, "fiel_access_key", None):
|
|
clave_fiel_value = self._encrypt_fiel(vu.fiel_access_key or "")
|
|
elif company_fiel_certificate:
|
|
# Fallback en base de datos: certificado FIEL de la empresa
|
|
company_fiel_secret = (
|
|
getattr(company_fiel_certificate, "access_key", None)
|
|
or getattr(company_fiel_certificate, "password", None)
|
|
or ""
|
|
)
|
|
clave_fiel_value = self._encrypt_fiel(str(company_fiel_secret))
|
|
|
|
# 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.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 DODA del agente, "
|
|
"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 para COVE no está configurada ni en VU ni en la empresa.",
|
|
solution=[
|
|
"Captura la clave FIEL en la configuración VU del agente aduanal o en el certificado FIEL 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 VU",
|
|
solution=[
|
|
"Sube el certificado (.cer) y la llave (.key) en la configuración VU del agente aduanal o en los certificados digitales de la empresa."
|
|
],
|
|
code="MISSING_VU_CERT_KEY",
|
|
)
|
|
return None
|
|
|
|
# Convertir archivos .cer y .key de S3 a base64
|
|
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 de objetos",
|
|
solution=["Vuelve a subir el certificado en la configuración VU del agente aduanal o en los certificados digitales de la empresa."],
|
|
code="VU_CERT_NOT_FOUND",
|
|
)
|
|
else:
|
|
cer_bytes = get_object_bytes(certificate_path)
|
|
cer_b64 = base64.b64encode(cer_bytes).decode("ascii")
|
|
|
|
if not object_exists(key_path):
|
|
errors.add_error(
|
|
field="vu.key_path",
|
|
message="La llave VU no existe en el almacenamiento de objetos",
|
|
solution=["Vuelve a subir la llave en la configuración VU del agente aduanal o en los certificados digitales de la empresa."],
|
|
code="VU_KEY_NOT_FOUND",
|
|
)
|
|
else:
|
|
key_bytes = get_object_bytes(key_path)
|
|
key_b64 = base64.b64encode(key_bytes).decode("ascii")
|
|
except Exception as exc: # pragma: no cover - errores de IO externos
|
|
errors.add_error(
|
|
field="vu",
|
|
message="Error leyendo certificados VU desde almacenamiento de objetos.",
|
|
solution=["Verifica la configuración de MinIO/S3 y las rutas de certificados/llaves en VU o en los certificados digitales de la empresa."],
|
|
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 ""
|
|
)
|
|
|
|
# Clave/token del webservice: usar el valor de VU si existe, o una
|
|
# clave fija de pruebas mientras se termina la configuración real.
|
|
return ConfiguracionVU(
|
|
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,
|
|
)
|
|
|
|
def _clientprovider_to_persona(self, cp: ClientProvider) -> PersonaCove:
|
|
"""
|
|
Construye una PersonaCove a partir de un ClientProvider + su dirección.
|
|
No expone IDs internos; solo valores normalizados.
|
|
"""
|
|
addr = cp.address
|
|
|
|
tipo_nat = (cp.type_nat_foreign or "").strip().upper()
|
|
tipo_identificador = "0" if tipo_nat == "E" else "1"
|
|
identificacion = (cp.rfc or "").strip().upper()
|
|
|
|
# País: normalizar a código de 3 caracteres (ISO o catálogo VU).
|
|
raw_country = (addr.country or "") if addr and getattr(addr, "country", None) else ""
|
|
country_code = raw_country.strip().upper()[:3] if raw_country else ""
|
|
|
|
return PersonaCove(
|
|
tipo_identificador=tipo_identificador,
|
|
identificacion=identificacion,
|
|
apellido_paterno="",
|
|
apellido_materno="",
|
|
nombre=(cp.name or cp.short_name or "").strip(),
|
|
calle=(addr.streets or "").strip() if addr and addr.streets else "",
|
|
numero_exterior=(addr.exterior_number or "").strip()
|
|
if addr and addr.exterior_number
|
|
else "",
|
|
numero_interior=(addr.interior_number or "").strip()
|
|
if addr
|
|
else "",
|
|
colonia=(addr.neighborhood or "").strip()
|
|
if addr and addr.neighborhood
|
|
else "",
|
|
localidad=(addr.city or "").strip() if addr and addr.city else "",
|
|
municipio=(addr.municipality or "").strip()
|
|
if addr
|
|
else "",
|
|
entidad_federativa=(addr.state or "").strip()
|
|
if addr and addr.state
|
|
else "",
|
|
pais=country_code,
|
|
codigo_postal=(addr.postal_code or "").strip()
|
|
if addr and addr.postal_code
|
|
else "",
|
|
)
|
|
|
|
def _company_to_persona(self, company: Company) -> PersonaCove:
|
|
"""
|
|
Construye una PersonaCove a partir de Company + su dirección principal.
|
|
"""
|
|
# Tomar dirección 'main' si existe; si no, la primera.
|
|
addr = None
|
|
for a in company.addresses or []:
|
|
if getattr(a, "address_type", None) == "main":
|
|
addr = a
|
|
break
|
|
if addr is None and company.addresses:
|
|
addr = company.addresses[0]
|
|
|
|
# País: normalizar a código de 3 caracteres (ISO o catálogo VU).
|
|
raw_country = (addr.country or "") if addr and getattr(addr, "country", None) else ""
|
|
country_code = raw_country.strip().upper()[:3] if raw_country else ""
|
|
|
|
return PersonaCove(
|
|
tipo_identificador="1", # Empresa mexicana por defecto
|
|
identificacion=(company.rfc or "").strip().upper(),
|
|
apellido_paterno="",
|
|
apellido_materno="",
|
|
nombre=(company.name or "").strip(),
|
|
calle=(addr.street or "").strip() if addr and addr.street else "",
|
|
numero_exterior=(addr.exterior_number or "").strip()
|
|
if addr and addr.exterior_number
|
|
else "",
|
|
numero_interior=(addr.interior_number or "").strip()
|
|
if addr
|
|
else "",
|
|
colonia=(addr.neighborhood or "").strip()
|
|
if addr and addr.neighborhood
|
|
else "",
|
|
localidad=(addr.city or "").strip() if addr and addr.city else "",
|
|
municipio=(addr.municipality or "").strip()
|
|
if addr
|
|
else "",
|
|
entidad_federativa=(addr.state or "").strip()
|
|
if addr and addr.state
|
|
else "",
|
|
pais=country_code,
|
|
codigo_postal=(addr.postal_code or "").strip()
|
|
if addr and addr.postal_code
|
|
else "",
|
|
)
|
|
|
|
def _build_personas(
|
|
self, ctx: InvoiceContext, errors: ErrorCollector
|
|
) -> Tuple[PersonaCove | None, PersonaCove | None]:
|
|
"""
|
|
Construye emisor (exportador) y destinatario (importador) a partir de:
|
|
- InvoiceComplianceMx.provider_id / sold_to_id / shipped_to_id
|
|
- Catálogo de clientes/proveedores
|
|
- Company (datos de la propia empresa) como último recurso
|
|
"""
|
|
compliance = ctx.invoice.compliance_mx
|
|
tenant_id = getattr(ctx.invoice, "tenant_id", None)
|
|
company_id = getattr(ctx.invoice, "company_id", None)
|
|
|
|
emisor_persona: PersonaCove | None = None
|
|
destinatario_persona: PersonaCove | None = None
|
|
|
|
# --- Emisor: proveedor/exportador ---
|
|
if compliance and compliance.provider_id:
|
|
provider = (
|
|
self.db.query(ClientProvider)
|
|
.filter(
|
|
ClientProvider.id == compliance.provider_id,
|
|
ClientProvider.tenant_id == tenant_id,
|
|
ClientProvider.company_id == company_id,
|
|
)
|
|
.first()
|
|
)
|
|
if provider:
|
|
emisor_persona = self._clientprovider_to_persona(provider)
|
|
else:
|
|
errors.add_error(
|
|
field="emisor",
|
|
message="No se encontró el proveedor/exportador asociado a la factura",
|
|
solution=[
|
|
"Verifica que el proveedor/exportador exista en el catálogo y que el invoice_compliance_mx.provider_id sea válido."
|
|
],
|
|
code="EMISOR_PROVIDER_NOT_FOUND",
|
|
)
|
|
else:
|
|
errors.add_error(
|
|
field="emisor",
|
|
message="La factura no tiene proveedor/exportador configurado en cumplimiento (provider_id)",
|
|
solution=[
|
|
"Configura el proveedor/exportador (provider_id) en los datos de cumplimiento de la factura."
|
|
],
|
|
code="EMISOR_PROVIDER_MISSING",
|
|
)
|
|
|
|
# --- Destinatario: importador mexicano ---
|
|
dest_client: ClientProvider | None = None
|
|
if compliance and compliance.sold_to_id:
|
|
dest_client = (
|
|
self.db.query(ClientProvider)
|
|
.filter(
|
|
ClientProvider.id == compliance.sold_to_id,
|
|
ClientProvider.tenant_id == tenant_id,
|
|
ClientProvider.company_id == company_id,
|
|
)
|
|
.first()
|
|
)
|
|
elif compliance and compliance.shipped_to_id:
|
|
dest_client = (
|
|
self.db.query(ClientProvider)
|
|
.filter(
|
|
ClientProvider.id == compliance.shipped_to_id,
|
|
ClientProvider.tenant_id == tenant_id,
|
|
ClientProvider.company_id == company_id,
|
|
)
|
|
.first()
|
|
)
|
|
|
|
if dest_client:
|
|
destinatario_persona = self._clientprovider_to_persona(dest_client)
|
|
else:
|
|
# Fallback: usar la empresa de A76 como importador/destinatario
|
|
if company_id is not None:
|
|
company = (
|
|
self.db.query(Company)
|
|
.filter(Company.id == company_id, Company.tenant_id == tenant_id)
|
|
.first()
|
|
)
|
|
else:
|
|
company = None
|
|
|
|
if company:
|
|
destinatario_persona = self._company_to_persona(company)
|
|
else:
|
|
errors.add_error(
|
|
field="destinatario",
|
|
message="No se pudo determinar el destinatario (cliente/importador) para COVE",
|
|
solution=[
|
|
"Configura sold_to_id o shipped_to_id en los datos de cumplimiento de la factura, "
|
|
"o asegura que la compañía tenga datos de dirección configurados."
|
|
],
|
|
code="DESTINATARIO_NOT_FOUND",
|
|
)
|
|
|
|
return emisor_persona, destinatario_persona
|
|
|
|
def _build_mercancias(self, ctx: InvoiceContext, errors: ErrorCollector) -> list[MercanciaCove]:
|
|
"""
|
|
Construye la lista de mercancías COVE a partir de las partidas (LineItem)
|
|
asociadas a la factura.
|
|
"""
|
|
# Obtener todas las partidas de la factura
|
|
lines: list[LineItem] = (
|
|
self.db.query(LineItem)
|
|
.filter(LineItem.invoice_id == ctx.invoice.id)
|
|
.all()
|
|
)
|
|
|
|
if not lines:
|
|
errors.add_error(
|
|
field="mercancias",
|
|
message="La factura no tiene partidas (LineItem) asociadas",
|
|
solution=[
|
|
"Verifica que la factura tenga partidas capturadas antes de generar COVE."
|
|
],
|
|
code="NO_LINE_ITEMS_FOR_COVE",
|
|
)
|
|
return []
|
|
|
|
mercancias: list[MercanciaCove] = []
|
|
|
|
# Determinar moneda base: usamos la moneda de la factura tal como
|
|
# la maneja el módulo de invoices. En financials se normaliza:
|
|
# - currency: 'foreign' | 'local' | 'manual'
|
|
# - currency_type: código de catálogo (ej. 'USD', 'MXN'), upper.
|
|
fin = getattr(ctx.invoice, "financials", None)
|
|
raw_currency_type = getattr(fin, "currency_type", None)
|
|
invoice_currency = (raw_currency_type or "").strip().upper() or "USD"
|
|
|
|
for line in lines:
|
|
qty_model = line.quantity
|
|
fin_model = line.financial
|
|
desc_model = line.description
|
|
|
|
if not qty_model or qty_model.quantity is None or qty_model.quantity <= 0:
|
|
# Saltar partidas sin cantidad válida
|
|
continue
|
|
|
|
# Cantidad: normalizar a EXACTAMENTE 2 decimales (ej. 12.23)
|
|
cantidad = Decimal(str(qty_model.quantity)).quantize(Decimal("0.01"))
|
|
|
|
# Descripción genérica: priorizar descripción de parte / inglés / español
|
|
descripcion = ""
|
|
if desc_model:
|
|
descripcion = (
|
|
desc_model.part_description
|
|
or desc_model.description_english
|
|
or desc_model.description_spanish
|
|
or ""
|
|
).strip()
|
|
if not descripcion:
|
|
descripcion = (line.line_concept or "").strip()
|
|
if not descripcion:
|
|
descripcion = "SIN DESCRIPCION"
|
|
|
|
# Clave unidad de medida: usar OMA/customs si están disponibles
|
|
clave_unidad = ""
|
|
uom = line.unit_of_measure_info
|
|
if uom:
|
|
if uom.oma_unit and uom.oma_unit.code:
|
|
clave_unidad = uom.oma_unit.code
|
|
elif uom.customs_unit and uom.customs_unit.code:
|
|
clave_unidad = uom.customs_unit.code
|
|
elif uom.code:
|
|
clave_unidad = uom.code
|
|
|
|
if not clave_unidad:
|
|
errors.add_error(
|
|
field="mercancias",
|
|
message="No se pudo determinar la unidad de medida para una partida de la factura",
|
|
solution=[
|
|
"Asegúrate de que la partida tenga una unidad de medida configurada en el catálogo "
|
|
"y que esté ligada a una clave OMA/aduana válida."
|
|
],
|
|
code="MERCANCIA_UOM_MISSING",
|
|
)
|
|
continue
|
|
|
|
# Moneda y valores: usamos la moneda de la factura (3 caracteres)
|
|
tipo_moneda = invoice_currency
|
|
|
|
valor_total = Decimal("0")
|
|
valor_dolares = Decimal("0")
|
|
valor_unitario = Decimal("0")
|
|
|
|
if fin_model:
|
|
if tipo_moneda == "USD":
|
|
base_total = (
|
|
fin_model.value_total_usd
|
|
or fin_model.value_usd
|
|
or Decimal("0")
|
|
)
|
|
else:
|
|
# Para otras monedas, usamos el total en MXN/MC como respaldo
|
|
base_total = (
|
|
fin_model.value_total_mxn
|
|
or fin_model.value_mxn
|
|
or fin_model.value_total_mc
|
|
or fin_model.value_mc
|
|
or Decimal("0")
|
|
)
|
|
|
|
# Valor total: normalizar a EXACTAMENTE 2 decimales
|
|
valor_total = Decimal(str(base_total or 0)).quantize(Decimal("0.01"))
|
|
if cantidad > 0:
|
|
# Valor unitario también a EXACTAMENTE 2 decimales
|
|
valor_unitario = (valor_total / cantidad).quantize(Decimal("0.01"))
|
|
else:
|
|
valor_unitario = Decimal("0")
|
|
|
|
# Valor en dólares: si ya existe, lo usamos; si no, asumimos que los totales ya están en USD.
|
|
if fin_model.value_total_usd:
|
|
valor_dolares = Decimal(str(fin_model.value_total_usd))
|
|
elif tipo_moneda == "USD":
|
|
valor_dolares = valor_total
|
|
else:
|
|
valor_dolares = Decimal("0")
|
|
|
|
# Normalizar valor en dólares a EXACTAMENTE 2 decimales
|
|
valor_dolares = valor_dolares.quantize(Decimal("0.01"))
|
|
else:
|
|
errors.add_error(
|
|
field="mercancias",
|
|
message="La partida de la factura no tiene información financiera asociada",
|
|
solution=[
|
|
"Verifica que las partidas tengan datos financieros (LineFinancial) antes de generar COVE."
|
|
],
|
|
code="MERCANCIA_FINANCIAL_MISSING",
|
|
)
|
|
continue
|
|
|
|
mercancia = MercanciaCove(
|
|
descripcion_generica=descripcion[:500],
|
|
clave_unidad_medida=clave_unidad,
|
|
tipo_moneda=tipo_moneda,
|
|
cantidad=cantidad,
|
|
valor_unitario=valor_unitario,
|
|
valor_total=valor_total,
|
|
valor_dolares=valor_dolares,
|
|
descripcion_especifica=[],
|
|
)
|
|
mercancias.append(mercancia)
|
|
|
|
if not mercancias and not errors.has_errors():
|
|
errors.add_error(
|
|
field="mercancias",
|
|
message="No se generó ninguna mercancía COVE a partir de las partidas de la factura",
|
|
solution=[
|
|
"Verifica que las partidas tengan cantidad y datos financieros válidos antes de generar COVE."
|
|
],
|
|
code="NO_MERCANCIAS_GENERATED",
|
|
)
|
|
|
|
return mercancias
|
|
|
|
def build_factura_cove_request(
|
|
self,
|
|
invoice_id: int,
|
|
tenant_id: int,
|
|
company_id: int,
|
|
recipient_email: str | None = None,
|
|
) -> FacturaCoveRequest:
|
|
"""
|
|
Construye el FacturaCoveRequest completo a partir de una factura,
|
|
validando prerrequisitos de VU, factura y mapeos.
|
|
"""
|
|
ctx = self._load_context(invoice_id, tenant_id, company_id)
|
|
errors = ErrorCollector()
|
|
|
|
# Validaciones básicas de factura
|
|
if not ctx.invoice.invoice_number:
|
|
errors.add_error(
|
|
field="invoice.invoice_number",
|
|
message="La factura no tiene número de factura",
|
|
solution=["Captura el número de factura antes de generar COVE."],
|
|
code="MISSING_INVOICE_NUMBER",
|
|
)
|
|
|
|
# Construir configuración VU (puede agregar errores)
|
|
configuracion_vu = self._build_configuracion_vu(ctx, errors)
|
|
|
|
# Personas y mercancías (por ahora placeholders con errores explícitos)
|
|
emisor, destinatario = self._build_personas(ctx, errors)
|
|
mercancias = self._build_mercancias(ctx, errors)
|
|
|
|
if errors.has_errors():
|
|
# Levantamos ValidationException con todos los errores
|
|
raise ValidationException("No se puede generar COVE desde la factura", errors=errors.get_errors())
|
|
|
|
# Campos genéricos que se pueden poblar de forma segura
|
|
raw_tipo_operacion = (ctx.invoice.operation_type or "").strip().lower()
|
|
# Mapear tipo_operacion al código esperado por el API de COVE
|
|
# Ejemplos:
|
|
# - IMP / importación -> "TOCE.IMP"
|
|
# - EXP / exportación -> "TOCE.EXP"
|
|
if raw_tipo_operacion in {"imp", "import", "importacion", "importación"}:
|
|
tipo_operacion = "TOCE.IMP"
|
|
elif raw_tipo_operacion in {"exp", "export", "exportacion", "exportación"}:
|
|
tipo_operacion = "TOCE.EXP"
|
|
else:
|
|
# Fallback seguro: usar valor por defecto de importación
|
|
tipo_operacion = "TOCE.IMP"
|
|
numero_factura = (ctx.invoice.invoice_number or "").strip()[:50]
|
|
fecha_expedicion = ctx.invoice.invoice_date or ctx.invoice.emission_date or ctx.invoice.capture_date
|
|
|
|
if not fecha_expedicion:
|
|
raise ValidationException(
|
|
"Falta fecha de expedición de factura",
|
|
errors=[
|
|
{
|
|
"field": "invoice.invoice_date",
|
|
"message": "La factura no tiene fecha de expedición/emisión/captura",
|
|
"solution": ["Captura la fecha de la factura antes de generar COVE."],
|
|
}
|
|
],
|
|
)
|
|
|
|
# Patente aduanal en mayúsculas y acotada a 10 caracteres
|
|
raw_patente = (ctx.broker.license if ctx.broker else "") or ""
|
|
patente_aduanal = raw_patente.strip().upper()[:10]
|
|
|
|
# Normalizar tipo_figura desde VU: el API externo espera un código corto
|
|
# (en el ejemplo: "5" para agente aduanal). Hacemos un mapeo simple
|
|
# desde el texto configurado en la UI.
|
|
raw_figura = (
|
|
(ctx.vu.vu_figure_type or "").strip().upper()
|
|
if ctx.vu and ctx.vu.vu_figure_type
|
|
else ""
|
|
)
|
|
if "AGENTE" in raw_figura:
|
|
tipo_figura = "5"
|
|
elif "APODERADO" in raw_figura:
|
|
tipo_figura = "6"
|
|
elif "MANDATARIO" in raw_figura:
|
|
tipo_figura = "7"
|
|
else:
|
|
# Fallback: recortar a máximo 10 caracteres para cumplir el esquema
|
|
tipo_figura = raw_figura[:10]
|
|
|
|
correo_destino = (recipient_email or (ctx.vu.vu_email if ctx.vu else None) or "").strip() or None
|
|
|
|
return FacturaCoveRequest(
|
|
configuracion_vu=configuracion_vu,
|
|
# El RFC de consulta NO debe ser igual al RFC del que registra el comprobante.
|
|
# Usamos como RFC de consulta el RFC del agente aduanal (customs broker),
|
|
# y dejamos que configuracion_vu.rfc_usuario_vu represente al contribuyente.
|
|
rfc_consulta=(
|
|
(ctx.broker.tax_id or "").strip().upper() if ctx.broker and ctx.broker.tax_id else ""
|
|
),
|
|
tipo_figura=tipo_figura,
|
|
numero_factura=numero_factura,
|
|
tipo_operacion=tipo_operacion,
|
|
patente_aduanal=patente_aduanal,
|
|
fecha_expedicion=fecha_expedicion,
|
|
observaciones=ctx.invoice.vu_observations or None,
|
|
correo_electronico=correo_destino,
|
|
tiene_subdivision=bool(ctx.invoice.logistics and ctx.invoice.logistics.is_subdivision),
|
|
certificado_origen=False,
|
|
numero_exportador_autorizado=None,
|
|
emisor=emisor, # type: ignore[arg-type]
|
|
destinatario=destinatario, # type: ignore[arg-type]
|
|
mercancias=mercancias,
|
|
)
|
|
|
|
def check_eligibility(self, invoice_id: int, tenant_id: int, company_id: int) -> CoveEligibilityResponse:
|
|
"""
|
|
Versión "ligera" para frontend: evalúa si la factura puede generar COVE
|
|
e informa por qué no, sin disparar la tarea Celery.
|
|
"""
|
|
errors = ErrorCollector()
|
|
|
|
try:
|
|
ctx = self._load_context(invoice_id, tenant_id, company_id)
|
|
# Reutilizamos solo las validaciones, sin necesidad de devolver el request completo
|
|
if not ctx.invoice.invoice_number:
|
|
errors.add_error(
|
|
field="invoice.invoice_number",
|
|
message="La factura no tiene número de factura",
|
|
solution=["Captura el número de factura antes de generar COVE."],
|
|
code="MISSING_INVOICE_NUMBER",
|
|
)
|
|
|
|
self._build_configuracion_vu(ctx, errors)
|
|
self._build_personas(ctx, errors)
|
|
self._build_mercancias(ctx, errors)
|
|
except ValidationException as exc:
|
|
# Errores de load_context (factura no existe, compañía distinta, etc.)
|
|
return CoveEligibilityResponse(
|
|
can_generate=False,
|
|
reasons=[CoveEligibilityIssue(field=e.get("field", ""), message=e.get("message", "")) for e in exc.errors],
|
|
)
|
|
|
|
if not errors.has_errors():
|
|
return CoveEligibilityResponse(can_generate=True, reasons=[])
|
|
|
|
return CoveEligibilityResponse(
|
|
can_generate=False,
|
|
reasons=[
|
|
CoveEligibilityIssue(field=e.get("field", ""), message=e.get("message", ""))
|
|
for e in errors.get_errors()
|
|
],
|
|
)
|
|
|