fix/carga-csv-ergonomia

This commit is contained in:
2026-05-06 14:24:41 -06:00
parent 8c8cd0e247
commit daa3a252b9
53 changed files with 3164 additions and 391 deletions

View File

@@ -4,11 +4,10 @@ Construido a partir de los TEMPLATE_COLUMNS de cada módulo de imports.
"""
import csv
import io
from typing import Dict, List, Optional
from typing import Dict, List, Optional, Set
# Importar configs de cada módulo
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 (
@@ -140,6 +139,7 @@ ALWAYS_REQUIRED_HEADERS_BY_TEMPLATE: Dict[str, set[str]] = {
"imp_temp_series": {"NUMERO FACTURA", "LINEA FACTURA"},
"imp_def_series": {"NUMERO FACTURA", "LINEA FACTURA"},
"cmex_series": {"NUMERO FACTURA", "LINEA FACTURA"},
"exp_def_series": {"NUMERO FACTURA", "LINEA FACTURA"},
# Catálogos y transportes
"customs_brokers": {"TIPO", "CLAVE", "NOMBRE"},
"clients_providers": {"PROCEDENCIA", "SHORT_NAME", "NOMBRE", "RFC"},
@@ -191,23 +191,35 @@ ALWAYS_REQUIRED_HEADERS_BY_TEMPLATE: Dict[str, set[str]] = {
}
def _apply_required_prefix(template_id: str, headers: List[str]) -> List[str]:
"""Prefija '* ' a cabeceras siempre obligatorias para la plantilla."""
required_headers = ALWAYS_REQUIRED_HEADERS_BY_TEMPLATE.get(template_id)
if not required_headers:
return headers
def _compute_required_indices_map(base_rows: Dict[str, List[str]]) -> Dict[str, Set[int]]:
"""Índices de columnas obligatorias (según plantilla ES/base), para marcar * en cualquier idioma."""
out: Dict[str, Set[int]] = {}
for tid, headers in base_rows.items():
req = ALWAYS_REQUIRED_HEADERS_BY_TEMPLATE.get(tid)
if not req:
continue
req_norm = {_normalize_header_for_match(h) for h in req}
idx_set: Set[int] = set()
for i, h in enumerate(headers):
if not h:
continue
if _normalize_header_for_match(h) in req_norm:
idx_set.add(i)
out[tid] = idx_set
return out
required_norm = {_normalize_header_for_match(h) for h in required_headers}
def _apply_required_prefix_indices(template_id: str, headers: List[str]) -> List[str]:
"""Prefija '* ' usando índices fijos de la plantilla base (independiente del idioma de cabecera)."""
idx_set = _REQUIRED_INDICES.get(template_id)
if not idx_set:
return headers
out: List[str] = []
for header in headers:
norm_header = _normalize_header_for_match(header)
if norm_header and norm_header in required_norm:
if header.startswith("* "):
out.append(header)
else:
out.append(f"* {header}")
for i, h in enumerate(headers):
if i in idx_set and h and not str(h).startswith("* "):
out.append(f"* {h}")
else:
out.append(header)
out.append(h)
return out
@@ -231,6 +243,7 @@ def _build_registry() -> Dict[str, List[str]]:
"imp_def_series",
"exp_def_header",
"exp_def_details",
"exp_def_series",
"cmex_header",
"cmex_details",
"cmex_series",
@@ -292,7 +305,9 @@ def _build_registry() -> Dict[str, List[str]]:
return registry
_TEMPLATE_HEADERS: Dict[str, List[str]] = _build_registry()
_REGISTRY_BASE_ES: Dict[str, List[str]] = _build_registry()
_REQUIRED_INDICES: Dict[str, Set[int]] = _compute_required_indices_map(_REGISTRY_BASE_ES)
_TEMPLATE_HEADERS: Dict[str, List[str]] = _REGISTRY_BASE_ES
# Nombre de archivo sugerido para descarga (sin path)
TEMPLATE_FILENAMES: Dict[str, str] = {
@@ -320,15 +335,65 @@ TEMPLATE_FILENAMES: Dict[str, str] = {
"cmex_series": "EstructuraSeriesFacComprasMex.csv",
"exp_def_header": "EstructuraEncFacExpoCamReg.csv",
"exp_def_details": "EstructuraParExpoCamReg.csv",
"exp_def_series": "EstructuraSeriesFacExpoCamReg.csv",
}
def get_download_headers(template_id: str, locale: Optional[str] = "es") -> Optional[List[str]]:
"""
Cabeceras de descarga para la plantilla (sin prefijo *).
locale: es | en (default es).
"""
from api.v1.modules.a76.layouts_csv.common.template_locale import (
download_header_cell,
normalize_locale,
)
loc = normalize_locale(locale)
cols = resolve_imports_template(template_id)
if cols is not None:
return [download_header_cell(c, loc) for c in cols]
if template_id in ("part_numbers", "items"):
from api.v1.modules.a76.layouts_csv.parts import template_config as parts_tc
return parts_tc.download_headers_for_locale(loc)
if template_id == "material_classes":
from api.v1.modules.a76.layouts_csv.classes import template_config as classes_tc
return classes_tc.download_headers_for_locale(loc)
# template_id del registry → (module_columns, clave interna del dict de columnas)
catalog_resolvers: list[tuple[str, dict, str]] = [
("boms", BOMS_TEMPLATE_COLUMNS, "boms"),
("customs_brokers", CUSTOMS_BROKERS_TEMPLATE_COLUMNS, "customs_brokers"),
("clients_providers", CLIENTS_PROVIDERS_TEMPLATE_COLUMNS, "client_providers"),
("exchange_rates", EXCHANGE_RATE_TEMPLATE_COLUMNS, "exchange_rates"),
("american_fractions", US_TARIFF_FRACTIONS_TEMPLATE_COLUMNS, "us_tariff_fractions"),
("pedimentos", PEDIMENTOS_TEMPLATE_COLUMNS, "pedimentos"),
("transports", VEHICLES_TEMPLATE_COLUMNS, "vehicles"),
("drivers", DRIVERS_TEMPLATE_COLUMNS, "drivers"),
("trailers", TRAILERS_TEMPLATE_COLUMNS, "trailers"),
("transporters", TRANSPORTERS_TEMPLATE_COLUMNS, "transporters"),
]
for tid, mapping, inner_key in catalog_resolvers:
if tid != template_id:
continue
cols = mapping.get(inner_key)
if cols:
return [download_header_cell(col, loc) for col in cols]
return None
def get_template_headers(template_id: str) -> Optional[List[str]]:
"""Devuelve la lista de cabeceras canónicas para el template_id, o None si no existe."""
headers = _TEMPLATE_HEADERS.get(template_id)
if headers is None:
"""Cabeceras como en descarga ES histórica, con prefijo * en obligatorias."""
raw = get_download_headers(template_id, "es")
if raw is None:
return None
return _apply_required_prefix(template_id, headers)
return _apply_required_prefix_indices(template_id, raw)
def get_template_filename(template_id: str) -> str:
@@ -336,14 +401,19 @@ def get_template_filename(template_id: str) -> str:
return TEMPLATE_FILENAMES.get(template_id, f"plantilla_{template_id}.csv")
def generate_csv_content(template_id: str, include_bom: bool = True) -> Optional[bytes]:
def generate_csv_content(
template_id: str,
locale: str = "es",
include_bom: bool = True,
) -> Optional[bytes]:
"""
Genera el contenido CSV (solo fila de cabeceras) para el template_id.
UTF-8, opcionalmente con BOM para Excel.
"""
headers = get_template_headers(template_id)
headers = get_download_headers(template_id, locale)
if not headers:
return None
headers = _apply_required_prefix_indices(template_id, headers)
buf = io.StringIO()
writer = csv.writer(buf, lineterminator="\n")
writer.writerow(headers)