Refactor invoice validation logic and introduce export validators
- Moved common validation functions to a shared module for reusability. - Updated create and update validators for import invoices to use the new common validation functions. - Introduced new validators for export invoices, ensuring they adhere to the required fields and validation rules. - Enhanced the update logic to handle both import and export invoice types appropriately. - Cleaned up the code for better readability and maintainability.
This commit is contained in:
@@ -1,7 +1,21 @@
|
||||
from typing import Optional
|
||||
from core.exceptions import ErrorCollector
|
||||
from .. import models
|
||||
from sqlalchemy import func
|
||||
from sqlalchemy.orm import Session
|
||||
from .. import models
|
||||
from .. import schemas
|
||||
from ..models import InvoiceComplianceMx
|
||||
from ..models import TransportType, Currency, WeightUnit
|
||||
from api.v1.modules.a76.items.models import LineItem
|
||||
from api.v1.modules.a76.pedmientos.models.pedimentos import Pedimentos
|
||||
from api.v1.modules.a76.general_catalogs.exchange_rate.models import ExchangeRate
|
||||
from api.v1.modules.a76.clients_and_providers.models import ClientProvider
|
||||
from api.v1.modules.a76.customs_brokers.models import CustomsBroker
|
||||
from api.v1.modules.public.reference_data.incoterms.models import Incoterm
|
||||
from api.v1.modules.public.reference_data.currency_types.models import CurrencyType
|
||||
from api.v1.modules.public.reference_data.customs_sections.models import CustomsSection
|
||||
from core.exceptions import ErrorCollector
|
||||
from typing import Dict, Any
|
||||
|
||||
|
||||
def invoice_exists(
|
||||
@@ -58,7 +72,6 @@ def invoice_exists_by_id(
|
||||
return invoice
|
||||
return None
|
||||
|
||||
|
||||
def invoice_updated(
|
||||
db: Session,
|
||||
invoice_id: str,
|
||||
@@ -86,3 +99,519 @@ def invoice_updated(
|
||||
)
|
||||
return True
|
||||
return False
|
||||
|
||||
def validate_required_fields_by_operation(
|
||||
invoice_data: Dict[str, Any],
|
||||
operation_type: str,
|
||||
errors: ErrorCollector
|
||||
) -> None:
|
||||
"""
|
||||
Valida campos obligatorios según tipo de operación.
|
||||
Usar ANTES de guardar en BD.
|
||||
"""
|
||||
|
||||
# PROVEEDOR (SIEMPRE OBLIGATORIO - mensaje dinámico)
|
||||
if not invoice_data.get('provider_id'):
|
||||
# Mensaje dinámico según el header seleccionado
|
||||
provider_labels = {
|
||||
'proveedor': 'Proveedor',
|
||||
'exportador': 'Exportador'
|
||||
}
|
||||
provider_header = invoice_data.get('provider_header') or 'proveedor'
|
||||
field_label = provider_labels.get(provider_header, 'Proveedor')
|
||||
|
||||
errors.add_error(
|
||||
field="provider_id",
|
||||
message=f"Debe seleccionar {field_label}",
|
||||
solution=["Seleccione un proveedor de la lista desplegable"],
|
||||
code="REQUIRED",
|
||||
value=None
|
||||
)
|
||||
|
||||
# VENDIDO A / CONSIGNADO A (SIEMPRE OBLIGATORIO - mensaje dinámico)
|
||||
if not invoice_data.get('sold_to_id'):
|
||||
# Mensaje dinámico según el header seleccionado
|
||||
sold_to_labels = {
|
||||
'consignado_a': 'Consignado a',
|
||||
'vendido_a': 'Vendido a',
|
||||
'exportado_a': 'Exportado a',
|
||||
'importador': 'Importador'
|
||||
}
|
||||
sold_to_header = invoice_data.get('sold_to_header') or 'consignado_a'
|
||||
field_label = sold_to_labels.get(sold_to_header, 'Cliente')
|
||||
|
||||
errors.add_error(
|
||||
field="sold_to_id",
|
||||
message=f"Debe seleccionar {field_label}",
|
||||
solution=["Seleccione una opción de la lista desplegable"],
|
||||
code="REQUIRED",
|
||||
value=None
|
||||
)
|
||||
|
||||
# ENVIADO A (SIEMPRE OBLIGATORIO - mensaje fijo)
|
||||
if not invoice_data.get('shipped_to_id'):
|
||||
errors.add_error(
|
||||
field="shipped_to_id",
|
||||
message="Debe seleccionar el Destinatario",
|
||||
solution=["Seleccione un destinatario de la lista desplegable"],
|
||||
code="REQUIRED",
|
||||
value=None
|
||||
)
|
||||
|
||||
# AGENTE ADUANAL (OBLIGATORIO si hay pedimento)
|
||||
if invoice_data.get('pedimento_id') and not invoice_data.get('customs_broker_id'):
|
||||
errors.add_error(
|
||||
field="customs_broker_id",
|
||||
message="Debe seleccionar un Agente Aduanal",
|
||||
solution=["Seleccione un agente aduanal de la lista desplegable"],
|
||||
code="REQUIRED",
|
||||
value=None
|
||||
)
|
||||
|
||||
|
||||
def validate_common(
|
||||
db: Session,
|
||||
invoice: schemas.InvoiceHeaderUpdate,
|
||||
tenant_id: int,
|
||||
company_id: int,
|
||||
errors: ErrorCollector,
|
||||
):
|
||||
if invoice.compliance_mx.pedimento_id:
|
||||
pedimento = (
|
||||
db.query(Pedimentos)
|
||||
.filter(
|
||||
Pedimentos.id == invoice.compliance_mx.pedimento_id,
|
||||
Pedimentos.tenant_id == tenant_id,
|
||||
Pedimentos.company_id == company_id,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
if not pedimento:
|
||||
errors.add_error(
|
||||
field="compliance_mx.pedimento_id",
|
||||
message="El Pedimento no existe en el Catálogo de Pedimentos.",
|
||||
solution=["Verifica el ID", "Revisa el catálogo"],
|
||||
code="NOT_FOUND",
|
||||
value=invoice.compliance_mx.pedimento_id,
|
||||
)
|
||||
return # Stop here if pedimento not found
|
||||
|
||||
if not invoice.compliance_mx.is_regime_change:
|
||||
operacion = "Importación" if pedimento.operation_type == "imp" else "Exportación"
|
||||
# Validar que el pedimento sea de importación
|
||||
if pedimento.operation_type != invoice.operation_type:
|
||||
errors.add_error(
|
||||
field="compliance_mx.pedimento_id",
|
||||
message=f"El Pedimento seleccionado no corresponde a una {operacion}.",
|
||||
solution=[f"Selecciona un Pedimento de {operacion}"],
|
||||
code="INVALID_OPERATION_TYPE",
|
||||
value=pedimento.operation_type,
|
||||
)
|
||||
else:
|
||||
# Validar regímenes incompatibles
|
||||
only_regimes = ["EXD", "ETE", "ETR"]
|
||||
|
||||
if pedimento.regime in only_regimes:
|
||||
errors.add_error(
|
||||
field="compliance_mx.pedimento_id",
|
||||
message=f"El Pedimento seleccionado corresponde a una Exportación, no a una {operacion}.",
|
||||
solution=[f"Selecciona un Pedimento de {operacion}"],
|
||||
code="INVALID_REGIME",
|
||||
value=pedimento.regime,
|
||||
)
|
||||
|
||||
# Validar que el tipo de documento coincida con el régimen del pedimento
|
||||
# NOTA: Solo validamos si no hay errores previos y si document_type está presente
|
||||
if not errors.has_errors() and invoice.document_type:
|
||||
if invoice.document_type.upper().strip() != pedimento.regime:
|
||||
errors.add_error(
|
||||
field="document_type",
|
||||
message=f"El Tipo de Documento '{invoice.document_type}' no coincide con el Régimen '{pedimento.regime}' del Pedimento seleccionado.",
|
||||
solution=[
|
||||
"Ajusta el Tipo de Documento o selecciona otro Pedimento"
|
||||
],
|
||||
code="REGIME_MISMATCH",
|
||||
value=invoice.document_type,
|
||||
)
|
||||
else:
|
||||
# Cambio de Régimen - Generalmente es de Importación Temporal a Definitiva (IMD)
|
||||
# En el código legacy se comparaba pedimento.operation_type != 2.
|
||||
# Si asumimos que 2 era Importación en el sistema anterior:
|
||||
if pedimento.operation_type != "imp":
|
||||
errors.add_error(
|
||||
field="compliance_mx.pedimento_id",
|
||||
message="El Pedimento seleccionado no corresponde a una Importación (requerido para Cambio de Régimen).",
|
||||
solution=["Selecciona un Pedimento de Importación"],
|
||||
code="INVALID_OPERATION_TYPE",
|
||||
value=pedimento.operation_type,
|
||||
)
|
||||
else:
|
||||
if pedimento.regime != "IMD" and invoice.document_type == "IMD":
|
||||
# Si el destino es IMD, validamos que el pedimento original sea de importación
|
||||
# (aunque usualmente el pedimento que se asocia aquí es el nuevo, el de IMD)
|
||||
pass
|
||||
|
||||
if invoice.document_type.upper().strip() != pedimento.regime:
|
||||
errors.add_error(
|
||||
field="document_type",
|
||||
message=f"El Tipo de Documento '{invoice.document_type}' no coincide con el Régimen '{pedimento.regime}' del Pedimento seleccionado.",
|
||||
solution=[
|
||||
"Ajusta el Tipo de Documento o selecciona otro Pedimento"
|
||||
],
|
||||
code="REGIME_MISMATCH",
|
||||
value=invoice.document_type,
|
||||
)
|
||||
|
||||
if pedimento.pedimento_code not in ["A1", "A3"]:
|
||||
errors.add_error(
|
||||
field="compliance_mx.pedimento_id",
|
||||
message=f"El Pedimento seleccionado no es de tipo A1 o A3 requerido para Cambio de Régimen.",
|
||||
solution=["Selecciona un Pedimento de tipo A1 o A3"],
|
||||
code="INVALID_PEDEMENTO_CODE",
|
||||
value=pedimento.pedimento_code,
|
||||
)
|
||||
|
||||
if pedimento.pedimento_type == "consolidated":
|
||||
# Convertir invoice_date a date si es datetime para poder comparar
|
||||
invoice_date = (
|
||||
invoice.invoice_date.date()
|
||||
if hasattr(invoice.invoice_date, "date")
|
||||
else invoice.invoice_date
|
||||
)
|
||||
entry_date = (
|
||||
pedimento.pedimento_dates.entry_date.date()
|
||||
if hasattr(pedimento.pedimento_dates.entry_date, "date")
|
||||
else pedimento.pedimento_dates.entry_date
|
||||
)
|
||||
end_date = (
|
||||
pedimento.pedimento_dates.end_date.date()
|
||||
if hasattr(pedimento.pedimento_dates.end_date, "date")
|
||||
else pedimento.pedimento_dates.end_date
|
||||
)
|
||||
|
||||
if invoice_date < entry_date or invoice_date > end_date:
|
||||
errors.add_error(
|
||||
field="invoice_date",
|
||||
message=f"La Fecha de la Factura {invoice.invoice_date} no está dentro del rango de fechas del Pedimento {pedimento.customs_office}-{pedimento.license}-{pedimento.pedimento_number}.",
|
||||
solution=[
|
||||
f"Capturar una Fecha de Factura, entre la Fecha de Inicio: {pedimento.pedimento_dates.entry_date} y la Fecha Final: {pedimento.pedimento_dates.end_date} ."
|
||||
],
|
||||
code="DATE_OUT_OF_RANGE",
|
||||
value=invoice.invoice_date,
|
||||
)
|
||||
|
||||
# Remesa check
|
||||
if pedimento.pedimento_type == "consolidated":
|
||||
if not invoice.compliance_mx.remesa:
|
||||
errors.add_error(
|
||||
field="compliance_mx.remesa",
|
||||
message="El campo Remesa es obligatorio cuando se asocia un Pedimento consolidado.",
|
||||
solution=["Proporciona un valor para Remesa"],
|
||||
code="REQUIRED_FIELD",
|
||||
value=invoice.compliance_mx.remesa,
|
||||
)
|
||||
elif invoice.compliance_mx.remesa == 0:
|
||||
errors.add_error(
|
||||
field="compliance_mx.remesa",
|
||||
message="El campo Remesa no puede ser cero cuando se asocia un Pedimento.",
|
||||
solution=["Proporciona un valor válido para Remesa"],
|
||||
code="INVALID_VALUE",
|
||||
value=invoice.compliance_mx.remesa,
|
||||
)
|
||||
|
||||
duplicated_remesa = (
|
||||
db.query(InvoiceComplianceMx)
|
||||
.filter(
|
||||
InvoiceComplianceMx.remesa == invoice.compliance_mx.remesa,
|
||||
InvoiceComplianceMx.tenant_id == tenant_id,
|
||||
InvoiceComplianceMx.company_id == company_id,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
if duplicated_remesa and hasattr(invoice, "id"):
|
||||
if invoice.id != duplicated_remesa.invoice_id:
|
||||
errors.add_error(
|
||||
field="compliance_mx.remesa",
|
||||
message="El valor de Remesa ya está asociado a otro Pedimento.",
|
||||
solution=["Proporciona un valor único para Remesa"],
|
||||
code="DUPLICATE_VALUE",
|
||||
value=invoice.compliance_mx.remesa,
|
||||
)
|
||||
|
||||
# Financials checks (if provided)
|
||||
if invoice.financials:
|
||||
if not invoice.financials.exchange_rate or invoice.financials.exchange_rate <= 0:
|
||||
exchange_rate_exists = (
|
||||
db.query(ExchangeRate)
|
||||
.filter(
|
||||
func.date(ExchangeRate.date) == invoice.invoice_date,
|
||||
ExchangeRate.tenant_id == tenant_id,
|
||||
ExchangeRate.company_id == company_id,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
if not exchange_rate_exists:
|
||||
errors.add_error(
|
||||
field="financials.exchange_rate",
|
||||
message=f"No existe un Tipo de Cambio registrado para la fecha {invoice.invoice_date}.",
|
||||
solution=["Registra el Tipo de Cambio en el catálogo correspondiente"],
|
||||
code="EXCHANGE_RATE_NOT_FOUND",
|
||||
value=invoice.financials.exchange_rate,
|
||||
)
|
||||
else:
|
||||
invoice.financials.exchange_rate = exchange_rate_exists.value
|
||||
else:
|
||||
# If financials missing, we might want to error if it's required for this operation
|
||||
pass
|
||||
|
||||
if invoice.compliance_mx.is_regime_change:
|
||||
if invoice.document_type in ["EXD", "ETE", "ETR"]:
|
||||
errors.add_error(
|
||||
field="document_type",
|
||||
message="El Tipo de Documento no puede ser de Exportación cuando se trata de un Cambio de Régimen.",
|
||||
solution=[
|
||||
"Selecciona un Tipo de Documento válido para Cambio de Régimen"
|
||||
],
|
||||
code="INVALID_DOCUMENT_TYPE",
|
||||
value=invoice.document_type,
|
||||
)
|
||||
else:
|
||||
if invoice.document_type == "IMD":
|
||||
errors.add_error(
|
||||
field="document_type",
|
||||
message="El Tipo de Documento no puede ser 'IMD' a menos que sea un Cambio de Régimen.",
|
||||
solution=["Selecciona un Tipo de Documento válido"],
|
||||
code="INVALID_DOCUMENT_TYPE",
|
||||
value=invoice.document_type,
|
||||
)
|
||||
|
||||
# Validar proveedor solo si se proporciona
|
||||
if invoice.compliance_mx.provider_id:
|
||||
provider_exists = (
|
||||
db.query(ClientProvider)
|
||||
.filter(
|
||||
ClientProvider.id == invoice.compliance_mx.provider_id,
|
||||
ClientProvider.tenant_id == tenant_id,
|
||||
ClientProvider.company_id == company_id,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
if not provider_exists:
|
||||
errors.add_error(
|
||||
field="compliance_mx.provider_id",
|
||||
message="El Proveedor no existe en el Catálogo de Clientes y Proveedores.",
|
||||
solution=["Verifica el ID del Proveedor", "Revisa el catálogo"],
|
||||
code="NOT_FOUND",
|
||||
value=invoice.compliance_mx.provider_id,
|
||||
)
|
||||
|
||||
# Validar vendido a solo si se proporciona
|
||||
if invoice.compliance_mx.sold_to_id:
|
||||
selled_to_exists = (
|
||||
db.query(ClientProvider)
|
||||
.filter(
|
||||
ClientProvider.id == invoice.compliance_mx.sold_to_id,
|
||||
ClientProvider.tenant_id == tenant_id,
|
||||
ClientProvider.company_id == company_id,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
if not selled_to_exists:
|
||||
errors.add_error(
|
||||
field="compliance_mx.sold_to_id",
|
||||
message="El Cliente no existe en el Catálogo de Clientes y Proveedores.",
|
||||
solution=["Verifica el ID del Cliente", "Revisa el catálogo"],
|
||||
code="NOT_FOUND",
|
||||
value=invoice.compliance_mx.sold_to_id,
|
||||
)
|
||||
|
||||
# Validar destinatario solo si se proporciona
|
||||
if invoice.compliance_mx.shipped_to_id:
|
||||
shipped_to_exists = (
|
||||
db.query(ClientProvider)
|
||||
.filter(
|
||||
ClientProvider.id == invoice.compliance_mx.shipped_to_id,
|
||||
ClientProvider.tenant_id == tenant_id,
|
||||
ClientProvider.company_id == company_id,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
if not shipped_to_exists:
|
||||
errors.add_error(
|
||||
field="compliance_mx.shipped_to_id",
|
||||
message="El Destinatario no existe en el Catálogo de Clientes y Proveedores.",
|
||||
solution=["Verifica el ID del Destinatario", "Revisa el catálogo"],
|
||||
code="NOT_FOUND",
|
||||
value=invoice.compliance_mx.shipped_to_id,
|
||||
)
|
||||
|
||||
# Validar agente aduanal solo si se proporciona
|
||||
if invoice.compliance_mx.customs_broker_id:
|
||||
customs_broker_exists = (
|
||||
db.query(CustomsBroker)
|
||||
.filter(
|
||||
CustomsBroker.id == invoice.compliance_mx.customs_broker_id,
|
||||
CustomsBroker.tenant_id == tenant_id,
|
||||
CustomsBroker.company_id == company_id,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
if not customs_broker_exists:
|
||||
errors.add_error(
|
||||
field="compliance_mx.customs_broker_id",
|
||||
message="El Agente Aduanal no existe en el Catálogo de Clientes y Proveedores.",
|
||||
solution=["Verifica el ID del Agente Aduanal", "Revisa el catálogo"],
|
||||
code="NOT_FOUND",
|
||||
value=invoice.compliance_mx.customs_broker_id,
|
||||
)
|
||||
|
||||
if invoice.logistics:
|
||||
if invoice.logistics.transport_num and not invoice.logistics.transport_num:
|
||||
# logic ...
|
||||
pass
|
||||
|
||||
if invoice.logistics.transport_type not in [t.value for t in TransportType]:
|
||||
errors.add_error(
|
||||
field="logistics.transport_type",
|
||||
message="El Tipo de Transporte proporcionado no es válido.",
|
||||
solution=[
|
||||
f"Selecciona un Tipo de Transporte válido: {[t.value for t in TransportType]}"
|
||||
],
|
||||
code="INVALID_TRANSPORT_TYPE",
|
||||
value=invoice.logistics.transport_type,
|
||||
)
|
||||
else:
|
||||
if (
|
||||
invoice.logistics.transport_type == "none"
|
||||
and invoice.logistics.transport_num
|
||||
):
|
||||
errors.add_error(
|
||||
field="logistics.transport_num",
|
||||
message="El Número de Transporte no debe proporcionarse cuando el Tipo de Transporte es 'none'.",
|
||||
solution=[
|
||||
"Elimina el Número de Transporte o selecciona un Tipo de Transporte válido"
|
||||
],
|
||||
code="INVALID_VALUE",
|
||||
value=invoice.logistics.transport_num,
|
||||
)
|
||||
else:
|
||||
if (
|
||||
not invoice.logistics.transport_num
|
||||
and invoice.logistics.transport_type != "none"
|
||||
):
|
||||
errors.add_error(
|
||||
field="logistics.transport_num",
|
||||
message="El Número de Transporte es obligatorio cuando se proporciona un Tipo de Transporte distinto de 'none'.",
|
||||
solution=["Proporciona un Número de Transporte válido"],
|
||||
code="REQUIRED_FIELD",
|
||||
value=invoice.logistics.transport_num,
|
||||
)
|
||||
|
||||
invoice.financials.currency = invoice.financials.currency or "foreign"
|
||||
|
||||
if invoice.financials.currency not in [c.value for c in Currency]:
|
||||
errors.add_error(
|
||||
field="financials.currency",
|
||||
message="La Moneda proporcionada no es válida.",
|
||||
solution=[f"Selecciona una Moneda válida: {[c.value for c in Currency]}"],
|
||||
code="INVALID_CURRENCY",
|
||||
value=invoice.financials.currency,
|
||||
)
|
||||
else:
|
||||
# Only check for existing items during update operations (when invoice has an id)
|
||||
if hasattr(invoice, "id"):
|
||||
has_items = (
|
||||
db.query(LineItem)
|
||||
.filter(
|
||||
LineItem.invoice_id == invoice.id,
|
||||
LineItem.tenant_id == tenant_id,
|
||||
LineItem.company_id == company_id,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
if has_items:
|
||||
errors.add_error(
|
||||
field="items",
|
||||
message=f"La opcion tipo de moneda {invoice.financials.currency} no puede ser modificada ya que la factura tiene items asociados.",
|
||||
solution=[
|
||||
"Verifica la moneda de los items asociados a la factura."
|
||||
],
|
||||
code="CURRENCY_CANNOT_BE_CHANGED",
|
||||
value=invoice.financials.currency,
|
||||
)
|
||||
|
||||
if invoice.financials.currency == "foreign":
|
||||
invoice.financials.currency_type = "USD"
|
||||
elif invoice.financials.currency == "local":
|
||||
invoice.financials.currency_type = "MXN"
|
||||
elif invoice.financials.currency == "manual":
|
||||
if not invoice.financials.currency_type:
|
||||
errors.add_error(
|
||||
field="financials.currency_type",
|
||||
message="El Tipo de Moneda es obligatorio cuando la Moneda es 'manual'.",
|
||||
solution=["Proporciona un Tipo de Moneda válido"],
|
||||
code="REQUIRED_FIELD",
|
||||
value=invoice.financials.currency_type,
|
||||
)
|
||||
else:
|
||||
currency_exists = (
|
||||
db.query(CurrencyType)
|
||||
.filter(CurrencyType.code == invoice.financials.currency_type)
|
||||
.first()
|
||||
)
|
||||
if not currency_exists:
|
||||
errors.add_error(
|
||||
field="financials.currency_type",
|
||||
message="El Tipo de Moneda no existe en el Catálogo de Tipos de Moneda.",
|
||||
solution=[
|
||||
"Verifica el código del Tipo de Moneda",
|
||||
"Revisa el catálogo",
|
||||
],
|
||||
code="NOT_FOUND",
|
||||
value=invoice.financials.currency_type,
|
||||
)
|
||||
|
||||
if invoice.logistics.incoterm:
|
||||
incoterm_exists = (
|
||||
db.query(Incoterm)
|
||||
.filter(
|
||||
Incoterm.code == invoice.logistics.incoterm
|
||||
)
|
||||
.first()
|
||||
)
|
||||
if not incoterm_exists:
|
||||
errors.add_error(
|
||||
field="logistics.incoterm",
|
||||
message="El Incoterm no existe en el Catálogo de Incoterms.",
|
||||
solution=["Verifica el código del Incoterm", "Revisa el catálogo"],
|
||||
code="NOT_FOUND",
|
||||
value=invoice.logistics.incoterm,
|
||||
)
|
||||
|
||||
if invoice.logistics.weight_type not in [w.value for w in WeightUnit]:
|
||||
errors.add_error(
|
||||
field="logistics.weight_type",
|
||||
message="La Unidad de Peso proporcionada no es válida.",
|
||||
solution=[
|
||||
f"Selecciona una Unidad de Peso válida: {[w.value for w in WeightUnit]}"
|
||||
],
|
||||
code="INVALID_WEIGHT_UNIT",
|
||||
value=invoice.logistics.weight_type,
|
||||
)
|
||||
|
||||
if invoice.compliance_mx.aduana:
|
||||
custom_section_exists = (
|
||||
db.query(CustomsSection)
|
||||
.filter(
|
||||
CustomsSection.customs_code == invoice.compliance_mx.aduana,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
if not custom_section_exists:
|
||||
errors.add_error(
|
||||
field="compliance_mx.aduana",
|
||||
message="La Aduana no existe en el Catálogo de Secciones Aduaneras.",
|
||||
solution=["Verifica el código de la Aduana", "Revisa el catálogo"],
|
||||
code="NOT_FOUND",
|
||||
value=invoice.compliance_mx.aduana,
|
||||
)
|
||||
@@ -0,0 +1,95 @@
|
||||
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()
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
259
backend/api/v1/modules/a76/invoices/exports/validators/update.py
Normal file
259
backend/api/v1/modules/a76/invoices/exports/validators/update.py
Normal file
@@ -0,0 +1,259 @@
|
||||
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=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: 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 AA: 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)
|
||||
|
||||
# Validar que aduana sea obligatorio (excepto para MEX)
|
||||
if existing_invoice.invoice_type != "MEX":
|
||||
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 AD: 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 AD: 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)
|
||||
|
||||
@@ -1,531 +0,0 @@
|
||||
from sqlalchemy import func
|
||||
from sqlalchemy.orm import Session
|
||||
from ... import schemas
|
||||
from api.v1.modules.a76.pedmientos.models.pedimentos import Pedimentos
|
||||
from api.v1.modules.a76.general_catalogs.exchange_rate.models import ExchangeRate
|
||||
from api.v1.modules.a76.clients_and_providers.models import ClientProvider
|
||||
from api.v1.modules.a76.customs_brokers.models import CustomsBroker
|
||||
from api.v1.modules.public.reference_data.incoterms.models import Incoterm
|
||||
from ...models import InvoiceComplianceMx
|
||||
from api.v1.modules.a76.items.models import LineItem
|
||||
from api.v1.modules.public.reference_data.currency_types.models import CurrencyType
|
||||
from api.v1.modules.public.reference_data.customs_sections.models import CustomsSection
|
||||
from ...models import TransportType, Currency, WeightUnit
|
||||
from core.exceptions import ErrorCollector
|
||||
from typing import Dict, Any
|
||||
|
||||
|
||||
def validate_required_fields_by_operation(
|
||||
invoice_data: Dict[str, Any],
|
||||
operation_type: str,
|
||||
errors: ErrorCollector
|
||||
) -> None:
|
||||
"""
|
||||
Valida campos obligatorios según tipo de operación.
|
||||
Usar ANTES de guardar en BD.
|
||||
"""
|
||||
|
||||
# PROVEEDOR (SIEMPRE OBLIGATORIO - mensaje dinámico)
|
||||
if not invoice_data.get('provider_id'):
|
||||
# Mensaje dinámico según el header seleccionado
|
||||
provider_labels = {
|
||||
'proveedor': 'Proveedor',
|
||||
'exportador': 'Exportador'
|
||||
}
|
||||
provider_header = invoice_data.get('provider_header') or 'proveedor'
|
||||
field_label = provider_labels.get(provider_header, 'Proveedor')
|
||||
|
||||
errors.add_error(
|
||||
field="provider_id",
|
||||
message=f"Debe seleccionar {field_label}",
|
||||
solution=["Seleccione un proveedor de la lista desplegable"],
|
||||
code="REQUIRED",
|
||||
value=None
|
||||
)
|
||||
|
||||
# VENDIDO A / CONSIGNADO A (SIEMPRE OBLIGATORIO - mensaje dinámico)
|
||||
if not invoice_data.get('sold_to_id'):
|
||||
# Mensaje dinámico según el header seleccionado
|
||||
sold_to_labels = {
|
||||
'consignado_a': 'Consignado a',
|
||||
'vendido_a': 'Vendido a',
|
||||
'exportado_a': 'Exportado a',
|
||||
'importador': 'Importador'
|
||||
}
|
||||
sold_to_header = invoice_data.get('sold_to_header') or 'consignado_a'
|
||||
field_label = sold_to_labels.get(sold_to_header, 'Cliente')
|
||||
|
||||
errors.add_error(
|
||||
field="sold_to_id",
|
||||
message=f"Debe seleccionar {field_label}",
|
||||
solution=["Seleccione una opción de la lista desplegable"],
|
||||
code="REQUIRED",
|
||||
value=None
|
||||
)
|
||||
|
||||
# ENVIADO A (SIEMPRE OBLIGATORIO - mensaje fijo)
|
||||
if not invoice_data.get('shipped_to_id'):
|
||||
errors.add_error(
|
||||
field="shipped_to_id",
|
||||
message="Debe seleccionar el Destinatario",
|
||||
solution=["Seleccione un destinatario de la lista desplegable"],
|
||||
code="REQUIRED",
|
||||
value=None
|
||||
)
|
||||
|
||||
# AGENTE ADUANAL (OBLIGATORIO si hay pedimento)
|
||||
if invoice_data.get('pedimento_id') and not invoice_data.get('customs_broker_id'):
|
||||
errors.add_error(
|
||||
field="customs_broker_id",
|
||||
message="Debe seleccionar un Agente Aduanal",
|
||||
solution=["Seleccione un agente aduanal de la lista desplegable"],
|
||||
code="REQUIRED",
|
||||
value=None
|
||||
)
|
||||
|
||||
|
||||
def validate_common(
|
||||
db: Session,
|
||||
invoice: schemas.InvoiceHeaderUpdate,
|
||||
tenant_id: int,
|
||||
company_id: int,
|
||||
errors: ErrorCollector,
|
||||
):
|
||||
if invoice.compliance_mx.pedimento_id:
|
||||
pedimento = (
|
||||
db.query(Pedimentos)
|
||||
.filter(
|
||||
Pedimentos.id == invoice.compliance_mx.pedimento_id,
|
||||
Pedimentos.tenant_id == tenant_id,
|
||||
Pedimentos.company_id == company_id,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
if not pedimento:
|
||||
errors.add_error(
|
||||
field="compliance_mx.pedimento_id",
|
||||
message="El Pedimento no existe en el Catálogo de Pedimentos.",
|
||||
solution=["Verifica el ID", "Revisa el catálogo"],
|
||||
code="NOT_FOUND",
|
||||
value=invoice.compliance_mx.pedimento_id,
|
||||
)
|
||||
return # Stop here if pedimento not found
|
||||
|
||||
if not invoice.compliance_mx.is_regime_change:
|
||||
# Validar que el pedimento sea de importación
|
||||
if pedimento.operation_type != invoice.operation_type:
|
||||
errors.add_error(
|
||||
field="compliance_mx.pedimento_id",
|
||||
message="El Pedimento seleccionado no corresponde a una Importación.",
|
||||
solution=["Selecciona un Pedimento de Importación"],
|
||||
code="INVALID_OPERATION_TYPE",
|
||||
value=pedimento.operation_type,
|
||||
)
|
||||
else:
|
||||
# Validar regímenes incompatibles
|
||||
export_only_regimes = ["EXD", "ETE", "ETR"]
|
||||
|
||||
if pedimento.regime in export_only_regimes:
|
||||
errors.add_error(
|
||||
field="compliance_mx.pedimento_id",
|
||||
message="El Pedimento seleccionado corresponde a una Exportación, no a una Importación.",
|
||||
solution=["Selecciona un Pedimento de Importación"],
|
||||
code="INVALID_REGIME",
|
||||
value=pedimento.regime,
|
||||
)
|
||||
|
||||
# Validar que el tipo de documento coincida con el régimen del pedimento
|
||||
# NOTA: Solo validamos si no hay errores previos y si document_type está presente
|
||||
if not errors.has_errors() and invoice.document_type:
|
||||
if invoice.document_type.upper().strip() != pedimento.regime:
|
||||
errors.add_error(
|
||||
field="document_type",
|
||||
message=f"El Tipo de Documento '{invoice.document_type}' no coincide con el Régimen '{pedimento.regime}' del Pedimento seleccionado.",
|
||||
solution=[
|
||||
"Ajusta el Tipo de Documento o selecciona otro Pedimento"
|
||||
],
|
||||
code="REGIME_MISMATCH",
|
||||
value=invoice.document_type,
|
||||
)
|
||||
else:
|
||||
# Cambio de Régimen - Generalmente es de Importación Temporal a Definitiva (IMD)
|
||||
# En el código legacy se comparaba pedimento.operation_type != 2.
|
||||
# Si asumimos que 2 era Importación en el sistema anterior:
|
||||
if pedimento.operation_type != "imp":
|
||||
errors.add_error(
|
||||
field="compliance_mx.pedimento_id",
|
||||
message="El Pedimento seleccionado no corresponde a una Importación (requerido para Cambio de Régimen).",
|
||||
solution=["Selecciona un Pedimento de Importación"],
|
||||
code="INVALID_OPERATION_TYPE",
|
||||
value=pedimento.operation_type,
|
||||
)
|
||||
else:
|
||||
if pedimento.regime != "IMD" and invoice.document_type == "IMD":
|
||||
# Si el destino es IMD, validamos que el pedimento original sea de importación
|
||||
# (aunque usualmente el pedimento que se asocia aquí es el nuevo, el de IMD)
|
||||
pass
|
||||
|
||||
if invoice.document_type.upper().strip() != pedimento.regime:
|
||||
errors.add_error(
|
||||
field="document_type",
|
||||
message=f"El Tipo de Documento '{invoice.document_type}' no coincide con el Régimen '{pedimento.regime}' del Pedimento seleccionado.",
|
||||
solution=[
|
||||
"Ajusta el Tipo de Documento o selecciona otro Pedimento"
|
||||
],
|
||||
code="REGIME_MISMATCH",
|
||||
value=invoice.document_type,
|
||||
)
|
||||
|
||||
if pedimento.pedimento_code not in ["A1", "A3"]:
|
||||
errors.add_error(
|
||||
field="compliance_mx.pedimento_id",
|
||||
message=f"El Pedimento seleccionado no es de tipo A1 o A3 requerido para Cambio de Régimen.",
|
||||
solution=["Selecciona un Pedimento de tipo A1 o A3"],
|
||||
code="INVALID_PEDEMENTO_CODE",
|
||||
value=pedimento.pedimento_code,
|
||||
)
|
||||
|
||||
if pedimento.pedimento_type == "consolidated":
|
||||
# Convertir invoice_date a date si es datetime para poder comparar
|
||||
invoice_date = (
|
||||
invoice.invoice_date.date()
|
||||
if hasattr(invoice.invoice_date, "date")
|
||||
else invoice.invoice_date
|
||||
)
|
||||
entry_date = (
|
||||
pedimento.pedimento_dates.entry_date.date()
|
||||
if hasattr(pedimento.pedimento_dates.entry_date, "date")
|
||||
else pedimento.pedimento_dates.entry_date
|
||||
)
|
||||
end_date = (
|
||||
pedimento.pedimento_dates.end_date.date()
|
||||
if hasattr(pedimento.pedimento_dates.end_date, "date")
|
||||
else pedimento.pedimento_dates.end_date
|
||||
)
|
||||
|
||||
if invoice_date < entry_date or invoice_date > end_date:
|
||||
errors.add_error(
|
||||
field="invoice_date",
|
||||
message=f"La Fecha de la Factura {invoice.invoice_date} no está dentro del rango de fechas del Pedimento {pedimento.customs_office}-{pedimento.license}-{pedimento.pedimento_number}.",
|
||||
solution=[
|
||||
f"Capturar una Fecha de Factura, entre la Fecha de Inicio: {pedimento.pedimento_dates.entry_date} y la Fecha Final: {pedimento.pedimento_dates.end_date} ."
|
||||
],
|
||||
code="DATE_OUT_OF_RANGE",
|
||||
value=invoice.invoice_date,
|
||||
)
|
||||
|
||||
# Remesa check
|
||||
if pedimento.pedimento_type == "consolidated":
|
||||
if not invoice.compliance_mx.remesa:
|
||||
errors.add_error(
|
||||
field="compliance_mx.remesa",
|
||||
message="El campo Remesa es obligatorio cuando se asocia un Pedimento consolidado.",
|
||||
solution=["Proporciona un valor para Remesa"],
|
||||
code="REQUIRED_FIELD",
|
||||
value=invoice.compliance_mx.remesa,
|
||||
)
|
||||
elif invoice.compliance_mx.remesa == 0:
|
||||
errors.add_error(
|
||||
field="compliance_mx.remesa",
|
||||
message="El campo Remesa no puede ser cero cuando se asocia un Pedimento.",
|
||||
solution=["Proporciona un valor válido para Remesa"],
|
||||
code="INVALID_VALUE",
|
||||
value=invoice.compliance_mx.remesa,
|
||||
)
|
||||
|
||||
duplicated_remesa = (
|
||||
db.query(InvoiceComplianceMx)
|
||||
.filter(
|
||||
InvoiceComplianceMx.remesa == invoice.compliance_mx.remesa,
|
||||
InvoiceComplianceMx.tenant_id == tenant_id,
|
||||
InvoiceComplianceMx.company_id == company_id,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
if duplicated_remesa and hasattr(invoice, "id"):
|
||||
if invoice.id != duplicated_remesa.invoice_id:
|
||||
errors.add_error(
|
||||
field="compliance_mx.remesa",
|
||||
message="El valor de Remesa ya está asociado a otro Pedimento.",
|
||||
solution=["Proporciona un valor único para Remesa"],
|
||||
code="DUPLICATE_VALUE",
|
||||
value=invoice.compliance_mx.remesa,
|
||||
)
|
||||
|
||||
# Financials checks (if provided)
|
||||
if invoice.financials:
|
||||
if not invoice.financials.exchange_rate or invoice.financials.exchange_rate <= 0:
|
||||
exchange_rate_exists = (
|
||||
db.query(ExchangeRate)
|
||||
.filter(
|
||||
func.date(ExchangeRate.date) == invoice.invoice_date,
|
||||
ExchangeRate.tenant_id == tenant_id,
|
||||
ExchangeRate.company_id == company_id,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
if not exchange_rate_exists:
|
||||
errors.add_error(
|
||||
field="financials.exchange_rate",
|
||||
message=f"No existe un Tipo de Cambio registrado para la fecha {invoice.invoice_date}.",
|
||||
solution=["Registra el Tipo de Cambio en el catálogo correspondiente"],
|
||||
code="EXCHANGE_RATE_NOT_FOUND",
|
||||
value=invoice.financials.exchange_rate,
|
||||
)
|
||||
else:
|
||||
invoice.financials.exchange_rate = exchange_rate_exists.value
|
||||
else:
|
||||
# If financials missing, we might want to error if it's required for this operation
|
||||
pass
|
||||
|
||||
if invoice.compliance_mx.is_regime_change:
|
||||
if invoice.document_type in ["EXD", "ETE", "ETR"]:
|
||||
errors.add_error(
|
||||
field="document_type",
|
||||
message="El Tipo de Documento no puede ser de Exportación cuando se trata de un Cambio de Régimen.",
|
||||
solution=[
|
||||
"Selecciona un Tipo de Documento válido para Cambio de Régimen"
|
||||
],
|
||||
code="INVALID_DOCUMENT_TYPE",
|
||||
value=invoice.document_type,
|
||||
)
|
||||
else:
|
||||
if invoice.document_type == "IMD":
|
||||
errors.add_error(
|
||||
field="document_type",
|
||||
message="El Tipo de Documento no puede ser 'IMD' a menos que sea un Cambio de Régimen.",
|
||||
solution=["Selecciona un Tipo de Documento válido"],
|
||||
code="INVALID_DOCUMENT_TYPE",
|
||||
value=invoice.document_type,
|
||||
)
|
||||
|
||||
# Validar proveedor solo si se proporciona
|
||||
if invoice.compliance_mx.provider_id:
|
||||
provider_exists = (
|
||||
db.query(ClientProvider)
|
||||
.filter(
|
||||
ClientProvider.id == invoice.compliance_mx.provider_id,
|
||||
ClientProvider.tenant_id == tenant_id,
|
||||
ClientProvider.company_id == company_id,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
if not provider_exists:
|
||||
errors.add_error(
|
||||
field="compliance_mx.provider_id",
|
||||
message="El Proveedor no existe en el Catálogo de Clientes y Proveedores.",
|
||||
solution=["Verifica el ID del Proveedor", "Revisa el catálogo"],
|
||||
code="NOT_FOUND",
|
||||
value=invoice.compliance_mx.provider_id,
|
||||
)
|
||||
|
||||
# Validar vendido a solo si se proporciona
|
||||
if invoice.compliance_mx.sold_to_id:
|
||||
selled_to_exists = (
|
||||
db.query(ClientProvider)
|
||||
.filter(
|
||||
ClientProvider.id == invoice.compliance_mx.sold_to_id,
|
||||
ClientProvider.tenant_id == tenant_id,
|
||||
ClientProvider.company_id == company_id,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
if not selled_to_exists:
|
||||
errors.add_error(
|
||||
field="compliance_mx.sold_to_id",
|
||||
message="El Cliente no existe en el Catálogo de Clientes y Proveedores.",
|
||||
solution=["Verifica el ID del Cliente", "Revisa el catálogo"],
|
||||
code="NOT_FOUND",
|
||||
value=invoice.compliance_mx.sold_to_id,
|
||||
)
|
||||
|
||||
# Validar destinatario solo si se proporciona
|
||||
if invoice.compliance_mx.shipped_to_id:
|
||||
shipped_to_exists = (
|
||||
db.query(ClientProvider)
|
||||
.filter(
|
||||
ClientProvider.id == invoice.compliance_mx.shipped_to_id,
|
||||
ClientProvider.tenant_id == tenant_id,
|
||||
ClientProvider.company_id == company_id,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
if not shipped_to_exists:
|
||||
errors.add_error(
|
||||
field="compliance_mx.shipped_to_id",
|
||||
message="El Destinatario no existe en el Catálogo de Clientes y Proveedores.",
|
||||
solution=["Verifica el ID del Destinatario", "Revisa el catálogo"],
|
||||
code="NOT_FOUND",
|
||||
value=invoice.compliance_mx.shipped_to_id,
|
||||
)
|
||||
|
||||
# Validar agente aduanal solo si se proporciona
|
||||
if invoice.compliance_mx.customs_broker_id:
|
||||
customs_broker_exists = (
|
||||
db.query(CustomsBroker)
|
||||
.filter(
|
||||
CustomsBroker.id == invoice.compliance_mx.customs_broker_id,
|
||||
CustomsBroker.tenant_id == tenant_id,
|
||||
CustomsBroker.company_id == company_id,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
if not customs_broker_exists:
|
||||
errors.add_error(
|
||||
field="compliance_mx.customs_broker_id",
|
||||
message="El Agente Aduanal no existe en el Catálogo de Clientes y Proveedores.",
|
||||
solution=["Verifica el ID del Agente Aduanal", "Revisa el catálogo"],
|
||||
code="NOT_FOUND",
|
||||
value=invoice.compliance_mx.customs_broker_id,
|
||||
)
|
||||
|
||||
if invoice.logistics:
|
||||
if invoice.logistics.transport_num and not invoice.logistics.transport_num:
|
||||
# logic ...
|
||||
pass
|
||||
|
||||
if invoice.logistics.transport_type not in [t.value for t in TransportType]:
|
||||
errors.add_error(
|
||||
field="logistics.transport_type",
|
||||
message="El Tipo de Transporte proporcionado no es válido.",
|
||||
solution=[
|
||||
f"Selecciona un Tipo de Transporte válido: {[t.value for t in TransportType]}"
|
||||
],
|
||||
code="INVALID_TRANSPORT_TYPE",
|
||||
value=invoice.logistics.transport_type,
|
||||
)
|
||||
else:
|
||||
if (
|
||||
invoice.logistics.transport_type == "none"
|
||||
and invoice.logistics.transport_num
|
||||
):
|
||||
errors.add_error(
|
||||
field="logistics.transport_num",
|
||||
message="El Número de Transporte no debe proporcionarse cuando el Tipo de Transporte es 'none'.",
|
||||
solution=[
|
||||
"Elimina el Número de Transporte o selecciona un Tipo de Transporte válido"
|
||||
],
|
||||
code="INVALID_VALUE",
|
||||
value=invoice.logistics.transport_num,
|
||||
)
|
||||
else:
|
||||
if (
|
||||
not invoice.logistics.transport_num
|
||||
and invoice.logistics.transport_type != "none"
|
||||
):
|
||||
errors.add_error(
|
||||
field="logistics.transport_num",
|
||||
message="El Número de Transporte es obligatorio cuando se proporciona un Tipo de Transporte distinto de 'none'.",
|
||||
solution=["Proporciona un Número de Transporte válido"],
|
||||
code="REQUIRED_FIELD",
|
||||
value=invoice.logistics.transport_num,
|
||||
)
|
||||
|
||||
invoice.financials.currency = invoice.financials.currency or "foreign"
|
||||
|
||||
if invoice.financials.currency not in [c.value for c in Currency]:
|
||||
errors.add_error(
|
||||
field="financials.currency",
|
||||
message="La Moneda proporcionada no es válida.",
|
||||
solution=[f"Selecciona una Moneda válida: {[c.value for c in Currency]}"],
|
||||
code="INVALID_CURRENCY",
|
||||
value=invoice.financials.currency,
|
||||
)
|
||||
else:
|
||||
# Only check for existing items during update operations (when invoice has an id)
|
||||
if hasattr(invoice, "id"):
|
||||
has_items = (
|
||||
db.query(LineItem)
|
||||
.filter(
|
||||
LineItem.invoice_id == invoice.id,
|
||||
LineItem.tenant_id == tenant_id,
|
||||
LineItem.company_id == company_id,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
if has_items:
|
||||
errors.add_error(
|
||||
field="items",
|
||||
message=f"La opcion tipo de moneda {invoice.financials.currency} no puede ser modificada ya que la factura tiene items asociados.",
|
||||
solution=[
|
||||
"Verifica la moneda de los items asociados a la factura."
|
||||
],
|
||||
code="CURRENCY_CANNOT_BE_CHANGED",
|
||||
value=invoice.financials.currency,
|
||||
)
|
||||
|
||||
if invoice.financials.currency == "foreign":
|
||||
invoice.financials.currency_type = "USD"
|
||||
elif invoice.financials.currency == "local":
|
||||
invoice.financials.currency_type = "MXN"
|
||||
elif invoice.financials.currency == "manual":
|
||||
if not invoice.financials.currency_type:
|
||||
errors.add_error(
|
||||
field="financials.currency_type",
|
||||
message="El Tipo de Moneda es obligatorio cuando la Moneda es 'manual'.",
|
||||
solution=["Proporciona un Tipo de Moneda válido"],
|
||||
code="REQUIRED_FIELD",
|
||||
value=invoice.financials.currency_type,
|
||||
)
|
||||
else:
|
||||
currency_exists = (
|
||||
db.query(CurrencyType)
|
||||
.filter(CurrencyType.code == invoice.financials.currency_type)
|
||||
.first()
|
||||
)
|
||||
if not currency_exists:
|
||||
errors.add_error(
|
||||
field="financials.currency_type",
|
||||
message="El Tipo de Moneda no existe en el Catálogo de Tipos de Moneda.",
|
||||
solution=[
|
||||
"Verifica el código del Tipo de Moneda",
|
||||
"Revisa el catálogo",
|
||||
],
|
||||
code="NOT_FOUND",
|
||||
value=invoice.financials.currency_type,
|
||||
)
|
||||
|
||||
if invoice.logistics.incoterm:
|
||||
incoterm_exists = (
|
||||
db.query(Incoterm)
|
||||
.filter(
|
||||
Incoterm.code == invoice.logistics.incoterm
|
||||
)
|
||||
.first()
|
||||
)
|
||||
if not incoterm_exists:
|
||||
errors.add_error(
|
||||
field="logistics.incoterm",
|
||||
message="El Incoterm no existe en el Catálogo de Incoterms.",
|
||||
solution=["Verifica el código del Incoterm", "Revisa el catálogo"],
|
||||
code="NOT_FOUND",
|
||||
value=invoice.logistics.incoterm,
|
||||
)
|
||||
|
||||
if invoice.logistics.weight_type not in [w.value for w in WeightUnit]:
|
||||
errors.add_error(
|
||||
field="logistics.weight_type",
|
||||
message="La Unidad de Peso proporcionada no es válida.",
|
||||
solution=[
|
||||
f"Selecciona una Unidad de Peso válida: {[w.value for w in WeightUnit]}"
|
||||
],
|
||||
code="INVALID_WEIGHT_UNIT",
|
||||
value=invoice.logistics.weight_type,
|
||||
)
|
||||
|
||||
if invoice.compliance_mx.aduana:
|
||||
custom_section_exists = (
|
||||
db.query(CustomsSection)
|
||||
.filter(
|
||||
CustomsSection.customs_code == invoice.compliance_mx.aduana,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
if not custom_section_exists:
|
||||
errors.add_error(
|
||||
field="compliance_mx.aduana",
|
||||
message="La Aduana no existe en el Catálogo de Secciones Aduaneras.",
|
||||
solution=["Verifica el código de la Aduana", "Revisa el catálogo"],
|
||||
code="NOT_FOUND",
|
||||
value=invoice.compliance_mx.aduana,
|
||||
)
|
||||
@@ -3,7 +3,7 @@ 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 import validate_common, validate_required_fields_by_operation
|
||||
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 """
|
||||
|
||||
@@ -2,11 +2,12 @@ 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
|
||||
from .common import validate_required_fields_by_operation
|
||||
|
||||
|
||||
# Helper function para limpiar strings (equivalente a Clip())
|
||||
def clean_str(value: Optional[str]) -> Optional[str]:
|
||||
@@ -15,9 +16,12 @@ def clean_str(value: Optional[str]) -> Optional[str]:
|
||||
return value.strip()
|
||||
|
||||
def validate_update(
|
||||
invoice_data: InvoiceHeaderUpdate,
|
||||
db: Session,
|
||||
invoice: InvoiceHeaderUpdate,
|
||||
existing_invoice: InvoiceHeader,
|
||||
errors: ErrorCollector,
|
||||
tenant_id: int,
|
||||
company_id: int,
|
||||
errors: ErrorCollector,
|
||||
) -> None:
|
||||
"""
|
||||
Valida y procesa la actualización parcial de una factura de importación temporal.
|
||||
@@ -26,230 +30,230 @@ def validate_update(
|
||||
Si no, se mantiene el valor existente de la factura.
|
||||
|
||||
Args:
|
||||
invoice_data: Datos de la factura a validar/actualizar (modificado in-place)
|
||||
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_data in-place y acumula errores en errors)
|
||||
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_data.compliance_mx.provider_id if invoice_data.compliance_mx else (existing_invoice.compliance_mx.provider_id if existing_invoice.compliance_mx else None),
|
||||
'sold_to_id': invoice_data.compliance_mx.sold_to_id if invoice_data.compliance_mx else (existing_invoice.compliance_mx.sold_to_id if existing_invoice.compliance_mx else None),
|
||||
'sold_to_header': invoice_data.compliance_mx.sold_to_header if invoice_data.compliance_mx else (existing_invoice.compliance_mx.sold_to_header if existing_invoice.compliance_mx else None),
|
||||
'shipped_to_id': invoice_data.compliance_mx.shipped_to_id if invoice_data.compliance_mx else (existing_invoice.compliance_mx.shipped_to_id if existing_invoice.compliance_mx else None),
|
||||
'customs_broker_id': invoice_data.compliance_mx.customs_broker_id if invoice_data.compliance_mx else (existing_invoice.compliance_mx.customs_broker_id if existing_invoice.compliance_mx else None),
|
||||
'pedimento_id': invoice_data.compliance_mx.pedimento_id if invoice_data.compliance_mx else (existing_invoice.compliance_mx.pedimento_id if existing_invoice.compliance_mx else None),
|
||||
'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_data.operation_type or (existing_invoice.operation_type or 'imp'),
|
||||
invoice=invoice_dict,
|
||||
operation_type=invoice.operation_type or (existing_invoice.operation_type or 'imp'),
|
||||
errors=errors
|
||||
)
|
||||
|
||||
# Primero ejecutar validaciones comunes
|
||||
# validate_common(invoice_data, errors)
|
||||
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_data.compliance_mx:
|
||||
if invoice_data.compliance_mx.pedimento_id:
|
||||
invoice_data.compliance_mx.pedimento_id = invoice_data.compliance_mx.pedimento_id
|
||||
if invoice.compliance_mx:
|
||||
if invoice.compliance_mx.pedimento_id:
|
||||
invoice.compliance_mx.pedimento_id = invoice.compliance_mx.pedimento_id
|
||||
else:
|
||||
invoice_data.compliance_mx.pedimento_id = existing_invoice.compliance_mx.pedimento_id if existing_invoice.compliance_mx else None
|
||||
invoice.compliance_mx.pedimento_id = existing_invoice.compliance_mx.pedimento_id if existing_invoice.compliance_mx else None
|
||||
|
||||
# Columna B: Remesa
|
||||
if invoice_data.compliance_mx:
|
||||
if invoice_data.compliance_mx.remesa:
|
||||
invoice_data.compliance_mx.remesa = invoice_data.compliance_mx.remesa
|
||||
if invoice.compliance_mx:
|
||||
if invoice.compliance_mx.remesa:
|
||||
invoice.compliance_mx.remesa = invoice.compliance_mx.remesa
|
||||
else:
|
||||
invoice_data.compliance_mx.remesa = existing_invoice.compliance_mx.remesa if existing_invoice.compliance_mx else None
|
||||
invoice.compliance_mx.remesa = existing_invoice.compliance_mx.remesa if existing_invoice.compliance_mx else None
|
||||
|
||||
# Columna C: Factura (OBLIGATORIO)
|
||||
if invoice_data.invoice_number is not None:
|
||||
invoice_data.invoice_number = clean_str(invoice_data.invoice_number)
|
||||
if not invoice_data.invoice_number:
|
||||
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_data.invoice_number = existing_invoice.invoice_number
|
||||
invoice.invoice_number = existing_invoice.invoice_number
|
||||
|
||||
# Columna D: Fecha
|
||||
if not invoice_data.invoice_date:
|
||||
invoice_data.invoice_date = existing_invoice.invoice_date
|
||||
if not invoice.invoice_date:
|
||||
invoice.invoice_date = existing_invoice.invoice_date
|
||||
|
||||
# Columna E: Tipo Cambio
|
||||
if invoice_data.financials:
|
||||
if invoice_data.financials.exchange_rate is None:
|
||||
if invoice.financials:
|
||||
if invoice.financials.exchange_rate is None:
|
||||
if existing_invoice.financials:
|
||||
invoice_data.financials.exchange_rate = existing_invoice.financials.exchange_rate
|
||||
invoice.financials.exchange_rate = existing_invoice.financials.exchange_rate
|
||||
|
||||
# Columna F: Régimen
|
||||
if invoice_data.document_type:
|
||||
invoice_data.document_type = clean_str(invoice_data.document_type).upper()
|
||||
if invoice.document_type:
|
||||
invoice.document_type = clean_str(invoice.document_type).upper()
|
||||
else:
|
||||
invoice_data.document_type = existing_invoice.document_type
|
||||
invoice.document_type = existing_invoice.document_type
|
||||
|
||||
# Columna G: Clave Proveedor
|
||||
if invoice_data.compliance_mx:
|
||||
if invoice_data.compliance_mx.provider_id is None:
|
||||
invoice_data.compliance_mx.provider_id = existing_invoice.compliance_mx.provider_id if existing_invoice.compliance_mx else None
|
||||
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_data.compliance_mx:
|
||||
if invoice_data.compliance_mx.sold_to_id is None:
|
||||
invoice_data.compliance_mx.sold_to_id = existing_invoice.compliance_mx.sold_to_id if existing_invoice.compliance_mx else None
|
||||
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_data.compliance_mx:
|
||||
if invoice_data.compliance_mx.shipped_to_id is None:
|
||||
invoice_data.compliance_mx.shipped_to_id = existing_invoice.compliance_mx.shipped_to_id if existing_invoice.compliance_mx else None
|
||||
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_data.compliance_mx:
|
||||
if invoice_data.compliance_mx.customs_broker_id is None:
|
||||
invoice_data.compliance_mx.customs_broker_id = existing_invoice.compliance_mx.customs_broker_id if existing_invoice.compliance_mx else None
|
||||
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_data.logistics:
|
||||
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_data.logistics, 'carrier_id') and invoice_data.logistics.carrier_id is None:
|
||||
invoice_data.logistics.carrier_id = existing_invoice.logistics.carrier_id if existing_invoice.logistics else None
|
||||
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_data.logistics:
|
||||
if hasattr(invoice_data.logistics, 'driver_name') and not invoice_data.logistics.driver_name:
|
||||
invoice_data.logistics.driver_name = existing_invoice.logistics.driver_name if existing_invoice.logistics else None
|
||||
elif hasattr(invoice_data.logistics, 'driver_name'):
|
||||
invoice_data.logistics.driver_name = clean_str(invoice_data.logistics.driver_name)
|
||||
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_data.logistics:
|
||||
if hasattr(invoice_data.logistics, 'transport_type') and not invoice_data.logistics.transport_type:
|
||||
invoice_data.logistics.transport_type = existing_invoice.logistics.transport_type if existing_invoice.logistics else None
|
||||
elif hasattr(invoice_data.logistics, 'transport_type'):
|
||||
invoice_data.logistics.transport_type = clean_str(invoice_data.logistics.transport_type)
|
||||
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_data.logistics:
|
||||
if hasattr(invoice_data.logistics, 'transport_num') and not invoice_data.logistics.transport_num:
|
||||
invoice_data.logistics.transport_num = existing_invoice.logistics.transport_num if existing_invoice.logistics else None
|
||||
elif hasattr(invoice_data.logistics, 'transport_num'):
|
||||
invoice_data.logistics.transport_num = clean_str(invoice_data.logistics.transport_num)
|
||||
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_data.financials:
|
||||
if not invoice_data.financials.currency:
|
||||
invoice_data.financials.currency = existing_invoice.financials.currency if existing_invoice.financials else None
|
||||
if invoice.financials:
|
||||
if not invoice.financials.currency:
|
||||
invoice.financials.currency = existing_invoice.financials.currency if existing_invoice.financials else None
|
||||
else:
|
||||
invoice_data.financials.currency = clean_str(invoice_data.financials.currency).lower()
|
||||
invoice.financials.currency = clean_str(invoice.financials.currency).lower()
|
||||
|
||||
# Columna P: Clave Moneda
|
||||
if invoice_data.financials:
|
||||
if not invoice_data.financials.currency_type:
|
||||
invoice_data.financials.currency_type = existing_invoice.financials.currency_type if existing_invoice.financials else None
|
||||
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_data.financials.currency_type = clean_str(invoice_data.financials.currency_type).upper()
|
||||
invoice.financials.currency_type = clean_str(invoice.financials.currency_type).upper()
|
||||
|
||||
# Columna Q: Flete
|
||||
if invoice_data.financials:
|
||||
if invoice_data.financials.freight is None:
|
||||
invoice_data.financials.freight = existing_invoice.financials.freight if existing_invoice.financials else None
|
||||
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_data.financials:
|
||||
if invoice_data.financials.insurance_value is None:
|
||||
invoice_data.financials.insurance_value = existing_invoice.financials.insurance_value if existing_invoice.financials else None
|
||||
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_data.financials:
|
||||
if invoice_data.financials.insurance is None:
|
||||
invoice_data.financials.insurance = existing_invoice.financials.insurance if existing_invoice.financials else None
|
||||
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_data.financials:
|
||||
if invoice_data.financials.packaging is None:
|
||||
invoice_data.financials.packaging = existing_invoice.financials.packaging if existing_invoice.financials else None
|
||||
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_data.financials:
|
||||
if invoice_data.financials.other_increments is None:
|
||||
invoice_data.financials.other_increments = existing_invoice.financials.other_increments if existing_invoice.financials else None
|
||||
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_data.logistics:
|
||||
if hasattr(invoice_data.logistics, 'incoterm') and not invoice_data.logistics.incoterm:
|
||||
invoice_data.logistics.incoterm = existing_invoice.logistics.incoterm if existing_invoice.logistics else None
|
||||
elif hasattr(invoice_data.logistics, 'incoterm'):
|
||||
invoice_data.logistics.incoterm = clean_str(invoice_data.logistics.incoterm).upper()
|
||||
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_data.logistics:
|
||||
if hasattr(invoice_data.logistics, 'seal_number') and not invoice_data.logistics.seal_number:
|
||||
invoice_data.logistics.seal_number = existing_invoice.logistics.seal_number if existing_invoice.logistics else None
|
||||
elif hasattr(invoice_data.logistics, 'seal_number'):
|
||||
invoice_data.logistics.seal_number = clean_str(invoice_data.logistics.seal_number)
|
||||
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_data.emission_date:
|
||||
invoice_data.emission_date = existing_invoice.emission_date
|
||||
if not invoice.emission_date:
|
||||
invoice.emission_date = existing_invoice.emission_date
|
||||
|
||||
# Columna Y: Tipo de Peso (Opcional)
|
||||
if invoice_data.logistics:
|
||||
if hasattr(invoice_data.logistics, 'weight_type') and not invoice_data.logistics.weight_type:
|
||||
invoice_data.logistics.weight_type = existing_invoice.logistics.weight_type if existing_invoice.logistics else None
|
||||
elif hasattr(invoice_data.logistics, 'weight_type'):
|
||||
invoice_data.logistics.weight_type = clean_str(invoice_data.logistics.weight_type).upper()
|
||||
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: E-Document (Opcional)
|
||||
if invoice_data.compliance_mx:
|
||||
if not invoice_data.compliance_mx.edocument:
|
||||
invoice_data.compliance_mx.edocument = existing_invoice.compliance_mx.edocument if existing_invoice.compliance_mx else None
|
||||
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_data.compliance_mx.edocument = clean_str(invoice_data.compliance_mx.edocument)
|
||||
invoice.compliance_mx.edocument = clean_str(invoice.compliance_mx.edocument)
|
||||
|
||||
# Columna AA: Num. Operación (Opcional)
|
||||
if invoice_data.compliance_mx:
|
||||
if not invoice_data.compliance_mx.vucem_operation_num:
|
||||
invoice_data.compliance_mx.vucem_operation_num = existing_invoice.compliance_mx.vucem_operation_num if existing_invoice.compliance_mx else None
|
||||
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_data.compliance_mx.vucem_operation_num = clean_str(invoice_data.compliance_mx.vucem_operation_num)
|
||||
invoice.compliance_mx.vucem_operation_num = clean_str(invoice.compliance_mx.vucem_operation_num)
|
||||
|
||||
# Columna AB: Aduana (OBLIGATORIO)
|
||||
if invoice_data.compliance_mx:
|
||||
if not invoice_data.compliance_mx.aduana:
|
||||
invoice_data.compliance_mx.aduana = existing_invoice.compliance_mx.aduana if existing_invoice.compliance_mx else None
|
||||
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_data.compliance_mx.aduana = clean_str(invoice_data.compliance_mx.aduana)
|
||||
invoice.compliance_mx.aduana = clean_str(invoice.compliance_mx.aduana)
|
||||
|
||||
# Validar que aduana sea obligatorio (excepto para MEX)
|
||||
if existing_invoice.invoice_type != "MEX":
|
||||
current_aduana = invoice_data.compliance_mx.aduana if invoice_data.compliance_mx else (existing_invoice.compliance_mx.aduana if existing_invoice.compliance_mx else None)
|
||||
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_data.compliance_mx:
|
||||
if not invoice_data.compliance_mx.port_of_entry:
|
||||
invoice_data.compliance_mx.port_of_entry = existing_invoice.compliance_mx.port_of_entry if existing_invoice.compliance_mx else None
|
||||
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_data.compliance_mx.port_of_entry = clean_str(invoice_data.compliance_mx.port_of_entry)
|
||||
invoice.compliance_mx.port_of_entry = clean_str(invoice.compliance_mx.port_of_entry)
|
||||
|
||||
# Columna AD: Observación en Español (Opcional)
|
||||
if not invoice_data.observation_es:
|
||||
invoice_data.observation_es = existing_invoice.observation_es
|
||||
if not invoice.observation_es:
|
||||
invoice.observation_es = existing_invoice.observation_es
|
||||
else:
|
||||
invoice_data.observation_es = clean_str(invoice_data.observation_es)
|
||||
invoice.observation_es = clean_str(invoice.observation_es)
|
||||
|
||||
# Columna AD: Observación en Inglés (Opcional)
|
||||
if not invoice_data.observation_en:
|
||||
invoice_data.observation_en = existing_invoice.observation_en
|
||||
if not invoice.observation_en:
|
||||
invoice.observation_en = existing_invoice.observation_en
|
||||
else:
|
||||
invoice_data.observation_en = clean_str(invoice_data.observation_en)
|
||||
invoice.observation_en = clean_str(invoice.observation_en)
|
||||
|
||||
|
||||
@@ -5,8 +5,10 @@ from sqlalchemy import func
|
||||
from core.exceptions import ErrorCollector, DuplicateResourceException
|
||||
from core.context import get_user_context
|
||||
from .common.mappers import clean_dict
|
||||
from .imports.validators.create import validate_create
|
||||
from .imports.validators.update import validate_update
|
||||
from .imports.validators.create import validate_create as validate_create_import
|
||||
from .imports.validators.update import validate_update as validate_update_import
|
||||
from .exports.validators.create import validate_create as validate_create_export
|
||||
from .exports.validators.update import validate_update as validate_update_export
|
||||
from .common.common_validators import invoice_exists
|
||||
|
||||
from . import models, schemas
|
||||
@@ -124,7 +126,7 @@ class InvoiceService:
|
||||
|
||||
# Validar si la factura ya existe
|
||||
invoice_exists(db, invoice_data.invoice_number, tenant_id, company_id, errors)
|
||||
validate_create(db, invoice_data, tenant_id, company_id, errors)
|
||||
validate_create_import(db, invoice_data, tenant_id, company_id, errors)
|
||||
|
||||
# Si hay errores, lanzar excepción ANTES de intentar crear
|
||||
errors.raise_if_errors("Error al crear la factura")
|
||||
@@ -273,7 +275,11 @@ class InvoiceService:
|
||||
invoice_data.invoice_number,
|
||||
f"Ya existe otra factura con el número '{invoice_data.invoice_number}'",
|
||||
)
|
||||
validate_update(invoice_data, invoice, errors)
|
||||
|
||||
if invoice_data.operation_type == "exp":
|
||||
validate_update_export(invoice_data, invoice, errors)
|
||||
else:
|
||||
validate_update_import(invoice_data, invoice, errors)
|
||||
|
||||
# Si hay errores, lanzar excepción ANTES de actualizar
|
||||
errors.raise_if_errors("Error al actualizar la factura")
|
||||
|
||||
Reference in New Issue
Block a user