CLI Guidelines

clig.py

"""
clig.py — Reusable toolkit implementing https://clig.dev guidelines.

Covers:
  - TTY detection (stdout/stderr color/formatting awareness)
  - Styled output (info, success, warning, error) to correct streams
  - Structured exit codes
  - Spinner / progress feedback for long ops
  - Suggestions on typo / unknown subcommand
  - Stdin-is-TTY guard
  - --json output toggle
  - Confirmation prompts
  - Version callback (argparse + typer flavors)
  - Rich help formatter for argparse
  - Typer integration helpers

Usage with argparse:
    from clig import CLIGFormatter, out, err, ExitCode, confirm, spinner

Usage with typer:
    from clig import typer_app, out, err, ExitCode, confirm, spinner
"""

from __future__ import annotations

import argparse
import difflib
import json
import os
import shutil
import signal
import sys
import threading
import time
from contextlib import contextmanager
from enum import IntEnum
from typing import Any, Callable, Generator, Iterable, NoReturn, Sequence


# ---------------------------------------------------------------------------
# Exit codes
#
# Prefer os.EX_* (POSIX sysexits.h) where available (Unix/macOS).
# Windows doesn't define these constants, so fall back to the raw int
# values from sysexits.h so cross-platform code still compiles.
# os.EX_OK and the plain 1/2 convention are universally safe.
# ---------------------------------------------------------------------------

def _ex(posix_const: str, fallback: int) -> int:
    """Return os.EX_* value if the platform defines it, else the fallback int."""
    return getattr(os, posix_const, fallback)


class ExitCode(IntEnum):
    # Universal codes — safe on all platforms
    OK            = _ex("EX_OK",          0)
    GENERAL_ERROR = 1                          # no POSIX constant; convention
    MISUSE        = _ex("EX_USAGE",       64)  # bad CLI usage / invalid args

    # sysexits.h codes — fall back to canonical values on Windows
    DATA_ERR      = _ex("EX_DATAERR",     65)  # bad input data
    NO_INPUT      = _ex("EX_NOINPUT",     66)  # input file not found/readable
    NO_USER       = _ex("EX_NOUSER",      67)  # addressed user unknown
    NO_HOST       = _ex("EX_NOHOST",      68)  # host name unknown
    UNAVAILABLE   = _ex("EX_UNAVAILABLE", 69)  # service/resource unavailable
    SOFTWARE      = _ex("EX_SOFTWARE",    70)  # internal software error
    OS_ERR        = _ex("EX_OSERR",       71)  # OS error (fork/pipe/etc.)
    OS_FILE       = _ex("EX_OSFILE",      72)  # critical OS file missing
    CANT_CREATE   = _ex("EX_CANTCREAT",   73)  # output file cannot be created
    IO_ERROR      = _ex("EX_IOERR",       74)  # I/O error
    TEMP_FAIL     = _ex("EX_TEMPFAIL",    75)  # temporary failure; retry later
    PROTOCOL      = _ex("EX_PROTOCOL",    76)  # remote protocol error
    NO_PERM       = _ex("EX_NOPERM",      77)  # insufficient permissions
    CONFIG_ERR    = _ex("EX_CONFIG",      78)  # configuration error


# ---------------------------------------------------------------------------
# TTY detection
# ---------------------------------------------------------------------------

def is_tty(stream=None) -> bool:
    """Return True if the given stream (default stdout) is a TTY."""
    s = stream if stream is not None else sys.stdout
    return hasattr(s, "isatty") and s.isatty()


def supports_color(stream=None) -> bool:
    """Return True if the stream supports ANSI color codes."""
    if not is_tty(stream):
        return False
    if os.environ.get("NO_COLOR"):          # https://no-color.org
        return False
    if os.environ.get("TERM") == "dumb":
        return False
    return True


# ---------------------------------------------------------------------------
# ANSI helpers
# ---------------------------------------------------------------------------

_RESET  = "\033[0m"
_BOLD   = "\033[1m"
_DIM    = "\033[2m"
_RED    = "\033[31m"
_GREEN  = "\033[32m"
_YELLOW = "\033[33m"
_CYAN   = "\033[36m"
_WHITE  = "\033[97m"


