cli.main¶

@app.callback()
def main(
    version: bool = typer.Option(
        False,
        "--version",
        "-v",
        callback=_version_callback,
        is_eager=True,
        help="Show version and exit.",
    ),
    verbose: bool = typer.Option(
        False,
        "--verbose",
        help="Enable DEBUG logging with file/line info.",
    ),
) -> None:
    """Local-first task classifier CLI."""
    level = logging.DEBUG if verbose else logging.WARNING
    logging.basicConfig(
        level=level,
        format="%(levelname)s %(name)s %(pathname)s:%(lineno)d — %(message)s",
    )
    ensure_taskclf_dirs()

    from taskclf.core.logging import setup_file_logging

    setup_file_logging()

@ingest_app.command("aw")
def ingest_aw_cmd(
    input_file: str = typer.Option(
        ..., "--input", help="Path to an ActivityWatch export JSON file"
    ),
    out_dir: str = typer.Option(
        DEFAULT_RAW_AW_DIR,
        help="Output directory for normalized events (partitioned by date)",
    ),
    title_salt: str = typer.Option(
        DEFAULT_TITLE_SALT, "--title-salt", help="Salt for hashing window titles"
    ),
) -> None:
    """Ingest an ActivityWatch JSON export into privacy-safe normalized events."""
    from collections import defaultdict

    import pandas as pd

    from taskclf.adapters.activitywatch.client import (
        parse_aw_export,
        parse_aw_input_export,
    )
    from taskclf.core.store import write_parquet

    input_path = Path(input_file)
    if not input_path.exists():
        typer.echo(f"File not found: {input_path}", err=True)
        raise typer.Exit(code=1)

    events = parse_aw_export(input_path, title_salt=title_salt)
    if not events:
        typer.echo("No window events found in the export file.", err=True)
        raise typer.Exit(code=1)

    by_date: dict[str, list] = defaultdict(list)
    for ev in events:
        day = ev.timestamp.date().isoformat()
        by_date[day].append(ev.model_dump())

    out_base = Path(out_dir)
    for day, rows in sorted(by_date.items()):
        df = pd.DataFrame(rows)
        out_path = out_base / day / "events.parquet"
        write_parquet(df, out_path)
        typer.echo(f"  {day}: {len(rows)} events -> {out_path}")

    unique_apps = {ev.app_id for ev in events}
    date_range = f"{events[0].timestamp.date()} to {events[-1].timestamp.date()}"
    typer.echo(
        f"Ingested {len(events)} events across {len(by_date)} day(s) "
        f"({date_range}), {len(unique_apps)} unique apps"
    )

    # Ingest input events (aw-watcher-input) if present
    input_events = parse_aw_input_export(input_path)
    if input_events:
        input_by_date: dict[str, list] = defaultdict(list)
        for ie in input_events:
            day = ie.timestamp.date().isoformat()
            input_by_date[day].append(ie.model_dump())

        input_out_base = Path(out_dir).parent / "aw-input"
        for day, rows in sorted(input_by_date.items()):
            df = pd.DataFrame(rows)
            out_path = input_out_base / day / "events.parquet"
            write_parquet(df, out_path)
            typer.echo(f"  {day}: {len(rows)} input events -> {out_path}")

        ie_range = (
            f"{input_events[0].timestamp.date()} to {input_events[-1].timestamp.date()}"
        )
        typer.echo(
            f"Ingested {len(input_events)} input events across "
            f"{len(input_by_date)} day(s) ({ie_range})"
        )

