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

feature/clarion-validations-parts-csv

Browse Source
This commit is contained in:
hreyes
2026-03-04 14:57:26 -07:00
parent f6f23dacd2
commit afc0a35f1c
13 changed files with 1087 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -11,7 +11,10 @@ from api.v1.modules.a76.layouts_csv.facturas.template_config import (
TEMPLATE_COLUMNS as IMPORTS_TEMPLATE_COLUMNS,
_resolve_template_columns as resolve_imports_template,
)
from api.v1.modules.a76.layouts_csv.parts.template_config import TEMPLATE_COLUMNS as PARTS_TEMPLATE_COLUMNS
from api.v1.modules.a76.layouts_csv.parts.template_config import (
TEMPLATE_COLUMNS as PARTS_TEMPLATE_COLUMNS,
TEMPLATE_DOWNLOAD_HEADERS as PARTS_TEMPLATE_DOWNLOAD_HEADERS,
)
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,
@@ -59,9 +62,16 @@ def _build_registry() -> Dict[str, List[str]]:
registry[tid] = _canonicals_from_columns(cols)
# part_numbers (parts); "items" usa la misma plantilla
part_cols = PARTS_TEMPLATE_COLUMNS.get("part_numbers")
registry["part_numbers"] = _canonicals_from_columns(part_cols)
registry["items"] = _canonicals_from_columns(part_cols)
registry["part_numbers"] = (
PARTS_TEMPLATE_DOWNLOAD_HEADERS
if PARTS_TEMPLATE_DOWNLOAD_HEADERS
else _canonicals_from_columns(PARTS_TEMPLATE_COLUMNS.get("part_numbers"))
)
registry["items"] = (
PARTS_TEMPLATE_DOWNLOAD_HEADERS
if PARTS_TEMPLATE_DOWNLOAD_HEADERS
else _canonicals_from_columns(PARTS_TEMPLATE_COLUMNS.get("part_numbers"))
)
# boms
registry["boms"] = _canonicals_from_columns(BOMS_TEMPLATE_COLUMNS.get("boms"))

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

@@ -124,6 +124,27 @@ async def get_import_status(job_id: str):
if isinstance(result, dict) and result.get("status") in ("finished", "warning"):
return result
# Si Celery devolvió el resultado como string (p. ej. JSON), parsear y devolver como scan si aplica
if isinstance(result, str):
try:
parsed = json.loads(result)
if isinstance(parsed, dict) and (
parsed.get("status") == "waiting_confirmation"
or (parsed.get("job_id") and "total_rows" in parsed)
):
return parsed
if isinstance(parsed, dict) and parsed.get("status") in ("finished", "warning"):
return parsed
except (json.JSONDecodeError, TypeError):
pass
# Si el resultado tiene forma de escaneo (waiting_confirmation), devolverlo para que el front muestre el modal
if isinstance(result, dict) and (
result.get("status") == "waiting_confirmation"
or (result.get("job_id") and "total_rows" in result)
):
return result
logger.warning("Classes import task %s failed: state=%s", job_id, task_result.state)
err_msg = None
tb = getattr(task_result, "traceback", None)

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

