Merge pull request 'Enhance invoice validation logic to support existing invoices' (#255) from fix/currency-invoices into development

Reviewed-on: ADUANASOFT/anexo76#255
This commit is contained in:
2026-03-24 23:53:30 +00:00
3 changed files with 99 additions and 60 deletions

View File

@@ -1,4 +1,4 @@
from typing import Optional from typing import Any, Dict, Optional, Union
from core.exceptions import ErrorCollector from core.exceptions import ErrorCollector
from sqlalchemy import func from sqlalchemy import func
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
@@ -17,8 +17,14 @@ from api.v1.modules.a76.manifests.manifest.models import Manifest
from api.v1.modules.public.reference_data.incoterms.models import Incoterm 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.currency_types.models import CurrencyType
from api.v1.modules.public.reference_data.customs_sections.models import CustomsSection from api.v1.modules.public.reference_data.customs_sections.models import CustomsSection
from core.exceptions import ErrorCollector
from typing import Dict, Any
def _normalize_invoice_currency_value(value) -> str:
"""Lowercase currency code (foreign/local/manual) for comparisons."""
if value is None or value == "":
return ""
raw = getattr(value, "value", value)
return str(raw).lower()
def invoice_exists( def invoice_exists(
@@ -174,10 +180,11 @@ def validate_required_fields_by_operation(
def validate_common( def validate_common(
db: Session, db: Session,
invoice: schemas.InvoiceHeaderUpdate, invoice: Union[schemas.InvoiceHeaderCreate, schemas.InvoiceHeaderUpdate],
tenant_id: int, tenant_id: int,
company_id: int, company_id: int,
errors: ErrorCollector, errors: ErrorCollector,
existing_invoice: Optional[models.InvoiceHeader] = None,
): ):
if invoice.compliance_mx.pedimento_id: if invoice.compliance_mx.pedimento_id:
pedimento = ( pedimento = (
@@ -560,69 +567,97 @@ def validate_common(
value=invoice.logistics.transport_num, value=invoice.logistics.transport_num,
) )
invoice.financials.currency = invoice.financials.currency or "foreign" if invoice.financials:
submitted = invoice.financials.currency
if invoice.financials.currency not in [c.value for c in Currency]: stored_str = ""
errors.add_error( if existing_invoice and existing_invoice.financials is not None:
field="financials.currency", stored_str = _normalize_invoice_currency_value(
message="La Moneda proporcionada no es válida.", existing_invoice.financials.currency
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": if submitted is None or submitted == "":
invoice.financials.currency_type = "USD" if stored_str:
elif invoice.financials.currency == "local": invoice.financials.currency = stored_str
invoice.financials.currency_type = "MXN" else:
elif invoice.financials.currency == "manual": invoice.financials.currency = "foreign"
if not invoice.financials.currency_type: else:
invoice.financials.currency = (
_normalize_invoice_currency_value(submitted) or "foreign"
)
if invoice.financials.currency not in [c.value for c in Currency]:
errors.add_error( errors.add_error(
field="financials.currency_type", field="financials.currency",
message="El Tipo de Moneda es obligatorio cuando la Moneda es 'manual'.", message="La Moneda proporcionada no es válida.",
solution=["Proporciona un Tipo de Moneda válido"], solution=[f"Selecciona una Moneda válida: {[c.value for c in Currency]}"],
code="REQUIRED_FIELD", code="INVALID_CURRENCY",
value=invoice.financials.currency_type, value=invoice.financials.currency,
) )
else: else:
currency_exists = ( invoice_id = getattr(invoice, "id", None)
db.query(CurrencyType) if (
.filter(CurrencyType.code == invoice.financials.currency_type) invoice_id is not None
.first() and existing_invoice is not None
) and existing_invoice.financials is not None
if not currency_exists: ):
old_c = _normalize_invoice_currency_value(
existing_invoice.financials.currency
)
new_c = invoice.financials.currency
if old_c != new_c:
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=(
"La opción tipo de moneda no puede ser modificada "
"ya que la factura tiene partidas asociadas."
),
solution=[
"Verifica la moneda de las partidas asociadas 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( errors.add_error(
field="financials.currency_type", field="financials.currency_type",
message="El Tipo de Moneda no existe en el Catálogo de Tipos de Moneda.", message="El Tipo de Moneda es obligatorio cuando la Moneda es 'manual'.",
solution=[ solution=["Proporciona un Tipo de Moneda válido"],
"Verifica el código del Tipo de Moneda", code="REQUIRED_FIELD",
"Revisa el catálogo",
],
code="NOT_FOUND",
value=invoice.financials.currency_type, 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: if invoice.logistics.incoterm:
incoterm_exists = ( incoterm_exists = (

View File

@@ -55,7 +55,9 @@ def validate_update(
) )
# Primero ejecutar validaciones comunes # Primero ejecutar validaciones comunes
validate_common(db, invoice, tenant_id, company_id, errors) validate_common(
db, invoice, tenant_id, company_id, errors, existing_invoice=existing_invoice
)
# Mapeo de columnas CSV a campos de la factura # Mapeo de columnas CSV a campos de la factura
# Siguiendo la lógica del código Clarion original # Siguiendo la lógica del código Clarion original

View File

@@ -55,7 +55,9 @@ def validate_update(
) )
# Primero ejecutar validaciones comunes # Primero ejecutar validaciones comunes
validate_common(db, invoice, tenant_id, company_id, errors) validate_common(
db, invoice, tenant_id, company_id, errors, existing_invoice=existing_invoice
)
# Mapeo de columnas CSV a campos de la factura # Mapeo de columnas CSV a campos de la factura
# Siguiendo la lógica del código Clarion original # Siguiendo la lógica del código Clarion original