app/utils/file_handler.py

import io
import pandas as pd


def parse_csv(file_bytes: bytes) -> pd.DataFrame:
    """Parse CSV bytes into a cleaned DataFrame.

    Raises:
        ValueError: If the file is empty, exceeds 200 MB, or results in zero rows.
    """
    if len(file_bytes) == 0:
        raise ValueError("Uploaded file is empty")

    if len(file_bytes) > 200 * 1024 * 1024:
        raise ValueError("File exceeds 200 MB limit")

    # Read the raw header line to get original column names before pandas
    # auto-renames duplicates (e.g. col -> col.1, col.2).
    encoding = _detect_encoding(file_bytes)
    raw_header = _read_raw_header(file_bytes, encoding)
    deduped_names = _deduplicate_columns(raw_header)

    df = pd.read_csv(
        io.BytesIO(file_bytes),
        encoding=encoding,
        header=None,
        names=deduped_names,
        skiprows=1,
    )

    if len(df) == 0:
        raise ValueError("Uploaded file is empty")

    return df


def _detect_encoding(file_bytes: bytes) -> str:
    """Return 'utf-8' if the bytes decode cleanly, otherwise 'latin-1'."""
    try:
        file_bytes.decode("utf-8")
        return "utf-8"
    except UnicodeDecodeError:
        return "latin-1"


def _read_raw_header(file_bytes: bytes, encoding: str) -> list[str]:
    """Read the first line of the CSV and return column names as strings."""
    first_line = file_bytes.split(b"\n")[0].decode(encoding).rstrip("\r")
    # Use pandas to parse the header line so quoting/escaping is handled correctly
    header_df = pd.read_csv(io.StringIO(first_line), header=None)
    return [str(c) for c in header_df.iloc[0].tolist()]


def _deduplicate_columns(columns: list[str]) -> list[str]:
    """Deduplicate column names by appending _1, _2, etc. to duplicates.

    The first occurrence of a name keeps the original; subsequent occurrences
    get numeric suffixes.
    """
    seen: dict[str, int] = {}
    result: list[str] = []

    for col in columns:
        if col not in seen:
            seen[col] = 0
            result.append(col)
        else:
            seen[col] += 1
            new_name = f"{col}_{seen[col]}"
            # Ensure the generated name is also unique
            while new_name in seen:
                seen[col] += 1
                new_name = f"{col}_{seen[col]}"
            seen[new_name] = 0
            result.append(new_name)

    return result