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

feat: enhance invoice validation logic and add checks for remesa, exchange rate, and transport details

Browse Source
This commit is contained in:
AlexeerCT
2026-01-07 18:01:11 -06:00
parent 182f472370
commit 09bee94ae6
1 changed files with 330 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

354
backend/api/v1/modules/a76/invoices/imports/temporary/validators/common.py
View File

@@ -1,24 +1,40 @@
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.public.reference_data.incoterms.models import Incoterm
from api.v1.modules.a76.items.models import Item
from ....models import TransportType, Currency, WeightUnit
from core.exceptions import ErrorCollector
def validate_common(db: Session, invoice: schemas.InvoiceHeaderCreate, tenant_id: int, company_id: int, errors: ErrorCollector):
def validate_common(
db: Session,
invoice: schemas.InvoiceHeaderCreate,
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:
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
)
value=invoice.compliance_mx.pedimento_id,
)
if not invoice.compliance_mx.is_regime_change:
if not pedimento.operation_type == 1:
errors.add_error(
@@ -26,7 +42,7 @@ def validate_common(db: Session, invoice: schemas.InvoiceHeaderCreate, tenant_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
value=pedimento.operation_type,
)
else:
if pedimento.regime in ["EXD", "ETE", "ETR"]:
@@ -35,25 +51,27 @@ def validate_common(db: Session, invoice: schemas.InvoiceHeaderCreate, tenant_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
value=pedimento.regime,
)
else:
else:
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"],
solution=[
"Ajusta el Tipo de Documento o selecciona otro Pedimento"
],
code="REGIME_MISMATCH",
value=invoice.document_type
value=invoice.document_type,
)
else:
if pedimento.operation_type != 2:
else:
if pedimento.operation_type != 2:
errors.add_error(
field="compliance_mx.pedimento_id",
message="El Pedimento seleccionado no corresponde a una Importacion Definitiva.",
solution=["Selecciona un Pedimento de Importacion Definitiva"],
code="INVALID_OPERATION_TYPE",
value=pedimento.operation_type
value=pedimento.operation_type,
)
else:
if pedimento.regime != "IMD":
@@ -62,16 +80,18 @@ def validate_common(db: Session, invoice: schemas.InvoiceHeaderCreate, tenant_id
message=f"El Pedimento {pedimento.customs_office}-{pedimento.license}-{pedimento.pedimento_number} no corresponde a una Importacion Definitiva.",
solution=["Selecciona un Pedimento de Importacion Definitiva"],
code="INVALID_REGIME",
value=pedimento.regime
value=pedimento.regime,
)
else:
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"],
solution=[
"Ajusta el Tipo de Documento o selecciona otro Pedimento"
],
code="REGIME_MISMATCH",
value=invoice.document_type
value=invoice.document_type,
)
else:
if pedimento.pedimento_code not in ["A1", "A3"]:
@@ -80,8 +100,294 @@ def validate_common(db: Session, invoice: schemas.InvoiceHeaderCreate, tenant_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
value=pedimento.pedimento_code,
)
if pedimento.pedimento_type == "consolidated":
if (
invoice.invoice_date < pedimento.pedimento_dates.entry_date
or invoice.invoice_date > pedimento.pedimento_dates.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,
)
if not invoice.compliance_mx.remesa:
errors.add_error(
field="compliance_mx.remesa",
message="El campo Remesa es obligatorio cuando se asocia un Pedimento.",
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(Pedimentos)
.filter(
Pedimentos.remesa == invoice.compliance_mx.remesa,
Pedimentos.id != invoice.compliance_mx.pedimento_id,
Pedimentos.tenant_id == tenant_id,
Pedimentos.company_id == company_id,
)
.first()
)
if duplicated_remesa:
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,
)
else:
if invoice.compliance_mx.remesa and not invoice.compliance_mx.pedimento_id:
errors.add_error(
field="compliance_mx.pedimento_id",
message="El campo Pedimento es obligatorio cuando se proporciona Remesa.",
solution=["Proporciona un ID de Pedimento"],
code="REQUIRED_FIELD",
value=invoice.compliance_mx.pedimento_id,
)
if len(invoice.invoice_number) > 100:
errors.add_error(
field="invoice_number",
message="El número de factura excede la longitud máxima de 100 caracteres.",
solution=["Acorta el número de factura a 100 caracteres o menos"],
code="MAX_LENGTH_EXCEEDED",
value=invoice.invoice_number,
)
if not invoice.financials.exchange_rate or invoice.financials.exchange_rate <= 0:
exchange_rate_exists = (
db.query(ExchangeRate)
.filter(
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.date()}.",
solution=["Registra el Tipo de Cambio en el catálogo correspondiente"],
code="EXCHANGE_RATE_NOT_FOUND",
value=invoice.financials.exchange_rate,
)
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,
)
provider_exists = (
db.query(ClientProvider)
.filter(
ClientProvider.id == invoice.provider_id,
ClientProvider.tenant_id == tenant_id,
ClientProvider.company_id == company_id,
)
.first()
)
if not provider_exists:
errors.add_error(
field="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.provider_id,
)
selled_to_exists = (
db.query(ClientProvider)
.filter(
ClientProvider.id == invoice.selled_to_id,
ClientProvider.tenant_id == tenant_id,
ClientProvider.company_id == company_id,
)
.first()
)
if not selled_to_exists:
errors.add_error(
field="selled_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.selled_to_id,
)
shipped_to_exists = (
db.query(ClientProvider)
.filter(
ClientProvider.id == invoice.shipped_to_id,
ClientProvider.tenant_id == tenant_id,
ClientProvider.company_id == company_id,
)
.first()
)
if not shipped_to_exists:
errors.add_error(
field="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.shipped_to_id,
)
customs_broker_exists = (
db.query(ClientProvider)
.filter(
ClientProvider.id == invoice.customs_broker_id,
ClientProvider.tenant_id == tenant_id,
ClientProvider.company_id == company_id,
)
.first()
)
if not customs_broker_exists:
errors.add_error(
field="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.customs_broker_id,
)
if invoice.logistics.carrier_id:
carrier_exists = (
db.query(ClientProvider)
.filter(
ClientProvider.id == invoice.logistics.carrier_id,
ClientProvider.tenant_id == tenant_id,
ClientProvider.company_id == company_id,
)
.first()
)
if not carrier_exists:
errors.add_error(
field="logistics.carrier_id",
message="El Transportista no existe en el Catálogo de Clientes y Proveedores.",
solution=["Verifica el ID del Transportista", "Revisa el catálogo"],
code="NOT_FOUND",
value=invoice.logistics.carrier_id,
)
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 pedimento.pedimento_type == "normal":
pass
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:
has_items = db.query(Item).filter(
Item.invoice_id == invoice.id,
Item.tenant_id == tenant_id,
Item.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.logistics.incoterms:
incoterm_exists = (
db.query(Incoterm)
.filter(
Incoterm.code == invoice.logistics.incoterms,
Incoterm.tenant_id == tenant_id,
Incoterm.company_id == company_id,
)
.first()
)
if not incoterm_exists:
errors.add_error(
field="logistics.incoterms",
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.incoterms,
)
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,
)