@features_app.command("build")
def features_build(
    date: str = typer.Option(
        None,
        help="Single date in YYYY-MM-DD format (mutually exclusive with --date-from/--date-to)",
    ),
    date_from: str = typer.Option(
        None, "--date-from", help="Start of date range (YYYY-MM-DD, inclusive)"
    ),
    date_to: str = typer.Option(
        None, "--date-to", help="End of date range (YYYY-MM-DD, inclusive)"
    ),
    data_dir: str = typer.Option(
        DEFAULT_DATA_DIR, help="Processed data directory (derived from TASKCLF_HOME)"
    ),
    aw_host: str = typer.Option(
        DEFAULT_AW_HOST, "--aw-host", help="ActivityWatch server URL"
    ),
    title_salt: str = typer.Option(
        DEFAULT_TITLE_SALT, "--title-salt", help="Salt for hashing window titles"
    ),
    synthetic: bool = typer.Option(
        False, "--synthetic", help="Generate dummy features instead of fetching from AW"
    ),
) -> None:
    """Build per-minute feature rows by fetching from a running ActivityWatch server.

    Supports a single --date or a --date-from / --date-to range.
    Use --synthetic to generate dummy data for testing.
    """
    from taskclf.features.build import build_features_for_date

    if date and (date_from or date_to):
        typer.echo("Use either --date or --date-from/--date-to, not both.", err=True)
        raise typer.Exit(code=1)
    if not date and not date_from:
        typer.echo("Provide --date or --date-from/--date-to.", err=True)
        raise typer.Exit(code=1)

    if date:
        start = dt.date.fromisoformat(date)
        end = start
    else:
        start = dt.date.fromisoformat(date_from)  # type: ignore[arg-type]
        end = dt.date.fromisoformat(date_to) if date_to else start

    if end < start:
        typer.echo(f"--date-to ({end}) is before --date-from ({start}).", err=True)
        raise typer.Exit(code=1)

    effective_host = None if synthetic else aw_host
    data_path = Path(data_dir)
    current = start
    ok_days: list[dt.date] = []
    empty_days: list[dt.date] = []
    failed: list[tuple[dt.date, str]] = []

    while current <= end:
        try:
            out_path = build_features_for_date(
                current,
                data_path,
                aw_host=effective_host,
                title_salt=title_salt,
                synthetic=synthetic,
            )
            row_count = _parquet_row_count(out_path)
            if row_count == 0:
                empty_days.append(current)
            else:
                ok_days.append(current)
        except Exception as exc:
            failed.append((current, str(exc)))
        current += dt.timedelta(days=1)

    total = len(ok_days) + len(empty_days) + len(failed)
    out_dir = _feature_output_dir(data_path)
    typer.echo(
        f"Done — {total} day(s): {len(ok_days)} ok, "
        f"{len(empty_days)} empty, {len(failed)} failed."
    )
    typer.echo(f"  output: {out_dir}")
    if empty_days:
        dates_str = ", ".join(d.isoformat() for d in empty_days)
        typer.echo(f"  empty (no AW events): {dates_str}")
    if failed:
        for d, reason in failed:
            typer.echo(f"  FAILED {d}: {reason}", err=True)

@labels_app.command("import")
def labels_import_cmd(
    file: str = typer.Option(
        ..., "--file", help="Path to labels CSV (start_ts, end_ts, label, provenance)"
    ),
    data_dir: str = typer.Option(
        DEFAULT_DATA_DIR, help="Processed data directory (derived from TASKCLF_HOME)"
    ),
) -> None:
    """Import label spans from a CSV file and write to parquet."""
    from taskclf.labels.store import import_labels_from_csv, write_label_spans

    csv_path = Path(file)
    if not csv_path.exists():
        typer.echo(f"File not found: {csv_path}", err=True)
        raise typer.Exit(code=1)

    spans = import_labels_from_csv(csv_path)
    typer.echo(f"Validated {len(spans)} label spans")

    out_path = Path(data_dir) / "labels_v1" / "labels.parquet"
    write_label_spans(spans, out_path)
    typer.echo(f"Wrote labels to {out_path}")