@@ -1,12 +1,26 @@
"""
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).
Si fieldnames es None, cabeceras vacías se normalizan a _COL_0_, _COL_1_, ... para no colapsar columnas.
"""
import csv
import io
from typing import Iterator, Tuple, Dict, Any, Optional, List
def _normalize_empty_headers(headers: List[str]) -> List[str]:
"""Sustituye cabeceras vacías por _COL_0_, _COL_1_, ... para que DictReader no colapse columnas."""
result: List[str] = []
empty_idx = 0
for h in headers:
if (h or "").strip() == "":
result.append(f"_COL_{empty_idx}_")
empty_idx += 1
else:
result.append(h)
return result
def iter_csv_rows(
file_path: str,
fieldnames: Optional[List[str]] = None,
@@ -29,7 +43,15 @@ def iter_csv_rows(
for i, row in enumerate(reader, start=1):
yield i, dict(row)
else:
reader = csv.DictReader(f, dialect=dialect)
first_line = f.readline()
if not first_line:
return
row_reader = csv.reader(io.StringIO(first_line), dialect=dialect)
raw_headers = next(row_reader, None)
if not raw_headers:
return
normalized = _normalize_empty_headers(raw_headers)
reader = csv.DictReader(f, fieldnames=normalized, dialect=dialect, restval="")
for i, row in enumerate(reader, start=1):
yield i, row

113
backend/api/v1/modules/a76/layouts_csv/parts/common/fk_loader.py
View File

@@ -1,27 +1,65 @@
"""
Carga de conjuntos FK para validación/mapeo de import CSV de partes.
Clarion: Clases, UOM, monedas, fracción Mex (+ histórico), países, sectores autorizados, excepción RFC.
"""
from typing import Set, Tuple
from typing import Set, Tuple, Optional
import logging
from core.database import CoreSessionLocal
logger = logging.getLogger(__name__)
# RFCs con excepción: E (UOM) obligatorio solo cuando D (Clase) está vacía; Preferencia solo cuando D no vacía
RFC_EXCEPTION_SET = {"CLA940831AZ5", "CTE980130518"}
def load_parts_fk_sets(
tenant_id: int,
company_id: int,
) -> Tuple[Set[str], Set[str], Set[str]]:
) -> Tuple[
Set[str],
Set[str],
Set[str],
Set[str],
Set[str],
Set[str],
bool,
bool,
]:
"""
Carga valid_class_codes, valid_uom_codes, valid_currency_codes desde BD.
Devuelve (valid_class_codes, valid_uom_codes, valid_currency_codes).
Carga conjuntos para validación CSV de partes (paridad Clarion).
Devuelve:
- valid_class_codes
- valid_uom_codes
- valid_currency_codes (claves moneda para G=MC y genéricas)
- valid_fraction_mex_8 (fracción Mex 8 chars: TariffFraction + HistoricalTariffFraction)
- valid_country_m3 (códigos país m3_key)
- authorized_sector_keys (sectores con authorized=True)
- company_has_prosec (Company.prosec)
- is_rfc_exception (Company.rfc en RFC_EXCEPTION_SET)
"""
valid_class_codes: Set[str] = set()
valid_uom_codes: Set[str] = set()
valid_currency_codes: Set[str] = set()
valid_fraction_mex_8: Set[str] = set()
valid_country_m3: Set[str] = set()
authorized_sector_keys: Set[str] = set()
company_has_prosec = False
is_rfc_exception = False
try:
with CoreSessionLocal() as session:
from api.v1.modules.a76.classes.models import Class
from api.v1.modules.a76.general_catalogs.units_of_measure.models import UnitOfMeasure
from api.v1.modules.public.reference_data.currency_types.models import CurrencyType
from api.v1.modules.a76.general_catalogs.company.models import Company
from api.v1.modules.public.reference_data.countries.models import Country
from api.v1.modules.public.reference_data.sectors.models import Sector
from api.v1.modules.a76.general_catalogs.fractions.tariff_fractions.models import TariffFraction
from api.v1.modules.a76.general_catalogs.fractions.historical_tariff_fractions.models import (
HistoricalTariffFraction,
)
for c in (
session.query(Class.class_code)
.filter(
@@ -30,7 +68,9 @@ def load_parts_fk_sets(
)
.all()
):
valid_class_codes.add(c[0])
if c[0]:
valid_class_codes.add((c[0] or "").strip().upper())
for u in (
session.query(UnitOfMeasure.code)
.filter(
@@ -39,10 +79,63 @@ def load_parts_fk_sets(
)
.all()
):
valid_uom_codes.add(u[0])
if u[0]:
valid_uom_codes.add((u[0] or "").strip().upper())
for cur in session.query(CurrencyType.code).all():
valid_currency_codes.add(cur[0])
if cur[0]:
valid_currency_codes.add((cur[0] or "").strip().upper())
company = (
session.query(Company)
.filter(Company.id == company_id)
.first()
)
if company:
company_has_prosec = bool(company.prosec)
rfc = (company.rfc or "").strip().upper()
is_rfc_exception = rfc in RFC_EXCEPTION_SET
for row in session.query(Country.m3_key).all():
if row[0]:
valid_country_m3.add((row[0] or "").strip().upper())
for row in (
session.query(Sector.key)
.filter(Sector.authorized == True)
.all()
):
if row[0]:
authorized_sector_keys.add((row[0] or "").strip().upper())
for row in session.query(TariffFraction.code).all():
if row[0]:
code = (row[0] or "").strip()
valid_fraction_mex_8.add(code[:8])
for row in (
session.query(HistoricalTariffFraction.historical_fraction)
.filter(
HistoricalTariffFraction.tenant_id == tenant_id,
HistoricalTariffFraction.company_id == company_id,
HistoricalTariffFraction.historical_fraction.isnot(None),
)
.distinct()
.all()
):
if row[0] and (row[0] or "").strip():
valid_fraction_mex_8.add((row[0] or "").strip()[:8])
except Exception as e:
import logging
logging.getLogger(__name__).warning("Parts import: could not load FK sets: %s", e)
return valid_class_codes, valid_uom_codes, valid_currency_codes
logger.warning("Parts import: could not load FK sets: %s", e)
return (
valid_class_codes,
valid_uom_codes,
valid_currency_codes,
valid_fraction_mex_8,
valid_country_m3,
authorized_sector_keys,
company_has_prosec,
is_rfc_exception,
)

109
backend/api/v1/modules/a76/layouts_csv/parts/common/mappers.py
View File

@@ -1,6 +1,6 @@
"""
Mapeo fila CSV → datos para Part. Helpers de normalización de valores.
Referencia: a76/invoices common/mappers.py
Clarion: ME/USD, MN/MXP, MC/ClaveMoneda; apóstrofes omitidos en NUMPARTE; RFC excepción rellena desde Class.
"""
from decimal import Decimal, InvalidOperation
from typing import Dict, Any, Optional, Set
@@ -48,6 +48,42 @@ def _bool_from_row(val: Any) -> bool:
return True
def _normalize_num_parte(val: Optional[str], max_len: int = 70) -> Optional[str]:
"""NUMPARTE: mayúsculas, sin apóstrofes (Clarion: Se Omitirá el Apostrofe)."""
if val is None:
return None
s = str(val).strip().replace("'", "")
if not s:
return None
s = s.upper()
if max_len and len(s) > max_len:
return s[:max_len]
return s
def _currency_from_row(
row_norm: Dict[str, Any],
valid_currency_codes: Set[str],
) -> tuple:
"""
Deriva currency_type y currency_key desde TIPOMONEDA (Col G) y CLAVEMONEDA (Col H).
Clarion: G vacío -> ME/USD; G=ME -> USD; G=MN -> MXP; G=MC -> H obligatoria.
"""
g = (row_norm.get("TIPOMONEDA") or row_norm.get("MONEDA") or "").strip().upper()
h = (row_norm.get("CLAVEMONEDA") or "").strip().upper()
if not g:
return "ME", "USD"
if g == "ME":
return "ME", "USD"
if g == "MN":
return "MN", "MXP"
if g == "MC" and h and h in valid_currency_codes:
return "MC", h
if h and h in valid_currency_codes:
return g[:2], h
return None, None
def row_to_part_data(
row_norm: Dict[str, Any],
valid_class_codes: Set[str],
@@ -56,29 +92,33 @@ def row_to_part_data(
) -> Dict[str, Any]:
"""
Mapea una fila normalizada del CSV a un diccionario de datos para Part.
Ajusta FKs opcionales (part_class, unit_of_measure, currency_key) a None si no están en los conjuntos.
El caller debe añadir tenant_id, company_id, client_id al crear Part.
Clarion: NUMPARTE sin apóstrofes; Tipo Moneda ME/MN/MC con Clave; PAIS, PREFERENCIA, SECTOR, RUTA IMAGEN.
"""
part_number = _str_or_none(row_norm.get("NUMPARTE"), 70)
part_number = _normalize_num_parte(row_norm.get("NUMPARTE"), 70)
commercial = _str_or_none(row_norm.get("NUMPARTECOM"), 70)
desc_es = _str_or_none(row_norm.get("DESCRIPCIONE"), 500)
desc_en = _str_or_none(row_norm.get("DESCRIPCIONI"), 500)
part_class = _str_or_none(row_norm.get("CLASE"), 8)
if part_class and part_class not in valid_class_codes:
if part_class and part_class.upper() not in valid_class_codes:
part_class = 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:
if unit_of_measure and unit_of_measure.upper() not in valid_uom_codes:
unit_of_measure = None
currency_key = _str_or_none(row_norm.get("MONEDA"), 3)
currency_type, currency_key = _currency_from_row(row_norm, valid_currency_codes)
if currency_key and currency_key not in valid_currency_codes:
currency_key = None
currency_type = None
unit_cost = _decimal_or_none(row_norm.get("COSTOUNIT"))
currency_type = _str_or_none(row_norm.get("MONEDA"), 2) if currency_key else None
unit_weight = _decimal_or_none(row_norm.get("PESOUNIT"))
weight_type = _str_or_none(row_norm.get("TIPOPESO"), 6)
if weight_type and weight_type.upper() not in ("KILOS", "LIBRAS"):
weight_type = "KILOS"
fraction = _str_or_none(row_norm.get("FRACCION"), 10)
us_fraction = _str_or_none(row_norm.get("FRACCIONAME"), 16)
part_photo = _str_or_none(row_norm.get("RUTAIMAGEN"), 255)
fda_key = _str_or_none(row_norm.get("FDAKEY"), 20)
fcc_key = _str_or_none(row_norm.get("FCCKEY"), 30)
license_code = _str_or_none(row_norm.get("LICENCIA"), 3)
@@ -101,6 +141,7 @@ def row_to_part_data(
"weight_type": weight_type,
"fraction": fraction,
"us_fraction": us_fraction,
"part_photo": part_photo,
"fda_key": fda_key,
"fcc_key": fcc_key,
"license_code": license_code,
@@ -109,3 +150,55 @@ def row_to_part_data(
"exclusion_symbol": exclusion_symbol,
"is_active": is_active,
}
def row_to_part_data_merge_existing(
row_norm: Dict[str, Any],
existing_data: Dict[str, Any],
valid_class_codes: Set[str],
valid_uom_codes: Set[str],
valid_currency_codes: Set[str],
) -> Dict[str, Any]:
"""
Para modo Actualizar (parcial): valores del CSV si no vacíos, sino los de la parte existente (Clarion VALIDA_PARCIAL_PARTES).
"""
data = row_to_part_data(
row_norm, valid_class_codes, valid_uom_codes, valid_currency_codes
)
if not data.get("part_number"):
return data
merge_keys = (
"description_spanish",
"description_english",
"part_class",
"unit_of_measure",
"unit_cost",
"currency_type",
"currency_key",
"unit_weight",
"weight_type",
"fraction",
"us_fraction",
"part_photo",
)
for key in merge_keys:
val = data.get(key)
if val is None or (isinstance(val, str) and not val.strip()):
data[key] = existing_data.get(key)
return data
def apply_rfc_exception_from_class(
data: Dict[str, Any],
class_obj: Any,
) -> Dict[str, Any]:
"""
Clarion LLENA_PARTES: si empresa CLA/CTE y part_class informado, sobrescribir desde la clase:
UniMed, DescripcionE, DescripcionI, Fraccion, FraccionAme, TipoFraccion='GENERAL'.
"""
data["unit_of_measure"] = getattr(class_obj, "unit_of_measure", None) or data.get("unit_of_measure")
data["description_spanish"] = getattr(class_obj, "description_es", None) or data.get("description_spanish")
data["description_english"] = getattr(class_obj, "description_en", None) or data.get("description_english")
data["fraction"] = getattr(class_obj, "fraction", None) or data.get("fraction")
data["us_fraction"] = getattr(class_obj, "us_fraction", None) or data.get("us_fraction")
return data

25
backend/api/v1/modules/a76/layouts_csv/parts/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 parte existe"),
reemplazar_sin_preguntar: bool = Query(True, description="Si False, solo agregar nuevas (no sobrescribir existentes)"),
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": "part_numbers",
"actualizar": actualizar,
"reemplazar_sin_preguntar": reemplazar_sin_preguntar,
}
try:
@@ -120,6 +124,27 @@ async def get_import_status(job_id: str):
if isinstance(result, dict) and result.get("status") in ("finished", "warning"):
return result
# Si Celery devolvió el resultado como string (p. ej. JSON), parsear y devolver como scan si aplica
if isinstance(result, str):
try:
parsed = json.loads(result)
if isinstance(parsed, dict) and (
parsed.get("status") == "waiting_confirmation"
or (parsed.get("job_id") and "total_rows" in parsed)
):
return parsed
if isinstance(parsed, dict) and parsed.get("status") in ("finished", "warning"):
return parsed
except (json.JSONDecodeError, TypeError):
pass
# Si el resultado tiene forma de escaneo (waiting_confirmation), devolverlo para que el front muestre el modal
if isinstance(result, dict) and (
result.get("status") == "waiting_confirmation"
or (result.get("job_id") and "total_rows" in result)
):
return result
logger.warning("Parts import task %s failed: state=%s", job_id, task_result.state)
err_msg = None
tb = getattr(task_result, "traceback", None)

167
backend/api/v1/modules/a76/layouts_csv/parts/tasks.py
View File

@@ -1,7 +1,7 @@
"""
Tareas Celery para importación CSV de Números de Parte.
Flujo: scan_file (validación) → insert_valid_rows (commit).
Orquestación usa common (storage, normalize, csv_reader, meta, responses) y common.fk_loader.
Paridad Clarion: actualizar (ACT), validación full/parcial, merge existente, reemplazar_sin_preguntar, RFC desde clase.
"""
import json
import logging
@@ -16,9 +16,13 @@ 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 .validators import validate_row_part
from .common.mappers import row_to_part_data
from .template_config import row_from_template, detect_headers_or_data
from .validators import validate_row_part, get_row_warnings
from .common.mappers import (
row_to_part_data,
row_to_part_data_merge_existing,
apply_rfc_exception_from_class,
)
from .common.fk_loader import load_parts_fk_sets
logger = logging.getLogger(__name__)
@@ -43,8 +47,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)}
@@ -53,7 +58,32 @@ def scan_file(self, job_id: str, config: str = None):
except ValueError as e:
return {"status": "failed", "error": str(e)}
valid_class_codes, valid_uom_codes, valid_currency_codes = load_parts_fk_sets(tenant_id, company_id)
meta = common_meta.load_meta(file_path)
actualizar = meta.get("actualizar", False)
(
valid_class_codes,
valid_uom_codes,
valid_currency_codes,
valid_fraction_mex_8,
valid_country_m3,
authorized_sector_keys,
company_has_prosec,
is_rfc_exception,
) = load_parts_fk_sets(tenant_id, company_id)
from api.v1.modules.a76.parts.models import Part
existing_part_numbers = set()
try:
with CoreSessionLocal() as session:
for p in session.query(Part.part_number).filter(
Part.tenant_id == tenant_id,
Part.company_id == company_id,
).all():
if p[0]:
existing_part_numbers.add((p[0] or "").strip().upper())
except Exception as e:
logger.warning("Parts import: could not load existing part numbers: %s", e)
error_count = 0
processed_rows = 0
@@ -62,7 +92,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",
@@ -75,6 +105,13 @@ def scan_file(self, job_id: str, config: str = None):
valid_class_codes=valid_class_codes,
valid_uom_codes=valid_uom_codes,
valid_currency_codes=valid_currency_codes,
valid_fraction_mex_8=valid_fraction_mex_8,
valid_country_m3=valid_country_m3,
authorized_sector_keys=authorized_sector_keys,
company_has_prosec=company_has_prosec,
is_rfc_exception=is_rfc_exception,
actualizar=actualizar,
existing_part_numbers=existing_part_numbers,
)
if err:
error_count += 1
@@ -86,6 +123,14 @@ def scan_file(self, job_id: str, config: str = None):
"col": err.get("col", ""),
"msg": err.get("msg", ""),
})
else:
for w in get_row_warnings(row_norm, i):
if len(errors_detail) < 500:
errors_detail.append({
"line": w.get("line"),
"col": w.get("col", ""),
"msg": w.get("msg", ""),
})
processed_rows += 1
if error_lines_list:
@@ -120,9 +165,23 @@ def insert_valid_rows(self, job_id: str):
except ValueError as e:
return {"status": "failed", "error": str(e)}
from api.v1.modules.a76.parts.models import Part
meta = common_meta.load_meta(file_path)
actualizar = meta.get("actualizar", False)
reemplazar_sin_preguntar = meta.get("reemplazar_sin_preguntar", True)
valid_class_codes, valid_uom_codes, valid_currency_codes = load_parts_fk_sets(tenant_id, company_id)
from api.v1.modules.a76.parts.models import Part
from api.v1.modules.a76.classes.models import Class
(
valid_class_codes,
valid_uom_codes,
valid_currency_codes,
valid_fraction_mex_8,
valid_country_m3,
authorized_sector_keys,
company_has_prosec,
is_rfc_exception,
) = load_parts_fk_sets(tenant_id, company_id)
inserted_count = 0
skipped_invalid = 0
@@ -132,27 +191,45 @@ 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_part_number = {
p.part_number: p
for p in session.query(Part).filter(
Part.tenant_id == tenant_id,
Part.company_id == company_id,
).all()
}
existing_by_part_number = {}
for p in session.query(Part).filter(
Part.tenant_id == tenant_id,
Part.company_id == company_id,
).all():
key = (p.part_number or "").strip().upper()
if key:
existing_by_part_number[key] = p
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
row_norm = row_from_template(row, common_normalize.normalize_header)
part_number_raw = (row_norm.get("NUMPARTE") or "").strip().upper()
use_partial = (
actualizar
and part_number_raw
and part_number_raw in existing_by_part_number
and reemplazar_sin_preguntar
)
err = validate_row_part(
row_norm,
i,
valid_class_codes=valid_class_codes,
valid_uom_codes=valid_uom_codes,
valid_currency_codes=valid_currency_codes,
valid_fraction_mex_8=valid_fraction_mex_8,
valid_country_m3=valid_country_m3,
authorized_sector_keys=authorized_sector_keys,
company_has_prosec=company_has_prosec,
is_rfc_exception=is_rfc_exception,
actualizar=actualizar,
existing_part_numbers=set(existing_by_part_number.keys()),
)
if err:
skipped_invalid += 1
@@ -162,21 +239,63 @@ def insert_valid_rows(self, job_id: str):
})
continue
data = row_to_part_data(
row_norm,
valid_class_codes,
valid_uom_codes,
valid_currency_codes,
)
if use_partial:
existing = existing_by_part_number.get(part_number_raw)
existing_data = {
"description_spanish": existing.description_spanish,
"description_english": existing.description_english,
"part_class": existing.part_class,
"unit_of_measure": existing.unit_of_measure,
"unit_cost": existing.unit_cost,
"currency_type": existing.currency_type,
"currency_key": existing.currency_key,
"unit_weight": existing.unit_weight,
"weight_type": existing.weight_type,
"fraction": existing.fraction,
"us_fraction": existing.us_fraction,
"part_photo": existing.part_photo,
}
data = row_to_part_data_merge_existing(
row_norm,
existing_data,
valid_class_codes,
valid_uom_codes,
valid_currency_codes,
)
else:
data = row_to_part_data(
row_norm,
valid_class_codes,
valid_uom_codes,
valid_currency_codes,
)
part_number = data.get("part_number")
if not part_number:
skipped_invalid += 1
continue
if is_rfc_exception and data.get("part_class"):
class_obj = (
session.query(Class)
.filter(
Class.tenant_id == tenant_id,
Class.company_id == company_id,
Class.class_code == (data.get("part_class") or "").strip().upper(),
)
.first()
)
if class_obj:
data = apply_rfc_exception_from_class(data, class_obj)
existing = existing_by_part_number.get(part_number)
if existing:
if not reemplazar_sin_preguntar:
skipped_duplicate += 1
continue
for key, value in data.items():
setattr(existing, key, value)
if hasattr(existing, key):
setattr(existing, key, value)
session.add(existing)
inserted_count += 1
else:

