feature/abstraccion-de-funcionalidades-y-manejo-por-tareas-en-comun

backend/api/v1/modules/a76/layouts_csv/classes/common/__init__.py

@@ -0,0 +1 @@
+# common validators, mappers, fk_loader for classes CSV import

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

@@ -0,0 +1,64 @@
+"""
+Helpers reutilizables para validación de filas CSV (clases de materiales).
+"""
+from typing import Dict, Any, Optional
+
+
+def check_required(row: Dict[str, Any], col: str, line_num: int) -> Optional[Dict[str, Any]]:
+    val = (row.get(col) or "").strip()
+    if not val:
+        return {"line": line_num, "col": col, "msg": "Requerido"}
+    return None
+
+
+def check_max_length(
+    row: Dict[str, Any],
+    col: str,
+    max_len: int,
+    line_num: int,
+    required: bool = False,
+) -> Optional[Dict[str, Any]]:
+    val = (row.get(col) or "").strip()
+    if not val:
+        if required:
+            return {"line": line_num, "col": col, "msg": "Requerido"}
+        return None
+    if len(val) > max_len:
+        return {"line": line_num, "col": col, "msg": f"Máximo {max_len} caracteres"}
+    return None
+
+
+def check_int_range(
+    row: Dict[str, Any],
+    col: str,
+    line_num: int,
+    min_val: int,
+    max_val: int,
+) -> Optional[Dict[str, Any]]:
+    """Solo valida si hay valor; devuelve error si no es int o está fuera de rango."""
+    val = row.get(col)
+    if val is None or val == "":
+        return None
+    try:
+        v = int(val)
+        if v < min_val or v > max_val:
+            return {"line": line_num, "col": col, "msg": "Valor fuera de rango"}
+    except (ValueError, TypeError):
+        return {"line": line_num, "col": col, "msg": "Debe ser número entero"}
+    return None
+
+
+def check_in_set(
+    row: Dict[str, Any],
+    col: str,
+    line_num: int,
+    allowed: Optional[set],
+    msg: str = "No existe en el catálogo",
+) -> Optional[Dict[str, Any]]:
+    """Solo valida si hay valor y allowed no es None."""
+    val = (row.get(col) or "").strip()
+    if not val or allowed is None:
+        return None
+    if val not in allowed:
+        return {"line": line_num, "col": col, "msg": msg}
+    return None

backend/api/v1/modules/a76/layouts_csv/classes/common/fk_loader.py

@@ -0,0 +1,37 @@
+"""
+Carga de conjuntos FK para validación/mapeo de import CSV de clases de materiales.
+"""
+from typing import Set, Tuple
+
+from core.database import CoreSessionLocal
+
+
+def load_classes_fk_sets(
+    tenant_id: int,
+    company_id: int,
+) -> Tuple[Set[str], Set[str]]:
+    """
+    Carga valid_material_keys (MaterialType.key) y valid_uom_codes (UnitOfMeasure.code).
+    Devuelve (valid_material_keys, valid_uom_codes).
+    """
+    valid_material_keys: Set[str] = set()
+    valid_uom_codes: 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
+            for m in session.query(MaterialType.key).all():
+                valid_material_keys.add(m[0])
+            for u in (
+                session.query(UnitOfMeasure.code)
+                .filter(
+                    UnitOfMeasure.tenant_id == tenant_id,
+                    UnitOfMeasure.company_id == company_id,
+                )
+                .all()
+            ):
+                valid_uom_codes.add(u[0])
+    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

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

@@ -0,0 +1,62 @@
+"""
+Mapeo fila CSV → datos para Class (clases de materiales).
+"""
+from typing import Dict, Any, Optional, Set
+
+
+def _str_or_none(val: Any, max_len: Optional[int] = None) -> Optional[str]:
+    if val is None:
+        return None
+    s = str(val).strip()
+    if not s:
+        return None
+    if max_len and len(s) > max_len:
+        return s[:max_len]
+    return s
+
+
+def _int_or_none(val: Any) -> Optional[int]:
+    if val is None or val == "":
+        return None
+    try:
+        return int(val)
+    except (ValueError, TypeError):
+        return None
+
+
+def row_to_class_data(
+    row_norm: Dict[str, Any],
+    valid_material_keys: Set[str],
+    valid_uom_codes: Set[str],
+) -> 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 = _str_or_none(row_norm.get("CLASE"), 8)
+    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)
+    if material_key and material_key not in valid_material_keys:
+        material_key = None
+    unit_of_measure = _str_or_none(row_norm.get("UNIMED"), 5)
+    if unit_of_measure and unit_of_measure not in valid_uom_codes:
+        unit_of_measure = None
+    fraction = _str_or_none(row_norm.get("FRACCION"), 20)
+    us_fraction = _str_or_none(row_norm.get("FRACCIONAME"), 16)
+    sub_key = _str_or_none(row_norm.get("CLAVESUB"), 5)
+    physical_review = _int_or_none(row_norm.get("REVFISICA"))
+    iva_exempt_fraction = _str_or_none(row_norm.get("FRACCIONEXENTAIVA"), 4)
+
+    return {
+        "class_code": class_code,
+        "description_es": desc_es,
+        "description_en": desc_en,
+        "material_key": material_key,
+        "unit_of_measure": unit_of_measure,
+        "fraction": fraction,
+        "us_fraction": us_fraction,
+        "sub_key": sub_key,
+        "physical_review": physical_review,
+        "iva_exempt_fraction": iva_exempt_fraction,
+    }