@labels_app.command("add-block")
def labels_add_block_cmd(
    start: str = typer.Option(
        ..., "--start", help="Block start timestamp (ISO-8601 UTC)"
    ),
    end: str = typer.Option(..., "--end", help="Block end timestamp (ISO-8601 UTC)"),
    label: str = typer.Option(
        ...,
        "--label",
        help="Core label (Build, Debug, Review, Write, ReadResearch, Communicate, Meet, BreakIdle)",
    ),
    user_id: str = typer.Option(
        "default-user", "--user-id", help="User ID for this label"
    ),
    confidence: float | None = typer.Option(
        None, "--confidence", min=0.0, max=1.0, help="Labeler confidence (0-1)"
    ),
    data_dir: str = typer.Option(
        DEFAULT_DATA_DIR, help="Processed data directory (derived from TASKCLF_HOME)"
    ),
    model_dir: str | None = typer.Option(
        None, "--model-dir", help="Model bundle directory (for predicted label display)"
    ),
) -> None:
    """Create a manual label block for a time range."""
    import pandas as pd
    from rich.console import Console
    from rich.table import Table

    from taskclf.core.store import read_parquet
    from taskclf.core.types import LabelSpan
    from taskclf.labels.store import append_label_span, generate_label_summary

    console = Console()

    start_ts = dt.datetime.fromisoformat(start)
    end_ts = dt.datetime.fromisoformat(end)
    effective_confidence = confidence if confidence is not None else 1.0

    span = LabelSpan(
        start_ts=start_ts,
        end_ts=end_ts,
        label=label,
        provenance="manual",
        user_id=user_id,
        confidence=effective_confidence,
    )

    features_dfs: list[pd.DataFrame] = []
    data_path = Path(data_dir)
    current_date = start_ts.date()
    while current_date <= end_ts.date():
        fp = _feature_parquet_for_date(data_path, current_date)
        if fp is not None and fp.exists():
            features_dfs.append(read_parquet(fp))
        current_date += dt.timedelta(days=1)

    if features_dfs:
        feat_df = pd.concat(features_dfs, ignore_index=True)
        summary = generate_label_summary(feat_df, start_ts, end_ts)

        table = Table(title="Block Summary")
        table.add_column("Metric", style="bold")
        table.add_column("Value")
        table.add_row("Time range", f"{start_ts} → {end_ts}")
        table.add_row("Buckets", str(summary["total_buckets"]))
        table.add_row("Sessions", str(summary["session_count"]))
        if summary["mean_keys_per_min"] is not None:
            table.add_row("Avg keys/min", str(summary["mean_keys_per_min"]))
        if summary["mean_clicks_per_min"] is not None:
            table.add_row("Avg clicks/min", str(summary["mean_clicks_per_min"]))
        for app_info in summary["top_apps"][:3]:
            table.add_row(
                f"App: {app_info['app_id']}", f"{app_info['buckets']} buckets"
            )
        console.print(table)

    if model_dir is not None and features_dfs:
        try:
            from taskclf.core.model_io import load_model_bundle
            from taskclf.infer.batch import predict_labels

            model, _meta, cat_encoders = load_model_bundle(Path(model_dir))
            from sklearn.preprocessing import LabelEncoder

            from taskclf.core.types import LABEL_SET_V1

            le = LabelEncoder()
            le.fit(sorted(LABEL_SET_V1))
            preds = predict_labels(model, feat_df, le, cat_encoders=cat_encoders)
            from collections import Counter

            top_pred = Counter(preds).most_common(1)[0][0]
            console.print(f"[bold]Top predicted label:[/bold] {top_pred}")
        except Exception as exc:
            console.print(f"[dim]Could not predict: {exc}[/dim]")

    labels_path = Path(data_dir) / "labels_v1" / "labels.parquet"
    try:
        append_label_span(span, labels_path)
    except ValueError as exc:
        typer.echo(f"Error: {exc}", err=True)
        raise typer.Exit(code=1)

    typer.echo(
        f"Added label block: {label} [{start_ts} → {end_ts}] "
        f"user={user_id} confidence={confidence}"
    )

