diff --git a/backend/api/v1/modules/a76/csv_templates/registry.py b/backend/api/v1/modules/a76/csv_templates/registry.py index 2e933200..1f4dc3dc 100644 --- a/backend/api/v1/modules/a76/csv_templates/registry.py +++ b/backend/api/v1/modules/a76/csv_templates/registry.py @@ -13,7 +13,10 @@ from api.v1.modules.a76.layouts_csv.facturas.template_config import ( ) from api.v1.modules.a76.layouts_csv.parts.template_config import TEMPLATE_COLUMNS as PARTS_TEMPLATE_COLUMNS from api.v1.modules.a76.layouts_csv.boms.template_config import TEMPLATE_COLUMNS as BOMS_TEMPLATE_COLUMNS -from api.v1.modules.a76.layouts_csv.classes.template_config import TEMPLATE_COLUMNS as CLASSES_TEMPLATE_COLUMNS +from api.v1.modules.a76.layouts_csv.classes.template_config import ( + TEMPLATE_COLUMNS as CLASSES_TEMPLATE_COLUMNS, + TEMPLATE_DOWNLOAD_HEADERS as CLASSES_TEMPLATE_DOWNLOAD_HEADERS, +) from api.v1.modules.a76.layouts_csv.customs_brokers.template_config import ( TEMPLATE_COLUMNS as CUSTOMS_BROKERS_TEMPLATE_COLUMNS, ) @@ -63,8 +66,12 @@ def _build_registry() -> Dict[str, List[str]]: # boms registry["boms"] = _canonicals_from_columns(BOMS_TEMPLATE_COLUMNS.get("boms")) - # material_classes - registry["material_classes"] = _canonicals_from_columns(CLASSES_TEMPLATE_COLUMNS.get("material_classes")) + # material_classes: cabeceras de descarga según plantilla usuario (CLAVE, CLASE, DESCRIPCION, etc.) + registry["material_classes"] = ( + CLASSES_TEMPLATE_DOWNLOAD_HEADERS + if CLASSES_TEMPLATE_DOWNLOAD_HEADERS + else _canonicals_from_columns(CLASSES_TEMPLATE_COLUMNS.get("material_classes")) + ) # customs_brokers registry["customs_brokers"] = _canonicals_from_columns(CUSTOMS_BROKERS_TEMPLATE_COLUMNS.get("customs_brokers")) diff --git a/backend/api/v1/modules/a76/layouts_csv/classes/TEST_DATA_README.md b/backend/api/v1/modules/a76/layouts_csv/classes/TEST_DATA_README.md new file mode 100644 index 00000000..29cf1111 --- /dev/null +++ b/backend/api/v1/modules/a76/layouts_csv/classes/TEST_DATA_README.md @@ -0,0 +1,36 @@ +# Datos de prueba – Import CSV Clases de Materiales + +## Archivo: `test_data_clases.csv` + +### Cómo usar + +1. **Catálogos necesarios**: El CSV usa códigos que deben existir en tu tenant/empresa: + - **TIPO DE MATERIAL**: p. ej. `MAT01` (catálogo Tipos de Activo Fijo / MaterialType). + - **U.M. COMERCIAL**: p. ej. `KG`, `MTR` (catálogo Unidades de Medida por tenant/company). + - **FRACCION ARANCELARIA**: mínimo 8 caracteres y que exista en Fracciones Arancelarias Sifr@ o Histórico (p. ej. `12345678`, `87654321`). + - **FRACCION AMERICANA**: que exista en catálogo Fracciones Americanas (p. ej. `1234567890123456`, `6543210987654321`). + + Si no tienes esos códigos, crea al menos uno de cada catálogo o sustituye en el CSV por códigos reales de tu base. + +2. **Filas válidas (para probar carga correcta)** + - Líneas 2 y 3: `CLASE01`, `CLASE02` — completas y correctas (ajusta tipos de material, U.M. y fracciones a tus catálogos). + +3. **Filas que disparan errores (para probar mensajes)** + - **Línea 4**: Todas las celdas vacías → *"La columna de Clase esta vacio..."* (Col. A vacía). + - **Línea 5**: `CLASE04` sin B,D,E,F → *"Existen campos vacios que son obligatorios..."* (validación completa). + - **Línea 6**: Clase de más de 8 caracteres (`ABCDEFGHIJ`) → *"La Clase: ... supera la longitud de caracteres"*. + - **Línea 7**: Tipo de material `INVALIDO` (no en catálogo) → *"(Col. D) El Tipo de Activo Fijo: INVALIDO no existe..."*. + - **Línea 8**: U.M. `UMINE` (no en catálogo) → *"(Col. E) La Unidad de Medida Comercial: UMINE no existe..."*. + - **Línea 9**: Fracción mexicana `1234` (< 8 caracteres) → *"La Fraccion 1234 no alcanza la longitud de 8 caracteres"*. + - **Línea 10**: Fracción americana `INVALIDO` (no en catálogo) → *"(Col. G) La Fraccion Americana: INVALIDO no existe..."*. + - **Línea 11**: Tasa de depreciación `150` (> 100) → *"(Col. H) La Tasa de Depreciación: 150 no puede ser mayor al 100 %"*. + - **Línea 12**: Código producto CP `9999` (si existe catálogo CP y no está) → *"(Col. J) La clase: CLASE10 tiene asignado un código de producto inexistente"*. + +### CSV sin cabecera (opcional) + +Si quieres probar detección de “primera fila = datos”, usa un archivo cuya primera línea sea una fila de datos (no "CLAVE CLASE"...). El sistema usará `TEMPLATE_DOWNLOAD_HEADERS` como cabecera y la primera línea como dato. + +### Valores mínimos para una fila válida + +Con **Modo Actualizar** y clase ya existente, solo es obligatoria la Col. A (CLAVE CLASE). +Con **Modo Reemplazar** o clase nueva, son obligatorios: A, B, D, E, F (mínimo 8 caracteres y en catálogo), y el resto según reglas (H ≤ 100, J en catálogo CP si aplica). 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 9dfc8de8..eb765a83 100644 --- a/backend/api/v1/modules/a76/layouts_csv/classes/tasks.py +++ b/backend/api/v1/modules/a76/layouts_csv/classes/tasks.py @@ -16,7 +16,7 @@ from ..common import normalize as common_normalize 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 .template_config import row_from_template, detect_headers_or_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 @@ -43,8 +43,9 @@ def scan_file(self, job_id: str, config: str = None): error_path = common_storage.error_path_for_job(JOB_TYPE, job_id) + fieldnames, has_header = detect_headers_or_data(file_path, common_normalize.normalize_header) try: - total_rows = common_csv.count_csv_rows(file_path) + total_rows = common_csv.count_csv_rows(file_path, has_header=has_header) except Exception as e: return {"status": "failed", "error": str(e)} @@ -81,7 +82,7 @@ def scan_file(self, job_id: str, config: str = None): try: with open(error_path, "w", encoding="utf-8") as f_err: - for i, row in common_csv.iter_csv_rows(file_path): + for i, row in common_csv.iter_csv_rows(file_path, fieldnames=fieldnames): if i % 500 == 0: self.update_state( state="PROGRESS", @@ -160,6 +161,8 @@ def insert_valid_rows(self, job_id: str): response = None meta_path = common_meta.get_meta_path(file_path) + fieldnames, _ = detect_headers_or_data(file_path, common_normalize.normalize_header) + try: with CoreSessionLocal() as session: existing_by_code = {} @@ -171,7 +174,7 @@ def insert_valid_rows(self, job_id: str): if key: existing_by_code[key] = c - for i, row in common_csv.iter_csv_rows(file_path): + for i, row in common_csv.iter_csv_rows(file_path, fieldnames=fieldnames): if i in error_lines: continue diff --git a/backend/api/v1/modules/a76/layouts_csv/classes/template_config.py b/backend/api/v1/modules/a76/layouts_csv/classes/template_config.py index fc1fa635..b4c9623e 100644 --- a/backend/api/v1/modules/a76/layouts_csv/classes/template_config.py +++ b/backend/api/v1/modules/a76/layouts_csv/classes/template_config.py @@ -1,22 +1,77 @@ """ Configuración de plantilla CSV para Clases de Materiales (EstructuraCatClasesAF.xls). +Cabeceras de descarga = Clarion: CLAVE CLASE, DESCRIPCION ESPAÑOL, DESCRIPCION INGLES, etc. """ +import csv +import io +from typing import Dict, List, Any, Optional, Tuple -from typing import Dict, List, Any, Optional + +# Valores que indican que la primera fila es cabecera (primera columna normalizada) +FIRST_COLUMN_HEADER_VALUES = ("CLAVE CLASE", "CLASE") + + +def detect_headers_or_data( + file_path: str, + normalize_header_fn, + encoding: str = "utf-8-sig", +) -> Tuple[Optional[List[str]], bool]: + """ + Lee la primera línea del CSV y decide si es cabecera o dato. + Devuelve (fieldnames, has_header). + - Si la primera celda normalizada está en FIRST_COLUMN_HEADER_VALUES -> has_header=True, fieldnames=None + (la primera fila es cabecera; iter_csv_rows sin fieldnames). + - Si no -> has_header=False, fieldnames=TEMPLATE_DOWNLOAD_HEADERS (la primera fila es dato). + """ + try: + with open(file_path, "r", encoding=encoding) as f: + sample = f.read(2048) + except Exception: + return None, True + lines = sample.splitlines() + if not lines: + return None, True + try: + dialect = csv.Sniffer().sniff(sample, delimiters=",;\t") + except Exception: + dialect = csv.excel + reader = csv.reader(io.StringIO(lines[0]), dialect=dialect) + first_row = next(reader, None) + if not first_row: + return None, True + first_cell = (first_row[0] or "").strip() + first_cell_norm = normalize_header_fn(first_cell) + if first_cell_norm in FIRST_COLUMN_HEADER_VALUES: + return None, True + return list(TEMPLATE_DOWNLOAD_HEADERS), False + +# Cabeceras que se escriben al descargar la plantilla CSV (igual que Clarion EstructuraCatClasesAF) +TEMPLATE_DOWNLOAD_HEADERS: List[str] = [ + "CLAVE CLASE", + "DESCRIPCION ESPAÑOL", + "DESCRIPCION INGLES", + "TIPO DE MATERIAL", + "U.M. COMERCIAL", + "FRACCION ARANCELARIA", + "FRACCION AMERICANA", + "TASA DE DEPRECIACION", + "REVISION FISICA (1/0)", + "CODIGO DE PRODUCTO/SERVICIO CP", +] TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = { "material_classes": [ - {"canonical": "CLASE", "aliases": ["CLASS", "CODIGO", "CLASE CODIGO"]}, - {"canonical": "DESCRIPCIONE", "aliases": ["DESCRIPCION", "DESCRIPCION ES"]}, - {"canonical": "DESCRIPCIONI", "aliases": ["DESCRIPCION EN", "DESCRIPTION"]}, - {"canonical": "CLAVEMAT", "aliases": ["MATERIAL", "TIPOMAT", "CLAVE MATERIAL"]}, - {"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": "CLASE", "aliases": ["CLAVE CLASE", "CLASS", "CODIGO", "CLASE CODIGO"]}, + {"canonical": "DESCRIPCIONE", "aliases": ["DESCRIPCION ESPAÑOL", "DESCRIPCION", "DESCRIPCION ES", "DESCRIPCION ESPAÑOL"]}, + {"canonical": "DESCRIPCIONI", "aliases": ["DESCRIPCION INGLES", "DESCRIPCION EN", "DESCRIPTION", "DESCRIPCION INGLES"]}, + {"canonical": "CLAVEMAT", "aliases": ["TIPO DE MATERIAL", "MATERIAL", "TIPOMAT", "CLAVE MATERIAL"]}, + {"canonical": "UNIMED", "aliases": ["U.M. COMERCIAL", "UNIDAD MEDIDA", "UNIT", "UOM", "TIPO DE MU.M.", "U.M. COMERCIAL"]}, + {"canonical": "FRACCION", "aliases": ["FRACCION ARANCELARIA", "FRACCION MEX", "COM FRACCION"]}, + {"canonical": "FRACCIONAME", "aliases": ["FRACCION AMERICANA", "FRACCION USA", "US FRACTION", "FRACCION"]}, + {"canonical": "TASADEPRECIA", "aliases": ["TASA DE DEPRECIACION", "TASA DEPRECIACIÓN", "TASA DEPRECIACION"]}, {"canonical": "CLAVESUB", "aliases": ["SUB KEY", "CLAVE SUB"]}, - {"canonical": "REVFISICA", "aliases": ["REV FISICA", "PHYSICAL REVIEW"]}, - {"canonical": "FRACCIONEXENTAIVA", "aliases": ["EXENTA IVA", "FRACCION EXENTA IVA"]}, + {"canonical": "REVFISICA", "aliases": ["REVISION FISICA (1/0)", "REV FISICA", "PHYSICAL REVIEW", "TASA DE REVISION", "REVISION FISICA"]}, + {"canonical": "FRACCIONEXENTAIVA", "aliases": ["CODIGO DE PRODUCTO/SERVICIO CP", "EXENTA IVA", "FRACCION EXENTA IVA", "CODIGO PRODUCTO SERVICIO CP"]}, ], } @@ -43,4 +98,10 @@ def row_from_template(row: Dict[str, Any], normalize_header_fn) -> Dict[str, Any key_norm = normalize_header_fn(csv_header) if key_norm in lookup: out[lookup[key_norm]] = value + elif key_norm.startswith("CLAVE CLASE"): + # CSV leído con delimitador incorrecto: primera columna es "CLAVE CLASE,..." -> usar primer valor como CLASE + if "CLASE" not in out and value: + first_val = (value.split(",")[0] if "," in str(value) else value).strip() + if first_val: + out["CLASE"] = first_val return out 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 a500a4ab..5f4bd02b 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 @@ -6,9 +6,7 @@ from typing import Dict, Any, Optional, Set from ..common.common_validators import ( check_max_length, - check_min_length, check_int_range, - check_in_set, check_decimal_max, ) @@ -17,26 +15,37 @@ MSG_CLASE_VACIO_SOLUCION = "Capturar en la Columna A una Clase nueva o una ya ex def validate_row_required(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]: - """CLASE obligatorio; mensaje Clarion si está vacío.""" - val = (row.get("CLASE") or "").strip() + """CLASE obligatorio; mensaje Clarion si está vacío. Acepta clave 'CLASE' o 'CLAVE CLASE' (por si el CSV no se normalizó).""" + val = (row.get("CLASE") or row.get("CLAVE 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 { + "line": line_num, + "col": "CLASE", + "msg": f"Error: (Col. A) La Clase: {val} supera la longitud de caracteres. Capturar en la columna A una Clase de 8 caracteres como máximo.", + } 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 + cols_labels = [ + ("DESCRIPCIONE", "(Col.B) Descripción Español"), + ("CLAVEMAT", "(Col.D) Tipo de Material"), + ("UNIMED", "(Col.E) U.M. Comercial"), + ("FRACCION", "(Col.F) Fraccion Arancelaria Mex."), + ] + missing = [(col, label) for col, label in cols_labels if not (row.get(col) or "").strip()] + if not missing: + return None + campos = ", ".join(label for _, label in missing) + first_col = missing[0][0] + return { + "line": line_num, + "col": first_col, + "msg": f"Existen campos vacios que son obligatorios, es la {campos}. Revisar la línea del archivo y capturar los campos con la información correcta.", + } def validate_row_lengths(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]: @@ -59,10 +68,16 @@ def validate_row_lengths(row: Dict[str, Any], line_num: int) -> Optional[Dict[st 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.", - ) + val = (row.get("FRACCION") or "").strip() + if not val: + return None + if len(val) < 8: + return { + "line": line_num, + "col": "FRACCION", + "msg": f"La Fraccion {val} no alcanza la longitud de 8 caracteres. Capturar en la columna F una Fracción de 8 caracteres.", + } + return None def validate_row_types(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]: @@ -75,18 +90,20 @@ def validate_row_fks( valid_material_keys: Optional[Set[str]], valid_uom_codes: Optional[Set[str]], ) -> Optional[Dict[str, Any]]: - err = check_in_set( - row, "CLAVEMAT", line_num, valid_material_keys, - "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, - "Error: (Col. E) La Unidad de Medida Comercial no existe en el Catálogo de U.M.", - ) - if err: - return err + val_d = (row.get("CLAVEMAT") or "").strip() + if val_d and valid_material_keys is not None and val_d not in valid_material_keys: + return { + "line": line_num, + "col": "CLAVEMAT", + "msg": f"Error: (Col. D) El Tipo de Activo Fijo: {val_d} no existe en el Catálogo de Tipos de Activo Fijo. Revisar este Tipo de Activo Fijo en el archivo, en caso de ser correcto actualice los catálogos fijos.", + } + val_e = (row.get("UNIMED") or "").strip() + if val_e and valid_uom_codes is not None and val_e not in valid_uom_codes: + return { + "line": line_num, + "col": "UNIMED", + "msg": f"Error: (Col. E) La Unidad de Medida Comercial: {val_e} no existe en el Catálogo de U.M. Revisar esta Unidad de Medida en el archivo, en caso de ser correcta actualice los catálogos.", + } return None @@ -150,10 +167,11 @@ def validate_row_tasa_depreciacion(row: Dict[str, Any], line_num: int) -> Option 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.", - ) + err = check_decimal_max(row, "TASADEPRECIA", line_num, 100.0, msg=None) + if err and "100" in (err.get("msg") or ""): + v = (row.get("TASADEPRECIA") or "").strip() + err["msg"] = f"Error: (Col. H) La Tasa de Depreciación: {v} no puede ser mayor al 100 %. Ajustar la Tasa de Depreciacion." + return err def validate_row_codigo_producto_cp( diff --git a/backend/api/v1/modules/a76/layouts_csv/common/csv_reader.py b/backend/api/v1/modules/a76/layouts_csv/common/csv_reader.py index 3db621db..a0e783b4 100644 --- a/backend/api/v1/modules/a76/layouts_csv/common/csv_reader.py +++ b/backend/api/v1/modules/a76/layouts_csv/common/csv_reader.py @@ -1,14 +1,21 @@ """ Lectura de CSV con detección de delimitador (compartida por layouts_csv). +Si se pasa fieldnames, no se usa la primera fila como cabecera y se toma como dato (CSV sin cabeceras). """ import csv -from typing import Iterator, Tuple, Dict, Any +import io +from typing import Iterator, Tuple, Dict, Any, Optional, List -def iter_csv_rows(file_path: str) -> Iterator[Tuple[int, Dict[str, Any]]]: +def iter_csv_rows( + file_path: str, + fieldnames: Optional[List[str]] = None, +) -> Iterator[Tuple[int, Dict[str, Any]]]: """ Abre el CSV, detecta dialecto y devuelve (line_num, row_dict) por cada fila. line_num empieza en 1 (primera fila de datos). + Si fieldnames es None: la primera fila del archivo se usa como cabecera (comportamiento por defecto). + Si fieldnames es una lista: no se usa cabecera; la primera fila se considera dato y se usan fieldnames como columnas. """ with open(file_path, "r", encoding="utf-8-sig") as f: sample = f.read(2048) @@ -17,12 +24,18 @@ def iter_csv_rows(file_path: str) -> Iterator[Tuple[int, Dict[str, Any]]]: dialect = csv.Sniffer().sniff(sample, delimiters=",;\t") except Exception: dialect = "excel" - reader = csv.DictReader(f, dialect=dialect) - for i, row in enumerate(reader, start=1): - yield i, row + if fieldnames: + reader = csv.DictReader(f, fieldnames=fieldnames, dialect=dialect) + for i, row in enumerate(reader, start=1): + yield i, dict(row) + else: + reader = csv.DictReader(f, dialect=dialect) + for i, row in enumerate(reader, start=1): + yield i, row -def count_csv_rows(file_path: str) -> int: - """Cuenta filas del CSV (sin contar cabecera).""" +def count_csv_rows(file_path: str, has_header: bool = True) -> int: + """Cuenta filas del CSV. Si has_header=True (por defecto), no cuenta la cabecera.""" with open(file_path, "r", encoding="utf-8-sig") as f: - return sum(1 for _ in f) - 1 + total_lines = sum(1 for _ in f) + return total_lines if not has_header else max(0, total_lines - 1)