354 lines
12 KiB
Python
354 lines
12 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
|
|
|
|
# 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 (
|
|
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"},
|
|
# 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 _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
|
|
|
|
required_norm = {_normalize_header_for_match(h) for h in required_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}")
|
|
else:
|
|
out.append(header)
|
|
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",
|
|
"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
|
|
|
|
|
|
_TEMPLATE_HEADERS: Dict[str, List[str]] = _build_registry()
|
|
|
|
# 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",
|
|
}
|
|
|
|
|
|
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:
|
|
return None
|
|
return _apply_required_prefix(template_id, headers)
|
|
|
|
|
|
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, 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)
|
|
if not headers:
|
|
return None
|
|
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
|