Merge pull request 'feature/utf8-normalizado' (#275) from fix/carga-masiva-utf8 into development

Reviewed-on: ADUANASOFT/anexo76#275

backend/api/v1/modules/a76/layouts_csv/classes/tasks.py

@@ -33,6 +33,12 @@ CLS_IMPORT_ERROR_LINES_PREFIX = "cls_import_error_lines:"
 CLS_IMPORT_REDIS_TTL = common_storage.IMPORT_REDIS_TTL
 
 
+def _read_plan_for_classes(fieldnames):
+    if fieldnames:
+        return common_csv.CsvReadPlan(header_mode="headerless", fieldnames=fieldnames)
+    return common_csv.CsvReadPlan(header_mode="header")
+
+
 @celery_app.task(bind=True)
 def scan_file(self, job_id: str, config: str = None):
     logger.info("Classes import: starting scan for job %s", job_id)
@@ -45,8 +51,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)
+    read_plan = _read_plan_for_classes(fieldnames)
     try:
-        total_rows = common_csv.count_csv_rows(file_path, has_header=has_header)
+        total_rows = common_csv.count_csv_rows(file_path, has_header=has_header, read_plan=read_plan)
     except Exception as e:
         return {"status": "failed", "error": str(e)}
 
@@ -83,7 +90,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, fieldnames=fieldnames):
+            for i, row in common_csv.iter_csv_rows_with_plan(file_path, read_plan=read_plan):
                 self.update_state(
                         state="PROGRESS",
                         meta={"current": i, "total": total_rows, "errors": error_count},
@@ -171,6 +178,7 @@ def insert_valid_rows(self, job_id: str):
     meta_path = common_meta.get_meta_path(file_path)
 
     fieldnames, _ = detect_headers_or_data(file_path, common_normalize.normalize_header)
+    read_plan = _read_plan_for_classes(fieldnames)
 
     try:
         with CoreSessionLocal() as session:
@@ -183,7 +191,7 @@ def insert_valid_rows(self, job_id: str):
                 if key:
                     existing_by_code[key] = c
 
-            for i, row in common_csv.iter_csv_rows(file_path, fieldnames=fieldnames):
+            for i, row in common_csv.iter_csv_rows_with_plan(file_path, read_plan=read_plan):
                 if i in error_lines:
                     continue
 

backend/api/v1/modules/a76/layouts_csv/classes/template_config.py

@@ -7,28 +7,12 @@ import io
 from typing import Dict, List, Any, Optional, Tuple
 
 from ..common.cell_value import cell_to_str
+from ..common import csv_reader as common_csv_reader
 
 
 # Valores que indican que la primera fila es cabecera (primera columna normalizada)
 FIRST_COLUMN_HEADER_VALUES = ("CLAVE CLASE", "CLASE")
 
-_ENCODING_FALLBACKS: Tuple[str, ...] = ("utf-8-sig", "utf-8", "cp1252", "latin-1")
-
-
-def _read_text_sample(file_path: str, sample_bytes: int = 2048) -> str:
-    with open(file_path, "rb") as f:
-        raw = f.read(sample_bytes)
-    last_err: Optional[Exception] = None
-    for enc in _ENCODING_FALLBACKS:
-        try:
-            return raw.decode(enc)
-        except Exception as e:
-            last_err = e
-    if last_err:
-        raise last_err
-    return ""
-
-
 def detect_headers_or_data(
     file_path: str,
     normalize_header_fn,
@@ -42,24 +26,27 @@ def detect_headers_or_data(
     - Si no -> has_header=False, fieldnames=TEMPLATE_DOWNLOAD_HEADERS (la primera fila es dato).
     """
     try:
-        # `encoding` se mantiene por compatibilidad; si falla, hacemos fallback para CSVs tipo Excel (cp1252/latin-1).
-        if encoding and encoding.lower() not in ("auto", "detect"):
-            try:
-                with open(file_path, "r", encoding=encoding) as f:
-                    sample = f.read(2048)
-            except Exception:
-                sample = _read_text_sample(file_path, sample_bytes=2048)
-        else:
-            sample = _read_text_sample(file_path, sample_bytes=2048)
+        sample, _ = common_csv_reader.read_text_sample(
+            file_path,
+            requested_encoding=encoding,
+            sample_chars=2048,
+        )
     except Exception:
+        try:
+            sample, _ = common_csv_reader.read_text_sample(
+                file_path,
+                requested_encoding="auto",
+                sample_chars=2048,
+            )
+        except Exception:
+            return None, True
+
+    if not sample:
         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
+    dialect = common_csv_reader.detect_csv_dialect(sample, delimiters=",;\t")
     reader = csv.reader(io.StringIO(lines[0]), dialect=dialect)
     first_row = next(reader, None)
     if not first_row:
@@ -124,10 +111,15 @@ def row_from_template(row: Dict[str, Any], normalize_header_fn) -> Dict[str, Any
         if key_norm in lookup:
             out[lookup[key_norm]] = cell_to_str(value)
         elif key_norm.startswith("CLAVE CLASE"):
-            # CSV leído con delimitador incorrecto: primera columna es "CLAVE CLASE,..." -> usar primer valor como CLASE
+            # CSV leído con delimitador incorrecto: primera columna puede venir colapsada.
             if "CLASE" not in out and value:
                 val_str = cell_to_str(value)
-                first_val = (val_str.split(",")[0] if "," in val_str else val_str).strip()
+                first_val = val_str
+                for delimiter in (",", ";", "\t"):
+                    if delimiter in first_val:
+                        first_val = first_val.split(delimiter)[0]
+                        break
+                first_val = first_val.lstrip("\ufeff").strip()
                 if first_val:
                     out["CLASE"] = first_val
     return out

backend/api/v1/modules/a76/layouts_csv/common/csv_ingestion_rollout.md

@@ -0,0 +1,38 @@
+# CSV Ingestion Global Rollout
+
+## Goal
+- Apply one CSV ingestion policy across all upload flows.
+- Detect regressions early in scan/commit execution.
+
+## Rollout Stages
+- Stage 1: Enable common `CsvReadPlan` path for semiconsolidated layouts (`classes`, `parts`, `pedmientos`).
+- Stage 2: Enable common deduped-header iterator for legacy transportation layouts (`drivers`, `trailers`).
+- Stage 3: Keep `facturas` business parsing while enforcing common encoding + dialect detection.
+- Stage 4: Reuse common encoding detection in non-layout ingestion (`carta_porte_codes` seed).
+
+## Metrics To Track
+- `csv_scan_failed_total{layout}`: scan failures by layout.
+- `csv_commit_failed_total{layout}`: commit failures by layout.
+- `csv_decode_fallback_total{layout,encoding}`: resolved encoding different from UTF-8.
+- `csv_headerless_detected_total{layout}`: files treated as headerless.
+- `csv_rows_processed_total{layout,stage}`: processed rows in scan/commit.
+
+## Alerting Rules
+- Alert when `csv_scan_failed_total{layout}` spikes > 2x baseline for 15 minutes.
+- Alert when `csv_commit_failed_total{layout}` spikes > 2x baseline for 15 minutes.
+- Alert when `csv_decode_fallback_total` changes abruptly after deployment.
+- Alert when median `csv_rows_processed_total` drops sharply for active layouts.
+
+## Logging Requirements
+- Log resolved encoding and selected dialect once per job.
+- Log `header_mode` (`header`, `headerless`, `auto`) once per job.
+- Keep `job_id`, `layout`, `tenant_id`, `company_id` in structured logs.
+
+## Verification Checklist
+- Validate one UTF-8 and one CP1252 file per layout in staging.
+- Validate comma and semicolon delimiters in staging.
+- Validate scan and commit consistency for line numbers and stored values.
+
+## Rollback Strategy
+- Revert to previous release if scan/commit failures exceed threshold for 30 minutes.
+- Keep compatibility wrappers in `csv_reader` to reduce rollback blast radius.

backend/api/v1/modules/a76/layouts_csv/common/csv_reader.py

@@ -5,7 +5,9 @@ Si se pasa headerless_first_cell_values, se detecta si la primera fila es cabece
 """
 import csv
 import io
-from typing import Iterator, Tuple, Dict, Any, Optional, List, Set, Sequence
+import codecs
+from dataclasses import dataclass
+from typing import Iterator, Tuple, Dict, Any, Optional, List, Set, Sequence, Literal
 
 
 def _normalize_empty_headers(headers: List[str]) -> List[str]:
@@ -21,10 +23,72 @@ def _normalize_empty_headers(headers: List[str]) -> List[str]:
     return result
 
 
+def dedupe_duplicate_headers(headers: List[str], fallback_name: str = "COL") -> List[str]:
+    """
+    Hace únicos headers repetidos agregando sufijo incremental.
+    """
+    counts: Dict[str, int] = {}
+    unique: List[str] = []
+    for header in headers:
+        name = str(header or "").strip() or fallback_name
+        count = counts.get(name, 0) + 1
+        counts[name] = count
+        unique.append(name if count == 1 else f"{name} {count}")
+    return unique
+
+
 _ENCODING_FALLBACKS: Sequence[str] = ("utf-8-sig", "utf-8", "cp1252", "latin-1")
 
 
-def _detect_text_encoding(
+@dataclass(frozen=True)
+class CsvReadPlan:
+    """
+    Contrato declarativo para lectura de CSV.
+    - header_mode=header: primera fila siempre cabecera.
+    - header_mode=headerless: primera fila siempre dato.
+    - header_mode=auto: decide con headerless_first_cell_values.
+    """
+    header_mode: Literal["header", "headerless", "auto"] = "header"
+    fieldnames: Optional[List[str]] = None
+    headerless_first_cell_values: Optional[Set[str]] = None
+    encoding: Optional[str] = "auto"
+    delimiters: str = ",;\t"
+    sample_chars: int = 2048
+
+
+@dataclass(frozen=True)
+class CsvReadMetadata:
+    encoding: str
+    dialect: Any
+    has_header: bool
+    fieldnames: Optional[List[str]]
+
+
+def _sample_decodes_with_encoding(raw: bytes, encoding: str) -> bool:
+    """
+    Valida si un sample binario puede decodificarse con `encoding`.
+    Para UTF-8/UTF-8-SIG tolera corte al final de un multibyte (sample truncado).
+    """
+    try:
+        raw.decode(encoding)
+        return True
+    except UnicodeDecodeError as err:
+        if encoding not in ("utf-8", "utf-8-sig"):
+            return False
+        # Si el error es por sample truncado al final del buffer, validar con decodificador incremental.
+        if err.end != len(raw):
+            return False
+        try:
+            decoder = codecs.getincrementaldecoder(encoding)(errors="strict")
+            decoder.decode(raw, final=False)
+            return True
+        except Exception:
+            return False
+    except Exception:
+        return False
+
+
+def detect_text_encoding(
     file_path: str,
     encodings: Sequence[str] = _ENCODING_FALLBACKS,
     sample_bytes: int = 8192,
@@ -38,8 +102,8 @@ def _detect_text_encoding(
     last_err: Optional[Exception] = None
     for enc in encodings:
         try:
-            raw.decode(enc)
-            return enc
+            if _sample_decodes_with_encoding(raw, enc):
+                return enc
         except Exception as e:
             last_err = e
     if last_err:
@@ -47,6 +111,132 @@ def _detect_text_encoding(
     return "utf-8-sig"
 
 
+def _detect_text_encoding(
+    file_path: str,
+    encodings: Sequence[str] = _ENCODING_FALLBACKS,
+    sample_bytes: int = 8192,
+) -> str:
+    """
+    Compatibilidad retroactiva para imports internos antiguos.
+    """
+    return detect_text_encoding(file_path, encodings=encodings, sample_bytes=sample_bytes)
+
+
+def resolve_read_encoding(file_path: str, requested_encoding: Optional[str] = "auto") -> str:
+    if requested_encoding and requested_encoding.lower() not in ("auto", "detect"):
+        return requested_encoding
+    return detect_text_encoding(file_path)
+
+
+def read_text_sample(
+    file_path: str,
+    requested_encoding: Optional[str] = "auto",
+    sample_chars: int = 2048,
+) -> Tuple[str, str]:
+    """
+    Lee muestra de texto para detectar delimitador/primera fila.
+    Retorna (sample_text, resolved_encoding).
+    """
+    encoding = resolve_read_encoding(file_path, requested_encoding)
+    with open(file_path, "r", encoding=encoding) as f:
+        return f.read(sample_chars), encoding
+
+
+def detect_csv_dialect(sample: str, delimiters: str = ",;\t") -> Any:
+    try:
+        return csv.Sniffer().sniff(sample, delimiters=delimiters)
+    except Exception:
+        return "excel"
+
+
+def inspect_csv(file_path: str, read_plan: Optional[CsvReadPlan] = None) -> CsvReadMetadata:
+    plan = read_plan or CsvReadPlan()
+    sample, encoding = read_text_sample(
+        file_path,
+        requested_encoding=plan.encoding,
+        sample_chars=plan.sample_chars,
+    )
+    dialect = detect_csv_dialect(sample, delimiters=plan.delimiters)
+    has_header = True
+    fieldnames: Optional[List[str]] = None
+    if plan.header_mode == "headerless":
+        has_header = False
+        fieldnames = list(plan.fieldnames or [])
+    elif plan.header_mode == "auto" and plan.fieldnames and plan.headerless_first_cell_values is not None:
+        first_line = sample.splitlines()[0] if sample.splitlines() else ""
+        if first_line:
+            row_reader = csv.reader(io.StringIO(first_line), dialect=dialect)
+            first_cells = next(row_reader, None)
+            first_cell_clean = ((first_cells or [""])[0] or "").lstrip("\ufeff").strip().upper()
+            if first_cell_clean in plan.headerless_first_cell_values:
+                has_header = False
+                fieldnames = list(plan.fieldnames)
+    return CsvReadMetadata(
+        encoding=encoding,
+        dialect=dialect,
+        has_header=has_header,
+        fieldnames=fieldnames,
+    )
+
+
+def iter_csv_rows_with_plan(
+    file_path: str,
+    read_plan: Optional[CsvReadPlan] = None,
+) -> Iterator[Tuple[int, Dict[str, Any]]]:
+    """
+    Iterador común de filas usando un ReadPlan.
+    """
+    plan = read_plan or CsvReadPlan()
+    metadata = inspect_csv(file_path, plan)
+    with open(file_path, "r", encoding=metadata.encoding) as f:
+        if metadata.has_header:
+            first_line = f.readline()
+            if not first_line:
+                return
+            row_reader = csv.reader(io.StringIO(first_line), dialect=metadata.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=metadata.dialect, restval="")
+            for i, row in enumerate(reader, start=1):
+                yield i, dict(row)
+            return
+
+        fieldnames = metadata.fieldnames or list(plan.fieldnames or [])
+        if not fieldnames:
+            return
+        row_reader = csv.reader(f, dialect=metadata.dialect)
+        for i, cells in enumerate(row_reader, start=1):
+            if cells is None:
+                continue
+            pad = len(fieldnames) - len(cells)
+            normalized_cells = cells[: len(fieldnames)] + ([""] * pad if pad > 0 else [])
+            yield i, dict(zip(fieldnames, normalized_cells))
+
+
+def iter_csv_rows_deduped_headers(
+    file_path: str,
+    fallback_name: str = "COL",
+) -> Iterator[Tuple[int, Dict[str, Any]]]:
+    """
+    Itera filas asumiendo cabecera en primera línea y deduplicando nombres repetidos.
+    """
+    metadata = inspect_csv(file_path, CsvReadPlan(header_mode="header"))
+    with open(file_path, "r", encoding=metadata.encoding) as f:
+        first_line = f.readline()
+        if not first_line:
+            return
+        header_reader = csv.reader(io.StringIO(first_line), dialect=metadata.dialect)
+        raw_headers = next(header_reader, None)
+        if not raw_headers:
+            return
+        headers = dedupe_duplicate_headers(raw_headers, fallback_name=fallback_name)
+        dict_reader = csv.DictReader(f, fieldnames=headers, dialect=metadata.dialect, restval="")
+        for i, row in enumerate(dict_reader, start=1):
+            yield i, dict(row)
+
+
 def iter_csv_rows(
     file_path: str,
     fieldnames: Optional[List[str]] = None,
@@ -60,65 +250,33 @@ def iter_csv_rows(
     (quitando BOM, strip, upper) está en headerless_first_cell_values, se trata como dato y se usan fieldnames.
     headerless_second_cell_key_pattern se ignora si no se usa (reservado para otros layouts).
     """
-    encoding = _detect_text_encoding(file_path)
-    with open(file_path, "r", encoding=encoding) as f:
-        sample = f.read(2048)
-        f.seek(0)
-        try:
-            dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
-        except Exception:
-            dialect = "excel"
-        if fieldnames and headerless_first_cell_values is not None:
-            first_line = f.readline()
-            if not first_line:
-                return
-            row_reader = csv.reader(io.StringIO(first_line), dialect=dialect)
-            first_cells = next(row_reader, None)
-            if not first_cells:
-                return
-            first_cell_clean = (first_cells[0] or "").lstrip("\ufeff").strip().upper()
-            use_headerless = first_cell_clean in headerless_first_cell_values
-            if use_headerless:
-                pad = len(fieldnames) - len(first_cells)
-                cells = first_cells[: len(fieldnames)] + ([""] * pad if pad > 0 else [])
-                yield 1, dict(zip(fieldnames, cells))
-                reader = csv.DictReader(f, fieldnames=fieldnames, dialect=dialect, restval="")
-                for i, row in enumerate(reader, start=2):
-                    yield i, dict(row)
-                return
-            f.seek(0)
-            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
-        elif fieldnames:
-            reader = csv.DictReader(f, fieldnames=fieldnames, dialect=dialect)
-            for i, row in enumerate(reader, start=1):
-                yield i, dict(row)
-        else:
-            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
+    if fieldnames and headerless_first_cell_values is not None:
+        plan = CsvReadPlan(
+            header_mode="auto",
+            fieldnames=fieldnames,
+            headerless_first_cell_values=headerless_first_cell_values,
+        )
+    elif fieldnames:
+        plan = CsvReadPlan(
+            header_mode="headerless",
+            fieldnames=fieldnames,
+        )
+    else:
+        plan = CsvReadPlan(header_mode="header")
+
+    for item in iter_csv_rows_with_plan(file_path, plan):
+        yield item
 
 
-def count_csv_rows(file_path: str, has_header: bool = True) -> int:
+def count_csv_rows(
+    file_path: str,
+    has_header: bool = True,
+    read_plan: Optional[CsvReadPlan] = None,
+) -> int:
     """Cuenta filas del CSV. Si has_header=True (por defecto), no cuenta la cabecera."""
-    encoding = _detect_text_encoding(file_path)
+    metadata = inspect_csv(file_path, read_plan) if read_plan else None
+    encoding = metadata.encoding if metadata else detect_text_encoding(file_path)
+    effective_has_header = metadata.has_header if metadata else has_header
     with open(file_path, "r", encoding=encoding) as f:
         total_lines = sum(1 for _ in f)
-    return total_lines if not has_header else max(0, total_lines - 1)
+    return total_lines if not effective_has_header else max(0, total_lines - 1)

backend/api/v1/modules/a76/layouts_csv/drivers/tasks.py

@@ -4,7 +4,6 @@ Flujo: scan_file (validación) → insert_valid_rows (commit).
 Usa layouts_csv.common (storage, normalize, meta, responses); CSV con headers duplicados (dedupe) y clave de estado en Redis.
 Paridad Clarion: actualizar, existing_driver_keys, valid_transporter_keys, valid_country_ame.
 """
-import csv
 import json
 import logging
 import os
@@ -18,6 +17,7 @@ from ..common import storage as common_storage
 from ..common import normalize as common_normalize
 from ..common import meta as common_meta
 from ..common import responses as common_responses
+from ..common import csv_reader as common_csv_reader
 from .template_config import row_from_template
 from .validators import validate_row_driver, validate_row_driver_desfase
 from .common.mappers import row_to_driver_data
@@ -42,17 +42,6 @@ def _get_redis():
     return redis.Redis.from_url(url, decode_responses=False)
 
 
-def _dedupe_headers(headers: List[str]) -> List[str]:
-    counts: Dict[str, int] = {}
-    unique: List[str] = []
-    for header in headers:
-        name = str(header or "").strip() or "COL"
-        count = counts.get(name, 0) + 1
-        counts[name] = count
-        unique.append(name if count == 1 else f"{name} {count}")
-    return unique
-
-
 def _do_scan(job_id: str, progress_callback: Optional[Any] = None) -> Dict[str, Any]:
     file_path = common_storage.ensure_file_from_redis(JOB_TYPE, job_id, "Drivers import")
     if not file_path:
@@ -62,8 +51,7 @@ def _do_scan(job_id: str, progress_callback: Optional[Any] = None) -> Dict[str,
     error_path = common_storage.error_path_for_job(JOB_TYPE, job_id)
 
     try:
-        with open(file_path, "r", encoding="utf-8-sig") as f:
-            total_rows = sum(1 for _ in f) - 1
+        total_rows = common_csv_reader.count_csv_rows(file_path, has_header=True)
     except Exception as e:
         return {"status": "failed", "error": str(e)}
 
@@ -117,24 +105,8 @@ def _do_scan(job_id: str, progress_callback: Optional[Any] = None) -> Dict[str,
     error_lines_list: List[int] = []
 
     try:
-        with open(file_path, "r", encoding="utf-8-sig") as f_in, open(
-            error_path, "w", encoding="utf-8"
-        ) as f_err:
-            sample = f_in.read(2048)
-            f_in.seek(0)
-            try:
-                dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
-            except Exception:
-                dialect = "excel"
-            reader = csv.reader(f_in, dialect=dialect)
-            try:
-                headers = next(reader)
-            except StopIteration:
-                headers = []
-            headers = _dedupe_headers(headers)
-            dict_reader = csv.DictReader(f_in, fieldnames=headers, dialect=dialect)
-
-            for i, row in enumerate(dict_reader, start=1):
+        with open(error_path, "w", encoding="utf-8") as f_err:
+            for i, row in common_csv_reader.iter_csv_rows_deduped_headers(file_path):
                 if progress_callback:
                     progress_callback(i, total_rows, error_count)
 
@@ -303,22 +275,7 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
 
     try:
         with CoreSessionLocal() as session:
-            with open(file_path, "r", encoding="utf-8-sig") as f:
-                sample = f.read(2048)
-                f.seek(0)
-                try:
-                    dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
-                except Exception:
-                    dialect = "excel"
-                reader = csv.reader(f, dialect=dialect)
-                try:
-                    headers = next(reader)
-                except StopIteration:
-                    headers = []
-                headers = _dedupe_headers(headers)
-                dict_reader = csv.DictReader(f, fieldnames=headers, dialect=dialect)
-
-                for i, row in enumerate(dict_reader, start=1):
+            for i, row in common_csv_reader.iter_csv_rows_deduped_headers(file_path):
                     if i in error_lines:
                         continue
 

backend/api/v1/modules/a76/layouts_csv/facturas/tasks.py

@@ -26,6 +26,7 @@ from sqlalchemy import func
 from ..common import storage as common_storage
 from ..common import meta as common_meta
 from ..common import responses as common_responses
+from ..common import csv_reader as common_csv_reader
 from .template_config import row_from_template
 from .validators.encabezados_impo_temp import csv_tipo_moneda_es_me_mn_mc
 from api.v1.modules.a76.app_settings.service import AppSettingsService
@@ -64,6 +65,27 @@ def _ensure_worker_has_meta_from_redis(job_id: str, file_path: str) -> bool:
 def _delete_import_from_redis(job_id: str) -> None:
     common_storage.delete_import_from_redis(JOB_TYPE, job_id)
 
+
+def _ensure_utf8_compatible_import_file(file_path: str) -> None:
+    """
+    Homogeneiza a UTF-8-SIG cuando el archivo venga en otro encoding
+    para que todo el flujo legado de facturas lea exactamente lo mismo.
+    """
+    encoding = common_csv_reader.detect_text_encoding(file_path)
+    if encoding in ("utf-8", "utf-8-sig"):
+        return
+    with open(file_path, "r", encoding=encoding) as src:
+        content = src.read()
+    with open(file_path, "w", encoding="utf-8-sig") as dst:
+        dst.write(content)
+
+
+def _facturas_csv_encoding(file_path: str) -> str:
+    """
+    Encapsula la detección para centralizar la política de lectura CSV en facturas.
+    """
+    return common_csv_reader.detect_text_encoding(file_path)
+
 class ForeignKeyValidator:
     def __init__(self, session, tenant_id, company_id):
         self.session = session
@@ -462,6 +484,7 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
     if not file_path:
         return {"status": "failed", "error": "File not found (missing or expired in queue). Please upload again."}
     common_storage.ensure_meta_from_redis(effective_job_type, job_id, file_path, log_prefix)
+    _ensure_utf8_compatible_import_file(file_path)
 
     error_path = common_storage.error_path_for_job(effective_job_type, job_id)
 
@@ -472,7 +495,7 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
 
     # 3. Count Total (Quick Pass) or just estimate
     try:
-        with open(file_path, 'r', encoding='utf-8-sig') as f:
+        with open(file_path, 'r', encoding=_facturas_csv_encoding(file_path)) as f:
             total_rows = sum(1 for _ in f) - 1  # Minus header
     except Exception as e:
         return {"status": "failed", "error": f"Cannot read file: {e}"}
@@ -618,11 +641,11 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
 
                 csv_series_count_so_far: Dict[Tuple[str, str], int] = {}
 
-                with open(file_path, "r", encoding="utf-8-sig") as f_in, open(error_path, "w", encoding="utf-8") as f_err:
+                with open(file_path, "r", encoding=_facturas_csv_encoding(file_path)) as f_in, open(error_path, "w", encoding="utf-8") as f_err:
                     sample = f_in.read(2048)
                     f_in.seek(0)
                     try:
-                        dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
+                        dialect = common_csv_reader.detect_csv_dialect(sample, delimiters=",;\t")
                     except Exception:
                         dialect = "excel"
                     reader = csv.DictReader(f_in, dialect=dialect)
@@ -793,11 +816,11 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
                 csv_series_count_so_far: Dict[Tuple[str, str], int] = {}
 
             invoice_numbers_from_csv: Set[str] = set()
-            with open(file_path, "r", encoding="utf-8-sig") as f_in:
+            with open(file_path, "r", encoding=_facturas_csv_encoding(file_path)) as f_in:
                 sample = f_in.read(2048)
                 f_in.seek(0)
                 try:
-                    dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
+                    dialect = common_csv_reader.detect_csv_dialect(sample, delimiters=",;\t")
                 except Exception:
                     dialect = "excel"
                 reader = csv.DictReader(f_in, dialect=dialect)
@@ -811,10 +834,10 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
             else:
                 rfc_exception_updated = set()
 
-            with open(file_path, "r", encoding="utf-8-sig") as f_in, open(error_path, "w", encoding="utf-8") as f_err:
+            with open(file_path, "r", encoding=_facturas_csv_encoding(file_path)) as f_in, open(error_path, "w", encoding="utf-8") as f_err:
                 f_in.seek(0)
                 try:
-                    dialect = csv.Sniffer().sniff(f_in.read(2048), delimiters=",;\t")
+                    dialect = common_csv_reader.detect_csv_dialect(f_in.read(2048), delimiters=",;\t")
                 except Exception:
                     dialect = "excel"
                 f_in.seek(0)
@@ -992,11 +1015,11 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
 
                 csv_series_count_so_far: Dict[Tuple[str, str], int] = {}
 
-                with open(file_path, "r", encoding="utf-8-sig") as f_in, open(error_path, "w", encoding="utf-8") as f_err:
+                with open(file_path, "r", encoding=_facturas_csv_encoding(file_path)) as f_in, open(error_path, "w", encoding="utf-8") as f_err:
                     sample = f_in.read(2048)
                     f_in.seek(0)
                     try:
-                        dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
+                        dialect = common_csv_reader.detect_csv_dialect(sample, delimiters=",;\t")
                     except Exception:
                         dialect = "excel"
                     reader = csv.DictReader(f_in, dialect=dialect)
@@ -1146,11 +1169,11 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
                                 "number_id": nid or "",
                             })
 
-            with open(file_path, "r", encoding="utf-8-sig") as f_in, open(error_path, "w", encoding="utf-8") as f_err:
+            with open(file_path, "r", encoding=_facturas_csv_encoding(file_path)) as f_in, open(error_path, "w", encoding="utf-8") as f_err:
                 sample = f_in.read(2048)
                 f_in.seek(0)
                 try:
-                    dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
+                    dialect = common_csv_reader.detect_csv_dialect(sample, delimiters=",;\t")
                 except Exception:
                     dialect = "excel"
                 reader = csv.DictReader(f_in, dialect=dialect)
@@ -1382,11 +1405,11 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
             rfc_exception_updated: Set[str] = set()
             rfc_exception_num_parte: Set[str] = set()
 
-            with open(file_path, "r", encoding="utf-8-sig") as f_in:
+            with open(file_path, "r", encoding=_facturas_csv_encoding(file_path)) as f_in:
                 sample = f_in.read(2048)
                 f_in.seek(0)
                 try:
-                    dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
+                    dialect = common_csv_reader.detect_csv_dialect(sample, delimiters=",;\t")
                 except Exception:
                     dialect = "excel"
                 reader = csv.DictReader(f_in, dialect=dialect)
@@ -1766,11 +1789,11 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
             rfc_exception_updated: Set[str] = set()
             rfc_exception_num_parte: Set[str] = set()
 
-            with open(file_path, "r", encoding="utf-8-sig") as f_in:
+            with open(file_path, "r", encoding=_facturas_csv_encoding(file_path)) as f_in:
                 sample = f_in.read(2048)
                 f_in.seek(0)
                 try:
-                    dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
+                    dialect = common_csv_reader.detect_csv_dialect(sample, delimiters=",;\t")
                 except Exception:
                     dialect = "excel"
                 reader = csv.DictReader(f_in, dialect=dialect)
@@ -2063,11 +2086,11 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
                         valid_part_numbers.add((row[0] or "").strip().upper())
 
             invoice_numbers_from_csv = set()
-            with open(file_path, "r", encoding="utf-8-sig") as f_in:
+            with open(file_path, "r", encoding=_facturas_csv_encoding(file_path)) as f_in:
                 sample = f_in.read(2048)
                 f_in.seek(0)
                 try:
-                    dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
+                    dialect = common_csv_reader.detect_csv_dialect(sample, delimiters=",;\t")
                 except Exception:
                     dialect = "excel"
                 reader = csv.DictReader(f_in, dialect=dialect)
@@ -2344,11 +2367,11 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
             rfc_exception_updated: Set[str] = set()
             rfc_exception_num_parte: Set[str] = set()
 
-            with open(file_path, "r", encoding="utf-8-sig") as f_in:
+            with open(file_path, "r", encoding=_facturas_csv_encoding(file_path)) as f_in:
                 sample = f_in.read(2048)
                 f_in.seek(0)
                 try:
-                    dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
+                    dialect = common_csv_reader.detect_csv_dialect(sample, delimiters=",;\t")
                 except Exception:
                     dialect = "excel"
                 reader = csv.DictReader(f_in, dialect=dialect)
@@ -2699,11 +2722,11 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
                         else:
                             existing_tipo_moneda_by_number[str(num).strip()] = cur_str.upper()[:2]
 
-            with open(file_path, "r", encoding="utf-8-sig") as f_in:
+            with open(file_path, "r", encoding=_facturas_csv_encoding(file_path)) as f_in:
                 sample = f_in.read(2048)
                 f_in.seek(0)
                 try:
-                    dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
+                    dialect = common_csv_reader.detect_csv_dialect(sample, delimiters=",;\t")
                 except Exception:
                     dialect = "excel"
                 reader = csv.DictReader(f_in, dialect=dialect)
@@ -3184,11 +3207,11 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
                         else:
                             existing_tipo_moneda_by_number[str(num).strip()] = (cur_str or "").upper()[:2]
 
-            with open(file_path, "r", encoding="utf-8-sig") as f_in:
+            with open(file_path, "r", encoding=_facturas_csv_encoding(file_path)) as f_in:
                 sample = f_in.read(2048)
                 f_in.seek(0)
                 try:
-                    dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
+                    dialect = common_csv_reader.detect_csv_dialect(sample, delimiters=",;\t")
                 except Exception:
                     dialect = "excel"
                 reader = csv.DictReader(f_in, dialect=dialect)
@@ -3560,11 +3583,11 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
 
             date_format = _fc.get("dateFormat") or meta.get("date_format")
 
-            with open(file_path, "r", encoding="utf-8-sig") as f_in:
+            with open(file_path, "r", encoding=_facturas_csv_encoding(file_path)) as f_in:
                 sample = f_in.read(2048)
                 f_in.seek(0)
                 try:
-                    dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
+                    dialect = common_csv_reader.detect_csv_dialect(sample, delimiters=",;\t")
                 except Exception:
                     dialect = "excel"
                 reader = csv.DictReader(f_in, dialect=dialect)
@@ -3795,11 +3818,11 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
                         else:
                             existing_tipo_moneda_by_number[str(num).strip()] = (cur_str or "").upper()[:2]
 
-            with open(file_path, "r", encoding="utf-8-sig") as f_in:
+            with open(file_path, "r", encoding=_facturas_csv_encoding(file_path)) as f_in:
                 sample = f_in.read(2048)
                 f_in.seek(0)
                 try:
-                    dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
+                    dialect = common_csv_reader.detect_csv_dialect(sample, delimiters=",;\t")
                 except Exception:
                     dialect = "excel"
                 reader = csv.DictReader(f_in, dialect=dialect)
@@ -3911,7 +3934,7 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
         }
 
         with CoreSessionLocal() as session, \
-             open(file_path, 'r', encoding='utf-8-sig') as f_in, \
+             open(file_path, 'r', encoding=_facturas_csv_encoding(file_path)) as f_in, \
              open(error_path, 'w', encoding='utf-8') as f_err:
             validator = ForeignKeyValidator(session, tenant_id, company_id)
 
@@ -3955,7 +3978,7 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
             sample = f_in.read(2048)
             f_in.seek(0)
             try:
-                dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
+                dialect = common_csv_reader.detect_csv_dialect(sample, delimiters=",;\t")
             except:
                 dialect = 'excel'
                 
@@ -4666,6 +4689,7 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
         file_path = alt_path
     else:
         common_storage.ensure_meta_from_redis(effective_job_type, job_id, file_path, log_prefix)
+    _ensure_utf8_compatible_import_file(file_path)
 
     try:
         tenant_id, company_id = common_meta.require_tenant_context(file_path)
@@ -4734,11 +4758,11 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
                 skipped_invalid = 0
                 skipped_details: List[Dict[str, Any]] = []
 
-                with open(file_path, "r", encoding="utf-8-sig") as f:
+                with open(file_path, "r", encoding=_facturas_csv_encoding(file_path)) as f:
                     sample = f.read(2048)
                     f.seek(0)
                     try:
-                        dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
+                        dialect = common_csv_reader.detect_csv_dialect(sample, delimiters=",;\t")
                     except Exception:
                         dialect = "excel"
                     reader = csv.DictReader(f, dialect=dialect)
@@ -5009,11 +5033,11 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
                 skipped_invalid = 0
                 skipped_details: List[Dict[str, Any]] = []
 
-                with open(file_path, "r", encoding="utf-8-sig") as f:
+                with open(file_path, "r", encoding=_facturas_csv_encoding(file_path)) as f:
                     sample = f.read(2048)
                     f.seek(0)
                     try:
-                        dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
+                        dialect = common_csv_reader.detect_csv_dialect(sample, delimiters=",;\t")
                     except Exception:
                         dialect = "excel"
                     reader = csv.DictReader(f, dialect=dialect)
@@ -5280,11 +5304,11 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
                 skipped_invalid = 0
                 skipped_details: List[Dict[str, Any]] = []
 
-                with open(file_path, "r", encoding="utf-8-sig") as f:
+                with open(file_path, "r", encoding=_facturas_csv_encoding(file_path)) as f:
                     sample = f.read(2048)
                     f.seek(0)
                     try:
-                        dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
+                        dialect = common_csv_reader.detect_csv_dialect(sample, delimiters=",;\t")
                     except Exception:
                         dialect = "excel"
                     reader = csv.DictReader(f, dialect=dialect)
@@ -5692,12 +5716,12 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
                 except Exception:
                     pass
 
-            with open(file_path, 'r', encoding='utf-8-sig') as f:
+            with open(file_path, 'r', encoding=_facturas_csv_encoding(file_path)) as f:
                 # Detect Delimiter
                 sample = f.read(2048)
                 f.seek(0)
                 try:
-                    dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
+                    dialect = common_csv_reader.detect_csv_dialect(sample, delimiters=",;\t")
                 except:
                     dialect = 'excel'
                 

backend/api/v1/modules/a76/layouts_csv/parts/tasks.py

@@ -37,6 +37,12 @@ PART_IMPORT_ERROR_LINES_PREFIX = "part_import_error_lines:"
 PART_IMPORT_REDIS_TTL = common_storage.IMPORT_REDIS_TTL
 
 
+def _read_plan_for_parts(fieldnames):
+    if fieldnames:
+        return common_csv.CsvReadPlan(header_mode="headerless", fieldnames=fieldnames)
+    return common_csv.CsvReadPlan(header_mode="header")
+
+
 @celery_app.task(bind=True)
 def scan_file(self, job_id: str, config: str = None):
     logger.info("Parts import: starting scan for job %s", job_id)
@@ -49,8 +55,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)
+    read_plan = _read_plan_for_parts(fieldnames)
     try:
-        total_rows = common_csv.count_csv_rows(file_path, has_header=has_header)
+        total_rows = common_csv.count_csv_rows(file_path, has_header=has_header, read_plan=read_plan)
     except Exception as e:
         return {"status": "failed", "error": str(e)}
 
@@ -93,7 +100,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, fieldnames=fieldnames):
+            for i, row in common_csv.iter_csv_rows_with_plan(file_path, read_plan=read_plan):
                 self.update_state(
                         state="PROGRESS",
                         meta={"current": i, "total": total_rows, "errors": error_count},
@@ -204,6 +211,7 @@ def insert_valid_rows(self, job_id: str):
     meta_path = common_meta.get_meta_path(file_path)
 
     fieldnames, _ = detect_headers_or_data(file_path, common_normalize.normalize_header)
+    read_plan = _read_plan_for_parts(fieldnames)
 
     try:
         with CoreSessionLocal() as session:
@@ -216,7 +224,7 @@ def insert_valid_rows(self, job_id: str):
                 if key:
                     existing_by_part_number[key] = p
 
-            for i, row in common_csv.iter_csv_rows(file_path, fieldnames=fieldnames):
+            for i, row in common_csv.iter_csv_rows_with_plan(file_path, read_plan=read_plan):
                 if i in error_lines:
                     continue
 

backend/api/v1/modules/a76/layouts_csv/parts/template_config.py

@@ -7,6 +7,7 @@ import io
 from typing import Dict, List, Any, Optional, Tuple
 
 from ..common.cell_value import cell_to_str
+from ..common import csv_reader as common_csv_reader
 
 
 # Valores que indican que la primera fila es cabecera (primera columna normalizada)
@@ -24,17 +25,24 @@ def detect_headers_or_data(
     - 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)
+        sample, _ = common_csv_reader.read_text_sample(
+            file_path,
+            requested_encoding=encoding,
+            sample_chars=2048,
+        )
     except Exception:
-        return None, True
+        try:
+            sample, _ = common_csv_reader.read_text_sample(
+                file_path,
+                requested_encoding="auto",
+                sample_chars=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
+    dialect = common_csv_reader.detect_csv_dialect(sample, delimiters=",;\t")
     reader = csv.reader(io.StringIO(lines[0]), dialect=dialect)
     first_row = next(reader, None)
     if not first_row:

backend/api/v1/modules/a76/layouts_csv/pedmientos/tasks.py

@@ -41,6 +41,12 @@ PED_IMPORT_ERROR_LINES_PREFIX = "ped_import_error_lines:"
 PED_IMPORT_REDIS_TTL = common_storage.IMPORT_REDIS_TTL
 
 
+def _read_plan_for_pedimentos(fieldnames):
+    if fieldnames:
+        return common_csv_reader.CsvReadPlan(header_mode="headerless", fieldnames=fieldnames)
+    return common_csv_reader.CsvReadPlan(header_mode="header")
+
+
 def _norm_row(row: Dict[str, Any]) -> Dict[str, Any]:
     return row_from_template(row, common_normalize.normalize_header, TEMPLATE_ID)
 
@@ -60,8 +66,9 @@ def _do_scan(job_id: str, progress_callback: Optional[Any] = None) -> Dict[str,
         common_normalize.normalize_header,
         parse_pedimento_col_a,
     )
+    read_plan = _read_plan_for_pedimentos(fieldnames)
     try:
-        total_rows = common_csv_reader.count_csv_rows(file_path, has_header=has_header)
+        total_rows = common_csv_reader.count_csv_rows(file_path, has_header=has_header, read_plan=read_plan)
     except Exception as e:
         return {"status": "failed", "error": str(e)}
 
@@ -98,7 +105,7 @@ def _do_scan(job_id: str, progress_callback: Optional[Any] = None) -> Dict[str,
 
     try:
         with open(error_path, "w", encoding="utf-8") as f_err:
-            for i, row in common_csv_reader.iter_csv_rows(file_path, fieldnames=fieldnames):
+            for i, row in common_csv_reader.iter_csv_rows_with_plan(file_path, read_plan=read_plan):
                 if progress_callback:
                     progress_callback(i, total_rows, error_count)
 
@@ -217,6 +224,7 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
         common_normalize.normalize_header,
         parse_pedimento_col_a,
     )
+    read_plan_commit = _read_plan_for_pedimentos(fieldnames_commit)
 
     def _key_from_row(r: Dict[str, Any]) -> Optional[str]:
         if is_clarion_layout(r):
@@ -240,7 +248,7 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
 
     try:
         with CoreSessionLocal() as session:
-            for i, row in common_csv_reader.iter_csv_rows(file_path, fieldnames=fieldnames_commit):
+            for i, row in common_csv_reader.iter_csv_rows_with_plan(file_path, read_plan=read_plan_commit):
                 if i in error_lines:
                     continue
 

backend/api/v1/modules/a76/layouts_csv/pedmientos/template_config.py

@@ -10,6 +10,7 @@ from typing import Dict, List, Any, Optional, Tuple
 # Convierte valor de celda a str; si es lista (p. ej. CSV con columnas duplicadas), toma el primer elemento.
 # Re-exportado desde common para uso en validators; ver layouts_csv.common.cell_value.
 from ..common.cell_value import cell_to_str as _cell_to_str
+from ..common import csv_reader as common_csv_reader
 
 
 # Longitudes para validación (sin afectar modelos)
@@ -138,17 +139,24 @@ def detect_headers_or_data(
     - Si no -> has_header=True. Devuelve (fieldnames, has_header).
     """
     try:
-        with open(file_path, "r", encoding=encoding) as f:
-            sample = f.read(2048)
+        sample, _ = common_csv_reader.read_text_sample(
+            file_path,
+            requested_encoding=encoding,
+            sample_chars=2048,
+        )
     except Exception:
-        return None, True
+        try:
+            sample, _ = common_csv_reader.read_text_sample(
+                file_path,
+                requested_encoding="auto",
+                sample_chars=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
+    dialect = common_csv_reader.detect_csv_dialect(sample, delimiters=",;\t")
     reader = csv.reader(io.StringIO(lines[0]), dialect=dialect)
     first_row = next(reader, None)
     if not first_row:

backend/api/v1/modules/a76/layouts_csv/trailers/tasks.py

@@ -3,7 +3,6 @@ Tareas Celery para importación CSV de Trailers y Cajas.
 Flujo: scan_file (validación) → insert_valid_rows (commit).
 Usa layouts_csv.common (storage, normalize, meta, responses); CSV con headers duplicados (dedupe).
 """
-import csv
 import json
 import logging
 import os
@@ -16,6 +15,7 @@ from ..common import storage as common_storage
 from ..common import normalize as common_normalize
 from ..common import meta as common_meta
 from ..common import responses as common_responses
+from ..common import csv_reader as common_csv_reader
 from .template_config import row_from_template
 from .validators import validate_row_trailer, validate_row_trailer_desfase
 from .common.mappers import row_to_trailer_data, row_to_trailer_data_for_update
@@ -39,17 +39,6 @@ def _get_redis():
     return redis.Redis.from_url(url, decode_responses=False)
 
 
-def _dedupe_headers(headers: List[str]) -> List[str]:
-    counts: Dict[str, int] = {}
-    unique: List[str] = []
-    for header in headers:
-        name = str(header or "").strip() or "COL"
-        count = counts.get(name, 0) + 1
-        counts[name] = count
-        unique.append(name if count == 1 else f"{name} {count}")
-    return unique
-
-
 def _do_scan(job_id: str, progress_callback: Optional[Any] = None) -> Dict[str, Any]:
     file_path = common_storage.ensure_file_from_redis(JOB_TYPE, job_id, "Trailers import")
     if not file_path:
@@ -59,8 +48,7 @@ def _do_scan(job_id: str, progress_callback: Optional[Any] = None) -> Dict[str,
     error_path = common_storage.error_path_for_job(JOB_TYPE, job_id)
 
     try:
-        with open(file_path, "r", encoding="utf-8-sig") as f:
-            total_rows = sum(1 for _ in f) - 1
+        total_rows = common_csv_reader.count_csv_rows(file_path, has_header=True)
     except Exception as e:
         return {"status": "failed", "error": str(e)}
 
@@ -103,24 +91,8 @@ def _do_scan(job_id: str, progress_callback: Optional[Any] = None) -> Dict[str,
     error_lines_list: List[int] = []
 
     try:
-        with open(file_path, "r", encoding="utf-8-sig") as f_in, open(
-            error_path, "w", encoding="utf-8"
-        ) as f_err:
-            sample = f_in.read(2048)
-            f_in.seek(0)
-            try:
-                dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
-            except Exception:
-                dialect = "excel"
-            reader = csv.reader(f_in, dialect=dialect)
-            try:
-                headers = next(reader)
-            except StopIteration:
-                headers = []
-            headers = _dedupe_headers(headers)
-            dict_reader = csv.DictReader(f_in, fieldnames=headers, dialect=dialect)
-
-            for i, row in enumerate(dict_reader, start=1):
+        with open(error_path, "w", encoding="utf-8") as f_err:
+            for i, row in common_csv_reader.iter_csv_rows_deduped_headers(file_path):
                 if progress_callback:
                     progress_callback(i, total_rows, error_count)
 
@@ -270,22 +242,7 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
 
     try:
         with CoreSessionLocal() as session:
-            with open(file_path, "r", encoding="utf-8-sig") as f:
-                sample = f.read(2048)
-                f.seek(0)
-                try:
-                    dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
-                except Exception:
-                    dialect = "excel"
-                reader = csv.reader(f, dialect=dialect)
-                try:
-                    headers = next(reader)
-                except StopIteration:
-                    headers = []
-                headers = _dedupe_headers(headers)
-                dict_reader = csv.DictReader(f, fieldnames=headers, dialect=dialect)
-
-                for i, row in enumerate(dict_reader, start=1):
+            for i, row in common_csv_reader.iter_csv_rows_deduped_headers(file_path):
                     if i in error_lines:
                         continue