app.py

"""
app.py — tracer scan en tant que mini-app Gradio, à héberger sur un HuggingFace
Space (SDK: gradio). Permet d'uploader n'importe quel .jsonl compatible TRACER,
de choisir l'embedder / le target / le mode force, et de récupérer le rapport
HTML interactif + un résumé terminal-like du balayage de targets.

Déploiement :
  1. huggingface.co/new-space -> SDK "Gradio" -> hardware "CPU basic" suffit
  2. Uploader ce fichier (app.py) + requirements.txt à la racine du Space
  3. Le Space build automatiquement et expose l'UI

Aucun souci réseau/SSL attendu ici : le Space tourne sur l'infra HuggingFace,
donc le téléchargement du modèle sentence-transformers se fait en local au
Hub, sans proxy ni certificat tiers à valider.
"""

from __future__ import annotations

import tempfile
from pathlib import Path

import gradio as gr
import numpy as np

import tracer
from tracer.scanner import ThinDataError, load_scan_traces, scan_html
from tracer.embeddings.index import embed_texts

EMBED_MODEL_CHOICES = [
    "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2",  # défaut fit_tracer.py
    "BAAI/bge-m3",                                                   # exigé par la spec
    "all-MiniLM-L6-v2",                                              # défaut CLI tracer (EN only)
]

# Cache mémoire (process du Space) pour éviter de ré-embruter le même
# fichier+modèle entre deux clics (target/force/viz-layout différents).
_EMBED_CACHE: dict[str, np.ndarray] = {}


def _cache_key(path: Path, model: str) -> str:
    import hashlib
    h = hashlib.sha256()
    h.update(path.read_bytes())
    h.update(model.encode("utf-8"))
    return h.hexdigest()


def _get_embeddings(path: Path, model: str, inputs: list[str], progress: gr.Progress | None = None) -> np.ndarray:
    key = _cache_key(path, model)
    if key in _EMBED_CACHE:
        return _EMBED_CACHE[key]
    if progress:
        progress(0.1, desc=f"Téléchargement / chargement de {model}...")
    X = embed_texts(inputs, model=model, show_progress=False)
    _EMBED_CACHE[key] = X
    return X


def run_scan(
    files,
    embed_model: str,
    primary_target: float,
    sweep_targets_str: str,
    force_mode: str,
    price: float,
    monthly_calls: int,
    viz_layout: str,
    progress: gr.Progress = gr.Progress(),
):
    """Callback principal du bouton 'Lancer le scan'."""
    if not files:
        return "⚠ Aucun fichier fourni.", "", []

    try:
        sweep_targets = sorted(
            {float(t.strip()) for t in sweep_targets_str.split(",") if t.strip()}
            | {primary_target}
        )
    except ValueError:
        return "⚠ Targets de balayage invalides (attendu : ex. '0.8, 0.85, 0.9, 0.95').", "", []

    summary_lines: list[str] = []
    html_reports: list[str] = []
    html_files: list[str] = []

    for i, file_obj in enumerate(files):
        path = Path(file_obj.name if hasattr(file_obj, "name") else file_obj)
        progress((i) / len(files), desc=f"Lecture de {path.name}...")

        try:
            inputs, labels = load_scan_traces(path)
        except Exception as exc:
            summary_lines.append(f"=== {path.name} ===\n⚠ Erreur de lecture : {exc}\n")
            continue

        n = len(inputs)
        n_classes = len(set(labels))
        summary_lines.append(f"=== {path.name} ===")
        summary_lines.append(f"{n} traces utilisables · {n_classes} labels distincts")

        if force_mode == "auto":
            effective_force = n < 1000
            if effective_force:
                summary_lines.append(f"⚠ {n} < 1000 traces : force appliqué automatiquement (résultat best-effort).")
        elif force_mode == "force":
            effective_force = True
        else:
            effective_force = False

        progress((i + 0.3) / len(files), desc=f"Embeddings ({embed_model})...")
        try:
            X = _get_embeddings(path, embed_model, inputs, progress=progress)
        except Exception as exc:
            summary_lines.append(f"⚠ Échec du calcul des embeddings : {exc}\n")
            continue

        progress((i + 0.6) / len(files), desc="Balayage de targets...")
        for t in sweep_targets:
            try:
                result = tracer.scan(path, target=t, embeddings=X, force=effective_force, viz_layout=viz_layout)
                summary_lines.append(
                    f"  target {t:.0%}   certifiable = {result.certifiable_share:6.1%}   "
                    f"({result.n_clusters} cellules{'  [forced]' if result.forced else ''})"
                )
            except ThinDataError as exc:
                summary_lines.append(f"  target {t:.0%}   {exc}")

        progress((i + 0.9) / len(files), desc="Génération du rapport HTML...")
        try:
            result = tracer.scan(
                path,
                target=primary_target,
                embeddings=X,
                force=effective_force,
                teacher_price_per_1k=(price or None),
                monthly_calls=(monthly_calls or None),
                viz_layout=viz_layout,
            )
            html = scan_html(result, source_name=path.name)
            tmp = tempfile.NamedTemporaryFile(
                delete=False, suffix=".html", prefix=f"{path.stem}_scan_"
            )
            tmp.write(html.encode("utf-8"))
            tmp.close()
            html_files.append(tmp.name)
            html_reports.append(html)
            summary_lines.append(f"  → rapport HTML généré ({path.name})")
        except ThinDataError as exc:
            summary_lines.append(f"  ⚠ Rapport HTML non généré : {exc}")

        summary_lines.append("")

    terminal_summary = "\n".join(summary_lines)
    preview_html = html_reports[0] if html_reports else "<p>Aucun rapport généré.</p>"
    return terminal_summary, preview_html, html_files


