Processes and formats ECS logs from stdin using rich

pretty_ecs_logs.py

"""This script processes and formats ECS logs from stdin using rich"""

# /// script
# dependencies = ["rich"]
# ///

import json
import sys
import re
from datetime import datetime
from typing import Any, Dict

from rich.console import Console, Group
from rich.panel import Panel
from rich.rule import Rule
from rich.syntax import Syntax
from rich.theme import Theme

console = Console(theme=Theme({"info": "cyan", "warning": "yellow", "error": "red"}))


def _get_log_level(data: Dict[str, Any]) -> str:
    if "error" in data and isinstance(data["error"], dict):
        return "error"

    return (data.get("log", {}).get("level") or data.get("level") or "info").lower()


def _format_log_entry(data: Dict[str, Any]) -> Panel:
    log_level = _get_log_level(data)

    if "error" in data and isinstance(data["error"], dict):
        return _format_error_log(data)

    return _create_log_panel(_get_condensed_log(data), level=log_level)


def _format_error_log(data: Dict[str, Any]) -> Panel:
    error = data["error"]
    content = Group(
        _get_condensed_log(data),
        Rule(error.get("type", "Error"), style="red dim"),
        f"[red bold]{error.get('message', 'An error occurred')}[/red bold]",
        Syntax(
            error.get("stack_trace", ""),
            "python",
            theme="monokai",
            word_wrap=True,
            background_color="default",
        ),
    )
    return _create_log_panel(content, level="error")


def _create_log_panel(content, level="info"):
    return Panel(
        content,
        border_style=level,
        padding=(1, 2),
        title=level.upper(),
    )


def _format_timestamp(timestamp_str):
    try:
        dt = datetime.fromisoformat(timestamp_str.replace("Z", "+00:00"))
        local_dt = dt.astimezone()  # Convert to local timezone
        return local_dt.strftime("%H:%M:%S.%f")[:-3]
    except (ValueError, AttributeError):
        return timestamp_str


def _highlight_http_pattern(message: str) -> str:
    pattern = r"(GET|POST|PUT|DELETE|PATCH|HEAD|OPTIONS|TRACE|CONNECT) ([/\w\-._~:/?#\[\]@!$&\'()*+,;=%%]+)"
    return re.sub(
        pattern,
        r"[bold][on white]\1[/on white] \2[/bold]",
        message,
    )


def _get_condensed_log(data):
    # Extract relevant fields for condensed view
    timestamp = _format_timestamp(data.get("@timestamp", ""))
    level = _get_log_level(data)
    message = _highlight_http_pattern(data.get("message", ""))

    return Group(f"[dim]{timestamp}[/dim] [{level}]{message}[/{level}]")


def _create_compact_panel(content):
    return Panel(
        f"[dim]{content}[/dim]",
        border_style="dim",
        padding=(0, 1),
    )


def main() -> None:
    """Process and format logs from stdin."""
    for line in sys.stdin:
        if not line.strip():
            continue

        try:
            data = json.loads(line.strip())
            console.print(_format_log_entry(data))
        except (json.JSONDecodeError, TypeError):
            console.print(_create_compact_panel(line.strip()))


if __name__ == "__main__":
    main()