Textual TUI App for Log Viewer

Textual TUI App for Log Viewer

Problem Statement

Sifting through application logs during incidents is painful with tail, grep, and less — you can't easily filter, highlight, and follow a file simultaneously, and switching between tools breaks your flow. When a service is misbehaving, you need a single, interactive pane that tails a log, filters lines by severity, highlights keywords, and lets you scroll back without losing the live feed.

This script builds a Textual-based TUI that does exactly that: it streams a log file (or stdin) in real time, applies a severity filter, colorizes matches, and provides keyboard shortcuts for pausing, clearing, and exporting the current view. It's aimed at DevOps engineers who want a lightweight, dependency-minimal alternative to heavier log dashboards.

How It Works

The app uses Textual's App/Widget pattern with a worker thread that reads the log file line-by-line. A producer thread pushes new lines into a thread-safe queue.Queue; the UI polls that queue on a set_interval timer and appends formatted Rich markup to a RichLog widget. This decouples blocking I/O from the event loop, so the UI stays responsive even on busy files.

Filtering happens at two stages: a severity-level gate (parsed from common log formats like ERROR, WARN, INFO, DEBUG) and an optional substring/regex filter. Highlighting is applied by wrapping matched spans in Rich style tags. The --follow flag toggles tail-mode (seek to end, then stream); without it, the whole file is read once. Design choices favor simplicity: no external log parser library, regex-based severity detection, and a single-writer queue to avoid lock contention.

Full Script

#!/usr/bin/env python3
"""
Textual TUI App for Log Viewer

A terminal UI that tails a log file (or stdin), filters by severity level,
highlights keywords/regex matches, and lets you pause, clear, and export
the current view. Designed for incident triage on remote boxes where you
only have a shell and Python.

Usage:
    python3 log_tui.py --file /var/log/app.log --filter error --follow --highlight
"""

from __future__ import annotations

import argparse
import os
import queue
import re
import sys
import threading
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Iterable, Optional

from rich.text import Text
from textual.app import App, ComposeResult
from textual.containers import Horizontal
from textual.reactive import reactive
from textual.widgets import Footer, Header, Input, RichLog, Static

# ---------------------------------------------------------------------------
# Configuration (env vars with sensible defaults)
# ---------------------------------------------------------------------------
DEFAULT_LEVEL = os.environ.get("LOG_TUI_LEVEL", "INFO").upper()
DEFAULT_POLL_MS = int(os.environ.get("LOG_TUI_POLL_MS", "200"))   # UI drain interval
DEFAULT_TAIL_LINES = int(os.environ.get("LOG_TUI_TAIL_LINES", "0"))  # 0 = whole file
MAX_LINE_LEN = int(os.environ.get("LOG_TUI_MAX_LINE", "4000"))    # truncate huge lines

# Severity ordering — lower index = more severe
LEVELS = ["CRITICAL", "ERROR", "WARN", "WARNING", "INFO", "DEBUG", "TRACE"]
LEVEL_RANK = {name: i for i, name in enumerate(LEVELS)}

# Regex that finds a severity token anywhere in a line (common log formats).
LEVEL_RE = re.compile(
    r"\b(CRITICAL|FATAL|ERROR|WARN(?:ING)?|INFO|DEBUG|TRACE)\b",
    re.IGNORECASE,
)

# Rich styles per severity
LEVEL_STYLE = {
    "CRITICAL": "bold white on red",
    "FATAL": "bold white on red",
    "ERROR": "bold red",
    "WARN": "bold yellow",
    "WARNING": "bold yellow",
    "INFO": "cyan",
    "DEBUG": "dim",
    "TRACE": "dim italic",
}


@dataclass
class ViewerConfig:
    """Immutable-ish runtime configuration passed to the app."""
    path: Optional[Path]
    follow: bool
    min_level: str
    substring: Optional[str]
    pattern: Optional[re.Pattern[str]]
    highlight: Optional[re.Pattern[str]]
    tail_lines: int = DEFAULT_TAIL_LINES
    poll_ms: int = DEFAULT_POLL_MS
    stats: dict[str, int] = field(default_factory=dict)


def detect_level(line: str) -> Optional[str]:
    """Return the first severity token found in the line, or None."""
    m = LEVEL_RE.search(line)
    return m.group(1).upper() if m else None


