app/cli.py

"""CLI harness: python -m app.cli sample.csv --text-column body."""
from __future__ import annotations

import json
import sys
import time
import uuid
from pathlib import Path

import pandas as pd
import typer

from .config import DEFAULT_EMBED_MODEL, DEFAULT_REDUCER
from .db import (
    connect,
    ensure_project,
    fetch_points_for_labeling,
    set_status,
    update_cluster_labels,
    write_results,
)
from .labeling import label_clusters
from .pipeline import run_pipeline

cli = typer.Typer(add_completion=False, no_args_is_help=True)


@cli.command()
def run(
    csv_path: Path = typer.Argument(..., exists=True, readable=True),  # noqa: B008
    text_column: str = typer.Option(..., "--text-column", "-t"),  # noqa: B008
    name: str = typer.Option("cli-run", "--name", "-n"),  # noqa: B008
    embed_model: str = typer.Option(DEFAULT_EMBED_MODEL, "--embed-model"),  # noqa: B008
    reducer: str = typer.Option(DEFAULT_REDUCER, "--reducer"),  # noqa: B008
    project_id: str | None = typer.Option(None, "--project-id"),  # noqa: B008
    tenant_id: str | None = typer.Option(None, "--tenant-id"),  # noqa: B008
    limit: int | None = typer.Option(None, "--limit", help="Truncate CSV to N rows"),  # noqa: B008
) -> None:
    """Run the full pipeline against the cloud DB and print a summary."""
    df = pd.read_csv(csv_path)
    if text_column not in df.columns:
        typer.echo(f"error: text_column '{text_column}' not in {list(df.columns)}", err=True)
        raise typer.Exit(code=2)

    if limit:
        df = df.head(limit)

    texts = [str(v) for v in df[text_column].tolist() if v is not None and str(v).strip() != ""]
    if not texts:
        typer.echo("error: no non-empty text rows", err=True)
        raise typer.Exit(code=2)

    typer.echo(f"loaded {len(texts)} rows from {csv_path}")
    t0 = time.perf_counter()

    with connect() as conn:
        pid, tid = ensure_project(
            conn,
            project_id=project_id,
            name=name,
            embed_model=embed_model,
            reducer=reducer,
            tenant_id=tenant_id,
        )
        typer.echo(f"project_id={pid} tenant_id={tid}")
        try:
            result = run_pipeline(texts, embed_model=embed_model, reducer=reducer)
            summary = write_results(
                conn, project_id=pid, tenant_id=tid, texts=texts, result=result
            )
        except Exception as e:
            set_status(conn, pid, "error")
            typer.echo(f"error: pipeline failed: {e}", err=True)
            raise

    elapsed = time.perf_counter() - t0
    typer.echo("---")
    typer.echo(f"n_points    : {summary['n_points']}")
    typer.echo(f"n_clusters  : {summary['n_clusters']}")
    typer.echo(f"n_noise     : {summary['n_noise']}")
    typer.echo(f"used_pca    : {summary['used_pca']}")
    typer.echo(f"reducer     : {summary['reducer']}")
    typer.echo(f"embed_model : {summary['embed_model']}")
    typer.echo(f"runtime     : {elapsed:.2f}s")
    typer.echo(f"project_id  : {summary['project_id']}")


