src/csv_loader.py

import os
import sqlite3
from typing import Tuple, List, Optional, Union, Any

import chardet
import pandas as pd

# Prevent pandas from using pyarrow's string backend.
# The pyarrow DLL from user site-packages is incompatible with this conda env
# and causes an access violation (segfault) in Jupyter's multi-threaded kernel.
pd.options.mode.string_storage = "python"

try:
    from prompt_builder import normalize_text
except ImportError:
    from src.prompt_builder import normalize_text

def detect_encoding(file_input: Union[str, Any]) -> Optional[str]:
    """Detect the encoding of a CSV file or file-like object using chardet."""
    try:
        if isinstance(file_input, str):
            with open(file_input, 'rb') as f:
                raw_data = f.read(100000)
        else:
            # It's a file-like object (e.g., Streamlit UploadedFile)
            pos = file_input.tell()
            raw_data = file_input.read(100000)
            file_input.seek(pos)
        return chardet.detect(raw_data)['encoding']
    except Exception:
        return None

def load_csv_to_sqlite(file_input: Union[str, Any]) -> Tuple[sqlite3.Connection, str, List[str], List[str]]:
    """
    Loads a CSV file into an in-memory SQLite database.
    Normalizes column names and maps pandas dtypes to WikiSQL types (real/text).
    
    Args:
        file_input (str or file-like): Path to the CSV file, or a file-like object
        
    Returns:
        tuple: (conn, table_name, columns, types)
    """
    # 1. Try to load the CSV with detected encoding, fallback to others
    encodings_to_try = []
    detected = detect_encoding(file_input)
    if detected:
        encodings_to_try.append(detected)
    encodings_to_try.extend(['utf-8', 'latin-1', 'cp1252'])
    
    df = None
    for enc in encodings_to_try:
        try:
            if not isinstance(file_input, str):
                file_input.seek(0)
            df = pd.read_csv(file_input, encoding=enc)
            break
        except (UnicodeDecodeError, FileNotFoundError, pd.errors.ParserError, pd.errors.EmptyDataError):
            continue
            
    if df is None:
        file_desc = file_input if isinstance(file_input, str) else getattr(file_input, "name", "input buffer")
        raise ValueError(f"Could not read CSV file '{file_desc}'. Tried encodings: {encodings_to_try}")

    # 2. Normalize column names (essential for prompt building consistency)
    normalized_cols = [normalize_text(col) for col in df.columns]
    df.columns = normalized_cols

    # 3. Infer WikiSQL types (real vs text)
    wiki_types = []
    for col in df.columns:
        if pd.api.types.is_numeric_dtype(df[col]):
            wiki_types.append("real")
        else:
            wiki_types.append("text")

    # 4. Load into SQLite in-memory DB
    conn = sqlite3.connect(":memory:", check_same_thread=False)
    table_name = "data"
    
    # Write to SQLite
    # We use df.to_sql, which automatically maps pandas types to SQLite types safely
    df.to_sql(table_name, conn, index=False, if_exists="replace")

    return conn, table_name, normalized_cols, wiki_types