Log Viewer¶

def display_logs(
    self,
    *,
    limit: int = 200,
    level: Optional[Union[int, str]] = None,
    keyword: Optional[str] = None,
) -> bool:
    """Display log entries in a formatted Rich table.

    Parameters
    ----------
    limit : int, optional
        Maximum number of lines to display from the end. Default is 200.
    level : int or str or None, optional
        Minimum log level to display (e.g., "INFO" or 20). Default is None
        (show all levels).
    keyword : str or None, optional
        Only show lines containing this keyword (case-insensitive).
        Default is None (no filtering).

    Returns
    -------
    bool
        True if logs were rendered successfully, False if file not found.
    """
    if not self.log_file.exists():
        self.console.print(
            Panel(
                f"[bold]{self.log_file}[/bold]",
                title="[bold red] File Not Found [/bold red]",
                border_style="red",
                box=box.ROUNDED,
                expand=False,
            )
        )
        return False
    table = self._build_table(limit, level, keyword)
    self.console.print(table)
    if table.row_count == 0:
        self.console.print(
            "[yellow]No log entries matched the current filters.[/yellow]"
        )
    return True

def follow_logs(
    self,
    *,
    level: Optional[Union[int, str]] = None,
    keyword: Optional[str] = None,
    interval: float = 0.5,
    max_lines: int = 500,
) -> bool:
    """Live-follow the log file, displaying new lines as they are written.

    Similar to ``tail -f``, this method continuously monitors the log file
    and displays new entries in a Rich live-updating table.

    Parameters
    ----------
    level : int or str or None, optional
        Minimum log level to display. Default is None (show all levels).
    keyword : str or None, optional
        Only show lines containing this keyword (case-insensitive).
        Default is None (no filtering).
    interval : float, optional
        Refresh interval in seconds. Default is 0.5.
    max_lines : int, optional
        Maximum lines to keep in the rolling display window. Default is 500.

    Returns
    -------
    bool
        True if following completed (via Ctrl+C), False if file not found.

    Notes
    -----
    - Starts at end of file (only shows new lines).
    - Automatically handles log rotation (file truncation).
    - Press Ctrl+C to stop following.
    """
    if not self.log_file.exists():
        self.console.print(
            Panel(
                f"[bold]{self.log_file}[/bold]",
                title="[bold red] File Not Found [/bold red]",
                border_style="red",
                box=box.ROUNDED,
                expand=False,
            )
        )
        return False

    # Rolling buffer of *raw* lines to feed batch renderer
    window: deque[str] = deque(maxlen=max_lines)

    # Start with a headers-only table; show a "waiting" placeholder caption
    table = self._build_table_from_lines([], level, keyword)
    table.caption = "[dim]Waiting for new log entries…[/dim]"

    # Configure Live (only override refresh rate if interval > 0)
    live_kwargs = {"console": self.console, "transient": True}
    if interval > 0:
        live_kwargs["refresh_per_second"] = max(1, int(1.0 / interval))

    try:
        with self.log_file.open("r", encoding="utf-8", errors="replace") as f:
            # Start at EOF — only new lines will be shown
            f.seek(0, os.SEEK_END)
            try:
                last_size = os.fstat(f.fileno()).st_size
            except OSError:
                last_size = 0  # type: ignore[arg-type]

            with Live(table, **live_kwargs) as live:  # type: ignore[arg-type]
                while True:
                    try:
                        cur_size = os.fstat(f.fileno()).st_size
                    except OSError:
                        cur_size = 0
                    if cur_size < last_size:
                        f.seek(0, os.SEEK_SET)
                        last_size = cur_size

                    pos = f.tell()
                    line = f.readline()
                    if not line:
                        f.seek(pos)
                        last_size = cur_size
                        if interval > 0:
                            time.sleep(interval)
                        continue

                    window.append(line)
                    new_table = self._build_table_from_lines(
                        list(window), level, keyword
                    )
                    live.update(new_table)
                    table = new_table
                    last_size = cur_size
    except KeyboardInterrupt:
        return True

    return True