@labels_app.command("label-now")
def labels_label_now_cmd(
    minutes: int = typer.Option(
        10, "--minutes", min=1, help="How many minutes back from now to label"
    ),
    label: str = typer.Option(
        ...,
        "--label",
        help="Core label (Build, Debug, Review, Write, ReadResearch, Communicate, Meet, BreakIdle)",
    ),
    user_id: str = typer.Option(
        "default-user", "--user-id", help="User ID for this label"
    ),
    confidence: float | None = typer.Option(
        None, "--confidence", min=0.0, max=1.0, help="Labeler confidence (0-1)"
    ),
    aw_host: str = typer.Option(
        DEFAULT_AW_HOST, "--aw-host", help="ActivityWatch server URL for live summary"
    ),
    title_salt: str = typer.Option(
        DEFAULT_TITLE_SALT, "--title-salt", help="Salt for hashing window titles"
    ),
    data_dir: str = typer.Option(
        DEFAULT_DATA_DIR, help="Processed data directory (derived from TASKCLF_HOME)"
    ),
) -> None:
    """Label the last N minutes with a single command (no timestamps needed)."""
    from collections import Counter

    import pandas as pd
    from rich.console import Console
    from rich.table import Table

    from taskclf.core.store import read_parquet
    from taskclf.core.time import ts_utc_aware_get
    from taskclf.core.types import LabelSpan
    from taskclf.labels.store import append_label_span, generate_label_summary

    console = Console()

    end_ts = dt.datetime.now(dt.timezone.utc)
    start_ts = end_ts - dt.timedelta(minutes=minutes)

    console.print(
        f"[bold]Labeling last {minutes} minute(s):[/bold] "
        f"{start_ts:%H:%M:%S} → {end_ts:%H:%M:%S} UTC"
    )

    # Live AW summary (best-effort)
    try:
        from taskclf.adapters.activitywatch.client import (
            fetch_aw_events,
            find_window_bucket_id,
        )

        bucket_id = find_window_bucket_id(aw_host)
        events = fetch_aw_events(
            aw_host, bucket_id, start_ts, end_ts, title_salt=title_salt
        )
        if events:
            app_counts = Counter(ev.app_id for ev in events)
            table = Table(title="Live Activity (from ActivityWatch)")
            table.add_column("App", style="bold")
            table.add_column("Events", justify="right")
            for app, cnt in app_counts.most_common(5):
                table.add_row(app, str(cnt))
            table.add_row("Total events", str(len(events)), style="dim")
            console.print(table)
        else:
            console.print("[dim]No ActivityWatch events in this window.[/dim]")
    except Exception as exc:
        console.print(
            f"[dim]ActivityWatch not reachable ({exc}); skipping live summary.[/dim]"
        )

    start_utc = ts_utc_aware_get(start_ts)
    end_utc = ts_utc_aware_get(end_ts)

    # On-disk feature summary (best-effort)
    features_dfs: list[pd.DataFrame] = []
    data_path = Path(data_dir)
    current_date = start_utc.date()
    while current_date <= end_utc.date():
        fp = _feature_parquet_for_date(data_path, current_date)
        if fp is not None and fp.exists():
            features_dfs.append(read_parquet(fp))
        current_date += dt.timedelta(days=1)

    if features_dfs:
        feat_df = pd.concat(features_dfs, ignore_index=True)
        summary = generate_label_summary(feat_df, start_utc, end_utc)
        if summary["total_buckets"] > 0:
            table = Table(title="Feature Summary")
            table.add_column("Metric", style="bold")
            table.add_column("Value")
            table.add_row("Buckets", str(summary["total_buckets"]))
            table.add_row("Sessions", str(summary["session_count"]))
            if summary["mean_keys_per_min"] is not None:
                table.add_row("Avg keys/min", str(summary["mean_keys_per_min"]))
            if summary["mean_clicks_per_min"] is not None:
                table.add_row("Avg clicks/min", str(summary["mean_clicks_per_min"]))
            for app_info in summary["top_apps"][:3]:
                table.add_row(
                    f"App: {app_info['app_id']}", f"{app_info['buckets']} buckets"
                )
            console.print(table)

    effective_confidence = confidence if confidence is not None else 1.0
    span = LabelSpan(
        start_ts=start_utc,
        end_ts=end_utc,
        label=label,
        provenance="manual",
        user_id=user_id,
        confidence=effective_confidence,
    )

    labels_path = Path(data_dir) / "labels_v1" / "labels.parquet"
    try:
        append_label_span(span, labels_path)
    except ValueError as exc:
        typer.echo(f"Error: {exc}", err=True)
        raise typer.Exit(code=1)

    typer.echo(
        f"Labeled: {label} [{start_ts:%H:%M} → {end_ts:%H:%M} UTC] "
        f"user={user_id} confidence={effective_confidence}"
    )

