"""
Lectura de CSV con detección de delimitador (compartida por layouts_csv).
Si se pasa fieldnames, no se usa la primera fila como cabecera y se toma como dato (CSV sin cabeceras).
Si se pasa headerless_first_cell_values, se detecta si la primera fila es cabecera o dato por el valor de la primera celda.
"""
import csv
import io
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]:
    """Sustituye cabeceras vacías por _COL_0_, _COL_1_, ... para que DictReader no colapse columnas."""
    result: List[str] = []
    empty_idx = 0
    for h in headers:
        if (h or "").strip() == "":
            result.append(f"_COL_{empty_idx}_")
            empty_idx += 1
        else:
            result.append(h)
    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")


@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,
) -> str:
    """
    Detecta encoding por prueba de decodificación en un sample en binario.
    Nota: latin-1 decodifica cualquier byte; por eso debe ir al final.
    """
    with open(file_path, "rb") as f:
        raw = f.read(sample_bytes)
    last_err: Optional[Exception] = None
    for enc in encodings:
        try:
            if _sample_decodes_with_encoding(raw, enc):
                return enc
        except Exception as e:
            last_err = e
    if last_err:
        raise last_err
    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,
    headerless_first_cell_values: Optional[Set[str]] = None,
    headerless_second_cell_key_pattern: Optional[str] = None,
) -> Iterator[Tuple[int, Dict[str, Any]]]:
    """
    Abre el CSV, detecta dialecto y devuelve (line_num, row_dict) por cada fila.
    line_num empieza en 1 (primera fila de datos).
    Si fieldnames y headerless_first_cell_values se pasan: si la primera celda de la primera fila
    (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).
    """
    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,
    read_plan: Optional[CsvReadPlan] = None,
) -> int:
    """Cuenta filas del CSV. Si has_header=True (por defecto), no cuenta la cabecera."""
    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 effective_has_header else max(0, total_lines - 1)
