diff --git a/backend/api/v1/modules/a76/layouts_csv/classes/common/common_validators.py b/backend/api/v1/modules/a76/layouts_csv/classes/common/common_validators.py index 9a35fd86..eda632a3 100644 --- a/backend/api/v1/modules/a76/layouts_csv/classes/common/common_validators.py +++ b/backend/api/v1/modules/a76/layouts_csv/classes/common/common_validators.py @@ -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, diff --git a/backend/api/v1/modules/a76/layouts_csv/classes/common/mappers.py b/backend/api/v1/modules/a76/layouts_csv/classes/common/mappers.py index 5bef32eb..d7380170 100644 --- a/backend/api/v1/modules/a76/layouts_csv/classes/common/mappers.py +++ b/backend/api/v1/modules/a76/layouts_csv/classes/common/mappers.py @@ -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 diff --git a/backend/api/v1/modules/a76/layouts_csv/classes/routes.py b/backend/api/v1/modules/a76/layouts_csv/classes/routes.py index 0b61efb2..3af7cee1 100644 --- a/backend/api/v1/modules/a76/layouts_csv/classes/routes.py +++ b/backend/api/v1/modules/a76/layouts_csv/classes/routes.py @@ -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: diff --git a/backend/api/v1/modules/a76/layouts_csv/classes/tasks.py b/backend/api/v1/modules/a76/layouts_csv/classes/tasks.py index ba93da56..e5c128e1 100644 --- a/backend/api/v1/modules/a76/layouts_csv/classes/tasks.py +++ b/backend/api/v1/modules/a76/layouts_csv/classes/tasks.py @@ -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 diff --git a/backend/api/v1/modules/a76/layouts_csv/classes/validators/__init__.py b/backend/api/v1/modules/a76/layouts_csv/classes/validators/__init__.py index 83e82ef1..9d108067 100644 --- a/backend/api/v1/modules/a76/layouts_csv/classes/validators/__init__.py +++ b/backend/api/v1/modules/a76/layouts_csv/classes/validators/__init__.py @@ -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"] diff --git a/backend/api/v1/modules/a76/layouts_csv/classes/validators/common.py b/backend/api/v1/modules/a76/layouts_csv/classes/validators/common.py index 3bbc8cd9..14fa536b 100644 --- a/backend/api/v1/modules/a76/layouts_csv/classes/validators/common.py +++ b/backend/api/v1/modules/a76/layouts_csv/classes/validators/common.py @@ -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 diff --git a/backend/api/v1/modules/a76/layouts_csv/classes/validators/create.py b/backend/api/v1/modules/a76/layouts_csv/classes/validators/create.py index 50c6b922..4155d150 100644 --- a/backend/api/v1/modules/a76/layouts_csv/classes/validators/create.py +++ b/backend/api/v1/modules/a76/layouts_csv/classes/validators/create.py @@ -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)