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

Merge pull request 'feature/items-validations' (#199) from feature/items-validations into development

Browse Source 
Reviewed-on: ADUANASOFT/anexo76#199
This commit is contained in:
acazares
2026-03-11 16:10:31 +00:00
parent ff08994620 c1a9ee789b
commit 0416264fe9
11 changed files with 1366 additions and 937 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
backend/api/v1/modules/a76/invoices/services.py
View File

@@ -126,7 +126,10 @@ class InvoiceService:
# Validar si la factura ya existe
invoice_exists(db, invoice_data.invoice_number, tenant_id, company_id, errors)
validate_create_import(db, invoice_data, tenant_id, company_id, errors)
if invoice_data.operation_type == "exp":
validate_create_export(db, invoice_data, tenant_id, company_id, errors)
else:
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")

220
backend/api/v1/modules/a76/items/exports/validators/calculations.py Normal file
View File

@@ -0,0 +1,220 @@
from sqlalchemy.orm import Session
from api.v1.modules.a76.invoices.models import InvoiceFinancials, InvoiceHeader, InvoiceLogistics
from ...models import LineItem
from ...series.models import Serie
from api.v1.modules.a76.items.models import LineItem
from api.v1.modules.a24.fa.fa_item_lines.models import FaLineItem
from api.v1.modules.a76.classes.models import Class
def apply_calculations(
db: Session, line: LineItem, tenant_id: int, company_id: int, line_number: int
):
#TODO: SSisGen Logic
# if ssisgen.calcularcostounitarioenbaseavalortotalscaf = 1:
# unit_cost_capture = line.financial.total_value / line.financial.total_value <-- habria que revisar por que esta asi, por que para mi no tiene sentido, pero es lo que esta en clarion
calculate_values(db, line, tenant_id, company_id)
# ==========================================
# LLEVASERIE / LlevaCodFDA defaults
# ==========================================
# EqiPex:LlevaCodFDA = 'N'
line.has_fda_code = False
# ==========================================
# PAGO IMPUESTO default: 'N' (False)
# ==========================================
if line.tax_payment is None:
# TODO: Leer de SisExp:PagoImpuesto (preferencias del sistema)
line.tax_payment = False
# ==========================================
# FORMA DE PAGO default: '5'
# ==========================================
if not line.payment_method:
# TODO: Leer de SisExp:FormaPago (preferencias del sistema)
line.payment_method = "5"
# ==========================================
# SUBPARTIDAS: EsSubPartida / ContieneSubP / IncuyeSubPartidas
# ==========================================
fa_data: FaLineItem = getattr(line, "fa_data", None)
if fa_data is not None:
if fa_data.is_subitem is None:
fa_data.is_subitem = False
if not fa_data.is_subitem:
# Es partida principal — subitem_number se fuerza a 0
fa_data.subitem_number = 0
# ContieneSubP: verificar si ya existen subitems en DB que referencian esta línea
# (útil en updates; en create siempre será False porque la línea aún no existe)
existing_subitems = (
db.query(FaLineItem)
.join(LineItem, FaLineItem.id == LineItem.id)
.filter(
LineItem.invoice_id == line.invoice_id,
LineItem.line_number == line_number,
LineItem.tenant_id == tenant_id,
LineItem.company_id == company_id,
FaLineItem.is_subitem == True,
FaLineItem.subitem_number == line_number,
)
.count()
)
fa_data.contains_subitems = existing_subitems > 0
invoice_date = db.query(InvoiceHeader.invoice_date).filter(InvoiceHeader.id == line.invoice_id, InvoiceHeader.tenant_id == tenant_id, InvoiceHeader.company_id == company_id).scalar()
line.depreciation_date = invoice_date
if (not line.description.description_spanish and not line.description.description_english) and (line.part_info.description_spanish and line.part_info.description_english):
line.description.description_spanish = line.part_info.description_spanish
line.description.description_english = line.part_info.description_english
else:
if not line.description.description_spanish:
class_desc = (
db.query(Class.description_es, Class.description_en)
.filter(Class.id == line.class_id, Class.tenant_id == tenant_id, Class.company_id == company_id)
.first()
)
if class_desc:
line.description.description_spanish, line.description.description_english = class_desc
def calculate_values(
db: Session,
line: LineItem,
tenant_id: int,
company_id: int,
) -> None:
"""
Calcula valores financieros y copia campos de la línea de importación referenciada.
Traduce el CALCULOS ROUTINE de Clarion:
- Busca la factura de importación por fa_data.search_invoice (TEM → DEF como fallback)
- Copia clase, unidad de medida, fracción (si fa_data.download), país, tipo fracción,
bultos y descripción inglés desde la línea de importación encontrada
- Calcula valores en moneda (USD/MXN/MC) según la moneda de la factura
"""
fa_data: FaLineItem = getattr(line, "fa_data", None)
# ==========================================
# BUSCAR FACTURA DE IMPORTACIÓN (TEM → DEF)
# EqiFim:FacturaImpo = fa_data.search_invoice / EqiPim:LineaImpo = fa_data.search_line
# ==========================================
import_line = None
import_invoice_number = fa_data.search_invoice if fa_data else None
import_line_number = fa_data.search_line if fa_data else None
if import_invoice_number and import_line_number:
# 1. Intentar TEM (Importación Temporal)
tem_invoice = (
db.query(InvoiceHeader)
.filter(
InvoiceHeader.invoice_number == import_invoice_number,
InvoiceHeader.invoice_type == "TEM",
InvoiceHeader.tenant_id == tenant_id,
InvoiceHeader.company_id == company_id,
)
.first()
)
if tem_invoice:
import_line = (
db.query(LineItem)
.filter(
LineItem.invoice_id == tem_invoice.id,
LineItem.line_number == import_line_number,
LineItem.tenant_id == tenant_id,
LineItem.company_id == company_id,
)
.first()
)
# 2. Si no hay TEM, intentar DEF (Importación Definitiva)
if not import_line:
def_invoice = (
db.query(InvoiceHeader)
.filter(
InvoiceHeader.invoice_number == import_invoice_number,
InvoiceHeader.invoice_type == "DEF",
InvoiceHeader.tenant_id == tenant_id,
InvoiceHeader.company_id == company_id,
)
.first()
)
if def_invoice:
import_line = (
db.query(LineItem)
.filter(
LineItem.invoice_id == def_invoice.id,
LineItem.line_number == import_line_number,
LineItem.tenant_id == tenant_id,
LineItem.company_id == company_id,
)
.first()
)
# ==========================================
# COPIAR CAMPOS DESDE LÍNEA DE IMPORTACIÓN
# EqiPex:Clase, UnidadMedida, FraccionExpo (condicional), PaisOrigen,
# TipoFraccion, CantBultos, ClaveBultos, DescripcionE
# ==========================================
if import_line:
line.class_id = import_line.class_id
line.unit_of_measure = import_line.unit_of_measure
# Fracción: copiar solo si fa_data.download == True (≡ ColumnaV != '')
if fa_data and fa_data.download and import_line.customs:
line.customs.fraction = import_line.customs.fraction
if import_line.customs:
line.customs.origin_country = import_line.customs.origin_country
line.customs.fraction_type = import_line.customs.fraction_type
if import_line.quantity:
line.quantity.package_quantity = import_line.quantity.package_quantity
line.quantity.package_id = import_line.quantity.package_id
if import_line.description:
line.description.description_english = import_line.description.description_english
# ==========================================
# CÁLCULOS DE VALORES EN MONEDA
# foreign=ME, local=MN, manual=MC
# ==========================================
result = (
db.query(InvoiceFinancials.currency, InvoiceFinancials.exchange_rate)
.filter(
InvoiceFinancials.invoice_id == line.invoice_id,
InvoiceFinancials.tenant_id == tenant_id,
InvoiceFinancials.company_id == company_id,
)
.first()
)
if not result:
return
currency, exchange_rate = result
if currency == "foreign": # ME
line.financial.unit_cost_usd = line.financial.unit_cost_capture
line.financial.value_usd = line.financial.unit_cost_usd * line.quantity.quantity
line.financial.unit_cost_mxn = line.financial.unit_cost_capture * exchange_rate
line.financial.value_mxn = line.financial.unit_cost_mxn * line.quantity.quantity
line.financial.value_mc = line.financial.unit_cost_usd * line.quantity.quantity
elif currency == "local": # MN
line.financial.unit_cost_mxn = line.financial.unit_cost_capture
line.financial.value_mxn = line.financial.unit_cost_mxn * line.quantity.quantity
line.financial.unit_cost_usd = line.financial.unit_cost_capture / exchange_rate
line.financial.value_usd = line.financial.unit_cost_usd * line.quantity.quantity
line.financial.value_mc = line.financial.unit_cost_usd * line.quantity.quantity
elif currency == "manual": # MC
line.financial.unit_cost_usd = line.financial.unit_cost_capture / exchange_rate
line.financial.value_usd = line.financial.unit_cost_usd * line.quantity.quantity
line.financial.unit_cost_mxn = line.financial.unit_cost_usd * exchange_rate
line.financial.value_mxn = line.financial.unit_cost_mxn * line.quantity.quantity
line.financial.value_mc = line.financial.unit_cost_capture * line.quantity.quantity

417
backend/api/v1/modules/a76/items/exports/validators/common.py Normal file
View File

@@ -0,0 +1,417 @@
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
from ...common.fractions import search_fraction_preference
from ...common.common_validators import item_exists
from ...models import LineItem
from ...line_customs.models import FractionType, LineCustom
from api.v1.modules.a76.items.schemas import LineItemCreate
from api.v1.modules.a76.items.exports.validators.calculations import apply_calculations
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.units_of_measure.models import UnitOfMeasure
from api.v1.modules.a76.general_catalogs.packages.models import Package
from api.v1.modules.a24.fa.fa_item_lines.models import FaLineItem
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.a76.parts.models import Part
from api.v1.modules.a76.general_catalogs.company.models import Company
def validate_common(
db: Session,
line: LineItemCreate,
tenant_id: int,
company_id: int,
errors: ErrorCollector,
line_number: int,
):
invoice: InvoiceHeader = invoice_exists_by_id(
db, line.invoice_id, tenant_id, company_id, errors
)
line_item: LineItem = item_exists(db, line.line_number, tenant_id, company_id)
fecha_factura = invoice.invoice_date if invoice else None
fraction = None
class_ = db.query(Class).filter(Class.id == line.class_id).first()
if not class_:
errors.add_error(
field=f"line[{line_number}].class_id",
message="La clase especificada no existe.",
solution=["Darla de alta en el catalogo de clases."],
code="CLASS_NOT_FOUND",
)
else:
if not line.unit_of_measure and not class_.unit_of_measure:
errors.add_error(
field=f"line[{line_number}].unit_of_measure",
message="La unidad de medida es obligatoria para la clase especificada.",
solution=["Proporciona una unidad de medida valida."],
code="UNIT_OF_MEASURE_REQUIRED",
)
if not line.customs.fraction:
if not line_item:
if not class_.fraction:
errors.add_error(
field=f"line[{line_number}].customs.fraction",
message="La fracción arancelaria es obligatoria para la clase especificada.",
solution=["Proporciona una fracción arancelaria valida."],
code="FRACTION_REQUIRED",
)
else:
fraction = class_.fraction
else:
if not line.customs.fraction:
if not class_.fraction:
errors.add_error(
field=f"line[{line_number}].customs.fraction",
message="La fracción arancelaria es obligatoria para la clase especificada.",
solution=["Proporciona una fracción arancelaria valida."],
code="FRACTION_REQUIRED",
)
else:
fraction = class_.fraction
else:
if line_item:
fraction = line.customs.fraction
if not line.description.description_spanish and not class_.description_es:
errors.add_error(
field=f"line[{line_number}].description.description_spanish",
message="La descripción en español es obligatoria para la clase especificada.",
solution=["Proporciona una descripción en español valida."],
code="DESCRIPTION_SPANISH_REQUIRED",
)
if not line.description.description_english and not class_.description_en:
errors.add_error(
field=f"line[{line_number}].description.description_english",
message="La descripción en inglés es obligatoria para la clase especificada.",
solution=["Proporciona una descripción en inglés valida."],
code="DESCRIPTION_ENGLISH_REQUIRED",
)
if line.quantity.quantity and line.quantity.quantity <= 0:
errors.add_error(
field=f"line[{line_number}].quantity.quantity",
message="La cantidad debe ser mayor a cero.",
solution=["Proporciona una cantidad valida."],
code="QUANTITY_MUST_BE_GREATER_THAN_ZERO",
)
if line.unit_of_measure:
um = (
db.query(func.count(UnitOfMeasure.id))
.filter(
UnitOfMeasure.id == line.unit_of_measure,
UnitOfMeasure.tenant_id == tenant_id,
UnitOfMeasure.company_id == company_id,
)
.scalar()
)
if um == 0:
errors.add_error(
field=f"line[{line_number}].unit_of_measure",
message="La unidad de medida especificada no existe.",
solution=["Proporciona una unidad de medida valida."],
code="UNIT_OF_MEASURE_NOT_FOUND",
)
if line.quantity.package_id:
package = (
db.query(func.count(Package.id))
.filter(
Package.id == line.quantity.package_id,
Package.tenant_id == tenant_id,
Package.company_id == company_id,
)
.scalar()
)
if package == 0:
errors.add_error(
field=f"line[{line_number}].quantity.package_id",
message="El paquete especificado no existe.",
solution=["Proporciona un paquete valido."],
code="PACKAGE_NOT_FOUND",
)
if not line.quantity.package_quantity:
errors.add_error(
field=f"line[{line_number}].quantity.package_quantity",
message="La cantidad de paquetes es obligatoria cuando se proporciona el paquete.",
solution=["Proporciona una cantidad de paquetes valida."],
code="PACKAGE_QUANTITY_REQUIRED",
)
if line.quantity.package_quantity <= 0:
errors.add_error(
field=f"line[{line_number}].quantity.package_quantity",
message="La cantidad de paquetes debe ser mayor a cero.",
solution=["Proporciona una cantidad de paquetes valida."],
code="PACKAGE_QUANTITY_MUST_BE_GREATER_THAN_ZERO",
)
else:
if line.quantity.package_quantity and (line.quantity.package_quantity > 0 and not line.quantity.package_id):
errors.add_error(
field=f"line[{line_number}].quantity.package_id",
message="El paquete es obligatorio cuando se proporciona la cantidad de paquetes.",
solution=["Proporciona un paquete valido."],
code="PACKAGE_ID_REQUIRED",
)
# ==========================================
# FA DATA: PROCEDENCIA Y REFERENCIA DE IMPORTACIÓN (Col. C / D / E / H)
# ==========================================
fa_data: FaLineItem = getattr(line, "fa_data", None)
if fa_data and fa_data.movement_type_import:
# Col. C: debe ser 'TEM' o 'DEF'
if fa_data.movement_type_import.upper() not in ("TEM", "DEF"):
errors.add_error(
field=f"line[{line_number}].fa_data.movement_type_import",
message=f"La Procedencia '{fa_data.movement_type_import}' no es válida. Debe ser 'TEM' o 'DEF'.",
solution=["Capturar una procedencia válida como TEM o DEF."],
code="MOVEMENT_TYPE_IMPORT_INVALID",
)
elif fa_data.search_invoice:
tipo_label = "Temporales" if fa_data.movement_type_import.upper() == "TEM" else "Definitivas"
tipo_label_sg = "Temporal" if fa_data.movement_type_import.upper() == "TEM" else "Definitiva"
# Col. D: validar que la factura de importación exista
import_invoice: InvoiceHeader = (
db.query(InvoiceHeader)
.filter(
InvoiceHeader.invoice_number == fa_data.search_invoice,
InvoiceHeader.invoice_type == fa_data.movement_type_import.upper(),
InvoiceHeader.tenant_id == tenant_id,
InvoiceHeader.company_id == company_id,
)
.first()
)
if not import_invoice:
errors.add_error(
field=f"line[{line_number}].fa_data.search_invoice",
message=f"La Factura '{fa_data.search_invoice}' de Importación {tipo_label_sg} no existe.",
solution=[f"Capturar un Número de Factura que exista en el Catálogo de Importaciones {tipo_label}."],
code="IMPORT_INVOICE_NOT_FOUND",
)
elif fa_data.search_line:
# Col. D + E: validar que la línea de importación exista dentro de esa factura
import_line: LineItem = (
db.query(LineItem)
.filter(
LineItem.invoice_id == import_invoice.id,
LineItem.line_number == fa_data.search_line,
LineItem.tenant_id == tenant_id,
LineItem.company_id == company_id,
)
.first()
)
if not import_line:
errors.add_error(
field=f"line[{line_number}].fa_data.search_line",
message=f"La Factura '{fa_data.search_invoice}' con línea '{fa_data.search_line}' de Importación {tipo_label_sg} no existe.",
solution=["Capturar un Número de Factura con diferente línea que esté en el Catálogo de Importaciones correspondiente."],
code="IMPORT_LINE_NOT_FOUND",
)
elif (
# Col. H: valida unidad de medida sólo cuando hay descarga
fa_data.download is True
and line.unit_of_measure
and import_line.unit_of_measure
and line.unit_of_measure != import_line.unit_of_measure
):
errors.add_error(
field=f"line[{line_number}].unit_of_measure",
message=(
f"La Unidad de Medida '{line.unit_of_measure}' es diferente de "
f"'{import_line.unit_of_measure}', que es la U.M. de la Factura "
f"'{fa_data.search_invoice}' con línea '{fa_data.search_line}' "
f"de Importación {tipo_label_sg}."
),
solution=["Capturar la Unidad de Medida correcta para el descargo de esta línea de Importación."],
code="UNIT_OF_MEASURE_MISMATCH",
)
country = None
fraction_type = None
sector = None
if fraction:
fraction = line.customs.fraction if line.customs.fraction else fraction
country = line.customs.origin_country
if line_item and line_item.customs:
country = (
line_item.customs.origin_country
if line_item.customs.origin_country
else country
)
fraction_type = line.customs.fraction_type.upper()
if line_item and line_item.customs:
fraction_type = (
line_item.customs.fraction_type
if line_item.customs.fraction_type
else fraction_type
)
sector = line.customs.sector
if line_item and line_item.customs:
sector = line_item.customs.sector if line_item.customs.sector else sector
country_m3 = db.query(Country.m3_key).filter(Country.m3_key == country).scalar()
if not country_m3:
country_m3 = (
db.query(Country.m3_key).filter(Country.ame_key == country).scalar()
)
country = country_m3
if not country:
errors.add_error(
field=f"line[{line_number}].customs.origin_country",
message="El país de origen especificado no existe.",
solution=["Proporciona un país de origen valido."],
code="ORIGIN_COUNTRY_NOT_FOUND",
)
else:
if fraction_type.strip().upper() not in vars(FractionType).values():
valid_types = [
v
for k, v in vars(FractionType).items()
if not k.startswith("_") and isinstance(v, str)
]
errors.add_error(
field=f"line[{line_number}].customs.fraction_type",
message="El tipo de fracción especificado no es válido.",
solution=[
f"Proporciona un tipo de fracción válido. Valores permitidos: {', '.join(valid_types)}"
],
code="FRACTION_TYPE_INVALID",
value=fraction_type,
)
else:
if fraction_type.strip().upper() == FractionType.PROSEC and not sector:
errors.add_error(
field=f"line[{line_number}].customs.sector",
message="El sector es obligatorio cuando el tipo de fracción es 'PROSEC'.",
solution=["Proporciona un sector valido."],
code="SECTOR_REQUIRED_FOR_PROSEC",
)
elif fraction_type.strip().upper() != FractionType.PROSEC and sector:
errors.add_error(
field=f"line[{line_number}].customs.sector",
message="El sector solo es aplicable cuando el tipo de fracción es 'PROSEC'.",
solution=[
"Elimina el sector o cambia el tipo de fracción a 'PROSEC'."
],
code="SECTOR_ONLY_FOR_PROSEC",
)
elif fraction_type.strip().upper() == FractionType.PROSEC and sector:
sector_db: Sector = (
db.query(Sector).filter(Sector.key == sector).scalar()
)
if sector_db:
errors.add_error(
field=f"line[{line_number}].customs.sector",
message="El sector especificado no existe.",
solution=["Proporciona un sector valido."],
code="SECTOR_NOT_FOUND",
)
else:
if not sector_db.authorized:
errors.add_error(
field=f"line[{line_number}].customs.sector",
message="El sector especificado no está autorizado.",
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."
],
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,
)
if line.customs.american_fraction:
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",
message="La fracción americana especificada no existe.",
solution=["Proporciona una fracción americana valida."],
code="AMERICAN_FRACTION_NOT_FOUND",
)
if line.order:
if len(line.order) > 20:
errors.add_error(
field=f"item.order",
message="El campo orden no debe exceder los 20 caracteres.",
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
)
#TODO: SSisGen Logic Restringer cantidades decimales para piezas, revisar si es necesario agregar validación similar para otras unidades de medida
#TODO: SSisGen Logic Seguridad
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()
if not valuation_method_exists:
errors.add_error(
field=f"line[{line_number}].valuation_method",
message="El método de valoración especificado no existe.",
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:
errors.add_error(
field=f"line[{line_number}].part_number_id",
message="El número de parte especificado no existe.",
solution=["Proporciona un número de parte valido."],
code="PART_NUMBER_NOT_FOUND",
)
apply_calculations(db, line, tenant_id, company_id, line_number)

394
backend/api/v1/modules/a76/items/exports/validators/create.py Normal file
View File

@@ -0,0 +1,394 @@
from decimal import Decimal
from sqlalchemy import func, exists
from sqlalchemy.orm import Session
from ...common.common_validators import count_items
from core.exceptions import ErrorCollector
from ...models import LineItem
from api.v1.modules.a24.fa.fa_item_lines.models import FaLineItem
from ...models import LineItem
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 api.v1.modules.public.reference_data.payment_methods.models import PaymentMethod
from .common import validate_common
def validate_create(
db: Session,
line: LineItem,
tenant_id: int,
company_id: int,
errors: ErrorCollector,
line_number: int,
):
"""
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)
"""
# Access fa_data safely
fa_data: FaLineItem = getattr(line, "fa_data", None)
# Required field validations
if not line.class_id:
errors.add_required_error(field=f"line[{line_number}].class_id")
if not line.quantity.quantity or line.quantity.quantity <= 0:
errors.add_required_error(field=f"line[{line_number}].quantity.quantity")
# TODO: Añadir validacion SSisGen:CalcularCostoUnitarioEnBaseAValorTotalScaf <-- de la tabla de preferencias de el sistema
# if SSisGen:CalcularCostoUnitarioEnBaseAValorTotalScaf == False:
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")
if not line.customs.origin_country:
errors.add_required_error(field=f"line[{line_number}].customs.origin_country")
if not line.customs.fraction_type:
errors.add_required_error(field=f"line[{line_number}].customs.fraction_type")
# FA-specific validations
if fa_data:
# Col. C: Procedencia de la Importación (TipoMovImpo) — obligatorio
if not fa_data.movement_type_import:
errors.add_required_error(field=f"line[{line_number}].fa_data.movement_type_import")
# Col. F: ¿Descarga la línea? (DescargaPartida) — obligatorio
if fa_data.download is None:
errors.add_required_error(field=f"line[{line_number}].fa_data.download")
# Col. D / E: Factura y Línea de Impo — obligatorios sólo si hay descarga
if fa_data.download is True:
if not fa_data.search_invoice:
errors.add_required_error(field=f"line[{line_number}].fa_data.search_invoice")
if not fa_data.search_line:
errors.add_required_error(field=f"line[{line_number}].fa_data.search_line")
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"
)
# 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.id == LineItem.id)
& (LineItem.invoice_id == line.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"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)
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 == line.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á
# Calcular valores totales basados en cantidad y costo unitario
quantity = line.quantity.quantity or Decimal("0")
# Valor Comercial
if line.financial.unit_cost_usd is not None:
line.financial.value_usd = line.financial.unit_cost_usd * quantity
if line.financial.unit_cost_mxn is not None:
line.financial.value_mxn = line.financial.unit_cost_mxn * quantity
# Valor Aduanas (asumiendo que es igual al Valor Comercial por defecto)
line.financial.customs_value_usd = line.financial.value_usd
line.financial.customs_value_mxn = line.financial.value_mxn
# Valor MP Temp (Materia Prima Temporal)
line.financial.value_temp_material_usd = line.financial.value_usd
line.financial.value_temp_material_mxn = line.financial.value_mxn
# ==========================================
# 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 == "24" #KGS
unit_is_lbs = line.unit_of_measure and line.unit_of_measure == "25" #LBS
# 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_id:
package: Package = (
db.query(Package)
.filter(
Package.id == line.quantity.package_id,
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_id:
package: Package = (
db.query(Package)
.filter(
Package.id == line.quantity.package_id,
Package.tenant_id == tenant_id,
Package.company_id == company_id,
)
.first()
)
if package:
line.description.package_description = package.description_es
else:
line.quantity.package_quantity = 0
line.quantity.package_id = None
line.description.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()
# ==========================================
# VALIDAR Y ASIGNAR PAGO DE IMPUESTO
# ==========================================
# Col. L: Se Pagó Impuesto — opcional, defaults a preferencia del sistema
if line.tax_payment is not None:
# Ya viene como bool desde Pydantic; valor válido por definición de tipo
pass
else:
# TODO: Asignar desde SisExp:PagoImpuesto (preferencias del sistema)
pass
# ==========================================
# VALIDAR Y ASIGNAR FORMA DE PAGO
# ==========================================
# Col. M: Forma de Pago — opcional, debe existir en catálogo si se proporciona
if line.payment_method:
payment_method_exists = (
db.query(PaymentMethod)
.filter(PaymentMethod.key == line.payment_method)
.first()
)
if not payment_method_exists:
errors.add_error(
field=f"line[{line_number}].payment_method",
message=f"La Forma de Pago '{line.payment_method}' no es válida.",
solution=[
"Capturar una Forma de Pago dentro del Catálogo General de Formas de Pago."
],
code="PAYMENT_METHOD_INVALID",
)
else:
# TODO: Asignar desde SisExp:FormaPago (preferencias del sistema)
pass
# ==========================================
# ASIGNAR MÉTODO DE VALORACIÓN POR DEFECTO
# ==========================================
# TODO: Si no se especificó método de valoración, tomar de preferencias del sistema (SisImp:MetValor)

251
backend/api/v1/modules/a76/items/exports/validators/update.py Normal file
View File

@@ -0,0 +1,251 @@
from decimal import Decimal
from sqlalchemy.orm import Session
from core.exceptions import ErrorCollector
from ...models import LineItem
from api.v1.modules.a24.fa.fa_item_lines.models import FaLineItem
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: LineItem,
existing_line: LineItem,
tenant_id: int,
company_id: int,
errors: ErrorCollector,
line_number: int,
):
"""
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, 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 == line.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
# Recalcular valores monetarios si el costo o la cantidad cambian
currency_type = invoice.financials.currency_type
unit_cost_capture = line.financial.unit_cost_capture or Decimal("0")
if currency_type in ["USD", "ME"]:
line.financial.unit_cost_usd = unit_cost_capture
line.financial.unit_cost_mxn = unit_cost_capture * exchange_rate
elif currency_type in ["MXN", "MN"]:
line.financial.unit_cost_usd = (unit_cost_capture / exchange_rate) if exchange_rate else Decimal("0")
line.financial.unit_cost_mxn = unit_cost_capture
quantity = line.quantity.quantity if line.quantity.quantity is not None else existing_line.quantity.quantity
if line.financial.unit_cost_usd is not None:
line.financial.value_usd = line.financial.unit_cost_usd * quantity
if line.financial.unit_cost_mxn is not None:
line.financial.value_mxn = line.financial.unit_cost_mxn * quantity
line.financial.customs_value_usd = line.financial.value_usd
line.financial.customs_value_mxn = line.financial.value_mxn
line.financial.value_temp_material_usd = line.financial.value_usd
line.financial.value_temp_material_mxn = line.financial.value_mxn
# 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
print(f"After weight conversion: net_weight={line.quantity.net_weight}, gross_weight={line.quantity.gross_weight}, weight_type={invoice_weight_type}")
# 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_id:
line.quantity.package_id = existing_line.quantity.package_id
# 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.order:
line.order = existing_line.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
# ==========================================
# DATOS FA (Activo Fijo) — ACTUALIZACIÓN PARCIAL
# ==========================================
fa_data: FaLineItem = getattr(line, "fa_data", None)
existing_fa_data: FaLineItem = getattr(existing_line, "fa_data", None)
if fa_data is not None and existing_fa_data is not None:
# Col. C: Procedencia de la Importación (TipoMovImpo)
if not fa_data.movement_type_import:
fa_data.movement_type_import = existing_fa_data.movement_type_import
# Col. F: ¿Descarga la línea? (Descarga)
if fa_data.download is None:
fa_data.download = existing_fa_data.download
# Col. D: Factura de Importación — obligatoria sólo si hay descarga
if not fa_data.search_invoice:
fa_data.search_invoice = existing_fa_data.search_invoice
# Col. E: Línea de Importación — obligatoria sólo si hay descarga
if fa_data.search_line is None:
fa_data.search_line = existing_fa_data.search_line
# Subpartidas (EsSubPartida / SubPartida)
if fa_data.is_subitem is None:
fa_data.is_subitem = existing_fa_data.is_subitem
if fa_data.subitem_number is None:
fa_data.subitem_number = existing_fa_data.subitem_number
# Número de parte
if not line.part_number_id:
line.part_number_id = existing_line.part_number_id
# 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

14
backend/api/v1/modules/a76/items/imports/temporary/validators/calculations.py → backend/api/v1/modules/a76/items/imports/validators/calculations.py
View File

@@ -2,18 +2,8 @@ from sqlalchemy.orm import Session
from api.v1.modules.a76.invoices.models import InvoiceFinancials, InvoiceHeader, InvoiceLogistics
from core.exceptions import ErrorCollector
from ....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 LineItem
from ...models import LineItem
from ...models import LineItem
from api.v1.modules.a76.classes.models import Class

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

@@ -1,13 +1,14 @@
from sqlalchemy import exists
from sqlalchemy.orm import Session
from api.v1.modules.a76.invoices.common.common_validators import invoice_exists_by_id
from api.v1.modules.a76.items.imports.validators.calculations import apply_calculations
from core.exceptions import ErrorCollector
from sqlalchemy import func
from ....common.fractions import search_fraction_preference
from ....common.common_validators import item_exists
from ....models import LineItem
from ....line_customs.models import FractionType, LineCustom
from ...common.fractions import search_fraction_preference
from ...common.common_validators import item_exists
from ...models import LineItem
from ...line_customs.models import FractionType, LineCustom
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
@@ -304,6 +305,10 @@ def validate_common(
unit_of_measure = line.unit_of_measure or (
class_.unit_of_measure if class_ else None
)
#TODO: SSisGen Logic Restringer cantidades decimales para piezas, revisar si es necesario agregar validación similar para otras unidades de medida
#TODO: SSisGen Logic Seguridad
if unit_of_measure == "PZA" and line.quantity.quantity % 1 != 0:
errors.add_error(
field=f"line[{line_number}].quantity.quantity",
@@ -333,3 +338,5 @@ def validate_common(
solution=["Proporciona un número de parte valido."],
code="PART_NUMBER_NOT_FOUND",
)
apply_calculations(db, line, tenant_id, company_id, line_number)

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

@@ -1,21 +1,12 @@
from decimal import Decimal
from sqlalchemy import func, exists
from sqlalchemy.orm import Session
from ....common.common_validators import count_items
from ...common.common_validators import count_items
from core.exceptions import ErrorCollector
from ....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 ...models import LineItem
from api.v1.modules.a24.fa.fa_item_lines.models import FaLineItem
from ....models import LineItem
from ...models import LineItem
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
@@ -27,7 +18,7 @@ from .common import validate_common
def validate_create(
db: Session,
line: LineItem, # LineItemCreate schema (Pydantic)
line: LineItem,
tenant_id: int,
company_id: int,
errors: ErrorCollector,
@@ -44,7 +35,7 @@ def validate_create(
"""
# Access fa_data safely
fa_data = getattr(line, "fa_data", None)
fa_data: FaLineItem = getattr(line, "fa_data", None)
# Required field validations
if not line.class_id:

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

@@ -1,9 +1,8 @@
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 ....models import LineItem
from ...models import LineItem
from api.v1.modules.a76.invoices.models import InvoiceHeader
from api.v1.modules.a76.general_catalogs.fractions.us_tariff_fractions.models import (
USTariffFraction,

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

@@ -24,8 +24,10 @@ from api.v1.modules.a76.invoices.common.common_validators import (
invoice_updated,
)
from core.exceptions import ErrorCollector
from .imports.temporary.validators.create import validate_create
from .imports.temporary.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 .schemas import LineItemCreate, LineItemUpdate
from .line_financials.models import LineFinancial
@@ -333,14 +335,16 @@ 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")
errors.raise_if_errors("Error al crear el item")
errors.raise_if_errors("Error al crear el item - invoice_id es requerido")
if not invoice_exists_by_id(
invoice = invoice_exists_by_id(
db, item_data.invoice_id, tenant_id, company_id, errors
):
errors.raise_if_errors("Error al crear el item")
)
if not invoice:
errors.raise_if_errors("Error al encontra la factura para 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")
errors.raise_if_errors("Error al crear el item - la factura ya fue actualizada, no se pueden agregar items")
# Lock invoice and calculate line number
if not ItemService._lock_invoice(
@@ -370,14 +374,27 @@ class ItemService:
item_data.component_part_number_id = resolved_id
# Validar el item
validate_create(
db,
item_data, # Schema Pydantic completo
tenant_id,
company_id,
errors,
line_number,
)
if invoice.operation_type == "exp":
if invoice.invoice_type == "CR" and invoice.document_type == "AFIJO":
errors.add_error("invoice_id", "No se pueden agregar items a una factura de tipo CR con documento AFIJO", code="INVALID_INVOICE_TYPE")
validate_create_export(
db,
item_data, # Schema Pydantic completo
tenant_id,
company_id,
errors,
line_number,
)
else:
validate_create_import(
db,
item_data, # Schema Pydantic completo
tenant_id,
company_id,
errors,
line_number,
)
# Validaciones adicionales específicas del negocio
if item_data.fa_data and item_data.fa_data.is_subitem is None:
@@ -458,6 +475,12 @@ class ItemService:
# Validaciones con ErrorCollector
errors = ErrorCollector()
invoice = invoice_exists_by_id(
db, item_data.invoice_id, tenant_id, company_id, errors
)
if not invoice:
errors.raise_if_errors("Error al encontra la factura para el item")
# Lock invoice
invoice_id_to_lock = (
item_data.invoice_id if item_data.invoice_id else db_item.invoice_id
@@ -465,7 +488,7 @@ class ItemService:
if not ItemService._lock_invoice(
db, invoice_id_to_lock, tenant_id, company_id, errors
):
errors.raise_if_errors("Error al actualizar el item")
errors.raise_if_errors("Error al actualizar el item")
# Resolve part ID if a string is provided in part_number (alias for part_number_id)
if hasattr(item_data, 'part_number_id') and item_data.part_number_id and not isinstance(item_data.part_number_id, int):
@@ -485,17 +508,28 @@ class ItemService:
if resolved_id:
item_data.component_part_number_id = resolved_id
# Validar el item que se va a actualizar
validate_update(
db,
item_data, # Schema de update
db_item, # LineItem existente en DB
tenant_id,
company_id,
errors,
db_item.line_number,
)
if invoice.operation_type == "exp":
# Validar el item que se va a actualizar
validate_update_export(
db,
item_data, # Schema de update
db_item, # LineItem existente en DB
tenant_id,
company_id,
errors,
db_item.line_number,
)
else:
# Validar el item que se va a actualizar
validate_update_import(
db,
item_data, # Schema de update
db_item, # LineItem existente en DB
tenant_id,
company_id,
errors,
db_item.line_number,
)
# Validar tipo de partida
if hasattr(item_data, "item_type") and item_data.item_type:

877
backend/api/v1/modules/a76/reports/movements/invoices/services/query_builders.py.bak
View File

@@ -1,877 +0,0 @@
"""
SQL Query builders for invoice movement services.
Centralizes all SQL query construction logic.
"""
class TemporaryImportQueries:
"""SQL queries for temporary imports using PostgreSQL tables."""
@staticmethod
def build_aggregated_query(db_name: str, where_str: str) -> str:
"""Build optimized query for NORMAL mode (grouped by invoice with totals)."""
# Note: db_name parameter kept for compatibility but not used in PostgreSQL
return f"""
SELECT
ih.invoice_number AS C1,
COALESCE(ped.pedimento_number, '') AS C2,
TO_CHAR(ih.invoice_date, 'YYYYMMDD') AS C3,
CASE WHEN ih.is_updated THEN 'AC' ELSE 'NA' END AS C4,
COALESCE(ped.pedimento_code, '') AS C5,
COALESCE(ped.regime, '') AS C10,
COALESCE(TO_CHAR(log.entry_exit_date, 'YYYYMMDD'), '') AS C11,
COALESCE(TO_CHAR(log.delivery_date, 'YYYYMMDD'), '') AS C12,
COALESCE(TO_CHAR(log.payment_date, 'YYYYMMDD'), '') AS C13,
COALESCE(cmp.remesa, 0) AS C14,
COALESCE(fin.exchange_rate, 0) AS C15,
COALESCE(cmp.provider_id::text, '') AS C16,
COALESCE(cmp.sold_to_id::text, '') AS C17,
COALESCE(cmp.customs_broker_id::text, '') AS C18,
COALESCE(cmp.aduana, '') AS C38,
ih.id AS C39,
COALESCE(ped_r1.pedimento_number, '') AS C41,
COALESCE(cmp.edocument, '') AS C42,
COALESCE(cmp.vucem_operation_num, '') AS C43,
COALESCE(fin.exchange_rate, 0) AS C50,
COALESCE(TO_CHAR(ih.emission_date, 'YYYYMMDD'), '') AS C51,
COALESCE(ih.capture_user, '') AS C52,
COALESCE(ih.who_updated, '') AS C53,
COALESCE(log.carrier_id, '') AS C54,
COALESCE(log.transport_num || ' ' || log.license_plate, '') AS C55,
'' AS C56,
'' AS C57,
'' AS C58,
COALESCE(SUM(CASE WHEN COALESCE(fil.is_subitem, false) THEN 0 ELSE lf.value_usd END), 0) AS total_me,
COALESCE(SUM(CASE WHEN COALESCE(fil.is_subitem, false) THEN 0 ELSE lf.value_mxn END), 0) AS total_mn
FROM a76.invoice_header ih
LEFT JOIN a76.invoice_compliance_mx cmp ON cmp.invoice_id = ih.id
LEFT JOIN a76.invoice_financials fin ON fin.invoice_id = ih.id
LEFT JOIN a76.invoice_logistics log ON log.invoice_id = ih.id
LEFT JOIN a76.pedimentos ped ON ped.id = cmp.pedimento_id
LEFT JOIN a76.pedimentos ped_r1 ON ped_r1.id = cmp.pedimento_r1
LEFT JOIN a76.items i ON i.invoice_id = ih.id
LEFT JOIN a76.item_lines il ON il.item_id = i.id
LEFT JOIN a24.fa_item_lines fil ON fil.id = il.id
LEFT JOIN a76.item_line_financials lf ON lf.item_line_id = il.id
WHERE ih.operation_type = 'imp'
AND ih.invoice_type = 'TEM'
AND {where_str}
GROUP BY ih.id, ih.invoice_number, ped.pedimento_number, ped.pedimento_code, ped.regime,
log.entry_exit_date, log.delivery_date, log.payment_date, cmp.remesa, fin.exchange_rate,
cmp.provider_id, cmp.sold_to_id, cmp.customs_broker_id, cmp.aduana, ped_r1.pedimento_number,
cmp.edocument, cmp.vucem_operation_num, ih.emission_date, ih.capture_user, ih.who_updated,
log.carrier_id, log.transport_num, log.license_plate
ORDER BY ih.invoice_number
"""
@staticmethod
def build_main_query(db_name: str, where_str: str) -> str:
"""Build main SQL query for DETAILED mode (all partidas) from PostgreSQL."""
# Note: db_name parameter is kept for compatibility but not used in PostgreSQL
return f"""
SELECT
ih.invoice_number AS C1,
COALESCE(ped.pedimento_number, '') AS C2,
TO_CHAR(ih.invoice_date, 'YYYYMMDD') AS C3,
CASE WHEN ih.is_updated THEN 'AC' ELSE 'NA' END AS C4,
COALESCE(ped.pedimento_code, '') AS C5,
COALESCE(fin.value_me, 0) AS C6,
COALESCE(fin.value_mn, 0) AS C7,
COALESCE(cmp.provider_id::text, '') AS C8,
COALESCE(cmp.sold_to_id::text, '') AS C9,
COALESCE(ped.regime, '') AS C10,
COALESCE(TO_CHAR(log.entry_exit_date, 'YYYYMMDD'), '') AS C11,
COALESCE(TO_CHAR(log.delivery_date, 'YYYYMMDD'), '') AS C12,
COALESCE(TO_CHAR(log.payment_date, 'YYYYMMDD'), '') AS C13,
COALESCE(cmp.remesa, 0) AS C14,
COALESCE(fin.exchange_rate, 0) AS C15,
COALESCE(cmp.provider_id::text, '') AS C16,
COALESCE(cmp.sold_to_id::text, '') AS C17,
COALESCE(cmp.customs_broker_id::text, '') AS C18,
'' AS C19,
COALESCE(il.class_id::text, '') AS C20,
REPLACE(REPLACE(REPLACE(REPLACE(COALESCE(ld.description_spanish, ''), CHR(44), ' '), CHR(9), ' '), CHR(10), ' '), CHR(13), ' ') AS C21,
REPLACE(REPLACE(REPLACE(REPLACE(COALESCE(ld.description_english, ''), CHR(44), ' '), CHR(9), ' '), CHR(10), ' '), CHR(13), ' ') AS C22,
COALESCE(lq.quantity, 0) AS C23,
COALESCE(il.unit_of_measure::text, '') AS C24,
COALESCE(lf.value_mxn, 0) AS C25,
COALESCE(lf.customs_value_mxn, 0) AS C26,
COALESCE(lf.value_usd, 0) AS C27,
COALESCE(lf.customs_value_usd, 0) AS C28,
COALESCE(lq.net_weight, 0) AS C29,
COALESCE(lq.gross_weight, 0) AS C30,
COALESCE(ih.purchase_order, '') AS C31,
COALESCE(lc.fraction, '') AS C32,
COALESCE(lc.fraction_type, '') AS C33,
COALESCE(lc.advalorem_numeric, 0) AS C34,
COALESCE(lc.sector, '') AS C35,
COALESCE(lf.igi_amount_usd, 0) AS C36,
COALESCE(lc.origin_country, '') AS C37,
COALESCE(cmp.aduana, '') AS C38,
ih.id AS C39,
FALSE AS C40,
'' AS C41,
COALESCE(cmp.edocument, '') AS C42,
COALESCE(cmp.vucem_operation_num, '') AS C43,
COALESCE(il.line_number, 0) AS C44,
'' AS C45,
'' AS C46,
COALESCE(cls.us_fraction, '') AS C47,
COALESCE(prt.eccn, '') AS C48,
COALESCE(il.part_number::text, '') AS C49,
COALESCE(fin.exchange_rate, 0) AS C50,
COALESCE(TO_CHAR(ih.emission_date, 'YYYYMMDD'), '') AS C51,
COALESCE(ih.capture_user, '') AS C52,
COALESCE(ih.who_updated, '') AS C53,
COALESCE(log.carrier_id, '') AS C54,
COALESCE(log.transport_num || ' ' || log.license_plate, '') AS C55,
'' AS C56,
'' AS C57,
'' AS C58
FROM a76.invoice_header ih
LEFT JOIN a76.invoice_compliance_mx cmp ON cmp.invoice_id = ih.id
LEFT JOIN a76.invoice_financials fin ON fin.invoice_id = ih.id
LEFT JOIN a76.invoice_logistics log ON log.invoice_id = ih.id
LEFT JOIN a76.pedimentos ped ON ped.id = cmp.pedimento_id
LEFT JOIN a76.item_lines il ON il.item_id = (
SELECT id FROM a76.items WHERE invoice_id = ih.id LIMIT 1
)
LEFT JOIN a76.item_line_descriptions ld ON ld.item_line_id = il.id
LEFT JOIN a76.item_line_quantities lq ON lq.item_line_id = il.id
LEFT JOIN a76.item_line_financials lf ON lf.item_line_id = il.id
LEFT JOIN a76.item_line_customs lc ON lc.item_line_id = il.id
LEFT JOIN a76.classes cls ON cls.id = il.class_id
LEFT JOIN a76.parts prt ON prt.id = il.part_number
LEFT JOIN a76.units_of_measure um ON um.id = il.unit_of_measure
WHERE ih.operation_type = 'imp'
AND ih.invoice_type = 'TEM'
AND {where_str}
"""
@staticmethod
def build_totals_query(db_name: str) -> str:
"""Build query to get totals for an invoice."""
return f"""
SELECT
COALESCE(SUM(EqiPim.ValorImpoME), 0),
COALESCE(SUM(EqiPim.ValorImpoMN), 0)
FROM [{db_name}].dbo.QEqiMaq EqiPim
WHERE EqiPim.Consecutivo = :consecutivo
AND EqiPim.EsSubpartida = 'P'
"""
@staticmethod
def build_series_query(db_name: str) -> str:
"""Build query to get series information."""
return f"""
SELECT SerieImpo, ModeloImpo, ParteImpo
FROM [{db_name}].dbo.QSeriesImpo
WHERE Consecutivo = :consecutivo
AND LineaImpo = :linea
ORDER BY RenImpo
"""
@staticmethod
def build_driver_badge_query(db_name: str) -> str:
"""Build query to get driver badge number."""
return f"""
SELECT TOP 1 NUMGAFETEUNICO
FROM [{db_name}].dbo.GConductor
LEFT JOIN [{db_name}].dbo.QFacImp
ON QFacImp.CONDUCTOR = GConductor.CONDUCTOR
WHERE FacturaImpo = :factura
"""
class DefinitiveImportQueries:
"""SQL queries for definitive imports (PostgreSQL schema)."""
@staticmethod
def build_aggregated_query(db_name: str, where_clause: str) -> str:
"""Build optimized query for NORMAL mode (grouped by invoice with totals)."""
return f"""
SELECT
ih.invoice_number AS C1,
COALESCE(ped.pedimento_number, '') AS C2,
TO_CHAR(ih.invoice_date, 'YYYYMMDD') AS C3,
CASE WHEN ih.is_updated THEN 'AC' ELSE 'NA' END AS C4,
COALESCE(ped.pedimento_code, '') AS C5,
COALESCE(ped.regime, '') AS C10,
COALESCE(TO_CHAR(log.entry_exit_date, 'YYYYMMDD'), '') AS C11,
COALESCE(TO_CHAR(log.delivery_date, 'YYYYMMDD'), '') AS C12,
COALESCE(TO_CHAR(log.payment_date, 'YYYYMMDD'), '') AS C13,
COALESCE(log.payment_receipt_num, '') AS C14,
COALESCE(fin.exchange_rate, 0) AS C15,
COALESCE(cmp.provider_id::text, '') AS C16,
COALESCE(cmp.sold_to_id::text, '') AS C17,
COALESCE(cmp.customs_broker_id::text, '') AS C18,
COALESCE(ih.purchase_order, '') AS C31,
COALESCE(cmp.aduana, '') AS C39,
ih.id AS C35,
COALESCE(ped_r1.pedimento_number, '') AS C42,
COALESCE(cmp.edocument, '') AS C43,
COALESCE(cmp.vucem_operation_num, '') AS C44,
COALESCE(fin.exchange_rate, 0) AS C51,
COALESCE(TO_CHAR(ih.emission_date, 'YYYYMMDD'), '') AS C52,
COALESCE(ih.capture_user, '') AS C53,
COALESCE(ih.who_updated, '') AS C54,
COALESCE(log.transport_id || ' ' || log.transport_num, '') AS C56,
'' AS C57,
'' AS C58,
'' AS C59,
COALESCE(SUM(CASE WHEN COALESCE(fil.is_subitem, false) THEN 0 ELSE lf.value_usd END), 0) AS total_me,
COALESCE(SUM(CASE WHEN COALESCE(fil.is_subitem, false) THEN 0 ELSE lf.value_mxn END), 0) AS total_mn
FROM a76.invoice_header ih
LEFT JOIN a76.invoice_compliance_mx cmp ON cmp.invoice_id = ih.id
LEFT JOIN a76.invoice_financials fin ON fin.invoice_id = ih.id
LEFT JOIN a76.invoice_logistics log ON log.invoice_id = ih.id
LEFT JOIN a76.pedimentos ped ON ped.id = cmp.pedimento_id
LEFT JOIN a76.pedimentos ped_r1 ON ped_r1.id = cmp.pedimento_r1
LEFT JOIN a76.items i ON i.invoice_id = ih.id
LEFT JOIN a76.item_lines il ON il.item_id = i.id
LEFT JOIN a24.fa_item_lines fil ON fil.id = il.id
LEFT JOIN a76.item_line_financials lf ON lf.item_line_id = il.id
WHERE ih.operation_type = 'imp'
AND ih.invoice_type IN ('DEF', 'EXDEF', 'MATDE')
AND {where_clause}
GROUP BY ih.id, ih.invoice_number, ped.pedimento_number, ped.pedimento_code, ped.regime,
log.entry_exit_date, log.delivery_date, log.payment_date, log.payment_receipt_num,
fin.exchange_rate, cmp.provider_id, cmp.sold_to_id, cmp.customs_broker_id,
ih.purchase_order, cmp.aduana, ped_r1.pedimento_number, cmp.edocument,
cmp.vucem_operation_num, ih.emission_date, ih.capture_user, ih.who_updated,
log.transport_id, log.transport_num
ORDER BY ih.invoice_number
"""
@staticmethod
def build_main_query(db_name: str, where_clause: str) -> str:
return f"""
SELECT
ih.invoice_number AS C1, -- [0]
ped.pedimento_number AS C2, -- [1]
ih.invoice_date AS C3, -- [2]
ped.status AS C4, -- [3]
ped.pedimento_code AS C5, -- [4]
'' AS C6, '' AS C7, '' AS C8, '' AS C9, -- [5-8]
ped.regime AS C10, -- [9]
log.entry_exit_date AS C11, -- [10]
log.delivery_date AS C12, -- [11]
log.payment_date AS C13, -- [12]
log.payment_receipt_num AS C14, -- [13]
'' AS C15, -- [14]
cmp.provider_id AS C16, -- [15]
cmp.sold_to_id AS C17, -- [16]
cmp.customs_broker_id AS C18, -- [17]
'' AS C19, -- [18]
prt.part_number AS C20, -- [19]
ld.description_spanish AS C21, -- [20]
ld.description_english AS C22, -- [21]
lq.quantity AS C23, -- [22]
um.code AS C24, -- [23]
lf.value_mxn AS C25, -- [24]
'' AS C26, -- [25]
lf.value_usd AS C27, -- [26]
'' AS C28, -- [27]
lq.net_weight AS C29, -- [28]
lq.gross_weight AS C30, -- [29]
ih.purchase_order AS C31, -- [30]
lc.fraction AS C32, -- [31]
'' AS C33, '' AS C34, -- [32-33]
ih.id AS C35, -- [34]
'' AS C36, '' AS C37, -- [35-36]
lc.origin_country AS C38, -- [37]
cmp.aduana AS C39, -- [38]
il.material_type AS C40, -- [39]
il.id AS C41, -- [40]
'' AS C42, -- [41] rectification_id
cmp.edocument AS C43, -- [42]
cmp.vucem_operation_num AS C44, -- [43]
il.line_number AS C45, -- [44]
ld.brand AS C46, -- [45]
ld.model AS C47, -- [46]
prt.us_fraction AS C48, -- [47]
prt.eccn AS C49, -- [48]
prt.id AS C50, -- [49]
fin.exchange_rate AS C51, -- [50]
ih.emission_date AS C52, -- [51]
ih.capture_user AS C53, -- [52]
ih.who_updated AS C54, -- [53]
'' AS C55, -- [54]
COALESCE(log.transport_id || ' ' || log.transport_num, '') AS C56, -- [55]
'' AS C57, -- [56] Pedimento18 (row[56])
COALESCE(ld.lot, '') AS C58, -- [57] Lote (row[57])
'' AS C59, -- [58] TipoPed (row[58])
'' AS C60 -- [59] Relleno final
FROM a76.invoice_header ih
LEFT JOIN a76.invoice_compliance_mx cmp ON cmp.invoice_id = ih.id
LEFT JOIN a76.invoice_financials fin ON fin.invoice_id = ih.id
LEFT JOIN a76.invoice_logistics log ON log.invoice_id = ih.id
LEFT JOIN a76.pedimentos ped ON ped.id = cmp.pedimento_id
LEFT JOIN a76.items itm ON itm.invoice_id = ih.id
LEFT JOIN a76.item_lines il ON il.item_id = itm.id
LEFT JOIN a76.item_line_descriptions ld ON ld.item_line_id = il.id
LEFT JOIN a76.item_line_quantities lq ON lq.item_line_id = il.id
LEFT JOIN a76.item_line_financials lf ON lf.item_line_id = il.id
LEFT JOIN a76.item_line_customs lc ON lc.item_line_id = il.id
LEFT JOIN a76.classes cls ON cls.id = il.class_id
LEFT JOIN a76.parts prt ON prt.id = il.part_number
LEFT JOIN a76.units_of_measure um ON um.id = il.unit_of_measure
WHERE {where_clause}
ORDER BY ih.invoice_number, il.line_number
"""
@staticmethod
def build_totals_query(db_name: str) -> str:
"""Build query to get totals for a definitive import invoice."""
return f"""
SELECT
COALESCE(SUM(CASE WHEN COALESCE(fil.is_subitem, false) THEN 0 ELSE lf.value_usd END), 0),
COALESCE(SUM(CASE WHEN COALESCE(fil.is_subitem, false) THEN 0 ELSE lf.value_mxn END), 0)
FROM a76.item_line_financials lf
INNER JOIN a76.item_lines il ON il.id = lf.item_line_id
LEFT JOIN a24.fa_item_lines fil ON fil.id = il.id
INNER JOIN a76.items itm ON itm.id = il.item_id
WHERE itm.invoice_id = :consecutivo
"""
@staticmethod
def build_series_query(db_name: str) -> str:
"""Build query to get series information for definitive imports."""
# TODO: QSeriesDef table not migrated to PostgreSQL yet
return """
SELECT '' as serie, '' as modelo, '' as parte
WHERE 1=0
"""
@staticmethod
def build_driver_badge_query(db_name: str) -> str:
"""Build query to get driver badge number for definitive imports."""
# TODO: GConductor table not migrated to PostgreSQL yet
return """
SELECT '' as badge
WHERE 1=0
"""
class RepairImportQueries:
"""SQL queries for repair imports (PostgreSQL schema)."""
@staticmethod
def build_aggregated_query(db_name: str, where_str: str, discharge_clause: str = "") -> str:
"""Build optimized query for NORMAL mode (grouped by invoice with totals)."""
discharge_filter = "" # Temporarily disabled until schema migration
return f"""
SELECT
ih.invoice_number AS C2,
COALESCE(ped.pedimento_number, '') AS C3,
TO_CHAR(ih.invoice_date, 'YYYYMMDD') AS C4,
CASE WHEN ih.is_updated THEN 'AC' ELSE 'NA' END AS C5,
COALESCE(ped.pedimento_code, '') AS C6,
COALESCE(ped.regime, '') AS C7,
COALESCE(TO_CHAR(log.payment_date, 'YYYYMMDD'), '') AS C9,
COALESCE(cmp.remesa::text, '') AS C10,
COALESCE(fin.exchange_rate, 0) AS C11,
COALESCE(cmp.provider_id::text, '') AS C12,
COALESCE(cmp.sold_to_id::text, '') AS C13,
COALESCE(cmp.customs_broker_id::text, '') AS C14,
COALESCE(ih.purchase_order, '') AS C24,
COALESCE(ped.customs_office, '') AS C29,
ih.id AS C30,
COALESCE(cmp.edocument, '') AS C33,
COALESCE(cmp.vucem_operation_num, '') AS C34,
COALESCE(fin.exchange_rate, 0) AS C40,
COALESCE(TO_CHAR(ih.emission_date, 'YYYYMMDD'), '') AS C41,
COALESCE(ih.capture_user, '') AS C42,
COALESCE(ih.who_updated, '') AS C43,
COALESCE(log.carrier_id, '') AS C44,
COALESCE(log.transport_num, '') AS C45,
COALESCE(ped.pedimento_code, '') AS C47,
COALESCE(SUM(CASE WHEN COALESCE(fil.is_subitem, false) THEN 0 ELSE lf.value_usd END), 0) AS total_me,
COALESCE(SUM(CASE WHEN COALESCE(fil.is_subitem, false) THEN 0 ELSE lf.value_mxn END), 0) AS total_mn
FROM a76.invoice_header ih
LEFT JOIN a76.invoice_compliance_mx cmp ON cmp.invoice_id = ih.id
LEFT JOIN a76.invoice_financials fin ON fin.invoice_id = ih.id
LEFT JOIN a76.invoice_logistics log ON log.invoice_id = ih.id
LEFT JOIN a76.pedimentos ped ON ped.id = cmp.pedimento_id
LEFT JOIN a76.items i ON i.invoice_id = ih.id
LEFT JOIN a76.item_lines il ON il.item_id = i.id
LEFT JOIN a24.fa_item_lines fil ON fil.id = il.id
LEFT JOIN a76.item_line_financials lf ON lf.item_line_id = il.id
WHERE ih.operation_type = 'imp'
AND ih.invoice_type = 'REP'
AND COALESCE(cmp.is_regime_change, false) = false
{"AND " + where_str if where_str else ""}
{discharge_filter}
GROUP BY ih.id, ih.invoice_number, ped.pedimento_number, ped.pedimento_code, ped.regime,
log.payment_date, cmp.remesa, fin.exchange_rate, cmp.provider_id, cmp.sold_to_id,
cmp.customs_broker_id, ih.purchase_order, ped.customs_office, cmp.edocument,
cmp.vucem_operation_num, ih.emission_date, ih.capture_user, ih.who_updated,
log.carrier_id, log.transport_num
ORDER BY ih.invoice_number
"""
@staticmethod
def build_main_query(db_name: str, where_str: str, discharge_clause: str = "") -> str:
"""Build main SQL query for repair import data."""
# Note: is_discharged field not yet migrated to PostgreSQL schema
# discharge_filter = "AND il.is_discharged = true" if "descargado" in discharge_clause.lower() else ""
discharge_filter = "" # Temporarily disabled until schema migration
return f"""
SELECT
il.line_number,
ih.invoice_number,
COALESCE(ped.pedimento_number, ''),
TO_CHAR(ih.invoice_date, 'YYYYMMDD'),
CASE WHEN ih.is_updated THEN 'AC' ELSE 'NA' END,
COALESCE(ped.pedimento_code, ''),
COALESCE(ped.regime, ''),
'',
COALESCE(TO_CHAR(log.payment_date, 'YYYYMMDD'), ''),
COALESCE(cmp.remesa::text, ''),
COALESCE(fin.exchange_rate, 0),
COALESCE(cmp.provider_id::text, ''),
COALESCE(cmp.sold_to_id::text, ''),
COALESCE(cmp.customs_broker_id::text, ''),
COALESCE(il.part_number::text, ''),
REPLACE(REPLACE(REPLACE(REPLACE(COALESCE(ld.description_spanish, ''), CHR(44), ' '), CHR(9), ' '), CHR(10), ' '), CHR(13), ' '),
REPLACE(REPLACE(REPLACE(REPLACE(COALESCE(ld.description_english, ''), CHR(44), ' '), CHR(9), ' '), CHR(10), ' '), CHR(13), ' '),
COALESCE(lq.quantity, 0),
COALESCE(il.unit_of_measure, 0),
COALESCE(lf.value_mxn, 0),
COALESCE(lf.value_usd, 0),
COALESCE(lq.net_weight, 0),
COALESCE(lq.gross_weight, 0),
COALESCE(ih.purchase_order, ''),
COALESCE(lc.fraction, ''),
'',
COALESCE(lc.sector, ''),
COALESCE(lc.origin_country, ''),
COALESCE(ped.customs_office, ''),
ih.id,
'P',
'',
COALESCE(cmp.edocument, ''),
COALESCE(cmp.vucem_operation_num, ''),
REPLACE(REPLACE(REPLACE(REPLACE(COALESCE(ld.brand, ''), CHR(44), ' '), CHR(9), ' '), CHR(10), ' '), CHR(13), ' '),
REPLACE(REPLACE(REPLACE(REPLACE(COALESCE(ld.model, ''), CHR(44), ' '), CHR(9), ' '), CHR(10), ' '), CHR(13), ' '),
COALESCE(lc.american_fraction, ''),
COALESCE(prt.eccn, ''),
COALESCE(fin.exchange_rate, 0),
COALESCE(TO_CHAR(ih.emission_date, 'YYYYMMDD'), ''),
COALESCE(ih.capture_user, ''),
COALESCE(ih.who_updated, ''),
COALESCE(log.carrier_id, ''),
COALESCE(log.transport_num, ''),
'',
COALESCE(ped.pedimento_code, '')
FROM a76.invoice_header ih
LEFT JOIN a76.invoice_compliance_mx cmp ON cmp.invoice_id = ih.id
LEFT JOIN a76.invoice_financials fin ON fin.invoice_id = ih.id
LEFT JOIN a76.invoice_logistics log ON log.invoice_id = ih.id
LEFT JOIN a76.pedimentos ped ON ped.id = cmp.pedimento_id
LEFT JOIN a76.items itm ON itm.invoice_id = ih.id
LEFT JOIN a76.item_lines il ON il.item_id = itm.id
LEFT JOIN a76.item_line_descriptions ld ON ld.item_line_id = il.id
LEFT JOIN a76.item_line_quantities lq ON lq.item_line_id = il.id
LEFT JOIN a76.item_line_financials lf ON lf.item_line_id = il.id
LEFT JOIN a76.item_line_customs lc ON lc.item_line_id = il.id
LEFT JOIN a76.parts prt ON prt.id = il.part_number
LEFT JOIN a76.units_of_measure um ON um.id = il.unit_of_measure
WHERE ih.operation_type = 'imp'
AND ih.invoice_type = 'REP'
AND COALESCE(cmp.is_regime_change, false) = false
{"AND " + where_str if where_str else ""}
{discharge_filter}
ORDER BY ih.invoice_number, il.line_number
"""
@staticmethod
def build_totals_query(db_name: str, discharge_clause: str = "") -> str:
"""Build query to get totals for a repair import invoice."""
# Note: is_discharged field not yet migrated to PostgreSQL schema
# discharge_filter = "AND il.is_discharged = true" if "descargado" in discharge_clause.lower() else ""
discharge_filter = "" # Temporarily disabled until schema migration
return f"""
SELECT
COALESCE(SUM(CASE WHEN COALESCE(fil.is_subitem, false) THEN 0 ELSE lf.value_usd END), 0),
COALESCE(SUM(CASE WHEN COALESCE(fil.is_subitem, false) THEN 0 ELSE lf.value_mxn END), 0)
FROM a76.item_line_financials lf
INNER JOIN a76.item_lines il ON il.id = lf.item_line_id
LEFT JOIN a24.fa_item_lines fil ON fil.id = il.id
INNER JOIN a76.items itm ON itm.id = il.item_id
WHERE itm.invoice_id = :consecutivo
{discharge_filter}
"""
@staticmethod
def build_series_query(db_name: str) -> str:
"""Build query to get series information for repair imports."""
# TODO: QSeriesImpoRep table not migrated to PostgreSQL yet
return """
SELECT '' as serie, '' as modelo, '' as parte
WHERE 1=0
"""
@staticmethod
def build_driver_badge_query(db_name: str) -> str:
"""Build query to get driver badge number for repair imports."""
# TODO: GConductor table not migrated to PostgreSQL yet
return """
SELECT '' as badge
WHERE 1=0
"""
class ExportQueries:
"""SQL queries for exports (PostgreSQL schema)."""
@staticmethod
def build_aggregated_query(db_name: str, where_clause: str) -> str:
"""
Build optimized query for NORMAL mode (grouped by invoice with totals).
Args:
db_name: Database name (not used in PostgreSQL version)
where_clause: Additional WHERE conditions (without WHERE keyword)
"""
return f"""
SELECT
ih.invoice_number AS C1,
COALESCE(ped.pedimento_number, '') AS C2,
TO_CHAR(ih.invoice_date, 'YYYYMMDD') AS C3,
CASE WHEN ih.is_updated THEN 'AC' ELSE 'NA' END AS C6,
COALESCE(ped.pedimento_code, '') AS C7,
COALESCE(ped.regime, '') AS C8,
COALESCE(TO_CHAR(log.entry_exit_date, 'YYYYMMDD'), '') AS C9,
COALESCE(TO_CHAR(log.delivery_date, 'YYYYMMDD'), '') AS C10,
COALESCE(TO_CHAR(log.payment_date, 'YYYYMMDD'), '') AS C11,
COALESCE(log.payment_receipt_num, '') AS C12,
COALESCE(cmp.provider_id::text, '') AS C14,
COALESCE(cmp.sold_to_id::text, '') AS C15,
COALESCE(cmp.customs_broker_id::text, '') AS C16,
COALESCE(ih.purchase_order, '') AS C27,
COALESCE(cmp.aduana, '') AS C33,
COALESCE(ih.invoice_type, '') AS C34,
ih.id AS C35,
COALESCE(cmp.edocument, '') AS C40,
COALESCE(cmp.vucem_operation_num, '') AS C41,
COALESCE(fin.exchange_rate, 0) AS C48,
COALESCE(TO_CHAR(ih.emission_date, 'YYYYMMDD'), '') AS C49,
COALESCE(ih.capture_user, '') AS C50,
COALESCE(ih.who_updated, '') AS C51,
COALESCE(log.transport_id || ' ' || log.transport_num, '') AS C53,
COALESCE(SUM(CASE WHEN COALESCE(fil.is_subitem, false) THEN 0 ELSE lf.value_usd END), 0) AS total_me,
COALESCE(SUM(CASE WHEN COALESCE(fil.is_subitem, false) THEN 0 ELSE lf.value_mxn END), 0) AS total_mn
FROM a76.invoice_header ih
LEFT JOIN a76.invoice_compliance_mx cmp ON cmp.invoice_id = ih.id
LEFT JOIN a76.invoice_financials fin ON fin.invoice_id = ih.id
LEFT JOIN a76.invoice_logistics log ON log.invoice_id = ih.id
LEFT JOIN a76.pedimentos ped ON ped.id = cmp.pedimento_id
LEFT JOIN a76.items i ON i.invoice_id = ih.id
LEFT JOIN a76.item_lines il ON il.item_id = i.id
LEFT JOIN a24.fa_item_lines fil ON fil.id = il.id
LEFT JOIN a76.item_line_financials lf ON lf.item_line_id = il.id
WHERE {where_clause}
GROUP BY ih.id, ih.invoice_number, ih.is_updated, ped.pedimento_number, ped.pedimento_code, ped.regime,
log.entry_exit_date, log.delivery_date, log.payment_date, log.payment_receipt_num,
cmp.provider_id, cmp.sold_to_id, cmp.customs_broker_id, ih.purchase_order, cmp.aduana,
ih.invoice_type, cmp.edocument, cmp.vucem_operation_num, fin.exchange_rate,
ih.emission_date, ih.capture_user, ih.who_updated, log.transport_id, log.transport_num,
ih.invoice_date
ORDER BY ih.invoice_number
"""
@staticmethod
def build_main_query(db_name: str, where_clause: str) -> str:
return f"""
SELECT
ih.invoice_number AS C1, -- [0]
ped.pedimento_number AS C2, -- [1]
ih.invoice_date AS C3, -- [2]
'' AS C4, -- [3]
'' AS C5, -- [4]
ped.status AS C6, -- [5]
ped.pedimento_code AS C7, -- [6]
ped.regime AS C8, -- [7]
log.entry_exit_date AS C9, -- [8]
log.delivery_date AS C10, -- [9]
log.payment_date AS C11, -- [10]
log.payment_receipt_num AS C12, -- [11]
'' AS C13, -- [12]
cmp.provider_id AS C14, -- [13]
cmp.sold_to_id AS C15, -- [14]
cmp.customs_broker_id AS C16, -- [15]
'' AS C17, -- [16]
prt.part_number AS C18, -- [17]
ld.description_spanish AS C19, -- [18]
ld.description_english AS C20, -- [19]
lq.quantity AS C21, -- [20]
um.code AS C22, -- [21]
'' AS C23, -- [22]
'' AS C24, -- [23]
lq.net_weight AS C25, -- [24]
lq.gross_weight AS C26, -- [25]
ih.purchase_order AS C27, -- [26]
lc.fraction AS C28, -- [27]
'' AS C29, -- [28]
'' AS C30, -- [29]
'' AS C31, -- [30]
'' AS C32, -- [31]
cmp.aduana AS C33, -- [32]
ih.invoice_type AS C34, -- [33]
ih.id AS C35, -- [34]
lf.value_mxn AS C36, -- [35]
lf.value_usd AS C37, -- [36]
il.material_type AS C38, -- [37]
'' AS C39, -- [38] rectification_id
cmp.edocument AS C40, -- [39]
cmp.vucem_operation_num AS C41, -- [40]
il.line_number AS C42, -- [41]
ld.brand AS C43, -- [42]
ld.model AS C44, -- [43]
prt.us_fraction AS C45, -- [44]
prt.eccn AS C46, -- [45]
prt.id AS C47, -- [46]
fin.exchange_rate AS C48, -- [47]
ih.emission_date AS C49, -- [48]
ih.capture_user AS C50, -- [49]
ih.who_updated AS C51, -- [50]
'' AS C52, -- [51]
COALESCE(log.transport_id || ' ' || log.transport_num, '') AS C53, -- [52] NumCaja
'' AS C54, -- [53] Pedimento18
COALESCE(ld.lot, '') AS C55, -- [54] Lote
'' AS C56, -- [55] TipoPedimentoTransporte
'' AS C57, -- [56]
'' AS C58, -- [57]
'' AS C59, -- [58]
'' AS C60 -- [59] Relleno final
FROM a76.invoice_header ih
LEFT JOIN a76.invoice_compliance_mx cmp ON cmp.invoice_id = ih.id
LEFT JOIN a76.invoice_financials fin ON fin.invoice_id = ih.id
LEFT JOIN a76.invoice_logistics log ON log.invoice_id = ih.id
LEFT JOIN a76.pedimentos ped ON ped.id = cmp.pedimento_id
LEFT JOIN a76.items itm ON itm.invoice_id = ih.id
LEFT JOIN a76.item_lines il ON il.item_id = itm.id
LEFT JOIN a76.item_line_descriptions ld ON ld.item_line_id = il.id
LEFT JOIN a76.item_line_quantities lq ON lq.item_line_id = il.id
LEFT JOIN a76.item_line_financials lf ON lf.item_line_id = il.id
LEFT JOIN a76.item_line_customs lc ON lc.item_line_id = il.id
LEFT JOIN a76.classes cls ON cls.id = il.class_id
LEFT JOIN a76.parts prt ON prt.id = il.part_number
LEFT JOIN a76.units_of_measure um ON um.id = il.unit_of_measure
WHERE {where_clause}
ORDER BY ih.invoice_number, il.line_number
"""
@staticmethod
def build_totals_query(db_name: str, discharge_clause: str = "") -> str:
"""Build query to get totals for an export invoice.
Only sums partidas where is_subitem is false (main partidas, not sub-items).
"""
discharge_filter = "AND il.is_discharged = true" if "descargado" in discharge_clause.lower() else ""
return f"""
SELECT
COALESCE(SUM(lf.value_usd), 0),
COALESCE(SUM(lf.value_mxn), 0)
FROM a76.item_line_financials lf
INNER JOIN a76.item_lines il ON il.id = lf.item_line_id
INNER JOIN a76.items itm ON itm.id = il.item_id
WHERE itm.invoice_id = :consecutivo
{discharge_filter}
"""
@staticmethod
def build_series_query(db_name: str) -> str:
"""Build query to get series information for exports."""
# TODO: QSeriesExpo table not migrated to PostgreSQL yet
return """
SELECT '' as serie, '' as modelo, '' as parte
WHERE 1=0
"""
@staticmethod
def build_driver_badge_query(db_name: str) -> str:
"""Build query to get driver badge number for exports."""
# TODO: GConductor table not migrated to PostgreSQL yet
return """
SELECT '' as badge
WHERE 1=0
"""
class ExportRepairQueries:
"""SQL queries for export repairs (PostgreSQL schema)."""
@staticmethod
def build_aggregated_query(db_name: str, where_str: str, discharge_clause: str = "") -> str:
"""Build optimized query for NORMAL mode (grouped by invoice with totals)."""
# Note: discharge_clause temporarily disabled until is_discharged field migrated
discharge_filter = "" # Will be: " AND il.is_discharged = true/false" when ready
return f"""
SELECT
ih.invoice_number AS C1,
COALESCE(ped.pedimento_number, '') AS C2,
TO_CHAR(ih.invoice_date, 'YYYYMMDD') AS C3,
CASE WHEN ih.is_updated THEN 'AC' ELSE 'NA' END AS C6,
COALESCE(ped.pedimento_code, '') AS C7,
COALESCE(ped.regime, '') AS C8,
COALESCE(TO_CHAR(log.payment_date, 'YYYYMMDD'), '') AS C11,
COALESCE(cmp.remesa::text, '') AS C12,
COALESCE(fin.exchange_rate, 0) AS C13,
COALESCE(cmp.provider_id::text, '') AS C14,
COALESCE(cmp.sold_to_id::text, '') AS C15,
COALESCE(cmp.customs_broker_id::text, '') AS C16,
COALESCE(ih.purchase_order, '') AS C27,
COALESCE(ped.customs_office, '') AS C33,
COALESCE(ih.document_type, '') AS C34,
ih.id AS C35,
COALESCE(cmp.edocument, '') AS C40,
COALESCE(cmp.vucem_operation_num, '') AS C41,
COALESCE(fin.exchange_rate, 0) AS C48,
COALESCE(TO_CHAR(ih.invoice_date, 'YYYYMMDD'), '') AS C49,
COALESCE(ih.capture_user, '') AS C50,
COALESCE(ih.who_updated, '') AS C51,
COALESCE(log.carrier_id, '') AS C52,
COALESCE(log.transport_id || ' ' || log.transport_num, '') AS C53,
COALESCE(SUM(CASE WHEN COALESCE(fil.is_subitem, false) THEN 0 ELSE lf.value_usd END), 0) AS total_me,
COALESCE(SUM(CASE WHEN COALESCE(fil.is_subitem, false) THEN 0 ELSE lf.value_mxn END), 0) AS total_mn
FROM a76.invoice_header ih
LEFT JOIN a76.invoice_compliance_mx cmp ON cmp.invoice_id = ih.id
LEFT JOIN a76.invoice_financials fin ON fin.invoice_id = ih.id
LEFT JOIN a76.invoice_logistics log ON log.invoice_id = ih.id
LEFT JOIN a76.pedimentos ped ON ped.id = cmp.pedimento_id
LEFT JOIN a76.items i ON i.invoice_id = ih.id
LEFT JOIN a76.item_lines il ON il.item_id = i.id
LEFT JOIN a24.fa_item_lines fil ON fil.id = il.id
LEFT JOIN a76.item_line_financials lf ON lf.item_line_id = il.id
WHERE ih.operation_type = 'exp'
AND ih.invoice_type = 'REP'
{"AND " + where_str if where_str else ""}
GROUP BY ih.id, ih.invoice_number, ped.pedimento_number, ped.pedimento_code, ped.regime,
log.payment_date, cmp.remesa, fin.exchange_rate, cmp.provider_id, cmp.sold_to_id,
cmp.customs_broker_id, ih.purchase_order, ped.customs_office, ih.document_type,
cmp.edocument, cmp.vucem_operation_num, ih.invoice_date, ih.capture_user,
ih.who_updated, log.carrier_id, log.transport_id, log.transport_num
ORDER BY ih.invoice_number
"""
@staticmethod
def build_main_query(db_name: str, where_str: str) -> str:
"""Build main SQL query for export repair data."""
return f"""
SELECT
ih.invoice_number AS C1,
COALESCE(ped.pedimento_number, '') AS C2,
TO_CHAR(ih.invoice_date, 'YYYYMMDD') AS C3,
COALESCE(fin.value_me, 0) AS C4,
COALESCE(fin.value_mn, 0) AS C5,
CASE WHEN ih.is_updated THEN 'AC' ELSE 'NA' END AS C6,
COALESCE(ped.pedimento_code, '') AS C7,
COALESCE(ped.regime, '') AS C8,
'' AS C9,
'' AS C10,
COALESCE(TO_CHAR(log.payment_date, 'YYYYMMDD'), '') AS C11,
COALESCE(cmp.remesa::text, '') AS C12,
COALESCE(fin.exchange_rate, 0) AS C13,
COALESCE(cmp.provider_id::text, '') AS C14,
COALESCE(cmp.sold_to_id::text, '') AS C15,
COALESCE(cmp.customs_broker_id::text, '') AS C16,
'' AS C17,
COALESCE(cls.class_code, '') AS C18,
REPLACE(REPLACE(REPLACE(REPLACE(COALESCE(ld.description_spanish, ''), CHR(44), ' '), CHR(9), ' '), CHR(10), ' '), CHR(13), ' ') AS C19,
REPLACE(REPLACE(REPLACE(REPLACE(COALESCE(cls.description_en, ''), CHR(44), ' '), CHR(9), ' '), CHR(10), ' '), CHR(13), ' ') AS C20,
COALESCE(lq.quantity, 0) AS C21,
COALESCE(il.unit_of_measure, 0) AS C22,
COALESCE(lf.customs_value_mxn, 0) AS C23,
COALESCE(lf.customs_value_usd, 0) AS C24,
COALESCE(lq.net_weight, 0) AS C25,
COALESCE(lq.gross_weight, 0) AS C26,
COALESCE(ih.purchase_order, '') AS C27,
COALESCE(lc.fraction, '') AS C28,
COALESCE(lc.fraction_type, '') AS C29,
COALESCE(lc.advalorem_numeric, 0) AS C30,
COALESCE(lc.sector, '') AS C31,
COALESCE(lc.origin_country, '') AS C32,
COALESCE(ped.customs_office, '') AS C33,
COALESCE(ih.document_type, '') AS C34,
ih.id AS C35,
COALESCE(lf.value_mxn, 0) AS C36,
COALESCE(lf.value_usd, 0) AS C37,
'P' AS C38,
'' AS C39,
COALESCE(cmp.edocument, '') AS C40,
COALESCE(cmp.vucem_operation_num, '') AS C41,
COALESCE(il.line_number, 0) AS C42,
REPLACE(REPLACE(REPLACE(REPLACE(COALESCE(ld.brand, ''), CHR(44), ' '), CHR(9), ' '), CHR(10), ' '), CHR(13), ' ') AS C43,
REPLACE(REPLACE(REPLACE(REPLACE(COALESCE(ld.model, ''), CHR(44), ' '), CHR(9), ' '), CHR(10), ' '), CHR(13), ' ') AS C44,
COALESCE(cls.us_fraction, '') AS C45,
COALESCE(prt.eccn, '') AS C46,
COALESCE(il.part_number::text, '') AS C47,
COALESCE(fin.exchange_rate, 0) AS C48,
COALESCE(TO_CHAR(ih.invoice_date, 'YYYYMMDD'), '') AS C49,
COALESCE(ih.capture_user, '') AS C50,
COALESCE(ih.who_updated, '') AS C51,
COALESCE(log.carrier_id, '') AS C52,
COALESCE(log.transport_id || ' ' || log.transport_num, '') AS C53,
'' AS C54,
COALESCE(ld.lot, '') AS C55,
'' AS C56,
'' AS C57,
'' AS C58,
'' AS C59
FROM a76.invoice_header ih
LEFT JOIN a76.invoice_compliance_mx cmp ON cmp.invoice_id = ih.id
LEFT JOIN a76.invoice_financials fin ON fin.invoice_id = ih.id
LEFT JOIN a76.invoice_logistics log ON log.invoice_id = ih.id
LEFT JOIN a76.pedimentos ped ON ped.id = cmp.pedimento_id
LEFT JOIN a76.items itm ON itm.invoice_id = ih.id
LEFT JOIN a76.item_lines il ON il.item_id = itm.id
LEFT JOIN a76.item_line_descriptions ld ON ld.item_line_id = il.id
LEFT JOIN a76.item_line_quantities lq ON lq.item_line_id = il.id
LEFT JOIN a76.item_line_financials lf ON lf.item_line_id = il.id
LEFT JOIN a76.item_line_customs lc ON lc.item_line_id = il.id
LEFT JOIN a76.classes cls ON cls.id = il.class_id
LEFT JOIN a76.parts prt ON prt.id = il.part_number
LEFT JOIN a76.units_of_measure um ON um.id = il.unit_of_measure
WHERE UPPER(ih.operation_type) IN ('EXP', 'TRA', 'RET')
AND UPPER(ih.invoice_type) IN ('DEF', 'REP', 'EXDEF', 'MATDE')
AND {where_str}
ORDER BY ih.invoice_number, il.line_number
"""
@staticmethod
def build_totals_query(db_name: str, discharge_clause: str = "") -> str:
"""Build query to get totals for an export repair invoice."""
discharge_filter = "AND il.is_discharged = true" if "descargado" in discharge_clause.lower() else ""
return f"""
SELECT
COALESCE(SUM(lf.value_usd), 0),
COALESCE(SUM(lf.value_mxn), 0)
FROM a76.item_line_financials lf
INNER JOIN a76.item_lines il ON il.id = lf.item_line_id
INNER JOIN a76.items itm ON itm.id = il.item_id
WHERE itm.invoice_id = :consecutivo
{discharge_filter}
"""
@staticmethod
def build_series_query(db_name: str) -> str:
"""Build query to get series information for export repairs."""
# TODO: QSeriesExpoRep table not migrated to PostgreSQL yet
return """
SELECT '' as serie, '' as modelo, '' as parte
WHERE 1=0
"""
@staticmethod
def build_driver_badge_query(db_name: str) -> str:
"""Build query to get driver badge number for export repairs."""
# TODO: GConductor table not migrated to PostgreSQL yet
return """
SELECT '' as badge
WHERE 1=0
"""