src/io_utils.py

import os
import re

import pandas as pd


def _clean_columns(df: pd.DataFrame) -> pd.DataFrame:
    new_cols = []
    for c in df.columns:
        c = str(c).replace("\ufeff", "").strip()
        c = re.sub(r"\s+", " ", c)
        c = c.lower().replace(" ", "_")
        new_cols.append(c)
    df.columns = new_cols
    return df


def read_any(path: str, verbose: bool = True) -> pd.DataFrame:
    if not os.path.exists(path):
        raise FileNotFoundError(f"File not found: {path}")

    ext = os.path.splitext(path)[1].lower()

    if ext == ".csv":
        encodings_to_try = [
            "utf-8-sig",
            "utf-8",
            "cp1252",
            "latin1",
            "utf-16",  # requires BOM
            "utf-16-le",  # handles UTF-16LE without BOM
            "utf-16-be",  # handles UTF-16BE without BOM
        ]
        last_err = None

        for enc in encodings_to_try:
            try:
                df = pd.read_csv(path, encoding=enc)
                if verbose:
                    print(
                        f"✅ Loaded CSV: {os.path.basename(path)} | encoding={enc} | shape={df.shape}"
                    )
                return _clean_columns(df)

            except (UnicodeDecodeError, UnicodeError) as e:
                last_err = e
                continue

            except pd.errors.ParserError:
                try:
                    df = pd.read_csv(path, encoding=enc, engine="python")
                    if verbose:
                        print(
                            f"✅ Loaded CSV (python engine): {os.path.basename(path)} | encoding={enc} | shape={df.shape}"
                        )
                    return _clean_columns(df)
                except (UnicodeDecodeError, UnicodeError, Exception) as e:
                    last_err = e
                    continue

        raise RuntimeError(
            f"Could not decode CSV using encodings {encodings_to_try}. Last error: {last_err}"
        )

    if ext in [".xlsx", ".xls"]:
        df = pd.read_excel(path)
        if verbose:
            print(f"✅ Loaded Excel: {os.path.basename(path)} | shape={df.shape}")
        return _clean_columns(df)

    raise ValueError(f"Unsupported file type: {ext}. Use CSV or XLSX.")