104
backend/api/v1/modules/a76/layouts_csv/parts/template_config.py
View File

@@ -1,23 +1,111 @@
"""
Configuración de plantilla CSV para Números de Parte (EstructuraCatPartesAF.xls).
Configuración de plantilla CSV para Números de Parte (EstructuraCatPartesAF).
Cabeceras de descarga = Clarion: NUMERO DE PARTE, DESCRIPCION EN ESPAÑOL, etc.
"""
import csv
import io
from typing import Dict, List, Any, Optional, Tuple
from typing import Dict, List, Any
# Valores que indican que la primera fila es cabecera (primera columna normalizada)
FIRST_COLUMN_HEADER_VALUES = ("NUMERO DE PARTE", "NUMPARTE")
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.
- Si la primera celda normalizada está en FIRST_COLUMN_HEADER_VALUES -> has_header=True, fieldnames=None.
- Si no -> has_header=False, fieldnames=TEMPLATE_DOWNLOAD_HEADERS (primera fila = 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_FIELDNAMES_FOR_READING), False
# Cabeceras que se escriben al descargar la plantilla CSV (igual que Clarion EstructuraCatPartesAF)
# Columnas I y J vacías en Clarion: dos comas entre CLAVE MONEDA y PESO UNITARIO (espacios en blanco separados por coma)
TEMPLATE_DOWNLOAD_HEADERS: List[str] = [
"NUMERO DE PARTE",
"DESCRIPCION EN ESPAÑOL",
"DESCRIPCION EN INGLES",
"CLASE",
"UNIDAD DE MEDIDA COMERCIAL",
"COSTO UNITARIO",
"TIPO MONEDA COSTO",
"CLAVE MONEDA",
"", # Col I vacía
"", # Col J vacía
"PESO UNITARIO",
"TIPO PESO",
"FRACCION",
"PAIS",
"PREFERENCIA",
"SECTOR",
"RUTA DE LA IMAGEN",
]
# Para lectura cuando la primera fila es dato (sin cabecera): nombres únicos para columnas I y J
TEMPLATE_FIELDNAMES_FOR_READING: List[str] = [
"NUMERO DE PARTE",
"DESCRIPCION EN ESPAÑOL",
"DESCRIPCION EN INGLES",
"CLASE",
"UNIDAD DE MEDIDA COMERCIAL",
"COSTO UNITARIO",
"TIPO MONEDA COSTO",
"CLAVE MONEDA",
"_COL_I_",
"_COL_J_",
"PESO UNITARIO",
"TIPO PESO",
"FRACCION",
"PAIS",
"PREFERENCIA",
"SECTOR",
"RUTA DE LA IMAGEN",
]
TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
"part_numbers": [
{"canonical": "NUMPARTE", "aliases": ["NUMERO PARTE", "PART NUMBER", "NUM PARTE"]},
{"canonical": "NUMPARTE", "aliases": ["NUMERO DE PARTE", "NUMERO PARTE", "PART NUMBER", "NUM PARTE"]},
{"canonical": "NUMPARTECOM", "aliases": ["NUMERO PARTE COMERCIAL", "COMMERCIAL PART", "PARTE COMERCIAL"]},
{"canonical": "DESCRIPCIONE", "aliases": ["DESCRIPCION", "DESCRIPCION ES", "DESC ESPANOL"]},
{"canonical": "DESCRIPCIONI", "aliases": ["DESCRIPCION EN", "DESCRIPTION", "DESC INGLES"]},
{"canonical": "DESCRIPCIONE", "aliases": ["DESCRIPCION EN ESPAÑOL", "DESCRIPCION EN ESPANOL", "DESCRIPCION", "DESCRIPCION ES", "DESC ESPANOL"]},
{"canonical": "DESCRIPCIONI", "aliases": ["DESCRIPCION EN INGLES", "DESCRIPCION EN", "DESCRIPTION", "DESC INGLES"]},
{"canonical": "CLASE", "aliases": ["CLASS", "CLASE MATERIAL", "PART CLASS"]},
{"canonical": "UNIMED", "aliases": ["UNIDAD MEDIDA", "UNIT", "UOM", "UNIT OF MEASURE"]},
{"canonical": "UNIMED", "aliases": ["UNIDAD DE MEDIDA COMERCIAL", "UNIDAD MEDIDA", "UNIT", "UOM", "UNIT OF MEASURE"]},
{"canonical": "COSTOUNIT", "aliases": ["COSTO UNITARIO", "UNIT COST", "COSTO"]},
{"canonical": "MONEDA", "aliases": ["CURRENCY", "MONEDA CLAVE", "CURRENCY KEY"]},
{"canonical": "TIPOMONEDA", "aliases": ["TIPO MONEDA COSTO", "TIPO MONEDA", "MONEDA"]},
{"canonical": "CLAVEMONEDA", "aliases": ["CLAVE MONEDA", "CURRENCY", "MONEDA CLAVE", "CURRENCY KEY"]},
{"canonical": "PESOUNIT", "aliases": ["PESO UNITARIO", "UNIT WEIGHT", "PESO"]},
{"canonical": "TIPOPESO", "aliases": ["WEIGHT TYPE", "TIPO PESO"]},
{"canonical": "TIPOPESO", "aliases": ["TIPO PESO", "WEIGHT TYPE"]},
{"canonical": "FRACCION", "aliases": ["FRACCION MEX"]},
{"canonical": "FRACCIONAME", "aliases": ["FRACCION USA", "US FRACTION"]},
{"canonical": "PAIS", "aliases": ["PAIS", "COUNTRY"]},
{"canonical": "PREFERENCIA", "aliases": ["PREFERENCIA", "PREFERENCIA ARANCELARIA"]},
{"canonical": "SECTOR", "aliases": ["SECTOR"]},
{"canonical": "RUTAIMAGEN", "aliases": ["RUTA DE LA IMAGEN", "RUTA IMAGEN", "FOTO"]},
{"canonical": "FDAKEY", "aliases": ["FDA", "FDA KEY"]},
{"canonical": "FCCKEY", "aliases": ["FCC", "FCC KEY"]},
{"canonical": "LICENCIA", "aliases": ["LICENSE CODE", "LICENSE"]},

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

@@ -1,4 +1,4 @@
# validators for parts CSV import row validation
from .create import validate_row_part
from .create import validate_row_part, validate_row_part_partial, get_row_warnings
__all__ = ["validate_row_part"]
__all__ = ["validate_row_part", "validate_row_part_partial", "get_row_warnings"]

386
backend/api/v1/modules/a76/layouts_csv/parts/validators/common.py
View File

@@ -1,35 +1,99 @@
"""
Validaciones comunes de fila para import CSV de partes.
Usa helpers de common.common_validators; agrupa por tipo (requeridos, longitudes, tipos).
Referencia: a76/invoices imports/temporary/validators/common.py
Paridad Clarion: VALIDA_TODA_PARTES (obligatorios A, B, E salvo excepción RFC), VALIDACIONES_PARTES.
"""
from typing import Dict, Any, Optional
from typing import Dict, Any, Optional, Set
from ..common.common_validators import check_max_length, check_decimal
MSG_NUMPARTE_VACIO = (
"Error: (Col. A) La columna de Número de Parte esta vacio y no se pueden hacer las validaciones. "
"Capturar en la Columna A un Número de Parte nuevo o una ya existente a la cual desee actualizar campos"
)
NUMPARTE_MAX_LEN_CLARION = 30
NUMPARTE_MAX_LEN_DB = 70
def validate_row_required(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
"""Valida campos obligatorios de una fila de parte."""
return check_max_length(row, "NUMPARTE", 70, line_num, required=True)
"""Col A (NUMPARTE) obligatorio; mensaje Clarion si está vacío."""
val = (row.get("NUMPARTE") or "").strip()
if not val:
return {"line": line_num, "col": "NUMPARTE", "msg": MSG_NUMPARTE_VACIO}
if len(val) > NUMPARTE_MAX_LEN_DB:
return {
"line": line_num,
"col": "NUMPARTE",
"msg": f"Error: (Col. A) El Número de Parte: {val} supera la longitud de caracteres. "
f"Capturar en la columna A un Número de Parte de {NUMPARTE_MAX_LEN_DB} caracteres como máximo.",
}
return None
def validate_row_required_full(
row: Dict[str, Any],
line_num: int,
is_rfc_exception: bool = False,
) -> Optional[Dict[str, Any]]:
"""Obligatorios en validación completa: A, B; y E (U.M. Comercial) salvo excepción RFC con D vacía."""
err = validate_row_required(row, line_num)
if err:
return err
if not (row.get("DESCRIPCIONE") or "").strip():
return {
"line": line_num,
"col": "DESCRIPCIONE",
"msg": "Existen campos vacios que son obligatorios, es la (Col.B) Descripción Español. "
"Revisar la línea del archivo y capturar los campos con la información correcta.",
}
# E obligatorio salvo que empresa sea excepción RFC y Col D (Clase) esté vacía
col_d = (row.get("CLASE") or "").strip()
if not is_rfc_exception or col_d:
if not (row.get("UNIMED") or "").strip():
return {
"line": line_num,
"col": "UNIMED",
"msg": "Existen campos vacios que son obligatorios, es la (Col.E) U.M. Comercial. "
"Revisar la línea del archivo y capturar los campos con la información correcta.",
}
return None
def validate_row_required_act_part_exists(
row: Dict[str, Any],
line_num: int,
existing_part_numbers: Set[str],
) -> Optional[Dict[str, Any]]:
"""Cuando actualizar=True y se hace validación full: el número de parte debe existir en catálogo."""
num_parte = (row.get("NUMPARTE") or "").strip().upper()
if not num_parte:
return None
if num_parte not in existing_part_numbers:
return {
"line": line_num,
"col": "NUMPARTE",
"msg": "(Col.A) El número de parte no existe.",
}
return None
def validate_row_lengths(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
"""Valida longitudes máximas de todos los campos de texto."""
"""Longitudes máximas (Clarion: Col A 30; resto según modelo)."""
err = check_max_length(row, "NUMPARTE", NUMPARTE_MAX_LEN_DB, line_num)
if err:
return err
checks = [
("NUMPARTECOM", 70),
("DESCRIPCIONE", 500),
("DESCRIPCIONI", 500),
("CLASE", 8),
("UNIMED", 5),
("MONEDA", 3),
("TIPOMONEDA", 3),
("CLAVEMONEDA", 3),
("FRACCION", 10),
("FRACCIONAME", 16),
("FDAKEY", 20),
("FCCKEY", 30),
("LICENCIA", 3),
("ECCN", 20),
("EXPORTCODE", 2),
("EXCLUSION", 19),
("PAIS", 3),
("PREFERENCIA", 7),
("SECTOR", 5),
("RUTAIMAGEN", 255),
("TIPOPESO", 6),
]
for col, max_len in checks:
@@ -40,7 +104,7 @@ def validate_row_lengths(row: Dict[str, Any], line_num: int) -> Optional[Dict[st
def validate_row_types(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
"""Valida tipos numéricos (decimales) de la fila."""
"""Tipos numéricos (decimales) de la fila."""
err = check_decimal(row, "COSTOUNIT", line_num)
if err:
return err
@@ -48,3 +112,293 @@ def validate_row_types(row: Dict[str, Any], line_num: int) -> Optional[Dict[str,
if err:
return err
return None
def validate_row_class_fk(
row: Dict[str, Any],
line_num: int,
valid_class_codes: Optional[Set[str]],
) -> Optional[Dict[str, Any]]:
"""Col D (Clase): si no vacía, debe existir en catálogo de clases."""
val = (row.get("CLASE") or "").strip()
if not val or valid_class_codes is None:
return None
if val.upper() in valid_class_codes:
return None
return {
"line": line_num,
"col": "CLASE",
"msg": f"Error: (Col. D) La Clase: {val} no existe en el Catálogo de Clases de Activo Fijo. "
"Dar de alta la Clase en el Catálogo de Clases de Activo Fijo.",
}
def validate_row_uom_fk(
row: Dict[str, Any],
line_num: int,
valid_uom_codes: Optional[Set[str]],
) -> Optional[Dict[str, Any]]:
"""Col E (U.M. Comercial): si no vacía, debe existir en catálogo UOM."""
val = (row.get("UNIMED") or "").strip()
if not val or valid_uom_codes is None:
return None
if val.upper() in valid_uom_codes:
return None
return {
"line": line_num,
"col": "UNIMED",
"msg": f"Error: (Col. E) La Unidad de Medida Comercial: {val} no existe en el Catálogo de U.M. "
"Revisar esta Unidad de Medida en el archivo, en caso de ser correcta dar la de alta en el Catálogo de U.M.",
}
def validate_row_tipo_moneda(
row: Dict[str, Any],
line_num: int,
) -> Optional[Dict[str, Any]]:
"""Col G (Tipo Moneda): solo ME, MN o MC (o vacío)."""
val = (row.get("TIPOMONEDA") or "").strip().upper()
if not val:
return None
if val in ("ME", "MN", "MC"):
return None
return {
"line": line_num,
"col": "TIPOMONEDA",
"msg": f"Error: (Col. G) La opción de Tipo Moneda: {val} no es valida. "
"Capturar una opción valida: ME para Dolares, MN para Pesos, MC para Moneda de Captura o dejar el campo vacio.",
}
def validate_row_clave_moneda(
row: Dict[str, Any],
line_num: int,
valid_currency_codes: Optional[Set[str]],
) -> Optional[Dict[str, Any]]:
"""Col H (Clave Moneda): si G = MC, H obligatoria y debe existir en catálogo."""
g = (row.get("TIPOMONEDA") or "").strip().upper()
if g != "MC":
return None
val = (row.get("CLAVEMONEDA") or "").strip()
if not val:
return {
"line": line_num,
"col": "CLAVEMONEDA",
"msg": "Error: (Col. H) La Clave de la Moneda es obligatoria cuando Tipo Moneda es MC.",
}
if valid_currency_codes is not None and val.upper() not in valid_currency_codes:
return {
"line": line_num,
"col": "CLAVEMONEDA",
"msg": f"Error: (Col. H) La Clave de la Moneda: {val} no existe en el Catálogo de Claves de Moneda. "
"Revisar esta clave de la Moneda en el archivo, en caso de ser correcta Actualizar los Catálogos Fijos.",
}
return None
def validate_row_tipo_peso(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
"""Col L (Tipo Peso): solo KILOS o LIBRAS (o vacío)."""
val = (row.get("TIPOPESO") or "").strip()
if not val:
return None
if val.upper() in ("KILOS", "LIBRAS"):
return None
return {
"line": line_num,
"col": "TIPOPESO",
"msg": f"Error: (Col. L) La opción del Tipo de Peso {val} no es valida. "
"Capturar una opción valida: KILOS, LIBRAS o dejar el campo vacio y automaticamente se asigna KILOS.",
}
def _normalize_fraction_mex_8(value: str) -> str:
"""Primeros 8 caracteres si len>=10, sino hasta 8 (Clarion SUB)."""
if not value:
return ""
v = value.strip()
if len(v) >= 10:
return v[:8]
return v[:8] if len(v) > 8 else v
def validate_row_fraction_mex_catalog(
row: Dict[str, Any],
line_num: int,
valid_fraction_mex_8: Optional[Set[str]],
) -> Optional[Dict[str, Any]]:
"""Col M (Fracción Mex): si no vacía, debe existir en catálogo Mex o histórico."""
val = (row.get("FRACCION") or "").strip()
if not val or valid_fraction_mex_8 is None:
return None
code_8 = _normalize_fraction_mex_8(val)
if not code_8:
return None
if code_8 in valid_fraction_mex_8:
return None
return {
"line": line_num,
"col": "FRACCION",
"msg": f"Error: (Col. M) La Fraccion Mexicana: {val} no existe en el Catálogo de Fracciones Arancelarias Sifr@ ni en el Historico. "
"Revisar esta Fracción Arancelaria en el archivo, en caso de ser correcta Actualizar las Fracciones Arancelarias.",
}
def validate_row_country_fk(
row: Dict[str, Any],
line_num: int,
valid_country_m3: Optional[Set[str]],
) -> Optional[Dict[str, Any]]:
"""Col N (País): si no vacía, debe existir en catálogo de países (m3_key)."""
val = (row.get("PAIS") or "").strip()
if not val or valid_country_m3 is None:
return None
if val.upper() in valid_country_m3:
return None
return {
"line": line_num,
"col": "PAIS",
"msg": f"Error: (Col. N) La Clave del País: {val} no existe en el Catálogo de Paises. Dar lo de alta en el Catálogo de Paises.",
}
def validate_row_preferencia(
row: Dict[str, Any],
line_num: int,
is_rfc_exception: bool,
) -> Optional[Dict[str, Any]]:
"""Col O (Preferencia): solo GENERAL, TLCS, PROSEC o ALADI. Excepción RFC: solo aplica si D no vacía."""
col_d = (row.get("CLASE") or "").strip()
if is_rfc_exception and not col_d:
return None
val = (row.get("PREFERENCIA") or "").strip().upper()
if not val:
return None
if val in ("GENERAL", "TLCS", "PROSEC", "ALADI"):
return None
return {
"line": line_num,
"col": "PREFERENCIA",
"msg": "Error: (Col. O) La opción de Preferencia Arancelaria no es valida. "
"Capturar una opción valida: GENERAL, TLCS, PROSEC, ALADI.",
}
def validate_row_sector(
row: Dict[str, Any],
line_num: int,
company_has_prosec: bool,
authorized_sector_keys: Optional[Set[str]],
) -> Optional[Dict[str, Any]]:
"""Col P (Sector): si O=PROSEC entonces P obligatorio, empresa PROSEC, sector autorizado; si O≠PROSEC y P no vacío error."""
pref = (row.get("PREFERENCIA") or "").strip().upper()
sector = (row.get("SECTOR") or "").strip()
if pref == "PROSEC":
if not sector:
return {
"line": line_num,
"col": "SECTOR",
"msg": "Error: (Col. P) no hay un sector capturado y se tiene la opción de Preferencia Arancelaria: PROSEC. Capturar un Sector valido en la columna P",
}
if not company_has_prosec:
return {
"line": line_num,
"col": "SECTOR",
"msg": f"Error: (Col. P) Se quiere aplicar el sector {sector} pero la empresa no cuenta con Permiso PROSEC. "
"Marcar la opción que cuenta con un Permiso PROSEC en los Datos de la Empresa.",
}
if authorized_sector_keys is not None and sector.upper() not in authorized_sector_keys:
return {
"line": line_num,
"col": "SECTOR",
"msg": f"Error: (Col. P) El sector {sector} no esta promovido para esta Empresa. "
"Capturar un Sector en la columna P que este promovido en los Datos de la Empresa o active el Sector en el Catálogo de Sectores.",
}
else:
if sector:
return {
"line": line_num,
"col": "SECTOR",
"msg": f"Error: (Col. P) Hay un sector capturado y se tiene la opción de Preferencia Arancelaria: {pref or '(vacío)'}. "
"Borra el sector en la columna P o cambiar la preferencia en la columna O.",
}
return None
def validaciones_parte(
row: Dict[str, Any],
line_num: int,
valid_class_codes: Optional[Set[str]] = None,
valid_uom_codes: Optional[Set[str]] = None,
valid_currency_codes: Optional[Set[str]] = None,
valid_fraction_mex_8: Optional[Set[str]] = None,
valid_country_m3: Optional[Set[str]] = None,
authorized_sector_keys: Optional[Set[str]] = None,
company_has_prosec: bool = False,
is_rfc_exception: bool = False,
) -> Optional[Dict[str, Any]]:
"""Reglas compartidas Clarion (VALIDACIONES_PARTES): longitudes, tipos, FKs, preferencia, sector."""
err = validate_row_lengths(row, line_num)
if err:
return err
err = validate_row_types(row, line_num)
if err:
return err
err = validate_row_class_fk(row, line_num, valid_class_codes)
if err:
return err
err = validate_row_uom_fk(row, line_num, valid_uom_codes)
if err:
return err
err = validate_row_tipo_moneda(row, line_num)
if err:
return err
err = validate_row_clave_moneda(row, line_num, valid_currency_codes)
if err:
return err
err = validate_row_tipo_peso(row, line_num)
if err:
return err
err = validate_row_fraction_mex_catalog(row, line_num, valid_fraction_mex_8)
if err:
return err
err = validate_row_country_fk(row, line_num, valid_country_m3)
if err:
return err
err = validate_row_preferencia(row, line_num, is_rfc_exception)
if err:
return err
err = validate_row_sector(
row, line_num, company_has_prosec, authorized_sector_keys
)
if err:
return err
return None
def validate_row_warnings(
row: Dict[str, Any],
line_num: int,
) -> list:
"""
Devuelve lista de advertencias (no bloquean; no se añaden a error_lines).
Clarion: desfase cuando Col Q tiene valor; apóstrofes en Número de Parte.
"""
warnings: list = []
# Advertencia desfase: Clarion cuando CSVArc:ColumnaQ <> ''
if (row.get("RUTAIMAGEN") or "").strip():
warnings.append({
"line": line_num,
"col": "RUTAIMAGEN",
"msg": "Advertencia: Podría existir un desfase en esta línea.",
"severity": "warning",
})
# Advertencia apóstrofes en Número de Parte
num_parte = (row.get("NUMPARTE") or "").strip()
if "'" in num_parte:
warnings.append({
"line": line_num,
"col": "NUMPARTE",
"msg": f"Advertencia: El Número de Parte: {num_parte} Contiene Apostrofes. Se Omitirá el Apostrofe para Subir el Número de Parte.",
"severity": "warning",
})
return warnings

97
backend/api/v1/modules/a76/layouts_csv/parts/validators/create.py
View File

@@ -1,14 +1,15 @@
"""
Punto de entrada de validación para creación/import de una fila de parte.
Encadena validaciones comunes (requeridos, longitudes, tipos).
Referencia: a76/invoices imports/temporary/validators/create.py
Punto de entrada de validación para import de una fila de parte.
Flujo Clarion: no ACT → siempre VALIDA_TODA_PARTES; ACT y parte existe → VALIDA_PARCIAL_PARTES;
ACT y parte no existe → VALIDA_TODA_PARTES (obligatorios + validaciones) y en commit se crea (ADD).
"""
from typing import Dict, Any, Optional, Set
from .common import (
validate_row_required,
validate_row_lengths,
validate_row_types,
validate_row_required_full,
validaciones_parte,
validate_row_warnings,
)
@@ -18,22 +19,98 @@ def validate_row_part(
valid_class_codes: Optional[Set[str]] = None,
valid_uom_codes: Optional[Set[str]] = None,
valid_currency_codes: Optional[Set[str]] = None,
valid_fraction_mex_8: Optional[Set[str]] = None,
valid_country_m3: Optional[Set[str]] = None,
authorized_sector_keys: Optional[Set[str]] = None,
company_has_prosec: bool = False,
is_rfc_exception: bool = False,
actualizar: bool = False,
existing_part_numbers: Optional[Set[str]] = None,
) -> Optional[Dict[str, Any]]:
"""
Valida una fila de CSV de partes para import.
Encadena: requeridos → longitudes → tipos.
Los conjuntos valid_* se mantienen en la firma por compatibilidad con el caller.
- No actualizar: siempre validación completa (obligatorios A,B,E + validaciones_parte).
- Actualizar y número de parte ya existe: validación parcial (solo validaciones_parte).
- Actualizar y número de parte no existe: validación completa; si pasa, en commit se crea (ADD).
Si actualizar y validación full, además exige que el número de parte exista en catálogo.
"""
err = validate_row_required(row, line_num)
if err:
return err
err = validate_row_lengths(row, line_num)
part_number = (row.get("NUMPARTE") or "").strip().upper()
use_partial = (
actualizar
and existing_part_numbers is not None
and part_number in existing_part_numbers
)
if use_partial:
return validate_row_part_partial(
row,
line_num,
valid_class_codes=valid_class_codes,
valid_uom_codes=valid_uom_codes,
valid_currency_codes=valid_currency_codes,
valid_fraction_mex_8=valid_fraction_mex_8,
valid_country_m3=valid_country_m3,
authorized_sector_keys=authorized_sector_keys,
company_has_prosec=company_has_prosec,
is_rfc_exception=is_rfc_exception,
)
# Full validation (también cuando actualizar y parte no existe → ADD en commit)
err = validate_row_required_full(row, line_num, is_rfc_exception=is_rfc_exception)
if err:
return err
# No exigir que exista en actualizar: si no existe se valida completa y en commit se crea (ADD)
return validaciones_parte(
row,
line_num,
valid_class_codes=valid_class_codes,
valid_uom_codes=valid_uom_codes,
valid_currency_codes=valid_currency_codes,
valid_fraction_mex_8=valid_fraction_mex_8,
valid_country_m3=valid_country_m3,
authorized_sector_keys=authorized_sector_keys,
company_has_prosec=company_has_prosec,
is_rfc_exception=is_rfc_exception,
)
err = validate_row_types(row, line_num)
def validate_row_part_partial(
row: Dict[str, Any],
line_num: int,
valid_class_codes: Optional[Set[str]] = None,
valid_uom_codes: Optional[Set[str]] = None,
valid_currency_codes: Optional[Set[str]] = None,
valid_fraction_mex_8: Optional[Set[str]] = None,
valid_country_m3: Optional[Set[str]] = None,
authorized_sector_keys: Optional[Set[str]] = None,
company_has_prosec: bool = False,
is_rfc_exception: bool = False,
) -> Optional[Dict[str, Any]]:
"""
Validación parcial (modo Actualizar, parte existente): solo NUMPARTE + validaciones_parte.
Los campos vacíos se rellenan desde la parte existente en el mapper.
"""
err = validate_row_required(row, line_num)
if err:
return err
return validaciones_parte(
row,
line_num,
valid_class_codes=valid_class_codes,
valid_uom_codes=valid_uom_codes,
valid_currency_codes=valid_currency_codes,
valid_fraction_mex_8=valid_fraction_mex_8,
valid_country_m3=valid_country_m3,
authorized_sector_keys=authorized_sector_keys,
company_has_prosec=company_has_prosec,
is_rfc_exception=is_rfc_exception,
)
return None
def get_row_warnings(row: Dict[str, Any], line_num: int) -> list:
"""Devuelve lista de advertencias para la fila (desfase, apóstrofes). No bloquean."""
return validate_row_warnings(row, line_num)

7
frontend/src/lib/api.ts
View File

@@ -562,11 +562,14 @@ export const api = {
// CSV import for Números de parte (parts/imports)
partNumberImports: {
upload: (file: File, companyId: number) => {
upload: (file: File, companyId: number, options?: { actualizar?: boolean; reemplazar_sin_preguntar?: boolean }) => {
const formData = new FormData();
formData.append('file', file);
const params = new URLSearchParams({ company_id: String(companyId) });
if (options?.actualizar !== undefined) params.set('actualizar', String(options.actualizar));
if (options?.reemplazar_sin_preguntar !== undefined) params.set('reemplazar_sin_preguntar', String(options.reemplazar_sin_preguntar));
return fetchApi(
`/v1/a76/parts/imports/upload?company_id=${companyId}`,
`/v1/a76/parts/imports/upload?${params.toString()}`,
{ method: 'POST', body: formData }
);
},

111
frontend/src/routes/dashboard/csv-upload/+page.svelte
View File

@@ -16,6 +16,38 @@
import { toast } from 'svelte-sonner';
import { companyStore } from '$lib/stores/company.svelte';
/** Intenta extraer un objeto tipo scan desde string tipo repr de Python. */
function parsePythonReprScan(s: string): Record<string, unknown> | null {
const jobIdMatch = s.match(/'job_id':\s*'([^']*)'/);
const totalRowsMatch = s.match(/'total_rows':\s*(\d+)/);
if (!jobIdMatch || !totalRowsMatch) return null;
const job_id = jobIdMatch[1];
const total_rows = parseInt(totalRowsMatch[1], 10);
const errorCountMatch = s.match(/'error_count':\s*(\d+)/);
const validRowsMatch = s.match(/'valid_rows':\s*(\d+)/);
const error_count = errorCountMatch ? parseInt(errorCountMatch[1], 10) : 0;
const valid_rows = validRowsMatch ? parseInt(validRowsMatch[1], 10) : 0;
const errors: { line: number; col: string; msg: string }[] = [];
// Buscar cada bloque {'line': N, 'col': 'X', 'msg': '...'} en la cadena
const errRegex = /\{'line':\s*(\d+),\s*'col':\s*'([^']*)',\s*'msg':\s*'((?:[^'\\]|\\.)*)'\}/g;
let m: RegExpExecArray | null;
while ((m = errRegex.exec(s)) !== null) {
errors.push({
line: parseInt(m[1], 10),
col: m[2],
msg: m[3].replace(/\\'/g, "'")
});
}
return {
status: 'waiting_confirmation',
job_id,
total_rows,
error_count,
valid_rows,
errors
};
}
// We no longer need modal state
let activeTab = $state('catalogos');
@@ -293,7 +325,12 @@
if (usePartNumbersImport) {
try {
const res = await api.partNumberImports.upload(file, companyId);
const catalogosSettings = allSettings['catalogos'] || {};
const actualizar = catalogosSettings['mode'] === 'update';
const res = await api.partNumberImports.upload(file, companyId, {
actualizar,
reemplazar_sin_preguntar: true
});
if (res.data?.job_id) {
currentJobId = res.data.job_id;
pollStatus();
@@ -396,25 +433,71 @@
currentJobId = null;
return;
}
if (res.data?.status === 'waiting_confirmation') {
// Tratar como resultado de escaneo si viene status waiting_confirmation O si el payload tiene forma de scan (job_id + total_rows)
const looksLikeScanResult =
res.data?.status === 'waiting_confirmation' ||
(res.data?.job_id && typeof res.data?.total_rows === 'number');
if (looksLikeScanResult) {
scanResults = res.data;
showResultModal = true;
toast.success('Escaneo completado. Revisa los resultados.');
isUploading = false;
} else if (res.data?.status === 'failed' || res.data?.status === 'FAILURE') {
const errRaw = res.data.error;
const errText =
typeof errRaw === 'string'
? errRaw.includes('finished') && errRaw.includes('inserted')
? 'La importación pudo completarse. Revisa el listado de registros.'
: errRaw
: (errRaw?.message ?? 'Error desconocido');
toast.error('Error en el procesamiento: ' + errText);
isUploading = false;
currentJobId = null;
scanResults = null;
commitResults = null;
showResultModal = false;
// Si el backend devolvió el resultado del scan dentro de error (p. ej. string JSON), usarlo para mostrar el modal
let parsedScan: Record<string, unknown> | null = null;
if (typeof errRaw === 'string' && (errRaw.includes('waiting_confirmation') || (errRaw.includes('total_rows') && errRaw.includes('job_id')))) {
try {
const parsed = JSON.parse(errRaw) as Record<string, unknown>;
if (parsed && typeof parsed.job_id === 'string' && typeof parsed.total_rows === 'number') {
parsedScan = parsed;
}
} catch {
// No es JSON; intentar parsear como repr de Python
parsedScan = parsePythonReprScan(errRaw);
}
}
if (parsedScan) {
scanResults = parsedScan;
showResultModal = true;
toast.success('Escaneo completado. Revisa los resultados.');
isUploading = false;
} else {
// Mensaje parece resultado de escaneo pero no se pudo parsear (p. ej. repr Python) → no asustar con error
const looksLikeScanInError =
typeof errRaw === 'string' &&
(errRaw.includes('waiting_confirmation') || (errRaw.includes('total_rows') && errRaw.includes('job_id')));
if (looksLikeScanInError) {
toast.info('El escaneo terminó. Si no ves el modal, revisa el listado de registros.');
isUploading = false;
currentJobId = null;
return;
}
let errText: string;
if (typeof errRaw === 'string') {
errText =
errRaw.includes('finished') && errRaw.includes('inserted')
? 'La importación pudo completarse. Revisa el listado de registros.'
: errRaw.includes("'status'") && errRaw.includes('waiting_confirmation')
? 'Error al obtener el resultado. Revisa el modal de resultados.'
: errRaw;
} else if (typeof errRaw === 'object' && errRaw !== null) {
errText = (errRaw as { message?: string })?.message || 'Error en el procesamiento. Revisa el modal o los detalles.';
} else {
errText = 'Error desconocido';
}
// No mostrar como error si el mensaje indica éxito
if (errText.includes('pudo completarse') || errText.includes('Revisa el listado')) {
toast.success(errText);
} else {
toast.error('Error en el procesamiento: ' + errText);
}
isUploading = false;
currentJobId = null;
scanResults = null;
commitResults = null;
showResultModal = false;
}
} else if (res.data?.status === 'warning') {
// Caso cuando no se insertaron registros pero hay información de rechazo
commitResults = res.data;