profiler.py

#!/usr/bin/env python
"""Deterministic dataset profiler — the honest core of CSV -> Insights (#12).

Everything numeric a user ever sees is computed HERE, in plain pandas, not by an LLM. The narrator
(narrator.py) only turns this structured profile into prose; it never invents a statistic. This is
the same "grounded, no hallucinated numbers" discipline as the #10 RAG app.

    from profiler import profile_csv
    prof = profile_csv("samples/seattle-weather.csv")   # -> JSON-able dict
"""
import warnings
import numpy as np
import pandas as pd

MAX_TOP = 8          # top categorical values reported per column
OUTLIER_K = 1.5      # IQR multiplier for outlier flagging


def _round(x, n=3):
    """JSON-safe rounding: NaN/inf -> None, numpy scalars -> python floats/ints."""
    try:
        if x is None or (isinstance(x, float) and (np.isnan(x) or np.isinf(x))):
            return None
        if isinstance(x, (np.integer,)):
            return int(x)
        if isinstance(x, (np.floating, float)):
            v = float(x)
            return None if (np.isnan(v) or np.isinf(v)) else round(v, n)
        return x
    except Exception:
        return None


def _detect_datetime(s: pd.Series) -> pd.Series | None:
    """Return a parsed datetime Series if the column looks like dates, else None. Object columns are
    parsed leniently; a column qualifies only if most values parse (avoids mangling free text)."""
    if pd.api.types.is_datetime64_any_dtype(s):
        return s
    if s.dtype == object:
        name = str(s.name).lower()
        looks_datey = any(k in name for k in ("date", "time", "day", "month", "year", "timestamp"))
        sample = s.dropna().astype(str).head(50)
        if sample.empty:
            return None
        with warnings.catch_warnings():
            warnings.simplefilter("ignore")   # dateutil "could not infer format" is expected here
            # format="mixed" parses each value on its own terms — robust across pandas versions,
            # where bare inference could coerce valid ISO dates to NaT and mis-type the column.
            parsed = pd.to_datetime(sample, errors="coerce", format="mixed")
            if parsed.notna().mean() >= (0.6 if looks_datey else 0.95):
                return pd.to_datetime(s, errors="coerce", format="mixed")
    return None


def _numeric_summary(s: pd.Series) -> dict:
    d = s.dropna()
    if d.empty:
        return {"count": 0}
    q1, q3 = d.quantile(0.25), d.quantile(0.75)
    iqr = q3 - q1
    lo, hi = q1 - OUTLIER_K * iqr, q3 + OUTLIER_K * iqr
    outliers = int(((d < lo) | (d > hi)).sum()) if iqr > 0 else 0
    return {
        "count": int(d.count()),
        "mean": _round(d.mean()), "std": _round(d.std()),
        "min": _round(d.min()), "q1": _round(q1), "median": _round(d.median()),
        "q3": _round(q3), "max": _round(d.max()),
        "skew": _round(d.skew()) if d.count() > 2 else None,
        "outliers": outliers,
        "outlier_pct": _round(100 * outliers / len(d), 1),
    }


def _categorical_summary(s: pd.Series) -> dict:
    d = s.dropna().astype(str)
    vc = d.value_counts()
    top = [{"value": k[:60], "count": int(v), "pct": _round(100 * v / len(d), 1)}
           for k, v in vc.head(MAX_TOP).items()]
    return {"unique": int(s.nunique(dropna=True)), "top": top}


def profile_df(df: pd.DataFrame, name: str = "dataset") -> dict:
    n_rows, n_cols = df.shape
    columns, numeric_cols, cat_cols, dt_info = [], [], [], []

    for col in df.columns:
        s = df[col]
        missing = int(s.isna().sum())
        info = {
            "name": str(col),
            "dtype": str(s.dtype),
            "missing": missing,
            "missing_pct": _round(100 * missing / n_rows, 1) if n_rows else None,
            "role": None,
        }
        dt = _detect_datetime(s)
        if dt is not None and dt.notna().any():
            info["role"] = "datetime"
            valid = dt.dropna()
            span_days = int((valid.max() - valid.min()).days) if len(valid) > 1 else 0
            dt_info.append({"name": str(col),
                            "start": str(valid.min().date()), "end": str(valid.max().date()),
                            "span_days": span_days})
        elif pd.api.types.is_numeric_dtype(s):
            info["role"] = "numeric"
            info["stats"] = _numeric_summary(s)
            numeric_cols.append(str(col))
        else:
            info["role"] = "categorical"
            info["stats"] = _categorical_summary(s)
            cat_cols.append(str(col))
        columns.append(info)

    # correlations (strongest numeric pairs)
    corr_pairs = []
    if len(numeric_cols) >= 2:
        corr = df[numeric_cols].corr(numeric_only=True)
        seen = set()
        for a in numeric_cols:
            for b in numeric_cols:
                if a == b or (b, a) in seen:
                    continue
                seen.add((a, b))
                r = corr.loc[a, b]
                if pd.notna(r):
                    corr_pairs.append({"a": a, "b": b, "r": _round(r)})
        corr_pairs.sort(key=lambda p: abs(p["r"] or 0), reverse=True)

    # data-quality flags (deterministic, plain rules)
    flags = []
    for c in columns:
        if c["missing_pct"] and c["missing_pct"] >= 20:
            flags.append(f"'{c['name']}' is {c['missing_pct']}% missing")
        if c.get("role") == "categorical" and c["stats"]["unique"] <= 1 and n_rows:
            flags.append(f"'{c['name']}' has a single constant value")
        if c.get("role") == "numeric" and c["stats"].get("outlier_pct"):
            if c["stats"]["outlier_pct"] >= 5:
                flags.append(f"'{c['name']}' has {c['stats']['outlier_pct']}% outliers (IQR)")
    dup = int(df.duplicated().sum())
    if dup:
        flags.append(f"{dup} duplicate row(s)")

    return {
        "name": name,
        "shape": {"rows": int(n_rows), "cols": int(n_cols)},
        "columns": columns,
        "numeric_cols": numeric_cols,
        "categorical_cols": cat_cols,
        "datetime_cols": dt_info,
        "top_correlations": corr_pairs[:5],
        "duplicate_rows": dup,
        "flags": flags,
    }


def profile_csv(path_or_buffer, name: str = None, max_rows: int = 200_000) -> dict:
    df = pd.read_csv(path_or_buffer, nrows=max_rows)
    if isinstance(path_or_buffer, str) and name is None:
        name = path_or_buffer.replace("\\", "/").split("/")[-1]
    return profile_df(df, name=name or "dataset")


if __name__ == "__main__":
    import json, sys
    p = sys.argv[1] if len(sys.argv) > 1 else "samples/seattle-weather.csv"
    print(json.dumps(profile_csv(p), indent=2))