Textual RichLog 串流檢視器：有界緩衝、暫停追蹤與批次刷新

uv init pypy-log-viewer
cd pypy-log-viewer
uv add 'textual==8.2.8' 'rich==15.0.0'
uv add --dev pytest pytest-asyncio

from collections import deque
from threading import Lock
import re
import unicodedata

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

PENDING = 2000
HISTORY = 1000
VISIBLE = 500
BATCH = 100
MAX_RAW = 4096
SECRET = re.compile(r"(?i)\b(token|password)=\S+")


def clean(message: str) -> str:
    message = SECRET.sub(r"\1=[REDACTED]", message)
    message = message.replace("\n", " ↵ ").replace("\r", " ")
    return "".join(c for c in message
                   if not unicodedata.category(c).startswith("C"))[:240]


class Inbox:
    def __init__(self, capacity: int = PENDING):
        if capacity < 1:
            raise ValueError("capacity must be positive")
        self.capacity = capacity
        self.rows: deque[str] = deque()
        self.lock = Lock()
        self.dropped = 0
        self.rejected = 0

    def offer(self, message: str) -> None:
        # Only strings enter this boundary; reject oversized input before regex.
        if not isinstance(message, str) or len(message) > MAX_RAW:
            with self.lock:
                self.rejected += 1
            return
        row = clean(message)
        with self.lock:
            if len(self.rows) == self.capacity:
                self.rows.popleft()  # Keep the newest events, count the loss.
                self.dropped += 1
            self.rows.append(row)

    def drain(self, limit: int = BATCH):
        with self.lock:
            rows = [self.rows.popleft()
                    for _ in range(min(limit, len(self.rows)))]
            return rows, len(self.rows), self.dropped, self.rejected


class Viewer(App):
    CSS = """
    #stats { height: 2; }
    Input { height: 3; }
    RichLog { height: 1fr; border: solid $accent; }
    """
    BINDINGS = [
        ("ctrl+p", "pause_view", "暫停畫面"),
        ("ctrl+f", "follow", "追蹤尾端"),
        ("ctrl+q", "quit", "離開"),
    ]
    ENABLE_COMMAND_PALETTE = False

    def __init__(self, inbox: Inbox | None = None, demo: bool = True):
        super().__init__()
        self.inbox = inbox if inbox is not None else Inbox()
        self.history: deque[str] = deque(maxlen=HISTORY)
        self.retired = 0
        self.paused = False
        self.following = True
        self.dirty = True
        self.needle = ""
        self.demo = demo
        self.sequence = 0

    def compose(self) -> ComposeResult:
        yield Static(id="stats", markup=False)
        yield Input(placeholder="字面子字串過濾（不分大小寫）", max_length=80)
        yield RichLog(id="log", max_lines=VISIBLE, min_width=1,
                      markup=False, highlight=False, wrap=False,
                      auto_scroll=False)
        yield Footer()

    def on_mount(self) -> None:
        self.set_interval(0.1, self.flush)
        if self.demo:
            self.set_interval(0.1, self.produce)

    def produce(self) -> None:
        for _ in range(30):
            self.sequence += 1
            level = "ERROR" if self.sequence % 17 == 0 else "INFO"
            self.inbox.offer(f"{self.sequence:06d} {level} 拍拍君 task heartbeat")

    def on_input_changed(self, event: Input.Changed) -> None:
        self.needle = event.value.casefold()
        self.dirty = True

    def action_pause_view(self) -> None:
        self.paused = not self.paused
        if not self.paused:
            self.dirty = True

    def action_follow(self) -> None:
        self.following = not self.following
        if self.following and not self.paused:
            self.query_one(RichLog).scroll_end(animate=False)

    def flush(self) -> None:
        rows, pending, dropped, rejected = self.inbox.drain()
        self.retired += max(0, len(self.history) + len(rows) - HISTORY)
        self.history.extend(rows)
        state = "PAUSED" if self.paused else "LIVE"
        self.query_one("#stats", Static).update(
            f"{state} follow={self.following} pending={pending} "
            f"dropped={dropped} rejected={rejected}\n"
            f"history={len(self.history)}/{HISTORY} retired={self.retired}"
        )
        log = self.query_one(RichLog)
        if self.paused or not log.size.width or not log.size.height:
            self.dirty = True
            return
        source = self.history if self.dirty else rows
        selected = [row for row in source if self.needle in row.casefold()]
        if self.dirty:
            log.clear()
        if selected:
            log.write(Text("\n".join(selected[-VISIBLE:])),
                      scroll_end=self.following, animate=False)
        self.dirty = False


if __name__ == "__main__":
    Viewer().run()

uv run python viewer.py

import pytest
from textual.widgets import Input, RichLog
from viewer import Inbox, Viewer, clean


def text(log):
    return "\n".join(line.text for line in log.lines)


def test_overflow_and_admission():
    inbox = Inbox(capacity=2)
    for row in ["old", "middle", "new"]:
        inbox.offer(row)
    inbox.offer("x" * 4097)
    rows, pending, dropped, rejected = inbox.drain()
    assert (rows, pending, dropped, rejected) == (["middle", "new"], 0, 1, 1)
    assert clean("token=abc\n[red]hello\x1b") == "token=[REDACTED] ↵ [red]hello"


@pytest.mark.asyncio
async def test_freeze_filter_and_resume():
    app = Viewer(demo=False)
    async with app.run_test(size=(90, 25)) as pilot:
        await pilot.pause()
        app.inbox.offer("INFO [red]safe token=abc")
        app.flush()
        await pilot.pause()
        log = app.query_one(RichLog)
        frozen = text(log)
        assert "[red]safe token=[REDACTED]" in frozen
        await pilot.press("ctrl+p")
        app.inbox.offer("ERROR newest")
        app.flush()
        assert text(log) == frozen
        assert app.history[-1] == "ERROR newest"
        app.query_one(Input).value = "error"
        await pilot.pause()
        assert text(log) == frozen
        await pilot.press("ctrl+p")
        app.flush()
        await pilot.pause()
        assert text(log) == "ERROR newest"
        await pilot.press("ctrl+f")
        assert app.following is False

uv run pytest -q