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

6feature/clarion-template-validations

Browse Source
This commit is contained in:
hreyes
2026-03-04 12:49:34 -07:00
parent a081a61aa1
commit a6b60a95fe
6 changed files with 198 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
backend/api/v1/modules/a76/csv_templates/registry.py
View File

@@ -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"))

36
backend/api/v1/modules/a76/layouts_csv/classes/TEST_DATA_README.md Normal file
View File

@@ -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).

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

@@ -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

83
backend/api/v1/modules/a76/layouts_csv/classes/template_config.py
View File

@@ -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

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

@@ -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."}
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(

23
backend/api/v1/modules/a76/layouts_csv/common/csv_reader.py
View File

@@ -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"
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)