def _colorize(text: str, *codes: str, stream=None) -> str:
    if supports_color(stream):
        return "".join(codes) + text + _RESET
    return text


# ---------------------------------------------------------------------------
# Output helpers (stdout / stderr routing per CLIG)
# ---------------------------------------------------------------------------

class _Out:
    """
    Primary output writer.  Machine-readable data → stdout.
    Color/formatting only when stdout is a TTY.
    """

    def __call__(self, msg: str = "", end: str = "\n") -> None:
        print(msg, end=end, file=sys.stdout)

    def json(self, data: Any, indent: int = 2) -> None:
        """Emit JSON to stdout (always, regardless of TTY)."""
        print(json.dumps(data, indent=indent), file=sys.stdout)

    def success(self, msg: str) -> None:
        prefix = _colorize("✓ ", _GREEN, _BOLD, stream=sys.stdout)
        print(f"{prefix}{msg}", file=sys.stdout)

    def info(self, msg: str) -> None:
        prefix = _colorize("→ ", _CYAN, stream=sys.stdout)
        print(f"{prefix}{msg}", file=sys.stdout)

    def dim(self, msg: str) -> None:
        print(_colorize(msg, _DIM, stream=sys.stdout), file=sys.stdout)


class _Err:
    """
    Messaging writer (logs, warnings, errors) → stderr per CLIG.
    """

    def __call__(self, msg: str = "", end: str = "\n") -> None:
        print(msg, end=end, file=sys.stderr)

    def warning(self, msg: str) -> None:
        prefix = _colorize("⚠  Warning: ", _YELLOW, _BOLD, stream=sys.stderr)
        print(f"{prefix}{msg}", file=sys.stderr)

    def error(self, msg: str, hint: str | None = None) -> None:
        prefix = _colorize("✗ Error: ", _RED, _BOLD, stream=sys.stderr)
        print(f"{prefix}{msg}", file=sys.stderr)
        if hint:
            print(_colorize(f"  Hint: {hint}", _DIM, stream=sys.stderr), file=sys.stderr)

    def fatal(
        self,
        msg: str,
        hint: str | None = None,
        code: int = ExitCode.GENERAL_ERROR,
    ) -> NoReturn:
        self.error(msg, hint)
        sys.exit(code)


out = _Out()
err = _Err()


# ---------------------------------------------------------------------------
# Stdin guard (CLIG: if stdin is a TTY when piped input expected → show help)
# ---------------------------------------------------------------------------

def require_piped_stdin(parser: "argparse.ArgumentParser | None" = None) -> None:
    """
    Call this in commands that expect piped stdin.
    If stdin is interactive, print help (or a notice) and exit 2.
    """
    if is_tty(sys.stdin):
        if parser is not None:
            parser.print_help(sys.stderr)
        else:
            err("This command expects piped input (e.g. `cat file | mytool`).")
        sys.exit(ExitCode.MISUSE)


# ---------------------------------------------------------------------------
# Spinner / progress feedback
# ---------------------------------------------------------------------------

_SPINNER_FRAMES = ("⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏")


class Spinner:
    """
    Simple non-blocking terminal spinner.  Auto-disabled when stderr is
    not a TTY (e.g. piped/CI) so output stays clean.
    """

    def __init__(self, message: str = "Working…", stream=sys.stderr):
        self.message = message
        self.stream = stream
        self._active = False
        self._thread: threading.Thread | None = None
        self._enabled = is_tty(stream)

    def _spin(self) -> None:
        idx = 0
        while self._active:
            if self._enabled:
                frame = _colorize(_SPINNER_FRAMES[idx % len(_SPINNER_FRAMES)], _CYAN, stream=self.stream)
                self.stream.write(f"\r{frame} {self.message}")
                self.stream.flush()
            idx += 1
            time.sleep(0.08)

    def start(self) -> "Spinner":
        self._active = True
        self._thread = threading.Thread(target=self._spin, daemon=True)
        self._thread.start()
        return self

    def stop(self, final_msg: str | None = None) -> None:
        self._active = False
        if self._thread:
            self._thread.join()
        if self._enabled:
            self.stream.write("\r\033[K")   # clear line
            self.stream.flush()
        if final_msg:
            print(final_msg, file=self.stream)

    def __enter__(self) -> "Spinner":
        return self.start()

    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
        if exc_type:
            self.stop(_colorize("✗ Failed", _RED, _BOLD, stream=self.stream))
        else:
            self.stop(_colorize("✓ Done", _GREEN, _BOLD, stream=self.stream))


