feature/validaciones-clarion

This commit is contained in:
hreyes
2026-03-04 11:28:39 -07:00
parent 196b70924e
commit a081a61aa1
8 changed files with 252 additions and 25 deletions

View File

@@ -82,3 +82,27 @@ def check_in_set(
if val not in allowed:
return {"line": line_num, "col": col, "msg": msg}
return None
def check_decimal_max(
row: Dict[str, Any],
col: str,
line_num: int,
max_val: float,
msg: Optional[str] = None,
) -> Optional[Dict[str, Any]]:
"""Solo valida si hay valor; error si no es numérico o si es mayor que max_val (Clarion Col H)."""
val = row.get(col)
if val is None or val == "":
return None
try:
v = float(val)
if v > max_val:
return {
"line": line_num,
"col": col,
"msg": msg or f"El valor no puede ser mayor a {max_val}.",
}
except (ValueError, TypeError):
return {"line": line_num, "col": col, "msg": "Debe ser un número."}
return None

View File

@@ -1,5 +1,6 @@
"""
Carga de conjuntos FK para validación/mapeo de import CSV de clases de materiales.
Clarion: Tipo Activo Fijo, U.M., Fracción Mex (GFracGenSifra + histórico), Fracción Ame (GFracAme), Código Producto CP (si existe).
"""
from typing import Set, Tuple
@@ -9,19 +10,30 @@ from core.database import CoreSessionLocal
def load_classes_fk_sets(
tenant_id: int,
company_id: int,
) -> Tuple[Set[str], Set[str]]:
) -> Tuple[Set[str], Set[str], Set[str], Set[str], Set[str]]:
"""
Carga valid_material_keys (MaterialType.key) y valid_uom_codes (UnitOfMeasure.code).
Devuelve (valid_material_keys, valid_uom_codes).
Carga todos los conjuntos necesarios para validación CSV de clases (paridad Clarion).
Devuelve (valid_material_keys, valid_uom_codes, valid_fraction_mex_8, valid_fraction_ame, valid_product_codes_cp).
- valid_fraction_mex_8: códigos de 8 caracteres válidos (TariffFraction + HistoricalTariffFraction).
- valid_fraction_ame: códigos de fracción americana (USTariffFraction por tenant/company).
- valid_product_codes_cp: códigos de producto/servicio CP (vacío si no existe catálogo).
"""
valid_material_keys: Set[str] = set()
valid_uom_codes: Set[str] = set()
valid_fraction_mex_8: Set[str] = set()
valid_fraction_ame: Set[str] = set()
valid_product_codes_cp: Set[str] = set()
try:
with CoreSessionLocal() as session:
from api.v1.modules.public.reference_data.material_types.models import MaterialType
from api.v1.modules.a76.general_catalogs.units_of_measure.models import UnitOfMeasure
from api.v1.modules.a76.general_catalogs.fractions.tariff_fractions.models import TariffFraction
from api.v1.modules.a76.general_catalogs.fractions.historical_tariff_fractions.models import HistoricalTariffFraction
from api.v1.modules.a76.general_catalogs.fractions.us_tariff_fractions.models import USTariffFraction
for m in session.query(MaterialType.key).all():
valid_material_keys.add(m[0])
if m[0]:
valid_material_keys.add(m[0])
for u in (
session.query(UnitOfMeasure.code)
.filter(
@@ -30,8 +42,45 @@ def load_classes_fk_sets(
)
.all()
):
valid_uom_codes.add(u[0])
if u[0]:
valid_uom_codes.add(u[0])
for row in session.query(TariffFraction.code).all():
if row[0]:
code = row[0].strip()
valid_fraction_mex_8.add(code[:8])
for row in (
session.query(HistoricalTariffFraction.historical_fraction)
.filter(
HistoricalTariffFraction.tenant_id == tenant_id,
HistoricalTariffFraction.company_id == company_id,
HistoricalTariffFraction.historical_fraction.isnot(None),
)
.distinct()
.all()
):
if row[0] and row[0].strip():
valid_fraction_mex_8.add(row[0].strip()[:8])
for row in (
session.query(USTariffFraction.code)
.filter(
USTariffFraction.tenant_id == tenant_id,
USTariffFraction.company_id == company_id,
)
.all()
):
if row[0]:
valid_fraction_ame.add(row[0].strip())
except Exception as e:
import logging
logging.getLogger(__name__).warning("Classes import: could not load FK sets: %s", e)
return valid_material_keys, valid_uom_codes
return (
valid_material_keys,
valid_uom_codes,
valid_fraction_mex_8,
valid_fraction_ame,
valid_product_codes_cp,
)

View File

@@ -57,7 +57,9 @@ def scan_file(self, job_id: str, config: str = None):
actualizar = meta.get("actualizar", False)
siempre_toda = meta.get("siempre_toda", False)
valid_material_keys, valid_uom_codes = load_classes_fk_sets(tenant_id, company_id)
valid_material_keys, valid_uom_codes, valid_fraction_mex_8, valid_fraction_ame, valid_product_codes_cp = load_classes_fk_sets(
tenant_id, company_id
)
from api.v1.modules.a76.classes.models import Class
existing_class_codes = set()
@@ -94,6 +96,9 @@ def scan_file(self, job_id: str, config: str = None):
actualizar=actualizar,
siempre_toda=siempre_toda,
existing_class_codes=existing_class_codes,
valid_fraction_mex_8=valid_fraction_mex_8,
valid_fraction_ame=valid_fraction_ame,
valid_product_codes_cp=valid_product_codes_cp,
)
if err:
error_count += 1
@@ -145,7 +150,9 @@ def insert_valid_rows(self, job_id: str):
from api.v1.modules.a76.classes.models import Class
valid_material_keys, valid_uom_codes = load_classes_fk_sets(tenant_id, company_id)
valid_material_keys, valid_uom_codes, valid_fraction_mex_8, valid_fraction_ame, valid_product_codes_cp = load_classes_fk_sets(
tenant_id, company_id
)
inserted_count = 0
skipped_invalid = 0
@@ -177,6 +184,9 @@ def insert_valid_rows(self, job_id: str):
row_norm, i,
valid_material_keys=valid_material_keys,
valid_uom_codes=valid_uom_codes,
valid_fraction_mex_8=valid_fraction_mex_8,
valid_fraction_ame=valid_fraction_ame,
valid_product_codes_cp=valid_product_codes_cp,
)
else:
err = validate_row_class(
@@ -187,6 +197,9 @@ def insert_valid_rows(self, job_id: str):
actualizar=actualizar,
siempre_toda=siempre_toda,
existing_class_codes=set(existing_by_code.keys()),
valid_fraction_mex_8=valid_fraction_mex_8,
valid_fraction_ame=valid_fraction_ame,
valid_product_codes_cp=valid_product_codes_cp,
)
if err:
skipped_invalid += 1

