424 lines
14 KiB
Python
424 lines
14 KiB
Python
"""
|
|
Registro central de plantillas CSV: template_id -> lista de cabeceras canónicas.
|
|
Construido a partir de los TEMPLATE_COLUMNS de cada módulo de imports.
|
|
"""
|
|
import csv
|
|
import io
|
|
from typing import Dict, List, Optional, Set
|
|
|
|
# Importar configs de cada módulo
|
|
from api.v1.modules.a76.layouts_csv.facturas.template_config import (
|
|
_resolve_template_columns as resolve_imports_template,
|
|
)
|
|
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,
|
|
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,
|
|
)
|
|
from api.v1.modules.a76.layouts_csv.clients_and_providers.template_config import (
|
|
TEMPLATE_COLUMNS as CLIENTS_PROVIDERS_TEMPLATE_COLUMNS,
|
|
)
|
|
from api.v1.modules.a76.layouts_csv.exchange_rate.template_config import (
|
|
TEMPLATE_COLUMNS as EXCHANGE_RATE_TEMPLATE_COLUMNS,
|
|
)
|
|
from api.v1.modules.a76.layouts_csv.us_tariff_fractions.template_config import (
|
|
TEMPLATE_COLUMNS as US_TARIFF_FRACTIONS_TEMPLATE_COLUMNS,
|
|
)
|
|
from api.v1.modules.a76.layouts_csv.pedmientos.template_config import (
|
|
TEMPLATE_COLUMNS as PEDIMENTOS_TEMPLATE_COLUMNS,
|
|
)
|
|
from api.v1.modules.a76.layouts_csv.vehicles.template_config import (
|
|
TEMPLATE_COLUMNS as VEHICLES_TEMPLATE_COLUMNS,
|
|
)
|
|
from api.v1.modules.a76.layouts_csv.drivers.template_config import (
|
|
TEMPLATE_COLUMNS as DRIVERS_TEMPLATE_COLUMNS,
|
|
)
|
|
from api.v1.modules.a76.layouts_csv.trailers.template_config import (
|
|
TEMPLATE_COLUMNS as TRAILERS_TEMPLATE_COLUMNS,
|
|
)
|
|
from api.v1.modules.a76.layouts_csv.transportistas.template_config import (
|
|
TEMPLATE_COLUMNS as TRANSPORTERS_TEMPLATE_COLUMNS,
|
|
)
|
|
|
|
|
|
def _normalize_header_for_match(header: str) -> str:
|
|
"""Normaliza cabeceras para comparación interna (sin alterar salida)."""
|
|
if not header:
|
|
return ""
|
|
cleaned = header.strip()
|
|
if cleaned.startswith("* "):
|
|
cleaned = cleaned[2:]
|
|
return cleaned.strip().upper()
|
|
|
|
|
|
ALWAYS_REQUIRED_HEADERS_BY_TEMPLATE: Dict[str, set[str]] = {
|
|
# Facturas encabezados
|
|
"imp_temp_header": {
|
|
"NUMERO FACTURA",
|
|
"FECHA FACTURA",
|
|
"REGIMEN",
|
|
"CLAVE PROVEEDOR",
|
|
"CLAVE VENDIDO A",
|
|
"CLAVE ENVIADO A",
|
|
"AGENTE ADUANAL",
|
|
"ADUANA DE CRUCE",
|
|
},
|
|
"imp_def_header": {
|
|
"NUMERO FACTURA",
|
|
"FECHA FACTURA",
|
|
"REGIMEN",
|
|
"CLAVE PROVEEDOR",
|
|
"CLAVE VENDIDO A",
|
|
"CLAVE ENVIADO A",
|
|
"AGENTE ADUANAL",
|
|
},
|
|
"exp_def_header": {
|
|
"NUMERO FACTURA",
|
|
"FECHA FACTURA",
|
|
"REGIMEN",
|
|
"CLAVE PROVEEDOR",
|
|
"CLAVE VENDIDO A",
|
|
"CLAVE ENVIADO A",
|
|
"AGENTE ADUANAL",
|
|
"ADUANA DE CRUCE",
|
|
},
|
|
"cmex_header": {
|
|
"NUMERO FACTURA",
|
|
"FECHA FACTURA",
|
|
"CLAVE PROVEEDOR",
|
|
"CLAVE VENDIDO A",
|
|
"CLAVE ENVIADO A",
|
|
},
|
|
# Facturas partidas (plantillas del registry)
|
|
"imp_temp_details": {
|
|
"NUMERO FACTURA",
|
|
"CLASE",
|
|
"CANTIDAD IMPORTADA",
|
|
"COSTO UNITARIO",
|
|
"PESO NETO",
|
|
"PAIS ORIGEN",
|
|
"PREFERENCIA ARANCELARIA",
|
|
},
|
|
"imp_def_details": {
|
|
"NUMERO FACTURA",
|
|
"CLASE",
|
|
"CANTIDAD IMPORTADA",
|
|
"COSTO UNITARIO",
|
|
"PESO NETO",
|
|
"PAIS ORIGEN",
|
|
"PREFERENCIA ARANCELARIA",
|
|
"NUM. PARTE",
|
|
},
|
|
"exp_def_details": {
|
|
"NUMERO FACTURA",
|
|
"CLASE",
|
|
"CANTIDAD IMPORTADA",
|
|
"COSTO UNITARIO",
|
|
"PESO NETO",
|
|
"PAIS ORIGEN",
|
|
"PREFERENCIA ARANCELARIA",
|
|
},
|
|
"cmex_details": {
|
|
"NUMERO FACTURA",
|
|
"CLASE",
|
|
"CANTIDAD IMPORTADA",
|
|
"COSTO UNITARIO",
|
|
"PESO NETO",
|
|
"PAIS ORIGEN",
|
|
"PREFERENCIA ARANCELARIA",
|
|
"NUM. PARTE",
|
|
},
|
|
# Facturas series (plantillas del registry)
|
|
"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"},
|
|
"exchange_rates": {"FECHA", "VALOR"},
|
|
"american_fractions": {"FRACCION_ARANCELARIA", "DESCRIPCION"},
|
|
"material_classes": {
|
|
"CLAVE CLASE",
|
|
"CLASE",
|
|
"DESCRIPCION ESPAÑOL",
|
|
"DESCRIPCIONE",
|
|
"TIPO DE MATERIAL",
|
|
"CLAVEMAT",
|
|
"U.M. COMERCIAL",
|
|
"UNIMED",
|
|
"FRACCION ARANCELARIA",
|
|
"FRACCION",
|
|
},
|
|
"part_numbers": {
|
|
"NUMERO DE PARTE",
|
|
"NUMPARTE",
|
|
"DESCRIPCION EN ESPAÑOL",
|
|
"DESCRIPCIONE",
|
|
"UNIDAD DE MEDIDA COMERCIAL",
|
|
"UNIMED",
|
|
},
|
|
"items": {
|
|
"NUMERO DE PARTE",
|
|
"NUMPARTE",
|
|
"DESCRIPCION EN ESPAÑOL",
|
|
"DESCRIPCIONE",
|
|
"UNIDAD DE MEDIDA COMERCIAL",
|
|
"UNIMED",
|
|
},
|
|
"boms": {"NUMPARTE_PADRE", "NUMPARTE_COMPONENTE", "CANTIDAD"},
|
|
"pedimentos": {
|
|
"AÑO", "PATENTE", "NUMERO", "PEDIMENTO",
|
|
"TIPO_OPERACION",
|
|
"CLAVE_PEDIMENTO",
|
|
"REGIMEN",
|
|
"FECHA_INICIO",
|
|
"FECHA_FINAL",
|
|
"FECHA_PAGO",
|
|
"ADUANA_SECCION_CRUCE",
|
|
},
|
|
"transports": {"CLAVE", "CODIGO DE ENTIDAD"},
|
|
"drivers": {"TRANSPORTISTA", "LINEA", "CLAVE CONDUCTOR"},
|
|
"trailers": {"NUMERO TRAILER"},
|
|
"transporters": {"CLAVE TRANSPORTISTA", "NOMBRE"},
|
|
}
|
|
|
|
|
|
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
|
|
|
|
|
|
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 i, h in enumerate(headers):
|
|
if i in idx_set and h and not str(h).startswith("* "):
|
|
out.append(f"* {h}")
|
|
else:
|
|
out.append(h)
|
|
return out
|
|
|
|
|
|
def _canonicals_from_columns(cols: Optional[List[Dict]]) -> List[str]:
|
|
"""Extrae la lista de nombres canónicos en orden a partir de una lista de columnas."""
|
|
if not cols:
|
|
return []
|
|
return [item["canonical"] for item in cols]
|
|
|
|
|
|
def _build_registry() -> Dict[str, List[str]]:
|
|
registry: Dict[str, List[str]] = {}
|
|
|
|
# a76/imports (facturas): imp_temp_header, imp_temp_details, imp_def_*, exp_def_*, cmex_*, series
|
|
for tid in (
|
|
"imp_temp_header",
|
|
"imp_temp_details",
|
|
"imp_temp_series",
|
|
"imp_def_header",
|
|
"imp_def_details",
|
|
"imp_def_series",
|
|
"exp_def_header",
|
|
"exp_def_details",
|
|
"exp_def_series",
|
|
"cmex_header",
|
|
"cmex_details",
|
|
"cmex_series",
|
|
):
|
|
cols = resolve_imports_template(tid)
|
|
registry[tid] = _canonicals_from_columns(cols)
|
|
|
|
# part_numbers (parts); "items" usa la misma plantilla
|
|
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"))
|
|
|
|
# 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"))
|
|
|
|
# clients_providers
|
|
registry["clients_providers"] = _canonicals_from_columns(CLIENTS_PROVIDERS_TEMPLATE_COLUMNS.get("client_providers"))
|
|
|
|
# exchange_rates
|
|
registry["exchange_rates"] = _canonicals_from_columns(EXCHANGE_RATE_TEMPLATE_COLUMNS.get("exchange_rates"))
|
|
|
|
# american_fractions
|
|
registry["american_fractions"] = _canonicals_from_columns(
|
|
US_TARIFF_FRACTIONS_TEMPLATE_COLUMNS.get("us_tariff_fractions")
|
|
)
|
|
|
|
# pedimentos
|
|
registry["pedimentos"] = _canonicals_from_columns(PEDIMENTOS_TEMPLATE_COLUMNS.get("pedimentos"))
|
|
|
|
# transports (vehicles)
|
|
registry["transports"] = _canonicals_from_columns(VEHICLES_TEMPLATE_COLUMNS.get("vehicles"))
|
|
|
|
# drivers
|
|
registry["drivers"] = _canonicals_from_columns(DRIVERS_TEMPLATE_COLUMNS.get("drivers"))
|
|
|
|
# trailers
|
|
registry["trailers"] = _canonicals_from_columns(TRAILERS_TEMPLATE_COLUMNS.get("trailers"))
|
|
|
|
# transporters (transportistas)
|
|
registry["transporters"] = _canonicals_from_columns(TRANSPORTERS_TEMPLATE_COLUMNS.get("transporters"))
|
|
|
|
return 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] = {
|
|
"customs_brokers": "EstructuraCatAgenteAduanal.csv",
|
|
"clients_providers": "EstructuraCatClienteProv.csv",
|
|
"exchange_rates": "EstructuraCatTiposCambio.csv",
|
|
"american_fractions": "EstructuraCatFraccAme.csv",
|
|
"material_classes": "EstructuraCatClasesAF.csv",
|
|
"part_numbers": "EstructuraCatPartesAF.csv",
|
|
"items": "EstructuraCatPartesAF.csv",
|
|
"boms": "EstructuraBOMS.csv",
|
|
"pedimentos": "EstructuraCatPedimentos.csv",
|
|
"transports": "EstructuraCatTransportes.csv",
|
|
"drivers": "EstructuraCatConductor.csv",
|
|
"trailers": "EstructuraCatTrailers.csv",
|
|
"transporters": "EstructuraCatTransportistas.csv",
|
|
"imp_temp_header": "EstructuraEncFacImpoTemp.csv",
|
|
"imp_temp_details": "EstructuraParFacImpoTempAF.csv",
|
|
"imp_temp_series": "EstructuraSeriesFacImpoTemp.csv",
|
|
"imp_def_header": "EstructuraEncFacImpoDef.csv",
|
|
"imp_def_details": "EstructuraParFacImpoDefAF.csv",
|
|
"imp_def_series": "EstructuraSeriesFacImpoDef.csv",
|
|
"cmex_header": "EstructuraEncFacComprasMex.csv",
|
|
"cmex_details": "EstructuraParFacComprasMex.csv",
|
|
"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]]:
|
|
"""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_indices(template_id, raw)
|
|
|
|
|
|
def get_template_filename(template_id: str) -> str:
|
|
"""Nombre de archivo sugerido para la descarga."""
|
|
return TEMPLATE_FILENAMES.get(template_id, f"plantilla_{template_id}.csv")
|
|
|
|
|
|
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_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)
|
|
content = buf.getvalue().encode("utf-8")
|
|
if include_bom:
|
|
content = b"\xef\xbb\xbf" + content
|
|
return content
|