with gr.Blocks(title="TRACER scan — testeur de dataset") as demo:
    gr.Markdown(
        "# tracer scan — testeur de dataset\n"
        "Upload un ou plusieurs `.jsonl` compatibles TRACER (`input`+`teacher`, "
        "ou alias tolérés) pour voir leur routabilité probable **avant** de lancer "
        "`fit`. Tourne sur l'infra HuggingFace : le modèle d'embedding se "
        "télécharge directement depuis le Hub, sans souci réseau/proxy local."
    )

    with gr.Row():
        with gr.Column(scale=1):
            files_in = gr.File(
                label="Fichier(s) .jsonl", file_count="multiple", file_types=[".jsonl"]
            )
            embed_model_in = gr.Dropdown(
                label="Embedder",
                choices=EMBED_MODEL_CHOICES,
                value=EMBED_MODEL_CHOICES[0],
                allow_custom_value=True,
            )
            target_in = gr.Slider(
                label="Target primaire (rapport HTML détaillé)",
                minimum=0.5, maximum=0.99, step=0.01, value=0.95,
            )
            sweep_in = gr.Textbox(
                label="Targets de balayage (séparés par des virgules)",
                value="0.80, 0.85, 0.90, 0.95",
            )
            force_in = gr.Radio(
                label="Mode thin-data",
                choices=[("auto (force si <1000 traces)", "auto"), ("forcer", "force"), ("refuser", "no_force")],
                value="auto",
            )
            with gr.Accordion("Options économiques (facultatif)", open=False):
                price_in = gr.Number(label="Prix teacher $/1k appels", value=None)
                monthly_in = gr.Number(label="Volume mensuel d'appels", value=None, precision=0)
            viz_in = gr.Radio(
                label="Layout 3D du rapport", choices=["pca", "umap", "tsne", "auto"], value="pca"
            )
            run_btn = gr.Button("Lancer le scan", variant="primary")

        with gr.Column(scale=2):
            summary_out = gr.Textbox(label="Résumé (terminal-like)", lines=24)
            html_preview = gr.HTML(label="Aperçu du rapport (1er fichier)")
            html_download = gr.Files(label="Télécharger le(s) rapport(s) HTML")

    run_btn.click(
        run_scan,
        inputs=[files_in, embed_model_in, target_in, sweep_in, force_in, price_in, monthly_in, viz_in],
        outputs=[summary_out, html_preview, html_download],
    )

if __name__ == "__main__":
    demo.launch(show_api=False)