Files
rich-print/src/rp/cli.py
Yunxiao Xu 0f93d3a095
All checks were successful
CI / lint (push) Successful in -26s
CI / typecheck (push) Successful in -23s
CI / test (3.12) (push) Successful in -25s
CI / test (3.13) (push) Successful in -24s
CI / test (3.14) (push) Successful in -24s
CI / lint (pull_request) Successful in -28s
CI / test (3.14) (pull_request) Successful in -24s
CI / typecheck (pull_request) Successful in -24s
CI / test (3.12) (pull_request) Successful in -25s
CI / test (3.13) (pull_request) Successful in -24s
Add watch mode for live file monitoring
Add --watch flag that polls a file and redraws the terminal when its
content changes. Useful for monitoring log files, configuration files,
or any file that changes over time.

Features:
- Polls file every 1 second for changes
- Disables pager (pager would block the watch loop)
- File-only: rejects stdin or - input
- Error recovery: retries when file is missing or temporarily unreadable
- State comparison: only redraws when content actually changes
2026-04-04 05:23:28 -07:00

267 lines
7.5 KiB
Python

"""CLI entry point for rp."""
from __future__ import annotations
from collections.abc import Callable
import sys
import time
from pathlib import Path
from typing import Annotated
import typer
from rich.console import Console
from rp import __author__, __license__, __version__
from rp.debug import DebugLogger, env_debug_enabled, log_debug, make_debug_logger
from rp.detect import detect_type
from rp.render import RenderOptions, render
_WATCH_POLL_INTERVAL = 1.0
app = typer.Typer(
name="rp",
help="Rich print files — markdown, JSON, code, and more — beautifully in your terminal.",
no_args_is_help=False,
rich_markup_mode="rich",
)
def version_callback(value: bool) -> None:
"""Print version information and exit.
Args:
value: Flag value from typer callback.
"""
if value:
console = Console()
console.print(f"Rich Print (rp) {__version__}")
console.print(f"Copyright (c) 2026 {__author__}")
console.print(f"License: {__license__}")
raise typer.Exit()
def _read_file_content(file: Path) -> str:
"""Read a UTF-8 text file for rendering."""
if not file.exists():
raise FileNotFoundError(f"File not found: {file}")
if file.is_dir():
raise IsADirectoryError(f"Is a directory: {file}")
try:
return file.read_text(encoding="utf-8")
except UnicodeDecodeError as exc:
raise ValueError(f"Cannot read binary file: {file}") from exc
def _render_content(
*,
console: Console,
debug: DebugLogger | None,
content: str,
file_path: str | None,
file_type: str | None,
guess_content: bool,
theme: str,
line_numbers: bool | None,
pager: bool | None,
) -> None:
"""Detect type and render content with the standard pipeline."""
detected_type = detect_type(
path=file_path,
content=content,
explicit_type=file_type,
guess_content=guess_content,
debug=debug,
)
options = RenderOptions(
theme=theme,
line_numbers=line_numbers,
pager=pager,
debug=debug,
)
render(content, detected_type, console, options)
def _watch_file(
*,
file: Path,
console: Console,
debug: DebugLogger | None,
file_type: str | None,
guess_content: bool,
theme: str,
line_numbers: bool | None,
sleep: Callable[[float], None] | None = None,
) -> None:
"""Poll a file and redraw when its visible state changes."""
if sleep is None:
sleep = time.sleep
previous_state: tuple[str, str] | None = None
while True:
try:
content = _read_file_content(file)
except (
FileNotFoundError,
IsADirectoryError,
PermissionError,
OSError,
ValueError,
) as exc:
state = ("error", str(exc))
if state != previous_state:
log_debug(debug, f"watch: detected error state {state[1]!r}")
console.clear(home=True)
console.print(f"[red]Error:[/red] {state[1]}")
previous_state = state
else:
state = ("content", content)
if state != previous_state:
log_debug(debug, f"watch: rendering update for {str(file)!r}")
console.clear(home=True)
_render_content(
console=console,
debug=debug,
content=content,
file_path=str(file),
file_type=file_type,
guess_content=guess_content,
theme=theme,
line_numbers=line_numbers,
pager=False,
)
previous_state = state
sleep(_WATCH_POLL_INTERVAL)
@app.command()
def main(
file: Annotated[
Path | None,
typer.Argument(help="File to rich-print. Use '-' or omit to read from stdin."),
] = None,
file_type: Annotated[
str | None,
typer.Option(
"--type", "-t", help="Force file type (e.g. json, python, markdown)."
),
] = None,
theme: Annotated[
str,
typer.Option("--theme", help="Pygments color theme."),
] = "monokai",
line_numbers: Annotated[
bool | None,
typer.Option(
"--line-numbers/--no-line-numbers",
help="Show line numbers. Default: auto.",
),
] = None,
pager: Annotated[
bool | None,
typer.Option(
"--pager/--no-pager",
help="Use pager for output. Default: auto-detect.",
),
] = None,
watch: Annotated[
bool,
typer.Option(
"--watch/--no-watch",
help="Watch a file for changes and redraw when it updates.",
),
] = False,
guess_content: Annotated[
bool,
typer.Option(
"--guess-content/--no-guess-content",
help="Use heuristic content-based type detection as a fallback.",
),
] = False,
version: Annotated[
bool | None,
typer.Option(
"--version",
"-v",
help="Show version.",
callback=version_callback,
is_eager=True,
),
] = None,
) -> None:
"""Render a file or stdin input in the terminal.
Args:
file: File to rich-print. Use '-' or omit to read from stdin.
file_type: Force file type (e.g. json, python, markdown).
theme: Pygments color theme.
line_numbers: Show line numbers. Default: auto.
pager: Use pager for output. Default: auto-detect.
watch: Watch a file for changes and redraw when it updates.
guess_content: Use heuristic content-based type detection as a fallback.
version: Show version and exit.
"""
console = Console()
debug = make_debug_logger(Console(stderr=True), env_debug_enabled())
if watch and (file is None or str(file) == "-"):
console.print("[red]Error:[/red] --watch only supports file input.")
raise typer.Exit(code=1)
if file is None or str(file) == "-":
log_debug(debug, "input: reading from stdin")
if sys.stdin.isatty():
console.print("[dim]Reading from stdin. Press Ctrl+D to end.[/dim]")
content = sys.stdin.read()
file_path = None
else:
log_debug(debug, f"input: reading file {str(file)!r}")
if watch:
log_debug(debug, f"watch: polling {str(file)!r}")
_watch_file(
file=file,
console=console,
debug=debug,
file_type=file_type,
guess_content=guess_content,
theme=theme,
line_numbers=line_numbers,
)
raise typer.Exit()
try:
content = _read_file_content(file)
except (
FileNotFoundError,
IsADirectoryError,
PermissionError,
OSError,
ValueError,
) as e:
console.print(f"[red]Error:[/red] {e}")
raise typer.Exit(code=1)
file_path = str(file)
if not content:
console.print("[dim]Empty input, nothing to display.[/dim]")
raise typer.Exit()
_render_content(
console=console,
debug=debug,
content=content,
file_path=file_path,
file_type=file_type,
guess_content=guess_content,
theme=theme,
line_numbers=line_numbers,
pager=pager,
)
if __name__ == "__main__":
app()