feature/csv-templates

backend/api/v1/modules/a76/csv_templates/registry.py

@@ -0,0 +1,147 @@
+"""
+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.imports.template_config import (
+    TEMPLATE_COLUMNS as IMPORTS_TEMPLATE_COLUMNS,
+    _resolve_template_columns as resolve_imports_template,
+)
+from api.v1.modules.a76.parts.imports.template_config import TEMPLATE_COLUMNS as PARTS_TEMPLATE_COLUMNS
+from api.v1.modules.a76.boms.imports.template_config import TEMPLATE_COLUMNS as BOMS_TEMPLATE_COLUMNS
+from api.v1.modules.a76.classes.imports.template_config import TEMPLATE_COLUMNS as CLASSES_TEMPLATE_COLUMNS
+from api.v1.modules.a76.customs_brokers.imports.template_config import (
+    TEMPLATE_COLUMNS as CUSTOMS_BROKERS_TEMPLATE_COLUMNS,
+)
+from api.v1.modules.a76.clients_and_providers.imports.template_config import (
+    TEMPLATE_COLUMNS as CLIENTS_PROVIDERS_TEMPLATE_COLUMNS,
+)
+from api.v1.modules.a76.general_catalogs.exchange_rate.imports.template_config import (
+    TEMPLATE_COLUMNS as EXCHANGE_RATE_TEMPLATE_COLUMNS,
+)
+from api.v1.modules.a76.general_catalogs.fractions.us_tariff_fractions.imports.template_config import (
+    TEMPLATE_COLUMNS as US_TARIFF_FRACTIONS_TEMPLATE_COLUMNS,
+)
+from api.v1.modules.a76.pedmientos.imports.template_config import (
+    TEMPLATE_COLUMNS as PEDIMENTOS_TEMPLATE_COLUMNS,
+)
+from api.v1.modules.a76.transportation.vehicles.imports.template_config import (
+    TEMPLATE_COLUMNS as VEHICLES_TEMPLATE_COLUMNS,
+)
+from api.v1.modules.a76.transportation.drivers.imports.template_config import (
+    TEMPLATE_COLUMNS as DRIVERS_TEMPLATE_COLUMNS,
+)
+from api.v1.modules.a76.transportation.trailers.imports.template_config import (
+    TEMPLATE_COLUMNS as TRAILERS_TEMPLATE_COLUMNS,
+)
+
+
+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_*
+    for tid in ("imp_temp_header", "imp_temp_details", "imp_def_header", "imp_def_details", "exp_def_header", "exp_def_details"):
+        cols = resolve_imports_template(tid)
+        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)
+
+    # boms
+    registry["boms"] = _canonicals_from_columns(BOMS_TEMPLATE_COLUMNS.get("boms"))
+
+    # material_classes
+    registry["material_classes"] = _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"))
+
+    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",
+    "imp_temp_header": "EstructuraEncFacImpoTemp.csv",
+    "imp_temp_details": "EstructuraParFacImpoTempAF.csv",
+    "imp_def_header": "EstructuraEncFacImpoDef.csv",
+    "imp_def_details": "EstructuraParFacImpoDefAF.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."""
+    return _TEMPLATE_HEADERS.get(template_id)
+
+
+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