def passes_level(line: str, min_level: str) -> bool:
    """True if the line has no level (assume it belongs to prior context)
    or its level is at least as severe as `min_level`."""
    lvl = detect_level(line)
    if lvl is None:
        return True  # continuation lines (stack traces) always pass
    # Normalize WARN/WARNING
    lvl = "WARN" if lvl == "WARNING" else lvl
    threshold = "WARN" if min_level == "WARNING" else min_level
    return LEVEL_RANK.get(lvl, 99) <= LEVEL_RANK.get(threshold, 99)


def style_line(line: str, cfg: ViewerConfig) -> Text:
    """Build a Rich Text object with severity + highlight styling applied."""
    if len(line) > MAX_LINE_LEN:
        line = line[:MAX_LINE_LEN] + f" …[truncated {len(line) - MAX_LINE_LEN} chars]"

    text = Text()
    lvl = detect_level(line)
    style = LEVEL_STYLE.get(lvl, "") if lvl else ""

    if cfg.highlight:
        last = 0
        for m in cfg.highlight.finditer(line):
            if m.start() > last:
                text.append(line[last:m.start()], style=style)
            text.append(m.group(0), style="bold black on yellow")
            last = m.end()
        if last < len(line):
            text.append(line[last:], style=style)
    else:
        text.append(line, style=style)
    return text


class LogReader(threading.Thread):
    """Background thread that reads a file (or stdin) and pushes lines to a queue."""

    def __init__(self, cfg: ViewerConfig, out_queue: "queue.Queue[Optional[str]]") -> None:
        super().__init__(daemon=True, name="log-reader")
        self.cfg = cfg
        self.q = out_queue
        self._stop = threading.Event()

    def stop(self) -> None:
        self._stop.set()

    def _open(self):
        if self.cfg.path is None or str(self.cfg.path) == "-":
            return sys.stdin, False
        # Open in binary-safe text mode with errors replaced so we never crash
        return open(self.cfg.path, "r", encoding="utf-8", errors="replace"), True

    def _seek_tail(self, fh) -> None:
        """Seek to the last N lines of the file (if tail_lines > 0)."""
        n = self.cfg.tail_lines
        if n <= 0:
            return
        try:
            fh.seek(0, os.SEEK_END)
            size = fh.tell()
            block = 4096
            data = b""
            while size > 0 and data.count(b"\n") <= n:
                size = max(0, size - block)
                fh.seek(size)
                data = fh.read(block) + data
            fh.seek(size + data.find(b"\n") + 1 if b"\n" in data else size)
        except OSError:
            pass

    def run(self) -> None:  # noqa: D401
        fh, should_close = self._open()
        try:
            if self.cfg.follow:
                self._seek_tail(fh)
                # If follow and no tail, jump to end so we only see new lines
                if self.cfg.tail_lines == 0 and self.cfg.path is not None:
                    fh.seek(0, os.SEEK_END)
            else:
                # One-shot read: optionally start from the last N lines
                if self.cfg.path is not None and self.cfg.tail_lines > 0:
                    self._seek_tail(fh)

            while not self._stop.is_set():
                line = fh.readline()
                if not line:
                    if self.cfg.follow:
                        time.sleep(0.25)
                        continue
                    break
                self.q.put(line.rstrip("\n"))
        except (OSError, UnicodeError) as exc:
            self.q.put(f"[log_tui] reader error: {exc}")
        finally:
            if should_close:
                try:
                    fh.close()
                except OSError:
                    pass
            self.q.put(None)  # sentinel: reader finished