View File

@@ -13,6 +13,7 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
{"canonical": "UNIMED", "aliases": ["UNIDAD MEDIDA", "UNIT", "UOM"]},
{"canonical": "FRACCION", "aliases": ["FRACCION MEX"]},
{"canonical": "FRACCIONAME", "aliases": ["FRACCION USA", "US FRACTION"]},
{"canonical": "TASADEPRECIA", "aliases": ["TASA DEPRECIACION", "TASA DEPRECIACIÓN"]},
{"canonical": "CLAVESUB", "aliases": ["SUB KEY", "CLAVE SUB"]},
{"canonical": "REVFISICA", "aliases": ["REV FISICA", "PHYSICAL REVIEW"]},
{"canonical": "FRACCIONEXENTAIVA", "aliases": ["EXENTA IVA", "FRACCION EXENTA IVA"]},

View File

@@ -1,5 +1,6 @@
"""
Validaciones comunes de fila para import CSV de clases de materiales.
Paridad con Clarion: VALIDACIONES_CLASE (Col A len, Col D/E/F/G en catálogo, Col F len≥8, Col H ≤100, Col J en catálogo CP).
"""
from typing import Dict, Any, Optional, Set
@@ -8,6 +9,7 @@ from ..common.common_validators import (
check_min_length,
check_int_range,
check_in_set,
check_decimal_max,
)
MSG_CLASE_VACIO = "Error: (Col. A) La columna de Clase esta vacio y no se pueden hacer las validaciones."
@@ -75,26 +77,120 @@ def validate_row_fks(
) -> Optional[Dict[str, Any]]:
err = check_in_set(
row, "CLAVEMAT", line_num, valid_material_keys,
"El Tipo de Activo Fijo no existe en el Catálogo de Tipos de Activo Fijo.",
"Error: (Col. D) El Tipo de Activo Fijo no existe en el Catálogo de Tipos de Activo Fijo.",
)
if err:
return err
err = check_in_set(
row, "UNIMED", line_num, valid_uom_codes,
"La Unidad de Medida Comercial no existe en el Catálogo de U.M.",
"Error: (Col. E) La Unidad de Medida Comercial no existe en el Catálogo de U.M.",
)
if err:
return err
return None
def _normalize_fraction_mex_8(value: str) -> str:
"""Primeros 8 caracteres si len>=10, sino hasta 8 (Clarion SUB(ColumnaF, 1, 8))."""
if not value:
return ""
v = value.strip()
if len(v) >= 10:
return v[:8]
return v[:8] if len(v) > 8 else v
def validate_row_fraction_mex_catalog(
row: Dict[str, Any],
line_num: int,
valid_fraction_mex_8: Optional[Set[str]],
) -> Optional[Dict[str, Any]]:
"""Col F: si no vacía, debe existir en catálogo Mex (GFracGenSifra) o Histórico (Clarion)."""
val = (row.get("FRACCION") or "").strip()
if not val or valid_fraction_mex_8 is None:
return None
code_8 = _normalize_fraction_mex_8(val)
if not code_8:
return None
if code_8 in valid_fraction_mex_8:
return None
return {
"line": line_num,
"col": "FRACCION",
"msg": (
f"Error: (Col. F) La Fraccion Mexicana: {val} no existe en el Catálogo de Fracciones Arancelarias Sifr@ ni en el Historico. "
"Revisar esta Fracción Arancelaria en el archivo, en caso de ser correcta Actualizar las Fracciones Arancelarias."
),
}
def validate_row_fraction_ame_catalog(
row: Dict[str, Any],
line_num: int,
valid_fraction_ame: Optional[Set[str]],
) -> Optional[Dict[str, Any]]:
"""Col G: si no vacía, debe existir en catálogo Fracciones Americanas (Clarion GFracAme)."""
val = (row.get("FRACCIONAME") or "").strip()
if not val or valid_fraction_ame is None:
return None
if val in valid_fraction_ame:
return None
return {
"line": line_num,
"col": "FRACCIONAME",
"msg": (
f"Error: (Col. G) La Fraccion Americana: {val} no existe en el Catálogo de Fracciones Americanas. "
"Dar de alta la Fracción Americana en el Catálogo de Fracciones Americanas."
),
}
def validate_row_tasa_depreciacion(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
"""Col H: si viene informada, no puede ser mayor al 100 % (Clarion)."""
val = row.get("TASADEPRECIA")
if val is None or val == "":
return None
return check_decimal_max(
row, "TASADEPRECIA", line_num, 100.0,
msg="Error: (Col. H) La Tasa de Depreciación no puede ser mayor al 100 %. Ajustar la Tasa de Depreciacion.",
)
def validate_row_codigo_producto_cp(
row: Dict[str, Any],
line_num: int,
valid_product_codes_cp: Optional[Set[str]],
class_code: str = "",
) -> Optional[Dict[str, Any]]:
"""Col J: si no vacía y existe catálogo CP, debe existir en GCodigosProductoCP (Clarion)."""
val = (row.get("FRACCIONEXENTAIVA") or "").strip()
if not val:
return None
if valid_product_codes_cp is None or len(valid_product_codes_cp) == 0:
return None
if val in valid_product_codes_cp:
return None
cl = class_code or (row.get("CLASE") or "").strip()
return {
"line": line_num,
"col": "FRACCIONEXENTAIVA",
"msg": (
f"Error: (Col. J) La clase: {cl} tiene asignado un código de producto inexistente. "
"Capturar un código de producto correcto."
),
}
def validaciones_clase(
row: Dict[str, Any],
line_num: int,
valid_material_keys: Optional[Set[str]],
valid_uom_codes: Optional[Set[str]],
valid_fraction_mex_8: Optional[Set[str]] = None,
valid_fraction_ame: Optional[Set[str]] = None,
valid_product_codes_cp: Optional[Set[str]] = None,
) -> Optional[Dict[str, Any]]:
"""Reglas compartidas Clarion (VALIDACIONES_CLASE): longitudes, tipos, FKs. Sin obligatorios B,D,E,F."""
"""Reglas compartidas Clarion (VALIDACIONES_CLASE): longitudes, tipos, FKs, fracciones, tasa, código CP."""
err = validate_row_lengths(row, line_num)
if err:
return err
@@ -105,6 +201,21 @@ def validaciones_clase(
if err:
return err
err = validate_row_fks(row, line_num, valid_material_keys, valid_uom_codes)
if err:
return err
err = validate_row_fraction_mex_catalog(row, line_num, valid_fraction_mex_8)
if err:
return err
err = validate_row_fraction_ame_catalog(row, line_num, valid_fraction_ame)
if err:
return err
err = validate_row_tasa_depreciacion(row, line_num)
if err:
return err
class_code = (row.get("CLASE") or "").strip()
err = validate_row_codigo_producto_cp(
row, line_num, valid_product_codes_cp, class_code
)
if err:
return err
return None

View File

@@ -1,7 +1,7 @@
"""
Punto de entrada de validación para import de una fila de clase de material.
Flujo Clarion: no ACT → siempre VALIDA_TODA_CLASE; ACT y clase existe → VALIDA_PARCIAL_CLASE;
ACT y clase no existe → VALIDA_TODA_CLASE y error "Clave no existe en catálogo" (no crear).
ACT y clase no existe → VALIDA_TODA_CLASE (validación completa) y si pasa se crea en el insert (ADD).
"""
from typing import Dict, Any, Optional, Set
@@ -20,12 +20,15 @@ def validate_row_class(
actualizar: bool = False,
siempre_toda: bool = False,
existing_class_codes: Optional[Set[str]] = None,
valid_fraction_mex_8: Optional[Set[str]] = None,
valid_fraction_ame: Optional[Set[str]] = None,
valid_product_codes_cp: Optional[Set[str]] = None,
) -> Optional[Dict[str, Any]]:
"""
Igual que Clarion:
- No ACT (actualizar=False): siempre TODA (B,D,E,F obligatorios + validaciones_clase).
- ACT y clase existe: PARCIAL (solo CLASE + validaciones_clase).
- ACT y clase no existe: TODA y error "Clave de la clase No Existe en el Catalogo" (no se crea).
- ACT y clase existe: PARCIAL (solo CLASE + validaciones_clase); en insert se actualiza (PUT).
- ACT y clase no existe: TODA (validación completa); si pasa, en insert se crea (ADD).
siempre_toda fuerza TODA en todos los casos.
"""
err = validate_row_required(row, line_num)
@@ -41,14 +44,17 @@ def validate_row_class(
err = validate_row_required_full(row, line_num)
if err:
return err
if actualizar and existing_class_codes is not None and class_code not in existing_class_codes:
return {
"line": line_num,
"col": "CLASE",
"msg": "Clave de la clase No Existe en el Catalogo. En modo actualizar la clase debe existir.",
}
# En Actualizar, si la clase no existe se valida completa y si pasa se crea en insert (como Clarion ADD).
return validaciones_clase(row, line_num, valid_material_keys, valid_uom_codes)
return validaciones_clase(
row,
line_num,
valid_material_keys,
valid_uom_codes,
valid_fraction_mex_8=valid_fraction_mex_8,
valid_fraction_ame=valid_fraction_ame,
valid_product_codes_cp=valid_product_codes_cp,
)
def validate_row_class_partial(
@@ -56,9 +62,20 @@ def validate_row_class_partial(
line_num: int,
valid_material_keys: Optional[Set[str]] = None,
valid_uom_codes: Optional[Set[str]] = None,
valid_fraction_mex_8: Optional[Set[str]] = None,
valid_fraction_ame: Optional[Set[str]] = None,
valid_product_codes_cp: Optional[Set[str]] = None,
) -> Optional[Dict[str, Any]]:
"""Validación parcial (modo Act, clase existente): solo CLASE + validaciones_clase."""
err = validate_row_required(row, line_num)
if err:
return err
return validaciones_clase(row, line_num, valid_material_keys, valid_uom_codes)
return validaciones_clase(
row,
line_num,
valid_material_keys,
valid_uom_codes,
valid_fraction_mex_8=valid_fraction_mex_8,
valid_fraction_ame=valid_fraction_ame,
valid_product_codes_cp=valid_product_codes_cp,
)