"""Extension marketplace — Data Lab : analyse de données (pandas / matplotlib).

Outils :
    - ``data_profile`` — profil complet d’un CSV / JSON / Parquet ;
    - ``data_query`` — filtre / groupby / agrégations ;
    - ``data_correlate`` — corrélations numériques + insights ;
    - ``data_outliers`` — détection d’outliers (IQR / z-score) ;
    - ``data_chart`` — graphique matplotlib écrit sur disque.

Deps : ``pip install pandas matplotlib`` (pyarrow optionnel pour parquet).
"""

from __future__ import annotations

import json
from pathlib import Path
from typing import Any


def _need_pandas() -> Any:
    try:
        import pandas as pd
    except ImportError as exc:  # pragma: no cover
        raise RuntimeError(
            "pandas requis — pip install 'pandas>=2.2' (ou miniagentic[extensions])"
        ) from exc
    return pd


def register(workspace: Path) -> list:
    """Enregistre les outils Data Lab."""

    def _resolve(path: str) -> Path:
        p = Path(path).expanduser()
        return p.resolve() if p.is_absolute() else (workspace / path).resolve()

    def _load(path: str):
        pd = _need_pandas()
        target = _resolve(path)
        if not target.is_file():
            raise FileNotFoundError(f"fichier introuvable: {target}")
        suffix = target.suffix.lower()
        if suffix == ".csv":
            return pd.read_csv(target), target
        if suffix == ".json":
            try:
                return pd.read_json(target), target
            except ValueError:
                return pd.read_json(target, lines=True), target
        if suffix in {".parquet", ".pq"}:
            return pd.read_parquet(target), target
        if suffix in {".tsv", ".txt"}:
            return pd.read_csv(target, sep="\t"), target
        raise ValueError(f"format non supporté: {suffix} (csv/json/parquet/tsv)")

    def data_profile(path: str, sample_rows: int = 5) -> str:
        """Profil exploratoire d’un jeu de données (dtypes, nulls, stats, sample).

        Args:
            path: CSV / JSON / Parquet / TSV relatif au workspace.
            sample_rows: Lignes d’aperçu.
        """
        try:
            df, target = _load(path)
        except Exception as exc:  # noqa: BLE001
            return json.dumps({"error": str(exc)}, ensure_ascii=False)

        cols = []
        for name in df.columns:
            s = df[name]
            info: dict[str, Any] = {
                "name": str(name),
                "dtype": str(s.dtype),
                "nulls": int(s.isna().sum()),
                "null_pct": round(float(s.isna().mean() * 100), 2),
                "unique": int(s.nunique(dropna=True)),
            }
            if s.dtype.kind in "iufc":
                desc = s.describe()
                info["stats"] = {
                    "min": _jsonable(desc.get("min")),
                    "max": _jsonable(desc.get("max")),
                    "mean": _jsonable(desc.get("mean")),
                    "std": _jsonable(desc.get("std")),
                    "p50": _jsonable(s.median()),
                }
            else:
                top = s.astype(str).value_counts(dropna=True).head(5)
                info["top_values"] = [
                    {"value": str(idx), "count": int(val)} for idx, val in top.items()
                ]
            cols.append(info)

        sample = df.head(max(0, sample_rows)).where(df.notna(), None)
        return json.dumps(
            {
                "path": str(target),
                "rows": int(len(df)),
                "cols": int(df.shape[1]),
                "memory_mb": round(float(df.memory_usage(deep=True).sum()) / 1e6, 3),
                "columns": cols,
                "sample": sample.to_dict(orient="records"),
                "dup_rows": int(df.duplicated().sum()),
            },
            ensure_ascii=False,
            indent=2,
            default=_jsonable,
        )

    def data_query(
        path: str,
        filter_expr: str = "",
        groupby: str = "",
        agg: str = "count",
        columns: str = "",
        limit: int = 50,
    ) -> str:
        """Requête tabulaire : filtre pandas, groupby, agrégation, projection.

        Args:
            path: Fichier de données.
            filter_expr: Expression ``DataFrame.query`` (ex: `revenue > 1000`).
            groupby: Colonnes groupby séparées par des virgules.
            agg: Agrégation (count, sum, mean, median, min, max) ou ``col:agg,...``.
            columns: Colonnes à garder (virgules) — vide = toutes.
            limit: Nombre max de lignes renvoyées.
        """
        pd = _need_pandas()
        try:
            df, target = _load(path)
            if filter_expr.strip():
                df = df.query(filter_expr, engine="python")
            if columns.strip():
                cols = [c.strip() for c in columns.split(",") if c.strip()]
                df = df.loc[:, cols]
            if groupby.strip():
                keys = [c.strip() for c in groupby.split(",") if c.strip()]
                if ":" in agg:
                    agg_map = {}
                    for part in agg.split(","):
                        c, _, a = part.partition(":")
                        agg_map[c.strip()] = a.strip()
                    out = df.groupby(keys, dropna=False).agg(agg_map).reset_index()
                else:
                    fn = agg.strip().lower()
                    if fn == "count":
                        out = df.groupby(keys, dropna=False).size().reset_index(name="count")
                    else:
                        grouped = df.groupby(keys, dropna=False)
                        if not hasattr(grouped, fn):
                            raise ValueError(f"agrégation inconnue: {fn}")
                        out = getattr(grouped, fn)(numeric_only=True).reset_index()
                df = out
            df = df.head(max(1, min(limit, 500)))
            return json.dumps(
                {
                    "path": str(target),
                    "rows": int(len(df)),
                    "columns": [str(c) for c in df.columns],
                    "records": df.where(df.notna(), None).to_dict(orient="records"),
                },
                ensure_ascii=False,
                indent=2,
                default=_jsonable,
            )
        except Exception as exc:  # noqa: BLE001
            return json.dumps({"error": str(exc)}, ensure_ascii=False)

    def data_correlate(path: str, method: str = "pearson", top: int = 15) -> str:
        """Matrice de corrélation + paires les plus liées.

        Args:
            path: Fichier de données.
            method: pearson | spearman | kendall.
            top: Nombre de paires à remonter.
        """
        try:
            df, target = _load(path)
            num = df.select_dtypes(include="number")
            if num.shape[1] < 2:
                return json.dumps({"error": "moins de 2 colonnes numériques"}, ensure_ascii=False)
            corr = num.corr(method=method)
            pairs: list[dict] = []
            cols = list(corr.columns)
            for i, a in enumerate(cols):
                for b in cols[i + 1 :]:
                    val = corr.loc[a, b]
                    if val != val:  # NaN
                        continue
                    pairs.append({"a": str(a), "b": str(b), "corr": round(float(val), 4)})
            pairs.sort(key=lambda x: abs(x["corr"]), reverse=True)
            return json.dumps(
                {
                    "path": str(target),
                    "method": method,
                    "matrix": {str(c): {str(k): _jsonable(v) for k, v in corr[c].items()} for c in corr.columns},
                    "top_pairs": pairs[: max(1, top)],
                },
                ensure_ascii=False,
                indent=2,
            )
        except Exception as exc:  # noqa: BLE001
            return json.dumps({"error": str(exc)}, ensure_ascii=False)

    def data_outliers(path: str, column: str, method: str = "iqr", z: float = 3.0) -> str:
        """Détecte les outliers sur une colonne numérique.

        Args:
            path: Fichier de données.
            column: Nom de colonne.
            method: ``iqr`` ou ``zscore``.
            z: Seuil z-score si method=zscore.
        """
        try:
            df, target = _load(path)
            if column not in df.columns:
                return json.dumps({"error": f"colonne absente: {column}"}, ensure_ascii=False)
            s = df[column]
            if s.dtype.kind not in "iufc":
                return json.dumps({"error": "colonne non numérique"}, ensure_ascii=False)
            if method == "zscore":
                mu, sigma = float(s.mean()), float(s.std(ddof=0) or 0)
                if sigma == 0:
                    mask = s != s
                else:
                    mask = ((s - mu).abs() / sigma) > z
                meta = {"mu": mu, "sigma": sigma, "z": z}
            else:
                q1, q3 = float(s.quantile(0.25)), float(s.quantile(0.75))
                iqr = q3 - q1
                lo, hi = q1 - 1.5 * iqr, q3 + 1.5 * iqr
                mask = (s < lo) | (s > hi)
                meta = {"q1": q1, "q3": q3, "iqr": iqr, "lo": lo, "hi": hi}
            hits = df.loc[mask].head(50)
            return json.dumps(
                {
                    "path": str(target),
                    "column": column,
                    "method": method,
                    "meta": meta,
                    "outlier_count": int(mask.sum()),
                    "sample": hits.where(hits.notna(), None).to_dict(orient="records"),
                },
                ensure_ascii=False,
                indent=2,
                default=_jsonable,
            )
        except Exception as exc:  # noqa: BLE001
            return json.dumps({"error": str(exc)}, ensure_ascii=False)

    def data_chart(
        path: str,
        chart: str = "hist",
        x: str = "",
        y: str = "",
        out_path: str = "data-lab-chart.png",
        title: str = "",
    ) -> str:
        """Génère un graphique matplotlib (hist, bar, line, scatter, box).

        Args:
            path: Fichier de données.
            chart: hist | bar | line | scatter | box.
            x: Colonne X (ou colonne unique pour hist/box).
            y: Colonne Y (scatter/line/bar).
            out_path: PNG de sortie.
            title: Titre du graphique.
        """
        try:
            import matplotlib

            matplotlib.use("Agg")
            import matplotlib.pyplot as plt
        except ImportError as exc:
            return json.dumps(
                {"error": f"matplotlib requis — pip install matplotlib ({exc})"},
                ensure_ascii=False,
            )
        try:
            df, target = _load(path)
            fig, ax = plt.subplots(figsize=(9, 5.2))
            kind = chart.lower().strip()
            if kind == "hist":
                col = x or df.select_dtypes(include="number").columns[0]
                df[col].dropna().plot(kind="hist", bins=30, ax=ax, color="#ff6b1a", alpha=0.85)
                ax.set_xlabel(col)
            elif kind == "box":
                col = x or df.select_dtypes(include="number").columns[0]
                df[[col]].plot(kind="box", ax=ax, color="#ff6b1a")
            elif kind == "bar":
                if not x:
                    raise ValueError("x requis pour bar")
                if y:
                    df.groupby(x)[y].mean(numeric_only=True).sort_values(ascending=False).head(20).plot(
                        kind="bar", ax=ax, color="#ff6b1a"
                    )
                else:
                    df[x].astype(str).value_counts().head(20).plot(kind="bar", ax=ax, color="#ff6b1a")
            elif kind == "line":
                if not x or not y:
                    raise ValueError("x et y requis pour line")
                df.plot(x=x, y=y, kind="line", ax=ax, color="#ff6b1a")
            elif kind == "scatter":
                if not x or not y:
                    raise ValueError("x et y requis pour scatter")
                df.plot(kind="scatter", x=x, y=y, ax=ax, alpha=0.7, color="#ff6b1a")
            else:
                raise ValueError(f"chart inconnu: {chart}")
            ax.set_title(title or f"{kind} · {target.name}")
            ax.grid(True, alpha=0.25)
            fig.tight_layout()
            out = _resolve(out_path)
            out.parent.mkdir(parents=True, exist_ok=True)
            fig.savefig(out, dpi=140)
            plt.close(fig)
            rel = str(out.relative_to(workspace)) if out.is_relative_to(workspace) else str(out)
            return json.dumps(
                {"written": rel, "source": str(target), "chart": kind, "x": x, "y": y},
                ensure_ascii=False,
                indent=2,
            )
        except Exception as exc:  # noqa: BLE001
            return json.dumps({"error": str(exc)}, ensure_ascii=False)

    return [data_profile, data_query, data_correlate, data_outliers, data_chart]


def _jsonable(value: Any) -> Any:
    """Convertit numpy/pandas scalars en types JSON-safe."""
    try:
        import numpy as np

        if isinstance(value, (np.integer,)):
            return int(value)
        if isinstance(value, (np.floating,)):
            f = float(value)
            return None if f != f else round(f, 6)
        if isinstance(value, (np.bool_,)):
            return bool(value)
    except ImportError:
        pass
    if hasattr(value, "item"):
        try:
            return value.item()
        except Exception:  # noqa: BLE001
            return str(value)
    return value