dataforge/cli/watch.py

"""CLI subcommand: ``dataforge watch``."""

from __future__ import annotations

import json
import time
from pathlib import Path
from typing import Annotated, Literal

import typer
from rich.console import Console
from rich.panel import Panel

from dataforge.cli.common import load_schema, read_csv, resolve_cli_path
from dataforge.detectors import run_all_detectors
from dataforge.detectors.base import Schema
from dataforge.ui.profile_view import render_profile_table
from dataforge.ui.repair_diff import render_repair_diff

_console = Console(stderr=True)

WatchAction = Literal["profile", "repair"]


def _load_optional_schema(schema_path: Path | None) -> Schema | None:
    if schema_path is None:
        return None
    resolved_schema = resolve_cli_path(schema_path)
    if not resolved_schema.exists():
        raise typer.BadParameter(f"Schema file '{schema_path}' does not exist.")
    return load_schema(resolved_schema)


def _profile_once(path: Path, schema: Schema | None, json_output: bool) -> None:
    df = read_csv(path)
    issues = run_all_detectors(df, schema)
    if json_output:
        typer.echo(
            json.dumps(
                {
                    "event": "profile",
                    "path": str(path),
                    "issues_count": len(issues),
                    "issues": [issue.model_dump(mode="json") for issue in issues],
                },
                indent=2,
                sort_keys=True,
            )
        )
        return
    render_profile_table(issues, Console(), file_path=str(path))


def _repair_once(path: Path, schema: Schema | None, apply: bool, json_output: bool) -> None:
    from dataforge.engine.repair import RepairPipelineRequest, run_repair_pipeline

    result = run_repair_pipeline(
        RepairPipelineRequest(
            source_path=path,
            mode="apply" if apply else "dry_run",
            schema=schema,
            interactive=False,
        )
    )
    if json_output:
        payload = result.model_dump(mode="json")
        payload["event"] = "repair"
        typer.echo(json.dumps(payload, indent=2, sort_keys=True))
        return
    render_repair_diff(result.fixes, Console(), file_path=str(path))


def _run_once(
    path: Path, schema: Schema | None, action: WatchAction, apply: bool, json: bool
) -> None:
    if action == "repair":
        _repair_once(path, schema, apply, json)
    else:
        _profile_once(path, schema, json)


def watch(
    path: Annotated[
        Path,
        typer.Argument(help="CSV or dbt artifact path to watch."),
    ],
    schema: Annotated[
        Path | None,
        typer.Option("--schema", help="Path to a YAML schema file with column types and FDs."),
    ] = None,
    action: Annotated[
        WatchAction,
        typer.Option("--action", help="Action to run when the file changes: profile or repair."),
    ] = "profile",
    apply: Annotated[
        bool,
        typer.Option("--apply", help="Apply repairs on change. Defaults to dry-run repair."),
    ] = False,
    interval: Annotated[
        float,
        typer.Option("--interval", min=0.1, help="Polling interval in seconds."),
    ] = 2.0,
    once: Annotated[
        bool,
        typer.Option("--once", help="Run once and exit, useful for CI acceptance."),
    ] = False,
    json_output: Annotated[
        bool,
        typer.Option("--json", help="Print watch events as JSON."),
    ] = False,
) -> None:
    """Poll a path and rerun profile or repair when it changes."""
    resolved_path = resolve_cli_path(path)
    if not resolved_path.exists():
        _console.print(f"[bold red]Watch path not found:[/bold red] {path}")
        raise typer.Exit(code=2)
    parsed_schema = _load_optional_schema(schema)

    if apply and action != "repair":
        _console.print(
            Panel(
                "--apply is only valid with --action repair.",
                title="Watch Error",
                style="red",
            )
        )
        raise typer.Exit(code=2)

    _run_once(resolved_path, parsed_schema, action, apply, json_output)
    if once:
        return

    last_mtime = resolved_path.stat().st_mtime_ns
    while True:
        time.sleep(interval)
        try:
            current_mtime = resolved_path.stat().st_mtime_ns
        except FileNotFoundError:
            _console.print(f"[bold red]Watch path disappeared:[/bold red] {resolved_path}")
            raise typer.Exit(code=2) from None
        if current_mtime == last_mtime:
            continue
        last_mtime = current_mtime
        _run_once(resolved_path, parsed_schema, action, apply, json_output)