class LogViewer(App):
    """Textual app: streams, filters, and highlights log lines."""

    CSS = """
    Screen { layout: vertical; }
    #status { height: 1; background: $panel; color: $text; padding: 0 1; }
    #search { dock: bottom; display: none; }
    #search.visible { display: block; }
    RichLog { border: round $primary; }
    """

    BINDINGS = [
        ("q", "quit", "Quit"),
        ("space", "toggle_pause", "Pause/Resume"),
        ("c", "clear", "Clear"),
        ("/", "focus_search", "Search"),
        ("escape", "close_search", "Close search"),
        ("e", "export", "Export"),
        ("g", "scroll_home", "Top"),
        ("G", "scroll_end", "Bottom"),
    ]

    paused: reactive[bool] = reactive(False)

    def __init__(self, cfg: ViewerConfig) -> None:
        super().__init__()
        self.cfg = cfg
        self.q: "queue.Queue[Optional[str]]" = queue.Queue(maxsize=10_000)
        self.reader: Optional[LogReader] = None
        self.lines_seen = 0
        self.lines_shown = 0
        self._search_term: Optional[re.Pattern[str]] = None

    # ------------------------------------------------------------------ UI
    def compose(self) -> ComposeResult:
        yield Header(show_clock=True)
        yield Static("", id="status")
        yield RichLog(id="log", highlight=False, markup=False, wrap=True)
        yield Input(placeholder="filter substring (Enter to apply)", id="search")
        yield Footer()

    def on_mount(self) -> None:
        self.reader = LogReader(self.cfg, self.q)
        self.reader.start()
        self.set_interval(self.cfg.poll_ms / 1000.0, self._drain_queue)
        self._update_status()

    # --------------------------------------------------------------- events
    def on_input_submitted(self, event: Input.Submitted) -> None:
        term = event.value.strip()
        self._search_term = re.compile(re.escape(term), re.IGNORECASE) if term else None
        self.query_one("#search", Input).remove_class("visible")
        self.set_focus(self.query_one("#log", RichLog))
        self._update_status()

    # -------------------------------------------------------------- actions
    def action_toggle_pause(self) -> None:
        self.paused = not self.paused
        self._update_status()

    def action_clear(self) -> None:
        self.query_one("#log", RichLog).clear()
        self.lines_shown = 0
        self._update_status()

    def action_focus_search(self) -> None:
        inp = self.query_one("#search", Input)
        inp.add_class("visible")
        self.set_focus(inp)

    def action_close_search(self) -> None:
        self.query_one("#search", Input).remove_class("visible")
        self.set_focus(self.query_one("#log", RichLog))

    def action_scroll_home(self) -> None:
        self.query_one("#log", RichLog).scroll_home(animate=False)

    def action_scroll_end(self) -> None:
        self.query_one("#log", RichLog).scroll_end(animate=False)

    def action_export(self) -> None:
        """Dump the visible widget contents to a timestamped file."""
        out = Path.cwd() / f"log_export_{int(time.time())}.txt"
        try:
            log = self.query_one("#log", RichLog)
            # RichLog doesn't expose raw text; re-render is impractical, so we
            # snapshot what we can via Rich's console capture.
            from rich.console import Console
            console = Console(record=True, width=200, file=open(os.devnull, "w"))
            console.print(log.renderable if hasattr(log, "renderable") else "")
            out.write_text(console.export_text(), encoding="utf-8")
            self.notify(f"Exported to {out.name}")
        except Exception as exc:  # noqa: BLE001 — surface any export failure
            self.notify(f"Export failed: {exc}", severity="error")

    # --------------------------------------------------------------- internals
    def _drain_queue(self) -> None:
        if self.paused:
            return
        drained = 0
        while drained < 500:  # cap work per tick to keep UI snappy
            try:
                line = self.q.get_nowait()
            except queue.Empty:
                break
            drained += 1
            if line is None:
                self._update_status(finished=True)
                continue
            self.lines_seen += 1
            if not passes_level(line, self.cfg.min_level):
                continue
            if self.cfg.substring and self.cfg.substring.lower() not in line.lower():
                continue
            if self._search_term and not self._search_term.search(line):
                continue
            self.query_one("#log", RichLog).write(style_line(line, self.cfg))
            self.lines_shown += 1
        if drained:
            self._update_status()

    def _update_status(self, finished: bool = False) -> None:
        state = "PAUSED" if self.paused else ("DONE" if finished else "LIVE")
        src = str(self.cfg.path) if self.cfg.path else "<stdin>"
        term = self._search_term.pattern if self._search_term else "-"
        self.query_one("#status", Static).update(
            f" {state} | {src} | level≥{self.cfg.min_level} | "
            f"search={term} | seen={self.lines_seen} shown={self.lines_shown} "
            f"| [space]pause [/]search [c]lear [e]xport [q]uit"
        )

    def on_unmount(self) -> None:
        if self.reader:
            self.reader.stop()


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def parse_args(argv: Optional[Iterable[str]] = None) -> argparse.Namespace:
    p = argparse.ArgumentParser(
        prog="log_tui",
        description="Interactive Textual log viewer with filtering and highlighting.",
    )
    p.add_argument("-f", "--file", default="-",
                   help="Log file to read ('-' for stdin). Default: stdin")
    p.add_argument("--follow", action="store_true",
                   help="Tail the file (like tail -f). Ignored for stdin.")
    p.add_argument("--filter", dest="substring", default=None,
                   help="Only show lines containing this substring (case-insensitive).")
    p.add_argument("--regex", default=None,
                   help="Only show lines matching this regex.")
    p.add_argument("--highlight", default=None,
                   help="Regex of terms to highlight in yellow.")
    p.add_argument("--level", default=DEFAULT_LEVEL,
                   choices=["CRITICAL", "ERROR", "WARN", "WARNING", "INFO", "DEBUG", "TRACE"],
                   help=f"Minimum severity to display. Default: {DEFAULT_LEVEL}")
    p.add_argument("--tail", type=int, default=DEFAULT_TAIL_LINES,
                   help="Start from last N lines (0 = whole file). Default: 0")
    p.add_argument("--poll-ms", type=int, default=DEFAULT_POLL_MS,
                   help=f"UI drain interval in ms. Default: {DEFAULT_POLL_MS}")
    return p.parse_args(argv)


