diff --git a/backend/api/v1/modules/a76/layouts_csv/classes/tasks.py b/backend/api/v1/modules/a76/layouts_csv/classes/tasks.py index 1d1abb43..c8146af0 100644 --- a/backend/api/v1/modules/a76/layouts_csv/classes/tasks.py +++ b/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 diff --git a/backend/api/v1/modules/a76/layouts_csv/classes/template_config.py b/backend/api/v1/modules/a76/layouts_csv/classes/template_config.py index cc6009bd..bd10a176 100644 --- a/backend/api/v1/modules/a76/layouts_csv/classes/template_config.py +++ b/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 diff --git a/backend/api/v1/modules/a76/layouts_csv/common/csv_ingestion_rollout.md b/backend/api/v1/modules/a76/layouts_csv/common/csv_ingestion_rollout.md new file mode 100644 index 00000000..c8974b12 --- /dev/null +++ b/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. diff --git a/backend/api/v1/modules/a76/layouts_csv/common/csv_reader.py b/backend/api/v1/modules/a76/layouts_csv/common/csv_reader.py index 33060545..f24a7ba1 100644 --- a/backend/api/v1/modules/a76/layouts_csv/common/csv_reader.py +++ b/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) diff --git a/backend/api/v1/modules/a76/layouts_csv/drivers/tasks.py b/backend/api/v1/modules/a76/layouts_csv/drivers/tasks.py index c03e88a1..d19405dd 100644 --- a/backend/api/v1/modules/a76/layouts_csv/drivers/tasks.py +++ b/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 diff --git a/backend/api/v1/modules/a76/layouts_csv/facturas/tasks.py b/backend/api/v1/modules/a76/layouts_csv/facturas/tasks.py index c364981e..143bec42 100644 --- a/backend/api/v1/modules/a76/layouts_csv/facturas/tasks.py +++ b/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 # Models are imported inside tasks to avoid circular dependencies and mapper initialization issues in the API process @@ -63,6 +64,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 @@ -461,6 +483,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) @@ -471,7 +494,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}"} @@ -617,11 +640,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) @@ -792,11 +815,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) @@ -810,10 +833,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) @@ -991,11 +1014,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) @@ -1145,11 +1168,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) @@ -1375,11 +1398,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) @@ -1753,11 +1776,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) @@ -2044,11 +2067,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) @@ -2319,11 +2342,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) @@ -2674,11 +2697,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) @@ -3159,11 +3182,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) @@ -3535,11 +3558,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) @@ -3770,11 +3793,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) @@ -3886,7 +3909,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) @@ -3930,7 +3953,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' @@ -4641,6 +4664,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) @@ -4709,11 +4733,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) @@ -4978,11 +5002,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) @@ -5243,11 +5267,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) @@ -5655,12 +5679,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' diff --git a/backend/api/v1/modules/a76/layouts_csv/parts/tasks.py b/backend/api/v1/modules/a76/layouts_csv/parts/tasks.py index 6d2fe8c2..6c4cf830 100644 --- a/backend/api/v1/modules/a76/layouts_csv/parts/tasks.py +++ b/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 diff --git a/backend/api/v1/modules/a76/layouts_csv/parts/template_config.py b/backend/api/v1/modules/a76/layouts_csv/parts/template_config.py index 2993f180..55910691 100644 --- a/backend/api/v1/modules/a76/layouts_csv/parts/template_config.py +++ b/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: diff --git a/backend/api/v1/modules/a76/layouts_csv/pedmientos/tasks.py b/backend/api/v1/modules/a76/layouts_csv/pedmientos/tasks.py index 5bd482a9..91112275 100644 --- a/backend/api/v1/modules/a76/layouts_csv/pedmientos/tasks.py +++ b/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 diff --git a/backend/api/v1/modules/a76/layouts_csv/pedmientos/template_config.py b/backend/api/v1/modules/a76/layouts_csv/pedmientos/template_config.py index b69bd5c2..b855d390 100644 --- a/backend/api/v1/modules/a76/layouts_csv/pedmientos/template_config.py +++ b/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: diff --git a/backend/api/v1/modules/a76/layouts_csv/trailers/tasks.py b/backend/api/v1/modules/a76/layouts_csv/trailers/tasks.py index 78a3ec24..f37f605f 100644 --- a/backend/api/v1/modules/a76/layouts_csv/trailers/tasks.py +++ b/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 diff --git a/backend/api/v1/modules/public/reference_data/carta_porte_codes/seed.py b/backend/api/v1/modules/public/reference_data/carta_porte_codes/seed.py index 7042748f..580e5425 100644 --- a/backend/api/v1/modules/public/reference_data/carta_porte_codes/seed.py +++ b/backend/api/v1/modules/public/reference_data/carta_porte_codes/seed.py @@ -1,6 +1,7 @@ import csv import os from sqlalchemy.orm import Session +from api.v1.modules.a76.layouts_csv.common import csv_reader as common_csv_reader from .models import CartaPorte def seed_carta_porte(db: Session): @@ -16,7 +17,8 @@ def seed_carta_porte(db: Session): print("Seeding Carta Porte catalog (this might take a while)...") - with open(csv_path, mode='r', encoding='utf-8-sig') as f: + encoding = common_csv_reader.detect_text_encoding(csv_path) + with open(csv_path, mode='r', encoding=encoding) as f: # User provided comma-separated data reader = csv.DictReader(f) diff --git a/backend/tests/unit/test_csv_reader_encoding.py b/backend/tests/unit/test_csv_reader_encoding.py new file mode 100644 index 00000000..b2f7487b --- /dev/null +++ b/backend/tests/unit/test_csv_reader_encoding.py @@ -0,0 +1,150 @@ +from pathlib import Path + +from api.v1.modules.a76.layouts_csv.common.csv_reader import ( + CsvReadPlan, + detect_text_encoding, + inspect_csv, + iter_csv_rows, + iter_csv_rows_with_plan, +) +from api.v1.modules.a76.layouts_csv.classes.template_config import ( + detect_headers_or_data as detect_classes_headers, + row_from_template as row_from_classes_template, +) +from api.v1.modules.a76.layouts_csv.parts.template_config import detect_headers_or_data as detect_parts_headers +from api.v1.modules.a76.layouts_csv.pedmientos.template_config import ( + detect_headers_or_data as detect_pedimentos_headers, + parse_pedimento_col_a, +) + + +def _write_bytes(tmp_path: Path, name: str, payload: bytes) -> Path: + file_path = tmp_path / name + file_path.write_bytes(payload) + return file_path + + +def test_detect_text_encoding_handles_truncated_utf8_sample(tmp_path: Path): + # Regression: "ó" in "Descripción" empieza en byte 21 (0xC3 0xB3). + # sample_bytes=22 lee bytes 0-21, terminando en 0xC3 (primer byte de ó, secuencia incompleta). + # El código viejo: raw.decode("utf-8-sig") fallaba → caía a cp1252 → mojibake. + # El código nuevo: decoder incremental tolera el corte → retorna utf-8-sig. + payload = "DESCRIPCION\nDescripción español\n".encode("utf-8") + file_path = _write_bytes(tmp_path, "truncated_utf8.csv", payload) + + enc = detect_text_encoding(str(file_path), sample_bytes=22) + + assert enc in ("utf-8", "utf-8-sig"), ( + f"Got {enc!r} — el archivo UTF-8 con corte de muestra a mitad de multibyte " + "fue detectado como cp1252, produciendo mojibake (español / Descripción)" + ) + + +def test_utf8_enie_at_sample_boundary_not_detected_as_cp1252(tmp_path: Path): + # Regresión directa del bug mojibake reportado en producción. + # Construye un payload donde 'ñ' (0xC3 0xB1 en UTF-8) cae exactamente en el byte 19, + # y sample_bytes=20 lee sólo 0xC3 (primer byte) — secuencia incompleta. + # Resultado esperado: utf-8 / utf-8-sig (no cp1252). + header = b"CLASE,DESC\n" # 11 bytes + row = "C01,español\n".encode("utf-8") # ñ en bytes 19-20 del payload total + payload = header + row + file_path = _write_bytes(tmp_path, "regression_mojibake.csv", payload) + + enc = detect_text_encoding(str(file_path), sample_bytes=20) + + assert enc in ("utf-8", "utf-8-sig"), ( + f"Got {enc!r} en lugar de utf-8 — leer como cp1252 produciría " + "'español' en lugar de 'español'" + ) + + +def test_iter_csv_rows_preserves_utf8_values(tmp_path: Path): + payload = "CLASE,DESCRIPCION ESPAÑOL\nCLASE01,Clase prueba español\n".encode("utf-8") + file_path = _write_bytes(tmp_path, "utf8_values.csv", payload) + + rows = list(iter_csv_rows(str(file_path))) + + assert len(rows) == 1 + _, row = rows[0] + assert row["DESCRIPCION ESPAÑOL"] == "Clase prueba español" + + +def test_iter_csv_rows_keeps_cp1252_compatibility(tmp_path: Path): + payload = "CLASE,DESCRIPCION ESPAÑOL\nCLASE01,Descripción\n".encode("cp1252") + file_path = _write_bytes(tmp_path, "cp1252_values.csv", payload) + + rows = list(iter_csv_rows(str(file_path))) + + assert len(rows) == 1 + _, row = rows[0] + assert row["DESCRIPCION ESPAÑOL"] == "Descripción" + + +def test_parts_detect_headers_or_data_handles_cp1252(tmp_path: Path): + payload = "NUMERO DE PARTE,DESCRIPCION EN ESPAÑOL\nP-01,Descripción\n".encode("cp1252") + file_path = _write_bytes(tmp_path, "parts_cp1252.csv", payload) + + fieldnames, has_header = detect_parts_headers(str(file_path), lambda s: (s or "").strip().upper()) + + assert has_header is True + assert fieldnames is None + + +def test_pedimentos_detect_headers_or_data_handles_cp1252_data_first_row(tmp_path: Path): + payload = "24,1234,1234567,I,A1\n".encode("cp1252") + file_path = _write_bytes(tmp_path, "pedimentos_cp1252_data.csv", payload) + + fieldnames, has_header = detect_pedimentos_headers( + str(file_path), + lambda s: (s or "").strip().upper(), + parse_pedimento_col_a, + ) + + assert has_header is False + assert fieldnames is not None + + +def test_classes_detect_headers_or_data_handles_cp1252(tmp_path: Path): + payload = "CLAVE CLASE;DESCRIPCION ESPAÑOL\nC01;Descripción\n".encode("cp1252") + file_path = _write_bytes(tmp_path, "classes_cp1252_semicolon.csv", payload) + + fieldnames, has_header = detect_classes_headers(str(file_path), lambda s: (s or "").strip().upper()) + + assert has_header is True + assert fieldnames is None + + +def test_classes_row_from_template_recovers_collapsed_header_with_semicolon(): + row = {"CLAVE CLASE,DESCRIPCION ESPAÑOL": "C01;Descripción;Description"} + mapped = row_from_classes_template(row, lambda s: (s or "").strip().upper()) + assert mapped["CLASE"] == "C01" + + +def test_iter_csv_rows_with_plan_headerless_and_semicolon(tmp_path: Path): + payload = "C01;Descripcion 1\nC02;Descripcion 2\n".encode("utf-8") + file_path = _write_bytes(tmp_path, "headerless_semicolon.csv", payload) + plan = CsvReadPlan( + header_mode="headerless", + fieldnames=["CLASE", "DESCRIPCIONE"], + ) + + rows = list(iter_csv_rows_with_plan(str(file_path), plan)) + + assert len(rows) == 2 + assert rows[0][1]["CLASE"] == "C01" + assert rows[1][1]["DESCRIPCIONE"] == "Descripcion 2" + + +def test_inspect_csv_auto_mode_switches_to_headerless(tmp_path: Path): + payload = "C01,Descripcion\n".encode("utf-8") + file_path = _write_bytes(tmp_path, "auto_mode.csv", payload) + plan = CsvReadPlan( + header_mode="auto", + fieldnames=["CLASE", "DESCRIPCIONE"], + headerless_first_cell_values={"C01"}, + ) + + metadata = inspect_csv(str(file_path), plan) + + assert metadata.has_header is False + assert metadata.fieldnames == ["CLASE", "DESCRIPCIONE"]