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

Enhance invoice processing features and update related models

Browse Source 
- Improved invoice processing logic and added new validation checks.
- Updated `InvoiceHeader` and `InvoiceFinancials` models to support additional data fields.
- Refined API methods for better handling of invoice statuses and error responses.
- Cleaned up deprecated code related to previous invoice processing implementations.
This commit is contained in:
AlexeerCT
2026-03-13 11:12:52 -05:00
parent 1cae798638
commit 57a55c7f00
2 changed files with 369 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

85
backend/api/v1/modules/a76/invoices/exports/validators/create.py Normal file
View File

@@ -0,0 +1,85 @@
from sqlalchemy.orm import Session
from api.v1.modules.a76.general_catalogs.exchange_rate.models import ExchangeRate
from core.exceptions import ErrorCollector
from ...schemas import InvoiceHeaderCreate
from ...common.common_validators import validate_common, validate_required_fields_by_operation
def validate_create(db: Session, invoice: InvoiceHeaderCreate, tenant_id: int, company_id: int, errors: ErrorCollector) -> None:
""" Valida la creación de una nueva factura de importe temporal """
if not invoice.operation_type:
errors.add_required_error("operation_type")
if not invoice.invoice_type:
errors.add_required_error("invoice_type")
if not invoice.document_type and invoice.invoice_type != "MEX":
errors.add_required_error("document_type")
if not invoice.invoice_number:
errors.add_required_error("invoice_number")
if not invoice.invoice_date:
errors.add_required_error("invoice_date")
# Validar campos obligatorios según tipo de operación
invoice_data = {
'provider_header': invoice.compliance_mx.provider_header if invoice.compliance_mx else None,
'provider_id': invoice.compliance_mx.provider_id if invoice.compliance_mx else None,
'sold_to_header': invoice.compliance_mx.sold_to_header if invoice.compliance_mx else None,
'sold_to_id': invoice.compliance_mx.sold_to_id if invoice.compliance_mx else None,
'shipped_to_header': invoice.compliance_mx.shipped_to_header if invoice.compliance_mx else None,
'shipped_to_id': invoice.compliance_mx.shipped_to_id if invoice.compliance_mx else None,
'customs_broker_id': invoice.compliance_mx.customs_broker_id if invoice.compliance_mx else None,
'pedimento_id': invoice.compliance_mx.pedimento_id if invoice.compliance_mx else None,
}
validate_required_fields_by_operation(
invoice_data=invoice_data,
operation_type=invoice.operation_type,
errors=errors
)
if errors.has_errors():
"""Se retorna porque hay campos obligatorios según el tipo de operación que deben ser llenados"""
return
validate_common(db, invoice, tenant_id, company_id, errors)
if errors.has_errors():
"""Se retorna por que fallaron las validaciones generales"""
return
if invoice.compliance_mx:
if not invoice.compliance_mx.pedimento_id:
invoice.compliance_mx.remesa = None
if invoice.financials:
if not invoice.financials.exchange_rate:
invoice.financials.exchange_rate = db.query(ExchangeRate.value).filter(ExchangeRate.date == invoice.invoice_date).scalar()
invoice.document_type = (invoice.document_type or "").upper()
if invoice.logistics:
if not invoice.logistics.transport_type:
invoice.logistics.transport_type = "none"
if invoice.logistics.transport_type == "none" and invoice.logistics.transport_num:
invoice.logistics.transport_num = None
invoice.logistics.incoterm = (invoice.logistics.incoterm or "").upper()
if not invoice.logistics.weight_type:
invoice.logistics.weight_type = "kgs"
if invoice.financials:
if not invoice.financials.currency:
invoice.financials.currency = "foreign"
if invoice.financials.currency == "local":
invoice.financials.currency_type = "MXN"
elif invoice.financials.currency == "foreign":
invoice.financials.currency_type = "USD"
elif invoice.financials.currency == "manual":
invoice.financials.currency_type = (invoice.financials.currency_type or "").upper()

284
backend/api/v1/modules/a76/invoices/exports/validators/update.py Normal file
View File

