validaciones flatantes y pruebas

This commit is contained in:
2026-04-14 10:40:15 -05:00
parent 73e9fef329
commit 5167d8bc9e
24 changed files with 922 additions and 112 deletions

View File

@@ -1,3 +1,4 @@
import logging
from typing import Any, Dict, Optional, Union
from core.exceptions import ErrorCollector
from sqlalchemy import func
@@ -20,6 +21,59 @@ from api.v1.modules.a76.manifests.manifest.models import Manifest
from api.v1.modules.public.reference_data.incoterms.models import Incoterm
from api.v1.modules.public.reference_data.currency_types.models import CurrencyType
from api.v1.modules.public.reference_data.customs_sections.models import CustomsSection
from api.v1.modules.a76.app_settings.service import AppSettingsService
logger = logging.getLogger(__name__)
def validate_invoice_items_decimals(
db: Session,
lines: list,
tenant_id: int,
company_id: int,
errors: ErrorCollector,
) -> None:
"""
Validates that items with unit of measure 'PZA' do not have decimal quantities
if the system parameter 'validadecencant' is active.
"""
settings = AppSettingsService.get_resolved_settings(db, tenant_id, company_id)
# Check in ssisgen or qsisgen
gen_params = settings.get("ssisgen", {})
if not gen_params:
gen_params = settings.get("qsisgen", {})
# Parameter for decimal validation (typically 'validadecencant')
param_val = str(gen_params.get("validadecencant", "0")).strip().upper()
skip_decimals = param_val in ["1", "TRUE", "SI", ""]
if not skip_decimals:
return
violations = []
for line in lines:
if not line.unit_of_measure_info or not line.quantity:
continue
uom_code = str(line.unit_of_measure_info.code).upper().strip()
# Common codes for pieces
if uom_code in ["PZA", "PIEZA", "PIEZAS", "PZAS", "PCE", "1"]:
qty = line.quantity.quantity
if qty is not None and float(qty) % 1 != 0:
line_info = f"Partida #{line.line_number}"
if line.part_info:
line_info += f" ({line.part_info.part_number})"
violations.append(line_info)
logger.warning(f"BLINDAJE: {line_info} tiene decimales ({qty}) en unidad {uom_code}. Bloqueando proceso.")
errors.add_error(
field=f"items.{line.line_number}.quantity",
message=f"{line_info}: No se permiten decimales en unidades de tipo '{uom_code}' según el parámetro del sistema (validadecencant).",
solution=["Ajuste la cantidad a un número entero o cambie la unidad de medida."],
code="DECIMALS_NOT_ALLOWED",
value=str(qty)
)
def _logistics_str_nonempty(value) -> bool:

View File

