dashboard/app/journal.py

import json
import re
from datetime import UTC, datetime
from typing import Any, cast

from app.utils.subprocess import run_async

CURSOR_RE = re.compile(r"^[A-Za-z0-9;:=+./_-]+$")
LEVELS = {"all": None, "warn": "warning", "err": "err"}


def parse_lines(text: str) -> list[dict[str, Any]]:
    """Parse `journalctl -o json` output into entry dicts.

    Each non-empty line is one JSON object. Multi-line messages are
    embedded as \\n escapes and control characters (e.g. NUL) are
    JSON-escaped, so no continuation-line handling is needed — the
    former -o export format required both.

    Args:
        text: raw `journalctl -o json` output.

    Returns:
        One dict per entry; lines that are not valid JSON objects are
        skipped.
    """
    entries: list[dict[str, Any]] = []
    for line in text.splitlines():
        line = line.strip()
        if not line:
            continue
        try:
            e = json.loads(line)
        except ValueError:
            continue
        if isinstance(e, dict):
            entries.append(cast("dict[str, Any]", e))
    return entries


def format_entries(entries: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """Reduce raw export entries to the fields the journal tab renders.

    Entries without a realtime timestamp are dropped. The local time is
    formatted as HH:MM:SS (invalid timestamps render as an empty string),
    PRIORITY defaults to 6 (info), and the identifier falls back
    SYSLOG_IDENTIFIER -> _COMM -> _PID.

    Args:
        entries: dicts from parse_lines.

    Returns:
        One row per kept entry with stamp, prio, ident, msg, cursor.
    """
    out: list[dict[str, Any]] = []
    for e in entries:
        ts = e.get("__REALTIME_TIMESTAMP")
        if ts is None:
            continue
        stamp = ""
        try:
            dt = datetime.fromtimestamp(int(ts) / 1e6, tz=UTC).astimezone()
            stamp = dt.strftime("%H:%M:%S")
        except (ValueError, OSError, TypeError):
            pass
        try:
            prio = int(e.get("PRIORITY", "6"))
        except ValueError:
            prio = 6
        msg = e.get("MESSAGE", "").rstrip("\n")
        out.append(
            {
                "stamp": stamp,
                "prio": prio,
                "ident": e.get("SYSLOG_IDENTIFIER") or e.get("_COMM") or e.get("_PID", "?"),
                "msg": msg,
                "cursor": e.get("__CURSOR", ""),
            }
        )
    return out


async def _journalctl(argv: list[str]) -> str:
    """Run a journalctl subprocess and return its stdout.

    Args:
        argv: full command, e.g. ["sudo", "journalctl", "-o", "json", "-n", "100"].

    Returns:
        The decoded stdout.

    Raises:
        RuntimeError: if journalctl exits non-zero (or cannot be
            spawned); the message is its stderr (or "journalctl failed"
            when stderr is empty).
    """
    rc, out, err = await run_async(argv)
    if rc != 0:
        raise RuntimeError(err.strip() or "journalctl failed")
    return out


async def tail(
    cursor: str | None,
    level: str,
    unit: str | None,
    search: str | None,
    lines: int,
    hide_sudo: bool = False,
) -> tuple[list[dict[str, Any]], str | None]:
    """Fetch a recent journal page, newest entries last.

    Runs `sudo journalctl -o json` with the requested filters. A
    non-empty cursor is validated against CURSOR_RE before being passed
    as --after-cursor (invalid cursors are silently ignored); level maps
    through LEVELS, the unit name is regex-checked, and the free-text
    search is truncated to 200 chars. When hide_sudo is set, twice as many
    lines are fetched (journalctl cannot express negated matches) and
    sudo's own entries are filtered out in Python afterwards.

    Args:
        cursor: opaque journal cursor to continue after, or None.
        level: one of "all" / "warn" / "err".
        unit: systemd unit to filter on, or None.
        search: free-text match, or None.
        lines: target number of entries.
        hide_sudo: drop entries logged by sudo itself.

    Returns:
        (formatted rows from format_entries, cursor of the newest row or
        None when nothing was returned).

    Raises:
        RuntimeError: if journalctl fails (see _journalctl).
    """
    fetch = lines * 2 if hide_sudo else lines
    args = ["--no-pager", "-o", "json", "-n", str(min(max(fetch, 1), 500))]
    lvl = LEVELS.get(level)
    if lvl:
        args += ["-p", lvl]
    if unit and re.match(r"^[A-Za-z0-9@:_.\-+]+\.\w+$", unit):
        args += ["-u", unit]
    if search:
        args += [search[:200]]
    if cursor and CURSOR_RE.match(cursor):
        args += ["--after-cursor", cursor]
    text = await _journalctl(["sudo", "journalctl"] + args)

    entries = parse_lines(text)
    if hide_sudo:
        entries = [e for e in entries if e.get("SYSLOG_IDENTIFIER") != "sudo"]
    entries = format_entries(entries)
    last_cursor = entries[-1]["cursor"] if entries else None
    return entries, last_cursor