@cli.command("bake-static")
def bake_static(
    csv_path: Path = typer.Argument(..., exists=True, readable=True),  # noqa: B008
    out_path: Path = typer.Option(..., "--out", "-o"),  # noqa: B008
    text_column: str = typer.Option(..., "--text-column", "-t"),  # noqa: B008
    name: str = typer.Option("Demo galaxy", "--name", "-n"),  # noqa: B008
    embed_model: str = typer.Option(DEFAULT_EMBED_MODEL, "--embed-model"),  # noqa: B008
    reducer: str = typer.Option(DEFAULT_REDUCER, "--reducer"),  # noqa: B008
    label_column: str | None = typer.Option(  # noqa: B008
        None, "--label-column", help="CSV column to use as a per-cluster label hint (majority vote)"
    ),
    min_cluster_size: int | None = typer.Option(  # noqa: B008
        None, "--min-cluster-size", help="HDBSCAN min_cluster_size override"
    ),
    cluster_method: str = typer.Option(  # noqa: B008
        "eom", "--cluster-method", help="HDBSCAN cluster_selection_method (eom or leaf)"
    ),
    limit: int | None = typer.Option(None, "--limit"),  # noqa: B008
) -> None:
    """Bake the pipeline output to a static JSON file (skips the DB).

    The payload matches the shape of `/api/projects/[id]/data` JSON so the web
    app's existing loadProject helper consumes it unchanged. Use this for the
    permanently-ready demo project.
    """
    df = pd.read_csv(csv_path)
    if text_column not in df.columns:
        typer.echo(f"error: text_column '{text_column}' not in {list(df.columns)}", err=True)
        raise typer.Exit(code=2)
    if limit:
        df = df.head(limit)

    mask = df[text_column].notna() & (df[text_column].astype(str).str.strip() != "")
    df = df.loc[mask].reset_index(drop=True)
    texts = [str(v) for v in df[text_column].tolist()]
    if not texts:
        typer.echo("error: no non-empty text rows", err=True)
        raise typer.Exit(code=2)

    labels = (
        [str(v) for v in df[label_column].tolist()] if label_column and label_column in df.columns
        else None
    )

    typer.echo(f"loaded {len(texts)} rows from {csv_path}")
    t0 = time.perf_counter()
    result = run_pipeline(
        texts,
        embed_model=embed_model,
        reducer=reducer,
        min_cluster_size=min_cluster_size,
        cluster_selection_method=cluster_method,
    )
    elapsed = time.perf_counter() - t0

    # Pre-mint stable UUIDs so future references (e.g. medoid_point_id) survive
    # round-trips through the static file.
    point_ids = [str(uuid.uuid4()) for _ in range(len(texts))]
    project_id = str(uuid.uuid4())

    cluster_label_hints: dict[int, str] = {}
    if labels is not None:
        from collections import Counter
        for c in result.clusters:
            members = [
                labels[i]
                for i in range(len(texts))
                if int(result.cluster_ids[i]) == c.cluster_id
            ]
            if members:
                top, _ = Counter(members).most_common(1)[0]
                cluster_label_hints[c.cluster_id] = top

    points_json = []
    for i, text in enumerate(texts):
        cid = int(result.cluster_ids[i])
        points_json.append({
            "id": point_ids[i],
            "text": text,
            "x": float(result.coords[i, 0]),
            "y": float(result.coords[i, 1]),
            "z": float(result.coords[i, 2]),
            "cluster_id": cid if cid != -1 else None,
            "cluster_probability": float(result.cluster_probabilities[i]),
        })

    clusters_json = [
        {
            "cluster_id": c.cluster_id,
            "label": cluster_label_hints.get(c.cluster_id, c.label),
            "cx": c.cx,
            "cy": c.cy,
            "cz": c.cz,
            "size": c.size,
            "medoid_point_id": point_ids[c.medoid_index],
        }
        for c in result.clusters
    ]

    payload = {
        "project": {
            "id": project_id,
            "name": name,
            "point_count": len(texts),
        },
        "points": points_json,
        "clusters": clusters_json,
    }

    out_path.parent.mkdir(parents=True, exist_ok=True)
    out_path.write_text(json.dumps(payload, separators=(",", ":")))
    size_mb = out_path.stat().st_size / (1024 * 1024)
    n_noise = int((result.cluster_ids == -1).sum())
    typer.echo("---")
    typer.echo(f"n_points   : {len(texts)}")
    typer.echo(f"n_clusters : {len(result.clusters)}")
    typer.echo(f"n_noise    : {n_noise}")
    typer.echo(f"used_pca   : {result.used_pca}")
    typer.echo(f"reducer    : {result.reducer}")
    typer.echo(f"runtime    : {elapsed:.2f}s")
    typer.echo(f"wrote      : {out_path} ({size_mb:.2f} MiB)")


@cli.command()
def relabel(
    project_id: str = typer.Argument(..., help="UUID of project to relabel"),  # noqa: B008
) -> None:
    """Recompute and persist cluster labels for an existing project.

    Reads texts + cluster_ids straight from points, runs c-TF-IDF (+ optional
    LLM upgrade if OPENAI_API_KEY is set), and updates clusters.label in place.
    No embedding work, no point/edge writes — just label repair.
    """
    with connect() as conn:
        texts, cluster_ids, snippets = fetch_points_for_labeling(conn, project_id)
        if not texts:
            typer.echo(f"project {project_id}: no points (or missing)")
            raise typer.Exit(code=1)
        labels = label_clusters(
            texts, cluster_ids, medoid_snippets_by_cluster=snippets or None
        )
        updated = update_cluster_labels(conn, project_id, labels)
    typer.echo(f"project {project_id}")
    typer.echo(f"  clusters : {len(labels)}")
    typer.echo(f"  updated  : {updated}")
    for cid, lbl in sorted(labels.items()):
        typer.echo(f"  {cid:>3}: {lbl}")


def main() -> None:
    cli()


if __name__ == "__main__":
    sys.exit(cli() or 0)