@contextmanager
def spinner(message: str = "Working…") -> Generator[Spinner, None, None]:
    """Context manager wrapper around Spinner."""
    s = Spinner(message)
    with s:
        yield s


# ---------------------------------------------------------------------------
# Confirmation prompt
# ---------------------------------------------------------------------------

def confirm(
    prompt: str,
    default: bool = False,
    destructive: bool = False,
) -> bool:
    """
    Prompt user for yes/no confirmation.  Returns bool.
    If stdin is not a TTY (scripted), returns `default` silently.
    Mark destructive=True to highlight the risk.
    """
    if not is_tty(sys.stdin):
        return default

    yn = "Y/n" if default else "y/N"
    if destructive:
        prompt = _colorize(prompt, _RED, _BOLD, stream=sys.stderr)
    try:
        answer = input(f"{prompt} [{yn}]: ").strip().lower()
    except (EOFError, KeyboardInterrupt):
        print(file=sys.stderr)
        return False

    if not answer:
        return default
    return answer in ("y", "yes")


# ---------------------------------------------------------------------------
# Typo / did-you-mean suggestions
# ---------------------------------------------------------------------------

def suggest(
    word: str,
    candidates: Iterable[str],
    cutoff: float = 0.6,
    n: int = 3,
) -> list[str]:
    """Return close matches from candidates for the given word."""
    return difflib.get_close_matches(word, list(candidates), n=n, cutoff=cutoff)


def die_with_suggestion(
    unknown: str,
    candidates: Iterable[str],
    noun: str = "command",
    code: int = ExitCode.MISUSE,
) -> NoReturn:
    """
    Print 'unknown <noun>' error, offer did-you-mean suggestions, exit.
    """
    matches = suggest(unknown, candidates)
    err.error(f"Unknown {noun}: {unknown!r}")
    if matches:
        if len(matches) == 1:
            print(_colorize(f"  Did you mean: {matches[0]}", _CYAN, stream=sys.stderr), file=sys.stderr)
        else:
            alts = ", ".join(matches)
            print(_colorize(f"  Did you mean one of: {alts}", _CYAN, stream=sys.stderr), file=sys.stderr)
    sys.exit(code)


# ---------------------------------------------------------------------------
# --json flag helper
# ---------------------------------------------------------------------------

def add_json_flag(parser: "argparse.ArgumentParser") -> None:
    """Add --json flag to an argparse parser."""
    parser.add_argument(
        "--json",
        action="store_true",
        default=False,
        help="Output result as JSON (machine-readable).",
    )


def output_or_json(data: dict | list, as_json: bool) -> None:
    """Emit data as JSON to stdout if --json, else return data for caller to format."""
    if as_json:
        out.json(data)
    return data


# ---------------------------------------------------------------------------
# Version callback (argparse)
# ---------------------------------------------------------------------------

def add_version_flag(
    parser: "argparse.ArgumentParser",
    version: str,
    prog: str | None = None,
) -> None:
    name = prog or parser.prog or "app"
    parser.add_argument(
        "-V", "--version",
        action="version",
        version=f"{name} {version}",
    )


# ---------------------------------------------------------------------------
# Rich help formatter for argparse (CLIG: formatted, scannable help text)
# ---------------------------------------------------------------------------

class CLIGFormatter(argparse.HelpFormatter):
    """
    Argparse formatter that:
      - Bolds section headings when stdout is a TTY
      - Shows default values
      - Widens the option column
      - Preserves line breaks in descriptions (use RawDescription variant)
    """

    def __init__(self, prog: str, indent_increment: int = 2, max_help_position: int = 32, width: int | None = None):
        w = width or min(shutil.get_terminal_size().columns, 100)
        super().__init__(prog, indent_increment, max_help_position, w)

    def start_section(self, heading: str | None) -> None:
        if heading and supports_color(sys.stdout):
            heading = _colorize(heading.upper(), _BOLD, stream=sys.stdout)
        super().start_section(heading)

    def _get_help_string(self, action: argparse.Action) -> str | None:
        h = action.help or ""
        if action.default not in (argparse.SUPPRESS, None, False):
            if "%(default)" not in h:
                h += f" (default: %(default)s)"
        return h