@@ -0,0 +1,284 @@
from typing import Optional
from datetime import date
from decimal import Decimal
from sqlalchemy.orm import Session
from ...common.common_validators import validate_common, validate_required_fields_by_operation
from core.exceptions import ErrorCollector
from ...schemas import InvoiceHeaderUpdate
from ...models import InvoiceHeader
# Helper function para limpiar strings (equivalente a Clip())
def clean_str(value: Optional[str]) -> Optional[str]:
if value is None or value == "":
return None
return value.strip()
def validate_update(
db: Session,
invoice: InvoiceHeaderUpdate,
existing_invoice: InvoiceHeader,
tenant_id: int,
company_id: int,
errors: ErrorCollector,
) -> None:
"""
Valida y procesa la actualización parcial de una factura de importación temporal.
Lógica: Si un campo viene con valor, se limpia/valida.
Si no, se mantiene el valor existente de la factura.
Args:
invoice: Datos de la factura a validar/actualizar (modificado in-place)
existing_invoice: Factura existente en la base de datos
errors: Colector de errores
Returns:
None (modifica invoice in-place y acumula errores en errors)
"""
# Validar campos requeridos según el tipo de operación
invoice_dict = {
'provider_id': invoice.compliance_mx.provider_id if invoice.compliance_mx else (existing_invoice.compliance_mx.provider_id if existing_invoice.compliance_mx else None),
'sold_to_id': invoice.compliance_mx.sold_to_id if invoice.compliance_mx else (existing_invoice.compliance_mx.sold_to_id if existing_invoice.compliance_mx else None),
'sold_to_header': invoice.compliance_mx.sold_to_header if invoice.compliance_mx else (existing_invoice.compliance_mx.sold_to_header if existing_invoice.compliance_mx else None),
'shipped_to_id': invoice.compliance_mx.shipped_to_id if invoice.compliance_mx else (existing_invoice.compliance_mx.shipped_to_id if existing_invoice.compliance_mx else None),
'customs_broker_id': invoice.compliance_mx.customs_broker_id if invoice.compliance_mx else (existing_invoice.compliance_mx.customs_broker_id if existing_invoice.compliance_mx else None),
'pedimento_id': invoice.compliance_mx.pedimento_id if invoice.compliance_mx else (existing_invoice.compliance_mx.pedimento_id if existing_invoice.compliance_mx else None),
}
validate_required_fields_by_operation(
invoice_data=invoice_dict,
operation_type=invoice.operation_type or (existing_invoice.operation_type or 'imp'),
errors=errors
)
# Primero ejecutar validaciones comunes
validate_common(db, invoice, tenant_id, company_id, errors)
# Mapeo de columnas CSV a campos de la factura
# Siguiendo la lógica del código Clarion original
# Columna A: Pedimento (si no viene en CSV, usar el existente)
if invoice.compliance_mx:
if invoice.compliance_mx.pedimento_id:
invoice.compliance_mx.pedimento_id = invoice.compliance_mx.pedimento_id
else:
invoice.compliance_mx.pedimento_id = existing_invoice.compliance_mx.pedimento_id if existing_invoice.compliance_mx else None
# Columna B: Remesa
if invoice.compliance_mx:
if invoice.compliance_mx.remesa:
invoice.compliance_mx.remesa = invoice.compliance_mx.remesa
else:
invoice.compliance_mx.remesa = existing_invoice.compliance_mx.remesa if existing_invoice.compliance_mx else None
# Columna C: Factura (OBLIGATORIO)
if invoice.invoice_number is not None:
invoice.invoice_number = clean_str(invoice.invoice_number)
if not invoice.invoice_number:
errors.add_required_error("invoice_number")
else:
invoice.invoice_number = existing_invoice.invoice_number
# Columna D: Fecha
if not invoice.invoice_date:
invoice.invoice_date = existing_invoice.invoice_date
# Columna E: Tipo Cambio
if invoice.financials:
if invoice.financials.exchange_rate is None:
if existing_invoice.financials:
invoice.financials.exchange_rate = existing_invoice.financials.exchange_rate
# Columna F: Régimen
if invoice.document_type:
invoice.document_type = clean_str(invoice.document_type).upper()
else:
invoice.document_type = existing_invoice.document_type
# Columna G: Clave Proveedor
if invoice.compliance_mx:
if invoice.compliance_mx.provider_id is None:
invoice.compliance_mx.provider_id = existing_invoice.compliance_mx.provider_id if existing_invoice.compliance_mx else None
# Columna H: Clave Vendido A
if invoice.compliance_mx:
if invoice.compliance_mx.sold_to_id is None:
invoice.compliance_mx.sold_to_id = existing_invoice.compliance_mx.sold_to_id if existing_invoice.compliance_mx else None
# Columna I: Clave Enviado A
if invoice.compliance_mx:
if invoice.compliance_mx.shipped_to_id is None:
invoice.compliance_mx.shipped_to_id = existing_invoice.compliance_mx.shipped_to_id if existing_invoice.compliance_mx else None
# Columna J: Clave A. Aduanal
if invoice.compliance_mx:
if invoice.compliance_mx.customs_broker_id is None:
invoice.compliance_mx.customs_broker_id = existing_invoice.compliance_mx.customs_broker_id if existing_invoice.compliance_mx else None
# Columna K: Clave Transportista
if invoice.logistics:
# Note: logistics in update schema seems to be a single object, but in model it's a list.
# This validator seems to expect a single object (InvoiceLogisticsUpdate).
# We'll stick to the existing logic but make it safe.
if hasattr(invoice.logistics, 'carrier_id') and invoice.logistics.carrier_id is None:
invoice.logistics.carrier_id = existing_invoice.logistics.carrier_id if existing_invoice.logistics else None
# Columna L: Nombre Conductor
if invoice.logistics:
if hasattr(invoice.logistics, 'driver_name') and not invoice.logistics.driver_name:
invoice.logistics.driver_name = existing_invoice.logistics.driver_name if existing_invoice.logistics else None
elif hasattr(invoice.logistics, 'driver_name'):
invoice.logistics.driver_name = clean_str(invoice.logistics.driver_name)
# Columna M: Tipo Transporte
if invoice.logistics:
if hasattr(invoice.logistics, 'transport_type') and not invoice.logistics.transport_type:
invoice.logistics.transport_type = existing_invoice.logistics.transport_type if existing_invoice.logistics else None
elif hasattr(invoice.logistics, 'transport_type'):
invoice.logistics.transport_type = clean_str(invoice.logistics.transport_type)
# Columna N: Número de Transporte
if invoice.logistics:
if hasattr(invoice.logistics, 'transport_num') and not invoice.logistics.transport_num:
invoice.logistics.transport_num = existing_invoice.logistics.transport_num if existing_invoice.logistics else None
elif hasattr(invoice.logistics, 'transport_num'):
invoice.logistics.transport_num = clean_str(invoice.logistics.transport_num)
# Columna O: Tipo de Moneda
if invoice.financials:
if not invoice.financials.currency:
invoice.financials.currency = existing_invoice.financials.currency if existing_invoice.financials else None
else:
invoice.financials.currency = clean_str(invoice.financials.currency).lower()
# Columna P: Clave Moneda
if invoice.financials:
if not invoice.financials.currency_type:
invoice.financials.currency_type = existing_invoice.financials.currency_type if existing_invoice.financials else None
else:
invoice.financials.currency_type = clean_str(invoice.financials.currency_type).upper()
# Columna Q: Flete
if invoice.financials:
if invoice.financials.freight is None:
invoice.financials.freight = existing_invoice.financials.freight if existing_invoice.financials else None
# Columna R: Val Seguros
if invoice.financials:
if invoice.financials.insurance_value is None:
invoice.financials.insurance_value = existing_invoice.financials.insurance_value if existing_invoice.financials else None
# Columna S: Seguros
if invoice.financials:
if invoice.financials.insurance is None:
invoice.financials.insurance = existing_invoice.financials.insurance if existing_invoice.financials else None
# Columna T: Embalaje
if invoice.financials:
if invoice.financials.packaging is None:
invoice.financials.packaging = existing_invoice.financials.packaging if existing_invoice.financials else None
# Columna U: Otros Incrementables
if invoice.financials:
if invoice.financials.other_increments is None:
invoice.financials.other_increments = existing_invoice.financials.other_increments if existing_invoice.financials else None
# Columna V: Incoterms
if invoice.logistics:
if hasattr(invoice.logistics, 'incoterm') and not invoice.logistics.incoterm:
invoice.logistics.incoterm = existing_invoice.logistics.incoterm if existing_invoice.logistics else None
elif hasattr(invoice.logistics, 'incoterm'):
invoice.logistics.incoterm = clean_str(invoice.logistics.incoterm).upper()
# Columna W: Precinto
if invoice.logistics:
if hasattr(invoice.logistics, 'seal_number') and not invoice.logistics.seal_number:
invoice.logistics.seal_number = existing_invoice.logistics.seal_number if existing_invoice.logistics else None
elif hasattr(invoice.logistics, 'seal_number'):
invoice.logistics.seal_number = clean_str(invoice.logistics.seal_number)
# Columna X: Fecha de Emisión
if not invoice.emission_date:
invoice.emission_date = existing_invoice.emission_date
# Columna Y: Tipo de Peso (Opcional)
if invoice.logistics:
if hasattr(invoice.logistics, 'weight_type') and not invoice.logistics.weight_type:
invoice.logistics.weight_type = existing_invoice.logistics.weight_type if existing_invoice.logistics else None
elif hasattr(invoice.logistics, 'weight_type'):
invoice.logistics.weight_type = clean_str(invoice.logistics.weight_type).upper()
# Columna Z: Número de Manifiesto (Opcional)
if invoice.compliance_mx.manifest_number:
if not invoice.compliance_mx.manifest_number:
invoice.compliance_mx.manifest_number = existing_invoice.compliance_mx.manifest_number if existing_invoice.compliance_mx else None
else:
invoice.compliance_mx.manifest_number = clean_str(invoice.compliance_mx.manifest_number)
# Columna AA: E-Document (Opcional)
if invoice.compliance_mx:
if not invoice.compliance_mx.edocument:
invoice.compliance_mx.edocument = existing_invoice.compliance_mx.edocument if existing_invoice.compliance_mx else None
else:
invoice.compliance_mx.edocument = clean_str(invoice.compliance_mx.edocument)
# Columna AB: Num. Operación (Opcional)
if invoice.compliance_mx:
if not invoice.compliance_mx.vucem_operation_num:
invoice.compliance_mx.vucem_operation_num = existing_invoice.compliance_mx.vucem_operation_num if existing_invoice.compliance_mx else None
else:
invoice.compliance_mx.vucem_operation_num = clean_str(invoice.compliance_mx.vucem_operation_num)
# Columna AB: Aduana (OBLIGATORIO)
if invoice.compliance_mx:
if not invoice.compliance_mx.aduana:
invoice.compliance_mx.aduana = existing_invoice.compliance_mx.aduana if existing_invoice.compliance_mx else None
else:
invoice.compliance_mx.aduana = clean_str(invoice.compliance_mx.aduana)
# Columna AC: Enviado Por (Obligatorio)
if invoice.compliance_mx:
if not invoice.compliance_mx.shipped_by_id:
invoice.compliance_mx.shipped_by_id = existing_invoice.compliance_mx.shipped_by_id if existing_invoice.compliance_mx else None
else:
invoice.compliance_mx.shipped_by_id = clean_str(invoice.compliance_mx.shipped_by_id)
# Columna AD: Aduana_Cruce (Obligatorio)
current_aduana = invoice.compliance_mx.aduana if invoice.compliance_mx else (existing_invoice.compliance_mx.aduana if existing_invoice.compliance_mx else None)
if not current_aduana:
errors.add_required_error("aduana")
# Columna AC: Sección de Despacho / Puerto de Entrada (Opcional)
if invoice.compliance_mx:
if not invoice.compliance_mx.port_of_entry:
invoice.compliance_mx.port_of_entry = existing_invoice.compliance_mx.port_of_entry if existing_invoice.compliance_mx else None
else:
invoice.compliance_mx.port_of_entry = clean_str(invoice.compliance_mx.port_of_entry)
# Columna AE: Observación en Español (Opcional)
if not invoice.observation_es:
invoice.observation_es = existing_invoice.observation_es
else:
invoice.observation_es = clean_str(invoice.observation_es)
# Columna AF: Observación en Inglés (Opcional)
if not invoice.observation_en:
invoice.observation_en = existing_invoice.observation_en
else:
invoice.observation_en = clean_str(invoice.observation_en)
# Columna AG: cfdi_uuid (Opcional)
if not invoice.cfdi_uuid:
invoice.cfdi_uuid = existing_invoice.cfdi_uuid if existing_invoice.compliance_mx else None
else:
invoice.cfdi_uuid = clean_str(invoice.cfdi_uuid)
# Columna AH: Localizacion (Opcional)
if invoice.compliance_mx.location:
if not invoice.compliance_mx.location:
invoice.compliance_mx.location = existing_invoice.compliance_mx.location if existing_invoice.compliance_mx.location else None
else:
invoice.compliance_mx.location = clean_str(invoice.compliance_mx.location)