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

feat: enhance invoice and item validation by adding new checks and updating schemas

Browse Source
This commit is contained in:
acazares
2026-02-12 17:00:12 -06:00
parent f8c38d998d
commit c011d7ad65
9 changed files with 637 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
backend/api/v1/modules/a76/invoices/common/common_validators.py
View File

@@ -31,10 +31,37 @@ def invoice_exists(
return invoice
return None
def invoice_exists_by_id(
db: Session,
invoice_id: str,
tenant_id: int,
company_id: int,
errors: Optional[ErrorCollector],
):
invoice = (
db.query(models.InvoiceHeader)
.filter(
models.InvoiceHeader.id == invoice_id,
models.InvoiceHeader.tenant_id == tenant_id,
models.InvoiceHeader.company_id == company_id,
)
.first()
)
if invoice:
if errors:
errors.add_duplicate_error(
"invoice_id",
invoice_id,
f"Ya existe una factura con el número '{invoice_id}'",
)
return invoice
return None
def invoice_updated(
db: Session,
invoice_number: str,
invoice_id: str,
tenant_id: int,
company_id: int,
errors: ErrorCollector,
@@ -42,7 +69,7 @@ def invoice_updated(
is_updated = (
db.query(models.InvoiceHeader.is_updated)
.filter(
models.InvoiceHeader.invoice_number == invoice_number,
models.InvoiceHeader.id == invoice_id,
models.InvoiceHeader.tenant_id == tenant_id,
models.InvoiceHeader.company_id == company_id,
)
@@ -52,10 +79,10 @@ def invoice_updated(
if is_updated:
errors.add_error(
field="invoice_number",
message=f"La factura con el número '{invoice_number}' ya ha sido actualizada y no se puede modificar.",
message=f"La factura con el número '{invoice_id}' ya ha sido actualizada y no se puede modificar.",
solution="Capturar otro número de Factura de Importación Temporal o Desactualizar la factura.",
code="INVOICE_UPDATED",
value=invoice_number,
value=invoice_id,
)
return True
return False

4
backend/api/v1/modules/a76/items/common/common_validators.py
View File

@@ -5,7 +5,6 @@ from sqlalchemy.orm import Session
def item_exists(
db: Session,
invoice_id: int,
item_line: int,
tenant_id: int,
company_id: int
@@ -13,8 +12,7 @@ def item_exists(
item_exists = (
db.query(models.LineItem.id)
.filter(
models.LineItem.invoice_id == invoice_id,
models.LineItem.LineItem == item_line,
models.LineItem.line_number == item_line,
models.LineItem.tenant_id == tenant_id,
models.LineItem.company_id == company_id,
)

95
backend/api/v1/modules/a76/items/imports/temporary/validators/common.py
View File

@@ -1,10 +1,6 @@
import logging
from typing import Optional, List, Tuple
from fastapi import HTTPException
from sqlalchemy import and_, exists, or_
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session, joinedload
from api.v1.modules.a76.invoices.common.common_validators import invoice_exists
from sqlalchemy import exists
from sqlalchemy.orm import Session
from api.v1.modules.a76.invoices.common.common_validators import invoice_exists_by_id
from core.exceptions import ErrorCollector
from sqlalchemy import func
@@ -12,12 +8,7 @@ from ....common.fractions import search_fraction_preference
from ....common.common_validators import item_exists
from ....models import Item
from ....line_items.models import LineItem
from ....line_financials.models import LineFinancial
from ....line_quantities.models import LineQuantity
from ....line_customs.models import FractionType, LineCustom
from ....line_descriptions.models import LineDescription
from ....line_references.models import LineReference
from api.v1.modules.a24.fa.fa_item_lines.models import FaLineItem
from api.v1.modules.a76.items.schemas import LineItemCreate
from api.v1.modules.a76.invoices.models import InvoiceHeader
from api.v1.modules.a76.classes.models import Class
@@ -25,25 +16,34 @@ from api.v1.modules.a76.general_catalogs.units_of_measure.models import UnitOfMe
from api.v1.modules.a76.general_catalogs.packages.models import Package
from api.v1.modules.public.reference_data.countries.models import Country
from api.v1.modules.public.reference_data.sectors.models import Sector
from api.v1.modules.public.reference_data.valuation_methods.models import ValuationMethod
from api.v1.modules.public.reference_data.valuation_methods.models import (
ValuationMethod,
)
from api.v1.modules.a76.parts.models import Part
from api.v1.modules.a76.general_catalogs.company.models import Company
def validate_common(
db: Session,
line: LineItemCreate,
invoice_id: int, # Para creación, se pasa directamente; para update, se consulta del item
tenant_id: int,
company_id: int,
errors: ErrorCollector,
line_number: int,
):
item_header = db.query(Item).filter(Item.id == line.item_id).first()
invoice: InvoiceHeader = invoice_exists(
db, item_header.invoice_id, tenant_id, company_id, errors
# Para updates, line.item_id existe; para creates, es None
item_header = None
if line.item_id:
item_header = db.query(Item).filter(Item.id == line.item_id).first()
if item_header:
invoice_id = item_header.invoice_id
invoice: InvoiceHeader = invoice_exists_by_id(
db, invoice_id, tenant_id, company_id, errors
)
line_item: LineItem = item_exists(
db, item_header.invoice_id, line.line_number, tenant_id, company_id
db, line.line_number, tenant_id, company_id
)
fecha_factura = invoice.invoice_date if invoice else None
@@ -118,9 +118,9 @@ def validate_common(
if line.unit_of_measure:
um = (
db.query(func.count(UnitOfMeasure))
db.query(func.count(UnitOfMeasure.id))
.filter(
UnitOfMeasure.code == line.unit_of_measure,
UnitOfMeasure.id == line.unit_of_measure,
UnitOfMeasure.tenant_id == tenant_id,
UnitOfMeasure.company_id == company_id,
)
@@ -136,13 +136,13 @@ def validate_common(
if line.quantity.package_id:
package = (
db.query(func.count(Package))
db.query(func.count(Package.id))
.filter(
Package.id == line.quantity.package_id,
Package.tenant_id == tenant_id,
Package.company_id == company_id,
)
.first()
.scalar()
)
if package == 0:
errors.add_error(
@@ -247,11 +247,7 @@ def validate_common(
)
elif fraction_type.strip().upper() == FractionType.PROSEC and sector:
sector_db: Sector = (
db.query(Sector)
.filter(
Sector.key == sector
)
.scalar()
db.query(Sector).filter(Sector.key == sector).scalar()
)
if sector_db:
errors.add_error(
@@ -268,21 +264,35 @@ def validate_common(
solution=["Proporciona un sector autorizado."],
code="SECTOR_NOT_AUTHORIZED",
)
company_db = db.query(Company).filter(Company.id == company_id).first()
if not company_db.prosec:
errors.add_error(
field=f"line[{line_number}].customs.sector",
message=" La empresa no cuenta con autorización PROSEC.",
solution=["Accese a los datos de la empresa y selecione la opción Pertenece al Programa de Promoción Sectorial y capture el número de permiso PROSEC."],
solution=[
"Accese a los datos de la empresa y selecione la opción Pertenece al Programa de Promoción Sectorial y capture el número de permiso PROSEC."
],
code="COMPANY_NOT_AUTHORIZED_FOR_PROSEC",
)
if fraction:
search_fraction_preference(db=db, country=country, fraccion=fraction, fraction_type=fraction_type, sector=sector, invoice_date=fecha_factura, errors=errors)
search_fraction_preference(
db=db,
country=country,
fraccion=fraction,
fraction_type=fraction_type,
sector=sector,
invoice_date=fecha_factura,
errors=errors,
)
if line.customs.american_fraction:
american_fraction_exists = db.query(exists().where(LineCustom.american_fraction == line.customs.american_fraction)).scalar()
american_fraction_exists = db.query(
exists().where(
LineCustom.american_fraction == line.customs.american_fraction
)
).scalar()
if not american_fraction_exists:
errors.add_error(
field=f"line[{line_number}].customs.american_fraction",
@@ -290,7 +300,7 @@ def validate_common(
solution=["Proporciona una fracción americana valida."],
code="AMERICAN_FRACTION_NOT_FOUND",
)
if item_header and item_header.order:
if len(item_header.order) > 20:
errors.add_error(
@@ -299,18 +309,22 @@ def validate_common(
solution=["Proporciona un valor valido para el campo orden."],
code="ORDER_EXCEEDS_MAX_LENGTH",
)
unit_of_measure = line.unit_of_measure or (class_.unit_of_measure if class_ else None)
if unit_of_measure == "PZA" and line.quantity.quantity % 1 != 0:
unit_of_measure = line.unit_of_measure or (
class_.unit_of_measure if class_ else None
)
if unit_of_measure == "PZA" and line.quantity.quantity % 1 != 0:
errors.add_error(
field=f"line[{line_number}].quantity.quantity",
message="La cantidad debe ser un número entero cuando la unidad de medida es PZA.",
solution=["Proporciona una cantidad entera."],
code="QUANTITY_MUST_BE_INTEGER_FOR_PIECES",
)
if line.valuation_method:
valuation_method_exists = db.query(exists().where(ValuationMethod.key == line.valuation_method)).scalar()
valuation_method_exists = db.query(
exists().where(ValuationMethod.key == line.valuation_method)
).scalar()
if not valuation_method_exists:
errors.add_error(
field=f"line[{line_number}].valuation_method",
@@ -318,7 +332,7 @@ def validate_common(
solution=["Proporciona un método de valoración valido."],
code="VALUATION_METHOD_NOT_FOUND",
)
if line.part_number_id:
part_exists = db.query(exists().where(Part.id == line.part_number_id)).scalar()
if not part_exists:
@@ -328,4 +342,3 @@ def validate_common(
solution=["Proporciona un número de parte valido."],
code="PART_NUMBER_NOT_FOUND",
)

351
backend/api/v1/modules/a76/items/imports/temporary/validators/create.py
View File

@@ -1,3 +1,4 @@
from decimal import Decimal
from sqlalchemy import func, exists
from sqlalchemy.orm import Session
from ....common.common_validators import count_items
@@ -5,27 +6,57 @@ from core.exceptions import ErrorCollector
from ....line_items.models import LineItem
from ....line_financials.models import LineFinancial
from ....line_financials.schemas import LineFinancialCreate
from ....line_quantities.models import LineQuantity
from ....line_quantities.schemas import LineQuantityCreate
from ....line_customs.models import LineCustom
from ....line_customs.schemas import LineCustomCreate
from ....line_descriptions.models import LineDescription
from ....line_descriptions.schemas import LineDescriptionCreate
from ....line_references.models import LineReference
from api.v1.modules.a24.fa.fa_item_lines.models import FaLineItem
from ....models import Item
from api.v1.modules.a76.invoices.models import InvoiceHeader
from api.v1.modules.a76.classes.models import Class
from api.v1.modules.a76.general_catalogs.packages.models import Package
from api.v1.modules.a76.general_catalogs.fractions.us_tariff_fractions.models import (
USTariffFraction,
)
from .common import validate_common
def validate_create(
db: Session,
line: LineItem,
line, # LineItemCreate schema (Pydantic)
invoice_id: int, # Passed from service
tenant_id: int,
company_id: int,
errors: ErrorCollector,
line_number: int,
):
item_header: Item = db.query(Item).filter(Item.id == line.item_id).first()
fa_line: FaLineItem = db.query(FaLineItem).filter(FaLineItem.id == line.id).first()
"""
Validates and calculates fields for a new line item before DB creation.
Works with Pydantic schemas, modifying them in-place.
Args:
line: LineItemCreate schema with nested data (financial, quantity, customs, etc.)
invoice_id: ID of the invoice this line belongs to
fa_data: FaLineItemCreateDTO or None (None for INV system)
"""
# Inicializar nested schemas si no existen (para poder validar y modificar)
if not line.financial:
line.financial = LineFinancialCreate()
if not line.quantity:
line.quantity = LineQuantityCreate()
if not line.customs:
line.customs = LineCustomCreate()
if not line.description:
line.description = LineDescriptionCreate()
# Access fa_data safely
fa_data = getattr(line, "fa_data", None)
# Required field validations
if not line.class_id:
errors.add_required_error(field=f"line[{line_number}].class_id")
@@ -34,12 +65,14 @@ def validate_create(
# TODO: Añadir validacion SSisGen:CalcularCostoUnitarioEnBaseAValorTotalScaf <-- de la tabla de preferencias de el sistema
# if SSisGen:CalcularCostoUnitarioEnBaseAValorTotalScaf == False:
if not fa_line.is_subitem and (
not line.financial.unit_cost_capture and line.financial.unit_cost_capture <= 0
):
errors.add_required_error(
field=f"line[{line_number}].financial.unit_cost_capture"
)
if fa_data and not fa_data.is_subitem:
if (
not line.financial.unit_cost_capture
or line.financial.unit_cost_capture <= 0
):
errors.add_required_error(
field=f"line[{line_number}].financial.unit_cost_capture"
)
if not line.quantity.net_weight or line.quantity.net_weight <= 0:
errors.add_required_error(field=f"line[{line_number}].quantity.net_weight")
@@ -50,48 +83,272 @@ def validate_create(
if not line.customs.fraction_type:
errors.add_required_error(field=f"line[{line_number}].customs.fraction_type")
if (
fa_line.is_subitem and fa_line.contains_subitems
) and not fa_line.subitem_number:
errors.add_required_error(field=f"line[{line_number}].subitem_number")
errors.raise_if_errors()
# Validar que si es un subitem, existe un item principal correspondiente
if fa_line.is_subitem and fa_line.subitem_number != 0:
principal_item_exists = db.query(
exists().where(
(LineItem.id == FaLineItem.id)
& (LineItem.item_id == item_header.id)
& (LineItem.line_number == line_number)
& (FaLineItem.is_subitem == False)
& (FaLineItem.contains_subitems == True)
& (LineItem.tenant_id == tenant_id)
& (LineItem.company_id == company_id)
# FA-specific validations
if fa_data:
if (
fa_data.is_subitem and fa_data.contains_subitems
) and not fa_data.subitem_number:
errors.add_required_error(
field=f"line[{line_number}].fa_data.subitem_number"
)
).scalar()
if not principal_item_exists:
# Validar que si es un subitem, existe un item principal correspondiente
if (
fa_data.is_subitem
and fa_data.subitem_number
and fa_data.subitem_number != 0
):
principal_item_exists = db.query(
exists().where(
(LineItem.id == FaLineItem.id)
& (LineItem.item_id == Item.id)
& (Item.invoice_id == invoice_id)
& (LineItem.line_number == line_number)
& (FaLineItem.is_subitem == False)
& (FaLineItem.contains_subitems == True)
& (LineItem.tenant_id == tenant_id)
& (LineItem.company_id == company_id)
)
).scalar()
if not principal_item_exists:
errors.add_error(
field=f"line[{line_number}]",
message=f"No existe un item principal registrado para esta linea {line_number} con subitem {fa_data.subitem_number}",
solution=[
"Registrar el item principal correspondiente a esta linea antes de registrar subitems."
],
code="SUBITEM_WITHOUT_PRINCIPAL_ITEM",
)
if fa_data.is_subitem and (
fa_data.subitem_number == 0 or not fa_data.subitem_number
):
errors.add_error(
field=f"line[{line_number}]",
message=f"No existe un item principal registrado para esta linea {line_number} con subitem {fa_line.subitem_number}",
solution=[
"Registrar el item principal correspondiente a esta linea antes de registrar subitems."
],
code="SUBITEM_WITHOUT_PRINCIPAL_ITEM",
message=f"El número de subitem no puede ser 0 si la línea es un subitem.",
solution=["Asignar un número de subitem mayor a 0 para esta línea."],
code="SUBITEM_NUMBER_INVALID",
)
if fa_line.is_subitem and (fa_line.subitem_number == 0 or not fa_line.subitem_number):
errors.add_error(
field=f"line[{line_number}]",
message=f"El número de subitem no puede ser 0 si la línea es un subitem.",
solution=[
"Asignar un número de subitem mayor a 0 para esta línea."
],
code="SUBITEM_NUMBER_INVALID",
)
validate_common(db, line, tenant_id, company_id, errors, line_number)
validate_common(db, line, invoice_id, tenant_id, company_id, errors, line_number)
if not errors.has_errors():
pass
# Obtener la factura para acceder a tipo de cambio, moneda y peso
invoice: InvoiceHeader = (
db.query(InvoiceHeader)
.filter(
InvoiceHeader.id == invoice_id,
InvoiceHeader.tenant_id == tenant_id,
InvoiceHeader.company_id == company_id,
)
.first()
)
if not invoice or not invoice.financials or not invoice.logistics:
errors.add_error(
field=f"line[{line_number}]",
message="No se pudo obtener información de la factura",
solution=[
"Verificar que la factura existe y tiene datos financieros y logísticos"
],
code="INVOICE_DATA_MISSING",
)
return
# Obtener la clase para valores por defecto
class_info: Class = (
db.query(Class)
.filter(
Class.id == line.class_id,
Class.tenant_id == tenant_id,
Class.company_id == company_id,
)
.first()
)
# ==========================================
# ASIGNAR TIPO DE CAMBIO
# ==========================================
exchange_rate = invoice.financials.exchange_rate or Decimal("1.0")
# ==========================================
# ASIGNAR UNIDAD DE MEDIDA
# ==========================================
# Si no se proporcionó unidad de medida, usar la de la clase
if not line.unit_of_measure and class_info:
line.unit_of_measure = class_info.unit_of_measure
# ==========================================
# ASIGNAR TIPOS DE MONEDA Y CALCULAR COSTOS
# ==========================================
currency_type = invoice.financials.currency_type
unit_cost_capture = line.financial.unit_cost_capture or Decimal("0")
# Calcular costos según tipo de moneda
if currency_type == "USD" or currency_type == "ME": # Moneda Extranjera (ME)
line.financial.unit_cost_capture = unit_cost_capture
line.financial.unit_cost_usd = unit_cost_capture
line.financial.unit_cost_mxn = unit_cost_capture * exchange_rate
elif currency_type == "MXN" or currency_type == "MN": # Moneda Nacional (MN)
line.financial.unit_cost_capture = unit_cost_capture
line.financial.unit_cost_usd = (
unit_cost_capture / exchange_rate if exchange_rate else Decimal("0")
)
line.financial.unit_cost_mxn = unit_cost_capture
# Si es otro tipo de moneda, dejamos el costo como está
# ==========================================
# VALIDAR Y CONVERTIR PESOS NETOS
# ==========================================
invoice_weight_type = invoice.logistics.weight_type # 'kgs' o 'lbs'
quantity = line.quantity.quantity or Decimal("0")
net_weight_input = line.quantity.net_weight or Decimal("0")
# Determinar si la unidad de medida es de peso
unit_is_kgs = line.unit_of_measure and line.unit_of_measure.upper() == "KGS"
unit_is_lbs = line.unit_of_measure and line.unit_of_measure.upper() == "LB"
# Calcular peso neto en kilogramos (estándar interno)
if unit_is_kgs:
if invoice_weight_type == "kgs":
line.quantity.net_weight = quantity
else: # invoice en libras
line.quantity.net_weight = quantity * Decimal("2.204624")
elif unit_is_lbs:
if invoice_weight_type == "kgs":
line.quantity.net_weight = quantity / Decimal("2.204624")
else: # invoice en libras
line.quantity.net_weight = quantity
else:
# Otra unidad de medida - usar peso capturado y convertir si es necesario
if invoice_weight_type == "kgs":
# El peso capturado está en kilos
line.quantity.net_weight = net_weight_input
else:
# El peso capturado está en libras, convertir a kilos
line.quantity.net_weight = net_weight_input / Decimal("2.204624")
# ==========================================
# CALCULAR PESO BRUTO
# ==========================================
gross_weight_input = line.quantity.gross_weight
package_quantity = line.quantity.package_quantity or 0
package_weight_unit = Decimal("0")
# Obtener peso unitario del bulto si existe
if line.quantity.package_key:
package: Package = (
db.query(Package)
.filter(
Package.key == line.quantity.package_key,
Package.tenant_id == tenant_id,
Package.company_id == company_id,
)
.first()
)
if package and package.weight_unit:
package_weight_unit = package.weight_unit
# Si no se proporcionó peso bruto, calcularlo
if not gross_weight_input or gross_weight_input == 0:
if invoice_weight_type == "kgs":
line.quantity.gross_weight = line.quantity.net_weight + (
package_weight_unit * package_quantity
)
else: # libras
line.quantity.gross_weight = line.quantity.net_weight + (
(package_weight_unit * Decimal("2.204624")) * package_quantity
)
else:
# Convertir peso bruto capturado según tipo de factura
if invoice_weight_type == "kgs":
line.quantity.gross_weight = gross_weight_input
else: # libras
line.quantity.gross_weight = gross_weight_input / Decimal("2.204624")
# ==========================================
# VALIDAR PESO BRUTO < PESO NETO
# ==========================================
if line.quantity.gross_weight < line.quantity.net_weight:
line.quantity.gross_weight = line.quantity.net_weight + (
package_weight_unit * package_quantity
)
# ==========================================
# ASIGNAR DESCRIPCIÓN DE BULTOS
# ==========================================
if package_quantity and package_quantity > 0 and line.quantity.package_key:
package: Package = (
db.query(Package)
.filter(
Package.key == line.quantity.package_key,
Package.tenant_id == tenant_id,
Package.company_id == company_id,
)
.first()
)
if package:
line.quantity.package_description = package.description_es
else:
line.quantity.package_quantity = 0
line.quantity.package_key = None
line.quantity.package_description = None
# ==========================================
# ASIGNAR FRACCIÓN AMERICANA POR DEFECTO
# ==========================================
if not line.customs.american_fraction and class_info and class_info.us_fraction:
line.customs.american_fraction = class_info.us_fraction
# Buscar el advalorem de la fracción americana
if line.customs.american_fraction:
us_fraction: USTariffFraction = (
db.query(USTariffFraction)
.filter(
USTariffFraction.code == line.customs.american_fraction,
USTariffFraction.tenant_id == tenant_id,
USTariffFraction.company_id == company_id,
)
.first()
)
if us_fraction:
# Si el tipo es 'ME' (Moneda Extranjera), usar costo fijo
# De lo contrario, usar ad valorem
if us_fraction.type_code == "foreign":
line.customs.advalorem_american = us_fraction.fixed_cost
else:
line.customs.advalorem_american = us_fraction.ad_valorem
# ==========================================
# ASIGNAR DESCRIPCIONES POR DEFECTO
# ==========================================
if not line.description.description_spanish and class_info:
line.description.description_spanish = class_info.description_es
if not line.description.description_english and class_info:
line.description.description_english = class_info.description_en
# ==========================================
# NORMALIZAR CAMPOS DE TEXTO
# ==========================================
# Convertir a mayúsculas campos que lo requieran
if line.description.brand:
line.description.brand = line.description.brand.upper().strip()
if line.description.model:
line.description.model = line.description.model.upper().strip()
# ==========================================
# ASIGNAR VALORES POR DEFECTO DE IMPUESTOS
# ==========================================
# Si no se especificó pago de impuesto, tomar de preferencias del sistema (SisImp)
# TODO: Implementar lectura de preferencias del sistema
# Por ahora dejamos None si no se proporcionó
# Si no se especificó forma de pago, tomar de preferencias del sistema
# TODO: Implementar lectura de preferencias del sistema
# Si no se especificó método de valoración, tomar de preferencias del sistema
# TODO: Implementar lectura de preferencias del sistema

200
backend/api/v1/modules/a76/items/imports/temporary/validators/update.py
View File

@@ -1,30 +1,202 @@
import logging
from typing import Optional, List, Tuple
from fastapi import HTTPException
from sqlalchemy import and_, or_
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session, joinedload
from decimal import Decimal
from sqlalchemy.orm import Session
from api.v1.modules.a76.invoices.common.common_validators import invoice_exists
from core.exceptions import ErrorCollector
from ....line_items.models import LineItem
from ....line_financials.models import LineFinancial
from ....line_quantities.models import LineQuantity
from ....line_customs.models import LineCustom
from ....line_descriptions.models import LineDescription
from ....line_references.models import LineReference
from api.v1.modules.a24.fa.fa_item_lines.models import FaLineItem
from ....models import Item
from api.v1.modules.a76.invoices.models import InvoiceHeader
from api.v1.modules.a76.general_catalogs.fractions.us_tariff_fractions.models import (
USTariffFraction,
)
from .common import validate_common
def validate_update(
db: Session,
line: Item,
line: LineItem,
existing_line: LineItem,
invoice_id: int, # Passed from service
tenant_id: int,
company_id: int,
errors: ErrorCollector,
line_number: int,
):
validate_common(db, line, tenant_id, company_id, errors, line_number)
"""
Validar y procesar actualización parcial de línea de importación temporal.
Si un campo no se proporciona, se mantiene el valor existente.
"""
validate_common(db, line, invoice_id, tenant_id, company_id, errors, line_number)
if not errors.has_errors():
# Obtener la factura para acceder a tipo de cambio, moneda y peso
invoice: InvoiceHeader = (
db.query(InvoiceHeader)
.filter(
InvoiceHeader.id == invoice_id,
InvoiceHeader.tenant_id == tenant_id,
InvoiceHeader.company_id == company_id
)
.first()
)
if not invoice or not invoice.financials or not invoice.logistics:
errors.add_error(
field=f"line[{line_number}]",
message="No se pudo obtener información de la factura",
solution=[
"Verificar que la factura existe y tiene datos financieros y logísticos"
],
code="INVOICE_DATA_MISSING",
)
return
# ==========================================
# ACTUALIZACIÓN PARCIAL DE CAMPOS
# Si no se proporciona, mantener valor existente
# ==========================================
# Tipo de cambio de la factura
exchange_rate = invoice.financials.exchange_rate or Decimal("1.0")
# Unidad de medida
if not line.unit_of_measure:
line.unit_of_measure = existing_line.unit_of_measure
# Costo unitario
if line.financial.unit_cost_capture is None:
line.financial.unit_cost_capture = existing_line.financial.unit_cost_capture
# Convertir peso neto si se proporcionó
invoice_weight_type = invoice.logistics.weight_type
if line.quantity.net_weight is not None:
# Se proporcionó nuevo peso neto, convertir según tipo
net_weight_input = line.quantity.net_weight
if invoice_weight_type == "kgs":
line.quantity.net_weight = net_weight_input
else: # libras, convertir a kilos
line.quantity.net_weight = net_weight_input / Decimal("2.204624")
else:
# Mantener peso existente
line.quantity.net_weight = existing_line.quantity.net_weight
# Convertir peso bruto si se proporcionó
if line.quantity.gross_weight is not None:
gross_weight_input = line.quantity.gross_weight
if invoice_weight_type == "kgs":
line.quantity.gross_weight = gross_weight_input
else: # libras, convertir a kilos
line.quantity.gross_weight = gross_weight_input / Decimal("2.204624")
else:
# Mantener peso existente
line.quantity.gross_weight = existing_line.quantity.gross_weight
# Cantidad de bultos
if line.quantity.package_quantity is None:
line.quantity.package_quantity = existing_line.quantity.package_quantity
# Clave de bultos
if not line.quantity.package_key:
line.quantity.package_key = existing_line.quantity.package_key
# País de origen
if not line.customs.origin_country:
line.customs.origin_country = existing_line.customs.origin_country
# Fracción arancelaria
if not line.customs.fraction:
line.customs.fraction = existing_line.customs.fraction
# Tipo de fracción
if not line.customs.fraction_type:
line.customs.fraction_type = existing_line.customs.fraction_type
# Sector
if not line.customs.sector:
line.customs.sector = existing_line.customs.sector
# Fracción americana y su advalorem
if line.customs.american_fraction:
# Se proporcionó nueva fracción americana, buscar su advalorem
us_fraction: USTariffFraction = (
db.query(USTariffFraction)
.filter(
USTariffFraction.code == line.customs.american_fraction,
USTariffFraction.tenant_id == tenant_id,
USTariffFraction.company_id == company_id,
)
.first()
)
if us_fraction:
if us_fraction.type_code == "ME":
line.customs.advalorem_american = us_fraction.fixed_cost
else:
line.customs.advalorem_american = us_fraction.ad_valorem
else:
# Mantener fracción americana existente
line.customs.american_fraction = existing_line.customs.american_fraction
line.customs.advalorem_american = existing_line.customs.advalorem_american
# Orden de compra
if not line.reference.purchase_order:
line.reference.purchase_order = existing_line.reference.purchase_order
# Descripciones
if not line.description.description_spanish:
line.description.description_spanish = (
existing_line.description.description_spanish
)
if not line.description.description_english:
line.description.description_english = (
existing_line.description.description_english
)
if not line.description.extra_description:
line.description.extra_description = (
existing_line.description.extra_description
)
# Marca y modelo
if line.description.brand:
line.description.brand = line.description.brand.upper().strip()
else:
line.description.brand = existing_line.description.brand
if line.description.model:
line.description.model = line.description.model.upper().strip()
else:
line.description.model = existing_line.description.model
# Subpartidas (si aplica)
# TODO: Implementar lógica de subpartidas si Loc:LevantarSubpartidas = 'S'
# Número de parte
if not line.part_number:
line.part_number = existing_line.part_number
# Pago de impuesto
if line.tax_payment is None:
line.tax_payment = existing_line.tax_payment
# Forma de pago
if not line.payment_method:
line.payment_method = existing_line.payment_method
# Método de valoración
if not line.valuation_method:
if existing_line.valuation_method:
line.valuation_method = existing_line.valuation_method
# else: TODO: Tomar de SisImp:MetValor (preferencias del sistema)
# Número de entrada
if not line.description.entry_number:
line.description.entry_number = existing_line.description.entry_number
# Lote
if not line.description.lot:
line.description.lot = existing_line.description.lot

18
backend/api/v1/modules/a76/items/line_items/schemas.py
View File

@@ -51,10 +51,16 @@ class LineItemBase(BaseModel):
# Part identification
part_number_id: Optional[int] = Field(
None, description="Part number", alias="part_number", serialization_alias="part_number_id"
None,
description="Part number",
alias="part_number",
serialization_alias="part_number_id",
)
component_part_number_id: Optional[int] = Field(
None, description="Component part number", alias="component_part_number", serialization_alias="component_part_number_id"
None,
description="Component part number",
alias="component_part_number",
serialization_alias="component_part_number_id",
)
class_id: Optional[int] = Field(None, description="Class code")
@@ -182,6 +188,12 @@ class LineItemBase(BaseModel):
class LineItemCreate(LineItemBase):
"""Schema for creating line item with all nested data"""
# Override base fields - estos se asignan automáticamente en el service
item_id: Optional[int] = Field(
None, description="ID of the parent item (auto-assigned)"
)
line_number: Optional[int] = Field(None, description="Line number (auto-assigned)")
financial: Optional[LineFinancialCreate] = Field(
None, description="Financial data for this line"
)
@@ -205,6 +217,8 @@ class LineItemCreate(LineItemBase):
class LineItemUpdate(LineItemBase):
"""Schema for updating line item with all nested data"""
# Override base fields - todos opcionales en updates
item_id: Optional[int] = Field(None, description="ID of the parent item")
line_number: Optional[int] = Field(None, description="Line number")
financial: Optional[LineFinancialUpdate] = Field(
None, description="Financial data for this line"

1
backend/api/v1/modules/a76/items/line_quantities/schemas.py
View File

@@ -26,6 +26,7 @@ class LineQuantityBase(BaseModel):
gross_weight: Optional[Decimal] = Field(None, description="Gross weight (PESOBRUTO)")
# Packaging
package_id: Optional[int] = Field(None, description="Package ID (IDBULTOS)")
package_key: Optional[str] = Field(None, max_length=5, description="Package key (CLAVEBULTOS)")
package_quantity: Optional[int] = Field(None, description="Package quantity (CANTBULTOS)")
package_description: Optional[str] = Field(None, max_length=40, description="Package description (DESCBULTOS)")

54
backend/api/v1/modules/a76/items/service.py
View File

@@ -17,7 +17,7 @@ from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session, joinedload
from api.v1.modules.a76.invoices.common.common_validators import (
invoice_exists,
invoice_exists_by_id,
invoice_updated,
)
from core.exceptions import ErrorCollector
@@ -205,7 +205,7 @@ class ItemService:
search_term = f"%{filters['search']}%"
query = query.filter(
or_(
Item.invoice_number.ilike(search_term),
Item.invoice_id.ilike(search_term),
Item.reference_number.ilike(search_term),
Item.order.ilike(search_term),
Item.guide_number.ilike(search_term),
@@ -262,14 +262,14 @@ class ItemService:
# Validar que la factura exista y no esté actualizada (si viene invoice_id)
if not item_data.invoice_id:
errors.add_required_error(field="invoice_id")
return
errors.raise_if_errors("Error al crear el item")
if not invoice_exists(db, item_data.invoice_id, tenant_id, company_id, errors):
return
if not invoice_updated(
db, item_data.invoice_number, tenant_id, company_id, errors
if not invoice_exists_by_id(
db, item_data.invoice_id, tenant_id, company_id, errors
):
return
errors.raise_if_errors("Error al crear el item")
if not invoice_updated(db, item_data.invoice_id, tenant_id, company_id, errors):
errors.raise_if_errors("Error al crear el item")
# Lock invoice and pre-calculate line_numbers
if not ItemService._lock_invoice(
@@ -287,25 +287,26 @@ class ItemService:
for idx, line_data in enumerate(item_data.lines):
line_number = line_numbers[idx] # Usar el line_number calculado
# Convertir a LineItemCreate para validar
line_create = LineItemCreate(**line_data.model_dump())
validate_create(
db, line_create, tenant_id, company_id, errors, line_number
db,
line_data, # Schema Pydantic completo
item_data.invoice_id, # invoice_id
tenant_id,
company_id,
errors,
line_number,
)
# Validaciones adicionales específicas del negocio
if not line_data.fa_data.is_subitem:
if line_data.fa_data and line_data.fa_data.is_subitem is None:
errors.add_required_error(
field=f"lines[{line_number}].fa_data.is_subitem"
)
return
if not line_data.fa_data.subitem_number:
if line_data.fa_data and line_data.fa_data.subitem_number is None:
errors.add_required_error(
field=f"lines[{line_number}].fa_data.subitem_number"
)
return
# Validar apóstrofes en número de parte
if line_data.part_number_id and "'" in str(line_data.part_number_id):
@@ -435,16 +436,27 @@ class ItemService:
None,
)
if existing_line:
# Convertir a LineItemUpdate para validar
line_update = LineItemUpdate(**line_data.model_dump())
# Validar update con línea existente
validate_update(
db, line_update, tenant_id, company_id, errors, line_number
db,
line_data, # Schema de update
existing_line, # LineItem existente en DB
invoice_id_to_lock, # invoice_id
tenant_id,
company_id,
errors,
line_number,
)
else:
# Es un nuevo line item, validar como creación
line_create = LineItemCreate(**line_data.model_dump())
validate_create(
db, line_create, tenant_id, company_id, errors, line_number
db,
line_data, # Schema Pydantic completo
invoice_id_to_lock, # invoice_id
tenant_id,
company_id,
errors,
line_number,
)
# Validaciones adicionales específicas del negocio

14
backend/api/v1/modules/sitar/fracciones_usa/schemas.py
View File

@@ -7,9 +7,17 @@ from pydantic import BaseModel, Field
class FraccionesUSAResponse(BaseModel):
"""USA tariff fractions"""
FRACCIONUSA: Optional[str] = Field(None, max_length=10)
DESCRIPCIONUSA: Optional[str] = None
CLAVEUM: Optional[str] = Field(None, max_length=2)
FRACCION_SIN_PUNTO: Optional[str] = None
FRACCION_CON_PUNTO: Optional[str] = None
FRACCION_MOSTRAR: Optional[str] = None
ESPECIFICO: Optional[str] = None
NIVEL: Optional[str] = None
DESCRIPCION: Optional[str] = None
UNIDADCANTIDAD: Optional[str] = None
TARIFA1: Optional[str] = None
TLC: Optional[str] = None
TARIFA2: Optional[str] = None
NOTAS: Optional[str] = None
CONSECUTIVO: int
class Config: