Fomrulario de factrua de exportacion implementada

This commit is contained in:
2026-02-10 10:59:11 -06:00
parent 0f50ae3d61
commit ae66879120
14 changed files with 972 additions and 534 deletions

View File

@@ -109,18 +109,30 @@ def validate_common(
code="NOT_FOUND",
value=invoice.compliance_mx.pedimento_id,
)
return # Stop here if pedimento not found
if not invoice.compliance_mx.is_regime_change:
if not pedimento.operation_type == "imp":
# Validar que el tipo de operación del pedimento coincida con el de la factura
# Nota: Para UPDATES, operation_type puede ser None. En ese caso usamos el del pedimento (asumimos que es correcto)
# o podríamos buscarlo en existing_invoice si tuviéramos acceso aquí.
current_operation = invoice.operation_type or pedimento.operation_type
if pedimento.operation_type != current_operation:
operation_labels = {"imp": "Importación", "exp": "Exportación"}
expected_label = operation_labels.get(current_operation, current_operation)
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"],
message=f"El Pedimento seleccionado no corresponde a una {expected_label}.",
solution=[f"Selecciona un Pedimento de {expected_label}"],
code="INVALID_OPERATION_TYPE",
value=pedimento.operation_type,
)
else:
if pedimento.regime in ["EXD", "ETE", "ETR"]:
# Validar regímenes incompatibles
export_only_regimes = ["EXD", "ETE", "ETR"]
import_only_regimes = ["IMD"] # Usualmente IMD es solo importación definitiva
if current_operation == "imp" and 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.",
@@ -128,7 +140,18 @@ def validate_common(
code="INVALID_REGIME",
value=pedimento.regime,
)
else:
elif current_operation == "exp" and pedimento.regime in import_only_regimes:
errors.add_error(
field="compliance_mx.pedimento_id",
message="El Pedimento seleccionado corresponde a una Importación, no a una Exportación.",
solution=["Selecciona un Pedimento de Exportació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",
@@ -140,43 +163,42 @@ def validate_common(
value=invoice.document_type,
)
else:
if pedimento.operation_type != 2:
# 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 Importacion Definitiva.",
solution=["Selecciona un Pedimento de Importacion Definitiva"],
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":
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 {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,
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,
)
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"
],
code="REGIME_MISMATCH",
value=invoice.document_type,
)
else:
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
@@ -207,22 +229,24 @@ def validate_common(
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,
)
# 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)
@@ -242,57 +266,32 @@ def validate_common(
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 not invoice.compliance_mx.is_pedimento_pending:
errors.add_error(
field="compliance_mx.pedimento_id",
message="El campo Pedimento es obligatorio cuando no se indica que el Pedimento está pendiente.",
solution=[
"Proporciona un ID de Pedimento",
"Marca el campo Pedimento Pendiente",
],
code="REQUIRED_FIELD",
value=invoice.compliance_mx.pedimento_id,
)
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(
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
# 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"]:
@@ -395,62 +394,47 @@ def validate_common(
value=invoice.compliance_mx.customs_broker_id,
)
# Validar transportista solo si se proporciona
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:
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.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'.",
field="logistics.transport_type",
message="El Tipo de Transporte proporcionado no es válido.",
solution=[
"Elimina el Número de Transporte o selecciona un Tipo de Transporte válido"
f"Selecciona un Tipo de Transporte válido: {[t.value for t in TransportType]}"
],
code="INVALID_VALUE",
value=invoice.logistics.transport_num,
code="INVALID_TRANSPORT_TYPE",
value=invoice.logistics.transport_type,
)
else:
if (
not invoice.logistics.transport_num
and invoice.logistics.transport_type != "none"
invoice.logistics.transport_type == "none"
and invoice.logistics.transport_num
):
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",
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"

View File

@@ -54,34 +54,38 @@ def validate_create(db: Session, invoice: InvoiceHeaderCreate, tenant_id: int, c
"""Se retorna por que fallaron las validaciones generales"""
return
if not invoice.compliance_mx.pedimento_id:
invoice.compliance_mx.remesa = None
if not invoice.financials.exchange_rate:
invoice.financials.exchange_rate = db.query(ExchangeRate.value).filter(ExchangeRate.date == invoice.invoice_date).scalar()
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 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
if not invoice.financials.currency:
invoice.financials.currency = "foreign"
if invoice.financials.currency == "local":
invoice.financials.currency_type = "MXN"
elif invoice.financials.currency_type == "foreign":
invoice.financials.currency = "USD"
elif invoice.financials.currency_type == "manual":
invoice.financials.currency_type = invoice.financials.currency_type.upper()
invoice.logistics.incoterm = (invoice.logistics.incoterm or "").upper()
if not invoice.logistics.weight_type:
invoice.logistics.weight_type = "kgs"
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()

View File

@@ -488,6 +488,9 @@ class InvoiceFinancials(Base, TenantScopedMixin, TimestampMixin):
other_increments: Mapped[Optional[float]] = mapped_column(
Numeric(19, 8), default=0, server_default="0"
) # OTROSINCREMENTA / Otros incrementables
other_deductibles: Mapped[Optional[float]] = mapped_column(
Numeric(19, 8), default=0, server_default="0"
) # OTROSDEDUCIBLES / Otros deducibles
total_increments_mn: Mapped[Optional[float]] = mapped_column(
Numeric(23, 8), default=0, server_default="0"
) # TOTALINCREMMN / Total incrementables MN

View File

@@ -1,7 +1,7 @@
from typing import Literal, Optional, List
from datetime import datetime, date
from decimal import Decimal
from pydantic import BaseModel, Field
from pydantic import BaseModel, Field, field_validator
from .models import (
DestinationOriginCove,
OperationType,
@@ -119,7 +119,7 @@ class InvoiceComplianceMxBase(BaseModel):
sold_to_id: Optional[int] = Field(None, description="Sold to ID")
shipped_to_header: Optional[str] = Field(None, max_length=20, description="Shipped to header")
shipped_to_id: Optional[int] = Field(None, description="Shipped to ID")
shipped_by_header: Optional[int] = Field(
shipped_by_header: Optional[str] = Field(
None, max_length=20, description="Shipped by header"
)
shipped_by_id: Optional[int] = Field(None, description="Shipped by ID")
@@ -248,6 +248,7 @@ class InvoiceFinancialsBase(BaseModel):
insurance_value: Optional[Decimal] = Field(None, description="Insurance value")
packaging: Optional[Decimal] = Field(None, description="Packaging")
other_increments: Optional[Decimal] = Field(None, description="Other increments")
other_deductibles: Optional[Decimal] = Field(None, description="Other deductibles")
total_increments_mn: Optional[Decimal] = Field(
None, description="Total increments in MXN"
)
@@ -318,6 +319,13 @@ class InvoiceLogisticsBase(BaseModel):
weight_type: WeightUnit = Field(
default="kgs", max_length=3, description="Weight type"
)
@field_validator("weight_type", mode="before")
@classmethod
def normalize_weight_type(cls, v):
if isinstance(v, str):
return v.lower()
return v
container_types: Optional[str] = Field(
None, max_length=500, description="Container types"
)

View File

@@ -301,6 +301,23 @@ class InvoiceService:
new_financials = models.InvoiceFinancials(**financials_dict)
db.add(new_financials)
# Update logistics if provided
if invoice_data.logistics is not None:
if invoice.logistics:
for key, value in invoice_data.logistics.model_dump(
exclude_unset=True
).items():
if value == "":
value = None
setattr(invoice.logistics, key, value)
else:
logistics_dict = invoice_data.logistics.model_dump()
logistics_dict["invoice_id"] = invoice.id
logistics_dict["tenant_id"] = tenant_id
logistics_dict["company_id"] = company_id
new_logistics = models.InvoiceLogistics(**logistics_dict)
db.add(new_logistics)
db.commit()
db.refresh(invoice)
return invoice