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

feature/validaciones-clarion-classes-cav

Browse Source
This commit is contained in:
hreyes
2026-03-04 10:11:23 -07:00
parent e83205caf9
commit ccafe8d16c
7 changed files with 218 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
backend/api/v1/modules/a76/layouts_csv/classes/common/common_validators.py
View File

@@ -28,6 +28,26 @@ def check_max_length(
return None
def check_min_length(
row: Dict[str, Any],
col: str,
min_len: int,
line_num: int,
msg: Optional[str] = None,
) -> Optional[Dict[str, Any]]:
"""Solo valida si hay valor; error si longitud menor que min_len."""
val = (row.get(col) or "").strip()
if not val:
return None
if len(val) < min_len:
return {
"line": line_num,
"col": col,
"msg": msg or f"Mínimo {min_len} caracteres",
}
return None
def check_int_range(
row: Dict[str, Any],
col: str,

24
backend/api/v1/modules/a76/layouts_csv/classes/common/mappers.py
View File

@@ -31,9 +31,10 @@ def row_to_class_data(
) -> Dict[str, Any]:
"""
Mapea una fila normalizada del CSV a un diccionario de datos para Class.
Ajusta material_key y unit_of_measure a None si no están en los conjuntos.
class_code se normaliza a mayúsculas (Clip(Upper) Clarion).
"""
class_code = _str_or_none(row_norm.get("CLASE"), 8)
raw_clase = _str_or_none(row_norm.get("CLASE"), 8)
class_code = raw_clase.upper() if raw_clase else None
desc_es = _str_or_none(row_norm.get("DESCRIPCIONE"), 500)
desc_en = _str_or_none(row_norm.get("DESCRIPCIONI"), 500)
material_key = _str_or_none(row_norm.get("CLAVEMAT"), 10)
@@ -60,3 +61,22 @@ def row_to_class_data(
"physical_review": physical_review,
"iva_exempt_fraction": iva_exempt_fraction,
}
def row_to_class_data_merge_existing(
row_norm: Dict[str, Any],
existing_data: Dict[str, Any],
valid_material_keys: Set[str],
valid_uom_codes: Set[str],
) -> Dict[str, Any]:
"""
Para modo actualizar (parcial): valores del CSV si no vacíos, sino los de la clase existente (Clarion VALIDA_PARCIAL_CLASE).
"""
data = row_to_class_data(row_norm, valid_material_keys, valid_uom_codes)
if not data["class_code"]:
return data
for key in ("description_es", "description_en", "material_key", "unit_of_measure",
"fraction", "us_fraction", "sub_key", "physical_review", "iva_exempt_fraction"):
if data.get(key) is None or (isinstance(data[key], str) and not data[key].strip()):
data[key] = existing_data.get(key)
return data

4
backend/api/v1/modules/a76/layouts_csv/classes/routes.py
View File

@@ -40,6 +40,8 @@ def _get_redis():
async def upload_import_file(
file: UploadFile = File(...),
company_id: int = Query(..., description="Company ID"),
actualizar: bool = Query(False, description="Modo actualizar (ACT): validación parcial si la clase existe"),
siempre_toda: bool = Query(False, description="Forzar siempre validación completa"),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
@@ -60,6 +62,8 @@ async def upload_import_file(
"company_id": company_id,
"user_id": current_user.get("id"),
"template_id": "material_classes",
"actualizar": actualizar,
"siempre_toda": siempre_toda,
}
try:

88
backend/api/v1/modules/a76/layouts_csv/classes/tasks.py
View File

@@ -17,8 +17,8 @@ from ..common import csv_reader as common_csv
from ..common import meta as common_meta
from ..common import responses as common_responses
from .template_config import row_from_template
from .validators import validate_row_class
from .common.mappers import row_to_class_data
from .validators import validate_row_class, validate_row_class_partial
from .common.mappers import row_to_class_data, row_to_class_data_merge_existing
from .common.fk_loader import load_classes_fk_sets
logger = logging.getLogger(__name__)
@@ -53,8 +53,25 @@ def scan_file(self, job_id: str, config: str = None):
except ValueError as e:
return {"status": "failed", "error": str(e)}
meta = common_meta.load_meta(file_path)
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)
from api.v1.modules.a76.classes.models import Class
existing_class_codes = set()
try:
with CoreSessionLocal() as session:
for c in session.query(Class.class_code).filter(
Class.tenant_id == tenant_id,
Class.company_id == company_id,
).all():
if c[0]:
existing_class_codes.add(c[0].strip().upper())
except Exception as e:
logger.warning("Classes import: could not load existing class codes: %s", e)
error_count = 0
processed_rows = 0
errors_detail: List[Dict[str, Any]] = []
@@ -74,6 +91,9 @@ def scan_file(self, job_id: str, config: str = None):
i,
valid_material_keys=valid_material_keys,
valid_uom_codes=valid_uom_codes,
actualizar=actualizar,
siempre_toda=siempre_toda,
existing_class_codes=existing_class_codes,
)
if err:
error_count += 1
@@ -119,6 +139,10 @@ def insert_valid_rows(self, job_id: str):
except ValueError as e:
return {"status": "failed", "error": str(e)}
meta = common_meta.load_meta(file_path)
actualizar = meta.get("actualizar", False)
siempre_toda = meta.get("siempre_toda", False)
from api.v1.modules.a76.classes.models import Class
valid_material_keys, valid_uom_codes = load_classes_fk_sets(tenant_id, company_id)
@@ -131,25 +155,39 @@ def insert_valid_rows(self, job_id: str):
try:
with CoreSessionLocal() as session:
existing_by_code = {
c.class_code: c
for c in session.query(Class).filter(
Class.tenant_id == tenant_id,
Class.company_id == company_id,
).all()
}
existing_by_code = {}
for c in session.query(Class).filter(
Class.tenant_id == tenant_id,
Class.company_id == company_id,
).all():
key = (c.class_code or "").strip().upper()
if key:
existing_by_code[key] = c
for i, row in common_csv.iter_csv_rows(file_path):
if i in error_lines:
continue
row_norm = row_from_template(row, common_normalize.normalize_header)
err = validate_row_class(
row_norm,
i,
valid_material_keys=valid_material_keys,
valid_uom_codes=valid_uom_codes,
)
class_code_raw = (row_norm.get("CLASE") or "").strip().upper()[:8]
use_partial = actualizar and class_code_raw and class_code_raw in existing_by_code and not siempre_toda
if use_partial:
err = validate_row_class_partial(
row_norm, i,
valid_material_keys=valid_material_keys,
valid_uom_codes=valid_uom_codes,
)
else:
err = validate_row_class(
row_norm,
i,
valid_material_keys=valid_material_keys,
valid_uom_codes=valid_uom_codes,
actualizar=actualizar,
siempre_toda=siempre_toda,
existing_class_codes=set(existing_by_code.keys()),
)
if err:
skipped_invalid += 1
skipped_details.append({
@@ -158,7 +196,25 @@ def insert_valid_rows(self, job_id: str):
})
continue
data = row_to_class_data(row_norm, valid_material_keys, valid_uom_codes)
if use_partial:
existing = existing_by_code.get(class_code_raw)
existing_data = {
"description_es": existing.description_es,
"description_en": existing.description_en,
"material_key": existing.material_key,
"unit_of_measure": existing.unit_of_measure,
"fraction": existing.fraction,
"us_fraction": existing.us_fraction,
"sub_key": existing.sub_key,
"physical_review": existing.physical_review,
"iva_exempt_fraction": existing.iva_exempt_fraction,
}
data = row_to_class_data_merge_existing(
row_norm, existing_data, valid_material_keys, valid_uom_codes,
)
else:
data = row_to_class_data(row_norm, valid_material_keys, valid_uom_codes)
class_code = data.get("class_code")
if not class_code:
skipped_invalid += 1

4
backend/api/v1/modules/a76/layouts_csv/classes/validators/__init__.py
View File

@@ -1,3 +1,3 @@
from .create import validate_row_class
from .create import validate_row_class, validate_row_class_partial
__all__ = ["validate_row_class"]
__all__ = ["validate_row_class", "validate_row_class_partial"]

61
backend/api/v1/modules/a76/layouts_csv/classes/validators/common.py
View File

@@ -5,13 +5,36 @@ from typing import Dict, Any, Optional, Set
from ..common.common_validators import (
check_max_length,
check_min_length,
check_int_range,
check_in_set,
)
MSG_CLASE_VACIO = "Error: (Col. A) La columna de Clase esta vacio y no se pueden hacer las validaciones."
MSG_CLASE_VACIO_SOLUCION = "Capturar en la Columna A una Clase nueva o una ya existente a la cual desee actualizar campos"
def validate_row_required(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
return check_max_length(row, "CLASE", 8, line_num, required=True)
"""CLASE obligatorio; mensaje Clarion si está vacío."""
val = (row.get("CLASE") or "").strip()
if not val:
return {"line": line_num, "col": "CLASE", "msg": f"{MSG_CLASE_VACIO} {MSG_CLASE_VACIO_SOLUCION}"}
if len(val) > 8:
return {"line": line_num, "col": "CLASE", "msg": "Máximo 8 caracteres"}
return None
def validate_row_required_full(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
"""Obligatorios en validación completa: B, D, E, F (Clarion VALIDA_TODA_CLASE)."""
for col, label in [
("DESCRIPCIONE", "Descripción Español"),
("CLAVEMAT", "Tipo de Material"),
("UNIMED", "U.M. Comercial"),
("FRACCION", "Fraccion Arancelaria Mex."),
]:
if not (row.get(col) or "").strip():
return {"line": line_num, "col": col, "msg": f"Campo obligatorio: {label}. Revisar la línea y capturar los campos correctos."}
return None
def validate_row_lengths(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
@@ -32,6 +55,14 @@ def validate_row_lengths(row: Dict[str, Any], line_num: int) -> Optional[Dict[st
return None
def validate_row_fraction_min(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
"""FRACCION (Col F): si no vacío, mínimo 8 caracteres (Clarion VALIDACIONES_CLASE)."""
return check_min_length(
row, "FRACCION", 8, line_num,
msg="La Fraccion no alcanza la longitud de 8 caracteres.",
)
def validate_row_types(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
return check_int_range(row, "REVFISICA", line_num, -32768, 32767)
@@ -43,13 +74,37 @@ def validate_row_fks(
valid_uom_codes: Optional[Set[str]],
) -> Optional[Dict[str, Any]]:
err = check_in_set(
row, "CLAVEMAT", line_num, valid_material_keys, "Tipo de material no existe"
row, "CLAVEMAT", line_num, valid_material_keys,
"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, "Unidad de medida no existe"
row, "UNIMED", line_num, valid_uom_codes,
"La Unidad de Medida Comercial no existe en el Catálogo de U.M.",
)
if err:
return err
return None
def validaciones_clase(
row: Dict[str, Any],
line_num: int,
valid_material_keys: Optional[Set[str]],
valid_uom_codes: Optional[Set[str]],
) -> Optional[Dict[str, Any]]:
"""Reglas compartidas Clarion (VALIDACIONES_CLASE): longitudes, tipos, FKs. Sin obligatorios B,D,E,F."""
err = validate_row_lengths(row, line_num)
if err:
return err
err = validate_row_fraction_min(row, line_num)
if err:
return err
err = validate_row_types(row, line_num)
if err:
return err
err = validate_row_fks(row, line_num, valid_material_keys, valid_uom_codes)
if err:
return err
return None

56
backend/api/v1/modules/a76/layouts_csv/classes/validators/create.py
View File

@@ -1,13 +1,14 @@
"""
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).
"""
from typing import Dict, Any, Optional, Set
from .common import (
validate_row_required,
validate_row_lengths,
validate_row_types,
validate_row_fks,
validate_row_required_full,
validaciones_clase,
)
@@ -16,25 +17,48 @@ def validate_row_class(
line_num: int,
valid_material_keys: Optional[Set[str]] = None,
valid_uom_codes: Optional[Set[str]] = None,
actualizar: bool = False,
siempre_toda: bool = False,
existing_class_codes: Optional[Set[str]] = None,
) -> Optional[Dict[str, Any]]:
"""
Valida una fila de CSV de clases de materiales.
Encadena: requeridos → longitudes → tipos → FKs.
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).
siempre_toda fuerza TODA en todos los casos.
"""
err = validate_row_required(row, line_num)
if err:
return err
err = validate_row_lengths(row, line_num)
class_code = (row.get("CLASE") or "").strip().upper()[:8]
use_full = siempre_toda or not actualizar
if actualizar and existing_class_codes is not None and class_code in existing_class_codes:
use_full = False
if use_full:
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.",
}
return validaciones_clase(row, line_num, valid_material_keys, valid_uom_codes)
def validate_row_class_partial(
row: Dict[str, Any],
line_num: int,
valid_material_keys: Optional[Set[str]] = None,
valid_uom_codes: 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
err = validate_row_types(row, line_num)
if err:
return err
err = validate_row_fks(row, line_num, valid_material_keys, valid_uom_codes)
if err:
return err
return None
return validaciones_clase(row, line_num, valid_material_keys, valid_uom_codes)