@labels_app.command("show-queue")
def labels_show_queue_cmd(
    user_id: str | None = typer.Option(
        None, "--user-id", help="Filter to a specific user"
    ),
    limit: int = typer.Option(10, "--limit", help="Maximum items to show"),
    data_dir: str = typer.Option(
        DEFAULT_DATA_DIR, help="Processed data directory (derived from TASKCLF_HOME)"
    ),
) -> None:
    """Show pending items in the active labeling queue."""
    from rich.console import Console
    from rich.table import Table

    from taskclf.labels.queue import ActiveLabelingQueue

    console = Console()
    queue_path = Path(data_dir) / "labels_v1" / "queue.json"

    if not queue_path.exists():
        console.print("[dim]No labeling queue found.[/dim]")
        return

    queue = ActiveLabelingQueue(queue_path)
    pending = queue.get_pending(user_id=user_id, limit=limit)

    if not pending:
        console.print("[dim]No pending labeling requests.[/dim]")
        return

    table = Table(title=f"Pending Labeling Requests ({len(pending)})")
    table.add_column("ID", style="dim", max_width=8)
    table.add_column("User")
    table.add_column("Time Range")
    table.add_column("Reason")
    table.add_column("Predicted")
    table.add_column("Confidence", justify="right")

    for req in pending:
        table.add_row(
            req.request_id[:8],
            req.user_id,
            f"{req.bucket_start_ts:%H:%M} → {req.bucket_end_ts:%H:%M}",
            req.reason,
            req.predicted_label or "—",
            f"{req.confidence:.2f}" if req.confidence is not None else "—",
        )

    console.print(table)

@labels_app.command("export")
def labels_export_cmd(
    out: str = typer.Option(
        "labels.csv", "--out", "-o", help="Destination CSV file path"
    ),
    data_dir: str = typer.Option(DEFAULT_DATA_DIR, help="Processed data directory"),
) -> None:
    """Export labels.parquet to CSV."""
    from taskclf.labels.store import export_labels_to_csv

    parquet_path = Path(data_dir) / "labels_v1" / "labels.parquet"
    csv_path = Path(out)
    try:
        result = export_labels_to_csv(parquet_path, csv_path)
        typer.echo(f"Exported labels -> {result}")
    except ValueError as exc:
        typer.echo(str(exc), err=True)
        raise typer.Exit(code=1)

@labels_app.command("project")
def labels_project_cmd(
    date_from: str = typer.Option(..., "--from", help="Start date (YYYY-MM-DD)"),
    date_to: str = typer.Option(..., "--to", help="End date (YYYY-MM-DD, inclusive)"),
    data_dir: str = typer.Option(
        DEFAULT_DATA_DIR, help="Processed data directory (derived from TASKCLF_HOME)"
    ),
    out_dir: str = typer.Option(
        DEFAULT_DATA_DIR, "--out-dir", help="Output directory for projected labels"
    ),
) -> None:
    """Project label blocks onto feature windows using strict containment rules."""
    import pandas as pd

    from taskclf.core.store import read_parquet, write_parquet
    from taskclf.labels.projection import project_blocks_to_windows
    from taskclf.labels.store import read_label_spans

    labels_path = Path(data_dir) / "labels_v1" / "labels.parquet"
    if not labels_path.exists():
        typer.echo(f"Labels file not found: {labels_path}", err=True)
        raise typer.Exit(code=1)

    spans = read_label_spans(labels_path)
    typer.echo(f"Loaded {len(spans)} label spans")

    start = dt.date.fromisoformat(date_from)
    end = dt.date.fromisoformat(date_to)

    all_features: list[pd.DataFrame] = []
    current = start
    while current <= end:
        fp = _feature_parquet_for_date(Path(data_dir), current)
        if fp is not None and fp.exists():
            all_features.append(read_parquet(fp))
        else:
            typer.echo(f"  skipping {current} (no features file)")
        current += dt.timedelta(days=1)

    if not all_features:
        typer.echo("No feature data found for the given date range.", err=True)
        raise typer.Exit(code=1)

    features_df = pd.concat(all_features, ignore_index=True)
    projected = project_blocks_to_windows(features_df, spans)

    out_path = Path(out_dir) / "labels_v1" / "projected_labels.parquet"
    write_parquet(projected, out_path)
    typer.echo(
        f"Projected {len(projected)} labeled windows "
        f"(from {len(features_df)} total) -> {out_path}"
    )