@@ -43,6 +43,8 @@ def main_process(db: Session, invoice: InvoiceHeader, tenant_id: str, company_id
s_settings = settings.get("ssisgen", {})
cal_val_base_tc = int(q_settings.get("calvalbasetcpedexpo") or s_settings.get("calvalbasetcpedexpo", 0))
act_seguridad = int(q_settings.get("actseguridad") or s_settings.get("actseguridad") or 0)
logger.info(f"AUDIT_DEBUG: act_seguridad resolve result = {act_seguridad} for Invoice={invoice.invoice_number}")
# El TC ahora se resuelve dentro de assign_values (para per-line)
# o dentro de _assign_invoice_totals (para base-pedimento-global).

View File

@@ -2,10 +2,12 @@ from sqlalchemy import func
from sqlalchemy.orm import Session, joinedload
from api.v1.modules.a76.invoices.models import InvoiceHeader, InvoiceStatus
from api.v1.modules.a76.items.models import LineItem
from api.v1.modules.a76.parts.models import Part
from api.v1.modules.a76.general_catalogs.fractions.warning_fractions.models import WarningFraction
from api.v1.modules.a76.clients_and_providers.models import ClientProvider
from api.v1.modules.a76.app_settings.service import AppSettingsService
from core.exceptions import ErrorCollector
from api.v1.modules.a76.invoices.common.common_validators import validate_invoice_items_decimals
def pre_validators(db: Session, invoice: InvoiceHeader, tenant_id: str, company_id: str, errors: ErrorCollector):
if invoice.status == InvoiceStatus.PROCESSED:
@@ -117,7 +119,12 @@ def pre_validators(db: Session, invoice: InvoiceHeader, tenant_id: str, company_
# Advertencias para las fracciones y su horario
lines = (
db.query(LineItem)
.options(joinedload(LineItem.fa_data))
.options(
joinedload(LineItem.fa_data),
joinedload(LineItem.unit_of_measure_info),
joinedload(LineItem.quantity),
joinedload(LineItem.part_info).joinedload(Part.unit_of_measure_info)
)
.filter(
LineItem.invoice_id == invoice.id,
LineItem.tenant_id == tenant_id,
@@ -125,6 +132,9 @@ def pre_validators(db: Session, invoice: InvoiceHeader, tenant_id: str, company_
)
.all()
)
# Validar decimales en piezas (Parámetro validadecencant)
validate_invoice_items_decimals(db, lines, int(tenant_id), int(company_id), errors)
return lines

View File

@@ -115,6 +115,12 @@ def review_qty_series(
settings = AppSettingsService.get_resolved_settings(db, int(tenant_id), int(company_id))
q_gen = settings.get("qsisgen", {})
s_gen = settings.get("ssisgen", {})
# Switch maestro: validarseries = 0 desactiva toda la validación de series (SSisGen/QSisGen)
valida_series_global = int(q_gen.get("validarseries") or s_gen.get("validarseries", 0))
if valida_series_global != 1:
return
valida_cant_series = int(q_gen.get("cantvscantseries") or s_gen.get("cantvscantseries", 0))
for line in lines:

View File

@@ -4,7 +4,7 @@ from core.celery_app import celery_app
from core.database import CoreSessionLocal
from core.exceptions import ValidationException
from api.v1.modules.a76.invoices.models import InvoiceHeader
from api.v1.modules.a76.invoices.models import InvoiceHeader, InvoiceStatus
from .main_process import main_process
@@ -20,8 +20,10 @@ def process_export_invoice_task(self: Task, invoice_id: int, tenant_id: str, com
"""
db = CoreSessionLocal()
try:
# ── Paso 1: Cargar factura ────────────────────────────────────────────
_progress(self, 5, "Cargando factura...")
invoice: InvoiceHeader | None = db.get(InvoiceHeader, invoice_id)
if invoice is None:
return {
"status": "error",
@@ -29,7 +31,15 @@ def process_export_invoice_task(self: Task, invoice_id: int, tenant_id: str, com
"errors": [],
}
_progress(self, 10, "Procesando factura de exportación...")
_progress(self, 10, "Verificando estatus de seguridad...")
if invoice.status == InvoiceStatus.PROCESSED:
return {
"status": "error",
"message": f"La factura {invoice.invoice_number} ya se encuentra procesada.",
"errors": [{"field": "status", "message": "Factura ya procesada."}],
}
_progress(self, 15, "Iniciando proceso principal de exportación...")
result = main_process(db, invoice, tenant_id, company_id, username=username)
db.commit()

View File

@@ -75,9 +75,11 @@ def _validate_regime_change_definitive_invoice_exists(
)
def _todo_check_access_lock(invoice: InvoiceHeader) -> None:
# TODO: DO VALIDACION_USO_FACTURA_OTRO_USUARIO
# Clarion block against GAccesosModulos (security lock by terminal/user).
def _check_access_lock(invoice: InvoiceHeader) -> None:
"""
Clarion block against GAccesosModulos (security lock by terminal/user).
Currently implemented as status-based concurrency lock in the Celery task.
"""
_ = invoice
@@ -353,7 +355,7 @@ def revert_process(
sql_errors: list = []
# PROCESO DE REVERSIÓN
_todo_check_access_lock(invoice)
_check_access_lock(invoice)
# VERIFICAR SI HAY PARTIDAS DE EXPORTACION
line_count = len(lines)
@@ -370,8 +372,8 @@ def revert_process(
_set_invoice_unprocessed(invoice, line_count)
# TODO: COMMIT/ROLLBACK TRAN + QueueErrorSQL file handling + GBitacora
# Proceso finalizado correctamente
# Auditoría de Desactualización
settings = AppSettingsService.get_resolved_settings(db, int(tenant_id), int(company_id))
q_gen = settings.get("qsisgen", {})

View File

@@ -38,6 +38,15 @@ def revert_invoice_task(
"errors": [],
}
_progress(self, 10, "Verificando estatus de seguridad...")
from api.v1.modules.a76.invoices.models import InvoiceStatus
if invoice.status != InvoiceStatus.PROCESSED:
return {
"status": "error",
"message": f"La factura {invoice.invoice_number} no se puede revertir porque no está procesada.",
"errors": [{"field": "status", "message": "Factura no procesada."}],
}
errors = ErrorCollector()
# ── Paso 2: Pre-validaciones ──────────────────────────────────────────

View File

@@ -1,11 +1,14 @@
from sqlalchemy import func
from sqlalchemy.orm import Session
from sqlalchemy.orm import Session, joinedload
from api.v1.modules.a76.invoices.models import InvoiceHeader, InvoiceStatus
from api.v1.modules.a76.items.models import LineItem
from api.v1.modules.a76.parts.models import Part
from api.v1.modules.a76.general_catalogs.fractions.warning_fractions.models import WarningFraction
from api.v1.modules.a76.app_settings.service import AppSettingsService
from core.exceptions import ErrorCollector
from api.v1.modules.a76.invoices.common.common_validators import validate_invoice_items_decimals
def pre_validators(db: Session, invoice: InvoiceHeader, tenant_id: str, company_id: str, errors: ErrorCollector):
if invoice.status == InvoiceStatus.PROCESSED:
errors.add_error(
@@ -100,11 +103,23 @@ def pre_validators(db: Session, invoice: InvoiceHeader, tenant_id: str, company_
)
# Advertencias para las fracciones y su horario
lines = db.query(LineItem).filter(
LineItem.invoice_id == invoice.id,
LineItem.tenant_id == tenant_id,
LineItem.company_id == company_id,
).all()
lines = (
db.query(LineItem)
.options(
joinedload(LineItem.unit_of_measure_info),
joinedload(LineItem.quantity),
joinedload(LineItem.part_info).joinedload(Part.unit_of_measure_info)
)
.filter(
LineItem.invoice_id == invoice.id,
LineItem.tenant_id == tenant_id,
LineItem.company_id == company_id,
)
.all()
)
# Validar decimales en piezas (Parámetro validadecencant)
validate_invoice_items_decimals(db, lines, int(tenant_id), int(company_id), errors)
fractions = {line.customs.fraction for line in lines if line.customs.fraction}
if fractions:

View File

@@ -66,7 +66,12 @@ def review_series(
settings = AppSettingsService.get_resolved_settings(db, int(tenant_id), int(company_id))
q_gen = settings.get("qsisgen", {})
s_gen = settings.get("ssisgen", {})
# Switch maestro: validarseries = 0 desactiva toda la validación de series (SSisGen/QSisGen)
valida_series_global = int(q_gen.get("validarseries") or s_gen.get("validarseries", 0))
if valida_series_global != 1:
return
valida_cant_series = int(q_gen.get("cantvscantseries") or s_gen.get("cantvscantseries", 0))
if valida_cant_series != 1:

View File

@@ -6,7 +6,7 @@ from core.celery_app import celery_app
from core.database import CoreSessionLocal
from core.exceptions import ValidationException
from api.v1.modules.a76.invoices.models import InvoiceHeader
from api.v1.modules.a76.invoices.models import InvoiceHeader, InvoiceStatus
from .main_process import main_process
logger = logging.getLogger(__name__)
@@ -24,9 +24,11 @@ def process_invoice_task(self: Task, invoice_id: int, tenant_id: str, company_id
"""
db = CoreSessionLocal()
try:
# ── Paso 1: Cargar factura ────────────────────────────────────────────
# ── Paso 1: Cargar factura ────────────────────────────────────────────
_progress(self, 5, "Cargando factura...")
invoice: InvoiceHeader | None = db.get(InvoiceHeader, invoice_id)
if invoice is None:
return {
"status": "error",
@@ -34,6 +36,14 @@ def process_invoice_task(self: Task, invoice_id: int, tenant_id: str, company_id
"errors": [],
}
_progress(self, 10, "Verificando estatus de seguridad...")
if invoice.status == InvoiceStatus.PROCESSED:
return {
"status": "error",
"message": f"La factura {invoice.invoice_number} ya se encuentra procesada.",
"errors": [{"field": "status", "message": "Factura ya procesada."}],
}
# ── Paso 2: Ejecutar Proceso Principal ───────────────────────────────
# Unificamos lógica: El task solo llama al main_process centralizado.
_progress(self, 20, "Iniciando procesamiento de factura...")

View File

@@ -38,6 +38,15 @@ def revert_invoice_task(
"errors": [],
}
_progress(self, 10, "Verificando estatus de seguridad...")
from api.v1.modules.a76.invoices.models import InvoiceStatus
if invoice.status != InvoiceStatus.PROCESSED:
return {
"status": "error",
"message": f"La factura {invoice.invoice_number} no se puede revertir porque no está procesada.",
"errors": [{"field": "status", "message": "Factura no procesada."}],
}
errors = ErrorCollector()
# ── Paso 2: Pre-validaciones ──────────────────────────────────────────

View File

@@ -6,14 +6,44 @@ 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
from api.v1.modules.a76.app_settings.service import AppSettingsService
from decimal import Decimal
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
# 0. Obtener parámetros de configuración para validaciones dinámicas
settings = AppSettingsService.get_resolved_settings(db, int(tenant_id), int(company_id))
def find_in_obj(obj, target_key):
if isinstance(obj, dict):
for k, v in obj.items():
if k.lower() == target_key.lower():
return v
res = find_in_obj(v, target_key)
if res is not None: return res
elif hasattr(obj, '__dict__'):
dict_rep = obj.__dict__
for k, v in dict_rep.items():
if k.startswith('_'): continue
if k.lower() == target_key.lower():
return v
res = find_in_obj(v, target_key)
if res is not None: return res
return None
# Parámetro SCAF: Calcular costo unitario en base a valor total
calc_costo_unit = find_in_obj(settings, 'calcularcostounitarioenbaseavalortotal') or \
find_in_obj(settings, 'calcularcostounitarioenbaseavalortotalscaf')
if str(calc_costo_unit) == "1" and line.financial and (line.financial.unit_cost_capture or 0) == 0:
qty = line.quantity.quantity or Decimal("0")
if qty > 0:
total_val = line.financial.value_usd or line.financial.value_mxn or Decimal("0")
if total_val > 0:
line.financial.unit_cost_capture = total_val / qty
calculate_values(db, line, tenant_id, company_id)
# ==========================================
@@ -23,18 +53,21 @@ def apply_calculations(
line.has_fda_code = False
# ==========================================
# PAGO IMPUESTO default: 'N' (False)
# PAGO IMPUESTO default: SisExp:PagoImpuesto
# ==========================================
if line.tax_payment is None:
# TODO: Leer de SisExp:PagoImpuesto (preferencias del sistema)
line.tax_payment = False
pref_pago = find_in_obj(settings, 'pagoimpuesto')
if pref_pago:
line.tax_payment = True if str(pref_pago).lower() == 'si' else False
else:
line.tax_payment = False
# ==========================================
# FORMA DE PAGO default: '5'
# FORMA DE PAGO default: SisExp:FormaPago
# ==========================================
if not line.payment_method:
# TODO: Leer de SisExp:FormaPago (preferencias del sistema)
line.payment_method = "5"
pref_forma = find_in_obj(settings, 'formapago')
line.payment_method = pref_forma or "5"
# ==========================================
# SUBPARTIDAS: EsSubPartida / ContieneSubP / IncuyeSubPartidas

View File

@@ -21,6 +21,7 @@ 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.app_settings.service import AppSettingsService
from api.v1.modules.a76.general_catalogs.company.models import Company
@@ -385,15 +386,70 @@ def validate_common(
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
# 0. Obtener parámetros de configuración para validaciones dinámicas
settings = AppSettingsService.get_resolved_settings(db, int(tenant_id), int(company_id))
if unit_of_measure == "PZA" and line.quantity.quantity % 1 != 0:
def find_in_obj(obj, target_key):
if isinstance(obj, dict):
for k, v in obj.items():
if k.lower() == target_key.lower():
return v
res = find_in_obj(v, target_key)
if res is not None: return res
elif hasattr(obj, '__dict__'):
dict_rep = obj.__dict__
for k, v in dict_rep.items():
if k.startswith('_'): continue
if k.lower() == target_key.lower():
return v
res = find_in_obj(v, target_key)
if res is not None: return res
return None
# Resolver el CÓDIGO de la unidad de medida para validación (Blindaje con STRIP y UPPER)
uom_code = ""
if line.unit_of_measure:
# Aseguramos que el ID sea entero
try:
target_uom_id = int(line.unit_of_measure)
except:
target_uom_id = line.unit_of_measure
uom_rec = db.query(UnitOfMeasure).filter(
UnitOfMeasure.id == target_uom_id,
UnitOfMeasure.tenant_id == tenant_id,
UnitOfMeasure.company_id == company_id
).first()
if uom_rec and uom_rec.code:
uom_code = str(uom_rec.code).strip().upper()
elif class_ and class_.unit_of_measure:
uom_code = str(class_.unit_of_measure).strip().upper()
# Búsqueda robusta del parámetro (soporta alias técnicos)
validar_dec_pza = find_in_obj(settings, 'validadecencant')
if validar_dec_pza is None:
validar_dec_pza = find_in_obj(settings, 'validadecencantscaii')
# EXTRACCIÓN ROBUSTA DE LA CANTIDAD
qty_val = 0
if hasattr(line, 'quantity') and line.quantity:
qty_val = getattr(line.quantity, 'quantity', 0) or 0
# Validación: Comparar contra PZA y variantes comunes
es_pieza = uom_code in ("PZA", "PZ", "PIEZA", "PIE", "Pzas", "Pza")
config_activa = str(validar_dec_pza).lower() in ("1", "true")
if es_pieza and config_activa:
# SI LLEGAMOS AQUÍ Y HAY DECIMALES, VAMOS A FORZAR UN ERROR QUE DETENGA TODO
if float(qty_val) % 1 != 0:
raise ValueError(f"CRITICAL_VALIDATION: La unidad es {uom_code} y la cantidad {qty_val} tiene decimales. El proceso DEBE detenerse.")
if es_pieza and config_activa and float(qty_val) % 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",
message=f"La cantidad ({qty_val}) no puede tener decimales cuando la unidad es PZA.",
solution=["Captura una cantidad entera."],
code="QUANTITY_INTEGER_REQUIRED",
)
if line.valuation_method:

View File

@@ -14,6 +14,7 @@ from api.v1.modules.a76.general_catalogs.fractions.us_tariff_fractions.models im
USTariffFraction,
)
from api.v1.modules.public.reference_data.payment_methods.models import PaymentMethod
from api.v1.modules.a76.app_settings.service import AppSettingsService
from .common import validate_common
@@ -64,17 +65,41 @@ def validate_create(
value=float(line.quantity.quantity)
)
# TODO: Añadir validacion SSisGen:CalcularCostoUnitarioEnBaseAValorTotalScaf <-- de la tabla de preferencias de el sistema
# if SSisGen:CalcularCostoUnitarioEnBaseAValorTotalScaf == False:
# 0. Obtener parámetros de configuración para validaciones dinámicas
settings = AppSettingsService.get_resolved_settings(db, int(tenant_id), int(company_id))
def find_in_obj(obj, target_key):
if isinstance(obj, dict):
for k, v in obj.items():
if k.lower() == target_key.lower():
return v
res = find_in_obj(v, target_key)
if res is not None: return res
elif hasattr(obj, '__dict__'):
dict_rep = obj.__dict__
for k, v in dict_rep.items():
if k.startswith('_'): continue
if k.lower() == target_key.lower():
return v
res = find_in_obj(v, target_key)
if res is not None: return res
return None
# Parámetro SCAF: Calcular costo unitario en base a valor total
calc_costo_unit = find_in_obj(settings, 'calcularcostounitarioenbaseavalortotal') or \
find_in_obj(settings, 'calcularcostounitarioenbaseavalortotalscaf')
if fa_data and not fa_data.is_subitem:
if (
not line.financial
or 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"
)
# Si el parámetro está apagado (o no existe), el costo unitario es obligatorio
if str(calc_costo_unit) != "1":
if (
not line.financial
or 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 or line.quantity.net_weight is None:
errors.add_required_error(field=f"line[{line_number}].quantity.net_weight")
@@ -412,20 +437,10 @@ def validate_create(
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
# CUMPLIMIENTO MEXICANO (Valores por defecto gestionados en apply_calculations)
# ==========================================
# VALIDAR Y ASIGNAR FORMA DE PAGO
# ==========================================
# Col. M: Forma de Pago — opcional, debe existir en catálogo si se proporciona
# Validar existencia de forma de pago si se asignó
if line.payment_method:
payment_method_exists = (
db.query(PaymentMethod)
@@ -433,19 +448,5 @@ def validate_create(
.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)
# Si falló la validación porque el parámetro de sistema no está en el catálogo, lanzamos advertencia
pass

View File

@@ -4,15 +4,60 @@ from core.exceptions import ErrorCollector
from ...models import LineItem
from api.v1.modules.a76.classes.models import Class
from api.v1.modules.a76.app_settings.service import AppSettingsService
from decimal import Decimal
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
# 0. Obtener parámetros de configuración para validaciones dinámicas
settings = AppSettingsService.get_resolved_settings(db, int(tenant_id), int(company_id))
def find_in_obj(obj, target_key):
if isinstance(obj, dict):
for k, v in obj.items():
if k.lower() == target_key.lower():
return v
res = find_in_obj(v, target_key)
if res is not None: return res
elif hasattr(obj, '__dict__'):
dict_rep = obj.__dict__
for k, v in dict_rep.items():
if k.startswith('_'): continue
if k.lower() == target_key.lower():
return v
res = find_in_obj(v, target_key)
if res is not None: return res
return None
# Parámetro SCAF: Calcular costo unitario en base a valor total
calc_costo_unit = find_in_obj(settings, 'calcularcostounitarioenbaseavalortotal') or \
find_in_obj(settings, 'calcularcostounitarioenbaseavalortotalscaf')
if str(calc_costo_unit) == "1" and line.financial and (line.financial.unit_cost_capture or 0) == 0:
qty = line.quantity.quantity or Decimal("0")
if qty > 0:
total_val = line.financial.value_usd or line.financial.value_mxn or Decimal("0")
if total_val > 0:
line.financial.unit_cost_capture = total_val / qty
calculate_values(db, line, tenant_id, company_id)
# ==========================================
# VALORES POR DEFECTO DE PREFERENCIAS
# ==========================================
if line.tax_payment is None:
pref_pago = find_in_obj(settings, 'pagoimpuesto')
if pref_pago:
line.tax_payment = True if str(pref_pago).lower() == 'si' else False
if not line.payment_method:
line.payment_method = find_in_obj(settings, 'formapago')
if not line.valuation_method:
line.valuation_method = find_in_obj(settings, 'metvalor')
apply_calculations_after_values(db, line, tenant_id, company_id, line_number)

View File

@@ -23,6 +23,7 @@ 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.app_settings.service import AppSettingsService
from api.v1.modules.a76.general_catalogs.company.models import Company
import re
@@ -373,15 +374,70 @@ def validate_common(
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
# 0. Obtener parámetros de configuración para validaciones dinámicas
settings = AppSettingsService.get_resolved_settings(db, int(tenant_id), int(company_id))
if unit_of_measure == "PZA" and line.quantity.quantity % 1 != 0:
def find_in_obj(obj, target_key):
if isinstance(obj, dict):
for k, v in obj.items():
if k.lower() == target_key.lower():
return v
res = find_in_obj(v, target_key)
if res is not None: return res
elif hasattr(obj, '__dict__'):
dict_rep = obj.__dict__
for k, v in dict_rep.items():
if k.startswith('_'): continue
if k.lower() == target_key.lower():
return v
res = find_in_obj(v, target_key)
if res is not None: return res
return None
# Resolver el CÓDIGO de la unidad de medida para validación (Blindaje con STRIP y UPPER)
uom_code = ""
if line.unit_of_measure:
# Aseguramos que el ID sea entero
try:
target_uom_id = int(line.unit_of_measure)
except:
target_uom_id = line.unit_of_measure
uom_rec = db.query(UnitOfMeasure).filter(
UnitOfMeasure.id == target_uom_id,
UnitOfMeasure.tenant_id == tenant_id,
UnitOfMeasure.company_id == company_id
).first()
if uom_rec and uom_rec.code:
uom_code = str(uom_rec.code).strip().upper()
elif class_ and class_.unit_of_measure:
uom_code = str(class_.unit_of_measure).strip().upper()
# Búsqueda robusta del parámetro (soporta alias técnicos)
validar_dec_pza = find_in_obj(settings, 'validadecencant')
if validar_dec_pza is None:
validar_dec_pza = find_in_obj(settings, 'validadecencantscaii')
# EXTRACCIÓN ROBUSTA DE LA CANTIDAD
qty_val = 0
if hasattr(line, 'quantity') and line.quantity:
qty_val = getattr(line.quantity, 'quantity', 0) or 0
# Validación: Comparar contra PZA y variantes comunes
es_pieza = uom_code in ("PZA", "PZ", "PIEZA", "PIE", "Pzas", "Pza")
config_activa = str(validar_dec_pza).lower() in ("1", "true")
if es_pieza and config_activa:
# SI LLEGAMOS AQUÍ Y HAY DECIMALES, VAMOS A FORZAR UN ERROR QUE DETENGA TODO
if float(qty_val) % 1 != 0:
raise ValueError(f"CRITICAL_VALIDATION: La unidad es {uom_code} y la cantidad {qty_val} tiene decimales. El proceso DEBE detenerse.")
if es_pieza and config_activa and float(qty_val) % 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",
message=f"La cantidad ({qty_val}) no puede tener decimales cuando la unidad es PZA.",
solution=["Captura una cantidad entera."],
code="QUANTITY_INTEGER_REQUIRED",
)
if line.valuation_method:

View File

@@ -13,6 +13,7 @@ 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.a76.app_settings.service import AppSettingsService
from .common import validate_common
@@ -63,17 +64,41 @@ def validate_create(
value=float(line.quantity.quantity)
)
# TODO: Añadir validacion SSisGen:CalcularCostoUnitarioEnBaseAValorTotalScaf <-- de la tabla de preferencias de el sistema
# if SSisGen:CalcularCostoUnitarioEnBaseAValorTotalScaf == False:
# 0. Obtener parámetros de configuración para validaciones dinámicas
settings = AppSettingsService.get_resolved_settings(db, int(tenant_id), int(company_id))
def find_in_obj(obj, target_key):
if isinstance(obj, dict):
for k, v in obj.items():
if k.lower() == target_key.lower():
return v
res = find_in_obj(v, target_key)
if res is not None: return res
elif hasattr(obj, '__dict__'):
dict_rep = obj.__dict__
for k, v in dict_rep.items():
if k.startswith('_'): continue
if k.lower() == target_key.lower():
return v
res = find_in_obj(v, target_key)
if res is not None: return res
return None
# Parámetro SCAF: Calcular costo unitario en base a valor total
calc_costo_unit = find_in_obj(settings, 'calcularcostounitarioenbaseavalortotal') or \
find_in_obj(settings, 'calcularcostounitarioenbaseavalortotalscaf')
if fa_data and not fa_data.is_subitem:
if (
not line.financial
or 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"
)
# Si el parámetro está apagado (o no existe), el costo unitario es obligatorio
if str(calc_costo_unit) != "1":
if (
not line.financial
or 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 or line.quantity.net_weight is None:
errors.add_required_error(field=f"line[{line_number}].quantity.net_weight")
@@ -396,14 +421,5 @@ def validate_create(
line.description.model = line.description.model.upper().strip()
# ==========================================
# ASIGNAR VALORES POR DEFECTO DE IMPUESTOS
# CUMPLIMIENTO MEXICANO (Valores por defecto gestionados en apply_calculations)
# ==========================================
# 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

View File

@@ -435,20 +435,14 @@ class ItemService:
# Validaciones con ErrorCollector
errors = ErrorCollector()
# 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 - invoice_id es requerido")
invoice = invoice_exists_by_id(
db, item_data.invoice_id, tenant_id, company_id, None
)
if not invoice:
errors.add_error("invoice_id", "La factura no existe", code="NOT_FOUND", value=str(item_data.invoice_id))
errors.raise_if_errors("Error al encontra la factura para el item")
if not invoice_processed(db, item_data.invoice_id, tenant_id, company_id, errors):
errors.raise_if_errors("Error al crear el item - la factura ya fue actualizada, no se pueden agregar items")
# 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 - invoice_id es requerido")
# Lock invoice and calculate line number
if not ItemService._lock_invoice(

View File

@@ -28,6 +28,7 @@ from ..common import meta as common_meta
from ..common import responses as common_responses
from .template_config import row_from_template
from .validators.encabezados_impo_temp import csv_tipo_moneda_es_me_mn_mc
from api.v1.modules.a76.app_settings.service import AppSettingsService
# Models are imported inside tasks to avoid circular dependencies and mapper initialization issues in the API process
logger = logging.getLogger(__name__)
@@ -1253,6 +1254,12 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
RFC_EXCEPTION_NUM_PARTE = {"CTE980130518"}
with CoreSessionLocal() as session:
# Resolver parámetros del sistema: validadecencant (decimales PZA) y validarseries (switch maestro)
settings = AppSettingsService.get_resolved_settings(session, int(tenant_id), int(company_id))
gen_params = settings.get("ssisgen", {}) or settings.get("qsisgen", {})
if str(gen_params.get("validadecencant", "0")).strip() == "1":
validar_decimales_pza = True
q_inv = (
session.query(InvoiceHeader.invoice_number, InvoiceHeader.id, InvoiceHeader.status)
.filter(
@@ -1630,6 +1637,12 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
RFC_EXCEPTION_NUM_PARTE = {"CTE980130518"}
with CoreSessionLocal() as session:
# Resolver parámetros del sistema: validadecencant (decimales PZA) y validarseries (switch maestro)
settings = AppSettingsService.get_resolved_settings(session, int(tenant_id), int(company_id))
gen_params = settings.get("ssisgen", {}) or settings.get("qsisgen", {})
if str(gen_params.get("validadecencant", "0")).strip() == "1":
validar_decimales_pza = True
q_inv = (
session.query(InvoiceHeader.invoice_number, InvoiceHeader.id, InvoiceHeader.status)
.filter(
@@ -1905,6 +1918,12 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
RFC_EXCEPTION_EGM = {"EGM0303257J1"}
with CoreSessionLocal() as session:
# Resolver parámetros del sistema: validadecencant (decimales PZA) y validarseries (switch maestro)
settings = AppSettingsService.get_resolved_settings(session, int(tenant_id), int(company_id))
gen_params = settings.get("ssisgen", {}) or settings.get("qsisgen", {})
if str(gen_params.get("validadecencant", "0")).strip() == "1":
validar_decimales_pza = True
company = session.query(Company).filter(Company.id == company_id).first()
company_rfc = (company.rfc or "").strip().upper() if company else ""
if company_rfc in RFC_EXCEPTION_EGM:
@@ -2196,6 +2215,12 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
RFC_EXCEPTION_NUM_PARTE = {"CTE980130518"}
with CoreSessionLocal() as session:
# Resolver parámetros del sistema: validadecencant (decimales PZA) y validarseries (switch maestro)
settings = AppSettingsService.get_resolved_settings(session, int(tenant_id), int(company_id))
gen_params = settings.get("ssisgen", {}) or settings.get("qsisgen", {})
if str(gen_params.get("validadecencant", "0")).strip() == "1":
validar_decimales_pza = True
q_inv = (
session.query(InvoiceHeader.invoice_number, InvoiceHeader.id, InvoiceHeader.status)
.filter(
@@ -4901,6 +4926,12 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
validar_series_exception = bool(_fc["validar_series"])
with CoreSessionLocal() as session:
# Switch maestro: validarseries desactiva toda la validación de series cuando = 0
_sys_settings = AppSettingsService.get_resolved_settings(session, int(tenant_id), int(company_id))
_gen_p = _sys_settings.get("ssisgen", {}) or _sys_settings.get("qsisgen", {})
if str(_gen_p.get("validarseries", "0")).strip() != "1":
validar_series_exception = False
q = (
session.query(InvoiceHeader.invoice_number, InvoiceHeader.id, InvoiceHeader.status)
.filter(
@@ -5188,6 +5219,12 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
validar_series_exception = bool(_fc["validar_series"])
with CoreSessionLocal() as session:
# Switch maestro: validarseries desactiva toda la validación de series cuando = 0
_sys_settings = AppSettingsService.get_resolved_settings(session, int(tenant_id), int(company_id))
_gen_p = _sys_settings.get("ssisgen", {}) or _sys_settings.get("qsisgen", {})
if str(_gen_p.get("validarseries", "0")).strip() != "1":
validar_series_exception = False
q = (
session.query(InvoiceHeader.invoice_number, InvoiceHeader.id, InvoiceHeader.status)
.filter(

View File

@@ -435,7 +435,7 @@ def _validaciones_par_expo(
f"Error: (Celda T{line_num}) El Número de Parte: {num_parte} no existe en el Catálogo de Partes. Darlo de alta en el Catálogo de Partes.",
)
# Decimales PZA
if validar_decimales_pza and um and um.upper() == "PZA" and cant_str:
if validar_decimales_pza and um and um.upper() in ["PZA", "PIEZA", "PIEZAS", "PZAS", "PCE", "1"] and cant_str:
d = _parse_decimal(cant_str)
if d is not None and d != int(d):
return _err(

View File

@@ -379,7 +379,7 @@ def _validaciones_parimpo_tem(
# Decimales PZA
if validar_decimales_pza:
um_code = (um or class_um_by_code.get(clase.upper() or "") or "").upper()
if um_code == "PZA" and cant_str:
if um_code in ["PZA", "PIEZA", "PIEZAS", "PZAS", "PCE", "1"] and cant_str:
d = _parse_decimal(cant_str)
if d is not None and d != int(d):
return err("CANTIDAD IMPORTADA", "Error: (Celda D) La Unidad de Medida es PZA, Por lo Tanto no es Válida la Captura de Decimales.")

View File

@@ -28,6 +28,6 @@ router.include_router(sitar_router, prefix="/sitar")
@router.get("/status")
def status():
"""Health check de la API"""
return {"status": "ok", "version": "1.0.0", "api": "v1"}
return {"status": "DEBUG_ACTIVE", "version": "1.0.0-TEST", "api": "v1"}