def build_config(args: argparse.Namespace) -> ViewerConfig:
    path = None if args.file in ("-", "") else Path(args.file).expanduser()
    if path is not None and not path.exists():
        raise FileNotFoundError(f"Log file not found: {path}")
    if path is not None and not os.access(path, os.R_OK):
        raise PermissionError(f"Log file not readable: {path}")

    try:
        pattern = re.compile(args.regex) if args.regex else None
        highlight = re.compile(args.highlight, re.IGNORECASE) if args.highlight else None
    except re.error as exc:
        raise ValueError(f"Invalid regex: {exc}") from exc

    return ViewerConfig(
        path=path,
        follow=args.follow,
        min_level=args.level.upper(),
        substring=args.substring,
        pattern=pattern,
        highlight=highlight,
        tail_lines=max(0, args.tail),
        poll_ms=max(50, args.poll_ms),
    )


def main(argv: Optional[Iterable[str]] = None) -> int:
    try:
        args = parse_args(argv)
        cfg = build_config(args)
    except (FileNotFoundError, PermissionError, ValueError) as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2

    # If stdin is a TTY and no file was given, there's nothing to read.
    if cfg.path is None and sys.stdin.isatty():
        print("error: no --file given and stdin is a TTY", file=sys.stderr)
        return 2

    try:
        LogViewer(cfg).run()
    except KeyboardInterrupt:
        return 130
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

Configuration

CLI flags: --file/-f, --follow, --filter, --regex, --highlight, --level, --tail, --poll-ms.

Usage Examples

1. Tail a live app log, only show ERROR and above, highlight stack traces:

python3 log_tui.py --file /var/log/app.log --follow --level ERROR \
    --highlight "Traceback|Exception|timeout"

Expected: a full-screen TUI with a header, a status bar reading LIVE | /var/log/app.log | level≥ERROR | ..., and only error lines appearing as they're written. Matches for Traceback, Exception, timeout are drawn with a yellow background.

2. Pipe a one-shot file through the viewer, filtering by substring:

python3 log_tui.py --file ./access.log --filter "500 " --tail 5000

Expected: the last 5000 lines are scanned; only lines containing 500 are displayed. The status bar shows DONE once the reader finishes, and you can scroll with arrow keys / PageUp / PageDown.

3. Read from a pipe (no --file):

journalctl -u nginx --since "10 min ago" | python3 log_tui.py --level WARN --highlight "refused|denied"

Expected: journal lines stream in, WARN and above are shown, and refused/denied are highlighted. Press space to pause, / to add a live search term, e to export, q to quit.

Error Handling

• Missing or unreadable file — build_config raises FileNotFoundError/PermissionError, caught in main, printed to stderr, exit code 2.
• Invalid regex — re.error is wrapped in ValueError with the original message, exit code 2.
• No file and stdin is a TTY — refuses to start rather than hanging on a terminal read.
• I/O errors mid-stream (log rotation, disk unplug) — the LogReader catches OSError/UnicodeError, pushes a [log_tui] reader error: … line into the queue, and sends the sentinel so the UI flips to DONE.
• Encoding garbage — file is opened with errors="replace" so binary noise becomes � instead of crashing.
• Queue backpressure — the queue is bounded at 10 000 entries; if the UI can't keep up, the reader blocks on put() rather than eating memory.
• Ctrl+C — caught in main, returns exit code 130 after the reader thread is stopped via on_unmount.

Extending Ideas

• Add a level filter widget: replace --level with a Select widget in the header so users can change severity at runtime without restarting.
• Persist filters and highlights: write the current ViewerConfig to ~/.config/log_tui.json and load it on startup for per-project defaults.
• Multi-file / multi-pane mode: run one LogReader per file and split the screen into TabbedContent panes so you can compare stdout and stderr logs side-by-side.
• Structured log support: detect JSON lines (line.lstrip().startswith("{")) and render a compact column view (timestamp, level, message) using DataTable instead of raw text.
• Remote tailing: swap the file reader for an SSH paramiko channel or a docker logs -f subprocess so the same TUI works against remote hosts and containers.