"""Carga robusta de CSV desde bytes (uploads).""" from __future__ import annotations import csv as _csv import os import tempfile from collections import Counter from typing import Optional import pandas as pd _CANDIDATE_DELIMITERS = ["|", ";", "\t", ",", ":"] def _detect_delimiter_from_sample(sample: str) -> str: if not sample: return "," try: dialect = _csv.Sniffer().sniff(sample, delimiters="".join(_CANDIDATE_DELIMITERS)) sniffed = dialect.delimiter except _csv.Error: sniffed = None lines = [ln for ln in sample.splitlines() if ln.strip()] if not lines: return sniffed or "," scores: dict[str, tuple[int, int]] = {} for delim in _CANDIDATE_DELIMITERS: counts = [ln.count(delim) for ln in lines[:50]] if not counts or max(counts) == 0: continue mode_count, mode_freq = Counter(counts).most_common(1)[0] if mode_count == 0: continue scores[delim] = (mode_count, mode_freq) if scores: best = max(scores.items(), key=lambda kv: (kv[1][0], kv[1][1])) if sniffed and sniffed in scores: sniffed_score = scores[sniffed] if sniffed_score[0] >= best[1][0] - 1: return sniffed return best[0] return sniffed or "," def load_csv_bytes(content: bytes, filename: str = "upload.csv") -> tuple[pd.DataFrame, str]: """Carga un CSV desde bytes. Devuelve `(df, delimiter)`.""" if not content: raise RuntimeError(f"El archivo {filename} está vacío") sample = "" for enc in ("utf-8-sig", "utf-8", "latin-1", "cp1252"): try: sample = content[:65536].decode(enc, errors="replace") break except Exception: continue delimiter = _detect_delimiter_from_sample(sample) suffix = os.path.splitext(filename)[1] or ".csv" with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp: tmp.write(content) tmp_path = tmp.name last_error: Optional[Exception] = None try: for enc in ("utf-8-sig", "utf-8", "latin-1", "cp1252"): try: df = pd.read_csv( tmp_path, delimiter=delimiter, dtype=str, encoding=enc, engine="python", on_bad_lines="skip", keep_default_na=False, ) df.columns = [str(c).strip() for c in df.columns] return df, delimiter except Exception as exc: # noqa: BLE001 last_error = exc raise RuntimeError(f"No se pudo leer el CSV {filename}: {last_error}") finally: try: os.unlink(tmp_path) except OSError: pass