class CLIGRawFormatter(CLIGFormatter, argparse.RawDescriptionHelpFormatter):
    """Combines CLIG styling with preserved description whitespace."""
    pass


# ---------------------------------------------------------------------------
# Argparse base parser factory
# ---------------------------------------------------------------------------

def make_parser(
    prog: str,
    description: str,
    version: str | None = None,
    epilog: str | None = None,
    formatter: type = CLIGRawFormatter,
    add_json: bool = False,
    support_url: str | None = None,
) -> argparse.ArgumentParser:
    """
    Create an argparse.ArgumentParser pre-configured for CLIG compliance.

    Args:
        prog:        Program name.
        description: Short description (shown at top of help).
        version:     If given, adds -V/--version flag.
        epilog:      Shown at bottom of help. Good place for examples.
        formatter:   Help formatter class (default: CLIGRawFormatter).
        add_json:    If True, adds --json flag.
        support_url: If given, appended to epilog.
    """
    full_epilog = epilog or ""
    if support_url:
        full_epilog += f"\n\nDocumentation / support: {support_url}"

    parser = argparse.ArgumentParser(
        prog=prog,
        description=description,
        epilog=full_epilog.strip() or None,
        formatter_class=formatter,
        add_help=True,
    )

    if version:
        add_version_flag(parser, version, prog)
    if add_json:
        add_json_flag(parser)

    return parser


# ---------------------------------------------------------------------------
# Graceful signal handling (CLIG: handle SIGPIPE, SIGINT cleanly)
# ---------------------------------------------------------------------------

def setup_signals() -> None:
    """
    Install clean signal handlers.
    - SIGPIPE: exit 0 silently (piped output closed by consumer)
    - SIGINT:  exit 130 with newline (Ctrl-C, standard convention)
    """
    if hasattr(signal, "SIGPIPE"):
        signal.signal(signal.SIGPIPE, signal.SIG_DFL)

    def _handle_interrupt(sig, frame):
        print(file=sys.stderr)   # move past ^C on the line
        sys.exit(130)

    signal.signal(signal.SIGINT, _handle_interrupt)


# ---------------------------------------------------------------------------
# Typer integration (optional — only imported when typer is available)
# ---------------------------------------------------------------------------

def get_typer_app(**kwargs) -> "Any":
    """
    Return a typer.Typer app pre-configured with CLIG-friendly settings.
    Requires `typer` to be installed.  Raises ImportError otherwise.

    Usage:
        app = get_typer_app(name="mytool", help="Does cool things.")

        @app.command()
        def main(name: str): ...
    """
    try:
        import typer
    except ImportError as exc:
        raise ImportError("Install typer: pip install typer") from exc

    defaults = dict(
        no_args_is_help=True,
        add_completion=True,
        pretty_exceptions_show_locals=False,
        context_settings={"help_option_names": ["-h", "--help"]},
    )
    defaults.update(kwargs)
    return typer.Typer(**defaults)


# ---------------------------------------------------------------------------
# Convenience: run a callable, handle exceptions, emit proper exit codes
# ---------------------------------------------------------------------------

def safe_run(fn: Callable, *args, **kwargs) -> NoReturn:
    """
    Call fn(*args, **kwargs).  Catches KeyboardInterrupt and common
    exceptions, routes them to stderr, exits with appropriate codes.
    """
    setup_signals()
    try:
        result = fn(*args, **kwargs)
        sys.exit(result if isinstance(result, int) else ExitCode.OK)
    except KeyboardInterrupt:
        print(file=sys.stderr)
        sys.exit(130)
    except BrokenPipeError:
        sys.exit(ExitCode.OK)
    except SystemExit:
        raise
    except Exception as exc:  # noqa: BLE001
        err.fatal(str(exc), code=ExitCode.GENERAL_ERROR)