cleaner.py

import pandas as pd
import numpy as np
import io
import re

def detect_column_types(df: pd.DataFrame) -> dict:
    types = {}
    for col in df.columns:
        # Default to text
        col_type = "text"

        # Check missing rate
        missing_rate = df[col].isna().mean()

        if df[col].dtype in ['int64', 'float64', 'Int64', 'Float64']:
            # Check if it's boolean-like (0, 1) or ID-like
            if df[col].dropna().nunique() <= 2 and set(df[col].dropna().unique()).issubset({0, 1, 0.0, 1.0}):
                col_type = "boolean"
            elif df[col].dtype == 'int64' and df[col].nunique() == len(df[col]) and df[col].min() >= 0:
                col_type = "id"
            else:
                col_type = "numeric"
        elif df[col].dtype == 'bool':
            col_type = "boolean"
        elif df[col].dtype == 'datetime64[ns]' or pd.api.types.is_datetime64_any_dtype(df[col]) or pd.api.types.is_timedelta64_dtype(df[col]):
            col_type = "datetime"
        else:
            # String / Object
            non_null_series = df[col].dropna()

            if len(non_null_series) == 0:
                col_type = "text"
            else:
                # Check boolean
                unique_vals = set(non_null_series.astype(str).str.lower().unique())
                if unique_vals.issubset({'true', 'false', 'yes', 'no', 't', 'f', 'y', 'n', '0', '1'}):
                    col_type = "boolean"
                else:
                    # Check if date
                    try:
                        # Only try parsing if it looks date-like (e.g. contains -, /, or year) to avoid false positives
                        if non_null_series.str.contains(r'\d{2,4}[-/]\d{1,2}[-/]\d{1,4}').any():
                             pd.to_datetime(non_null_series, errors='raise')
                             col_type = "datetime"
                    except:
                        pass

                    if col_type != "datetime":
                        # Pattern matching for email, phone, url
                        emails = non_null_series.str.contains(r'^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$')
                        if emails.mean() > 0.8:
                            col_type = "email"
                        else:
                            urls = non_null_series.str.contains(r'^https?://')
                            if urls.mean() > 0.8:
                                col_type = "url"
                            else:
                                phones = non_null_series.str.contains(r'^\+?\d[\d\-\s()]{7,}\d$')
                                if phones.mean() > 0.8:
                                    col_type = "phone"
                                else:
                                    # Categorical vs Text
                                    n_unique = non_null_series.nunique()
                                    if n_unique <= 20 or n_unique / len(non_null_series) < 0.05:
                                        col_type = "categorical"
                                    elif n_unique == len(df) or n_unique == len(non_null_series):
                                        col_type = "id"
                                    else:
                                        col_type = "text"

        types[col] = {
            "type": col_type,
            "missing_rate": float(missing_rate),
            "high_missing_warning": bool(missing_rate > 0.5 and col_type in ["text", "categorical"])
        }
    return types

def generate_profile(df: pd.DataFrame, col_types: dict) -> dict:
    profile = {
        "summary": {
            "total_rows": len(df),
            "total_columns": len(df.columns),
            "total_missing": int(df.isna().sum().sum()),
            "total_missing_percent": float(df.isna().sum().sum() / (df.shape[0] * df.shape[1])) * 100 if df.size > 0 else 0,
            "duplicate_rows": int(df.duplicated().sum()),
            "memory_usage_kb": float(df.memory_usage(deep=True).sum() / 1024),
        },
        "columns": [],
        "type_distribution": {
            "numeric": sum(1 for c in col_types.values() if c["type"] == "numeric"),
            "categorical": sum(1 for c in col_types.values() if c["type"] == "categorical"),
            "text": sum(1 for c in col_types.values() if c["type"] == "text"),
            "datetime": sum(1 for c in col_types.values() if c["type"] == "datetime"),
            "boolean": sum(1 for c in col_types.values() if c["type"] == "boolean"),
            "other": sum(1 for c in col_types.values() if c["type"] not in ["numeric", "categorical", "text", "datetime", "boolean"]),
        },
        "correlations": []
    }

    # Column details
    for col in df.columns:
        c_type = col_types[col]["type"]
        non_nulls = df[col].dropna()
        col_info = {
            "name": col,
            "type": c_type,
            "missing_count": int(df[col].isna().sum()),
            "missing_percent": float(df[col].isna().mean() * 100),
            "unique_count": int(non_nulls.nunique()),
            "samples": non_nulls.astype(str).head(3).tolist(),
            "distribution": []
        }

        # Sparkline data for distribution
        if len(non_nulls) > 0:
            if c_type == "numeric":
                try:
                    hist, bins = np.histogram(non_nulls, bins=10)
                    col_info["distribution"] = hist.tolist()
                except:
                    pass
            elif c_type in ["categorical", "boolean", "text", "email", "phone"]:
                counts = non_nulls.value_counts().head(10)
                col_info["distribution"] = counts.tolist()

        profile["columns"].append(col_info)

    # Correlation matrix for numeric only
    numeric_cols = [c for c, t in col_types.items() if t["type"] == "numeric" and pd.api.types.is_numeric_dtype(df[c])]
    if len(numeric_cols) > 1:
        corr_matrix = df[numeric_cols].corr().fillna(0).round(3)
        for i, col1 in enumerate(numeric_cols):
            for j, col2 in enumerate(numeric_cols):
                profile["correlations"].append({
                    "col1": col1,
                    "col2": col2,
                    "value": float(corr_matrix.iloc[i, j])
                })

    return profile

def apply_missing_strategy(df: pd.DataFrame, col: str, strategy: str, value: str = None, col_types: dict = None) -> dict:
    if col not in df.columns:
        raise ValueError(f"Column {col} not found")

    before_missing = int(df[col].isna().sum())
    c_type = col_types[col]["type"] if col_types and col in col_types else "text"

    # SAFETY CHECK: prevent "mean" or "median" on non-numeric
    if strategy in ["mean", "median", "interpolate"]:
        if not pd.api.types.is_numeric_dtype(df[col]):
            # Try converting first (e.g., if it's strings of numbers)
            try:
                df[col] = pd.to_numeric(df[col])
            except:
                raise ValueError(f"Strategy '{strategy}' cannot be applied to non-numeric column '{col}'. Please choose Mode or Fixed Value instead.")

    if strategy == "mean":
        df[col] = df[col].fillna(df[col].mean())
    elif strategy == "median":
        df[col] = df[col].fillna(df[col].median())
    elif strategy == "mode":
        mode_val = df[col].mode()
        if not mode_val.empty:
            df[col] = df[col].fillna(mode_val[0])
    elif strategy == "fixed":
        if value is None:
            raise ValueError("Value must be provided for 'fixed' strategy")
        # Try to cast value to appropriate type
        if c_type == "numeric":
            try:
                val = float(value)
            except:
                val = value
            df[col] = df[col].fillna(val)
        else:
            df[col] = df[col].fillna(value)
    elif strategy == "drop":
        df = df.dropna(subset=[col]).copy()
        df.reset_index(drop=True, inplace=True)
    elif strategy == "ffill":
        df[col] = df[col].ffill()
    elif strategy == "bfill":
        df[col] = df[col].bfill()
    elif strategy == "interpolate":
        df[col] = df[col].interpolate()
    else:
        raise ValueError(f"Unknown strategy: {strategy}")

    after_missing = int(df[col].isna().sum())

    return {
        "df": df,
        "rows_affected": before_missing - after_missing,
        "before_missing": before_missing,
        "after_missing": after_missing
    }