loggy/loggy.py

import logging
import os
import pickle
import socket
import struct
import threading
from datetime import datetime

from textual.app import App, ComposeResult
from textual.containers import Vertical
from textual.widgets import Footer, Header, Input, RichLog, Checkbox, Select

from mytui import MultiSelect

# File for logging errors
LOG_FILE = os.path.join(os.path.dirname(__file__), 'error.log')
logging.basicConfig(filename=LOG_FILE, level=logging.ERROR, format="%(message)s", filemode="w")

# Mapping von Log-Level-Namen zu Farben für die Anzeige
LEVEL_STYLES = {
    "CRITICAL": "bold white on red",
    "ERROR": "bold red",
    "WARNING": "bold yellow",
    "INFO": "bold green",
    "DEBUG": "bold blue",
}


class LogReceiver(threading.Thread):
    """Ein Thread, der auf eingehende Log-Nachrichten lauscht."""

    def __init__(self, app, host="", port=19996):
        super().__init__()
        self.app = app
        self.host = host
        self.port = port
        self.daemon = True

    def run(self):
        try:
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
                s.bind((self.host, self.port))
                s.listen()
                while True:
                    conn, _ = s.accept()
                    with conn:
                        while True:
                            chunk = conn.recv(4)
                            if len(chunk) < 4:
                                break
                            slen = struct.unpack(">L", chunk)[0]
                            chunk = conn.recv(slen)
                            while len(chunk) < slen:
                                chunk = chunk + conn.recv(slen - len(chunk))

                            obj = pickle.loads(chunk)
                            log_record = logging.makeLogRecord(obj)

                            self.app.call_from_thread(self.app.add_log, log_record)
        except Exception as e:
            logging.exception("LogReceiver-Error...")
            self.app.exit(return_code=1)


class LogViewerApp(App):
    """Eine Textual-App zum Anzeigen und Filtern von Python-Logs."""

    CSS_PATH = "style.tcss"
    BINDINGS = [
        ("q", "quit", "Quit"),
        ("c", "clear_screen", "Clear"),
        ("p", "pause", "Pause")
    ]
    MAX_LOGS = 5000

    def __init__(self):
        super().__init__()
        #
        self.all_logs = []
        self.__logging_enabled__ = True
        self.__module_selection__ = MultiSelect((), prompt="Module", id="module_filter")
        self.__level_selection__ = MultiSelect((), prompt="Level", id="level_filter")
        self.force_critical = True
        self.force_error = True
        self.force_warning = False
        self.log_display = RichLog(highlight=True, markup=True)

    def compose(self) -> ComposeResult:
        """Erstellt die Widgets für die App."""
        yield Header(name="Python Log Viewer")
        with Vertical(id="app-grid"):
            yield self.log_display
        with Vertical(id="bottom-bar"):
            yield self.__module_selection__
            yield Input(placeholder="Filter", id="select_filter")
            yield self.__level_selection__
            yield Checkbox("CRITICAL", self.force_critical, id="force_critical")
            yield Checkbox("ERROR", self.force_error, id="force_error")
            yield Checkbox("WARNING", self.force_warning, id="force_warning")
        yield Footer()

    def on_mount(self) -> None:
        """Startet den Log-Empfänger-Thread."""
        log_receiver = LogReceiver(self)
        log_receiver.start()

    def add_log(self, record: logging.LogRecord) -> None:
        """Fügt einen neuen Log-Eintrag hinzu und aktualisiert die Anzeige."""
        self.__module_selection__.AddEntry(record.name)
        self.__level_selection__.AddEntry(record.levelname)
        self.all_logs.append(record)
        if len(self.all_logs) > self.MAX_LOGS:
            self.all_logs = self.all_logs[-self.MAX_LOGS:]
        self._apply_filters_to_log(record)

    def _force(self, lvl: str) -> bool:
        rv = False
        rv |= lvl == 'critical' and self.force_critical
        rv |= lvl == 'error' and self.force_error
        rv |= lvl == 'warning' and self.force_warning
        return rv

    def _apply_filters_to_log(self, record: logging.LogRecord):
        """Prüft einen einzelnen Log-Eintrag gegen die Filter und zeigt ihn ggf. an."""
        module_match = self.__module_selection__.IsSelected(record.name)
        level_match = self.__level_selection__.IsSelected(record.levelname)

        if self._force(record.levelname.lower()) or (module_match and level_match):
            level_style = LEVEL_STYLES.get(record.levelname, "white")

            timestamp_str = datetime.fromtimestamp(record.created).strftime("%Y-%m-%d %H:%M:%S")
            asctime = f"{timestamp_str},{int(record.msecs):03d}"

            message = (
                f"[[dim]{asctime}[/dim]] "
                f"[{level_style}]{record.levelname:<8}[/{level_style}] "
                f"[bold]{record.name}[/bold] - "
                f"{record.getMessage()}"
            )
            if self.__logging_enabled__:
                self.log_display.write(message)

    def action_clear_screen(self):
        self.all_logs = []
        self._update_display()

    def action_pause(self):
        self.__logging_enabled__ = not self.__logging_enabled__
        if self.__logging_enabled__:
            self._update_display()

    def _update_display(self):
        """Löscht die Anzeige und rendert alle Logs basierend auf den aktuellen Filtern neu."""
        self.log_display.clear()
        for record in self.all_logs:
            self._apply_filters_to_log(record)

    def on_checkbox_changed(self, message: Checkbox.Changed) -> None:
        if message.checkbox.id == "force_critical":
            self.force_critical = message.value
        elif message.checkbox.id == "force_error":
            self.force_error = message.value
        elif message.checkbox.id == "force_warning":
            self.force_warning = message.value

        self._update_display()

    def on_input_changed(self, message: Input.Changed) -> None:
        """Update the tui inputs and execute task, if requireed."""
        if message.input.id == "select_filter":
            self.__module_selection__.SetSelectionRegEx(message.value)

    def on_select_changed(self, message: Select.Changed) -> None:
        if message.select.id in ('module_filter', 'level_filter'):
            self._update_display()


if __name__ == "__main__":
    app = LogViewerApp()
    app.run()
    if app.return_code != 0:
        with open(LOG_FILE, "r") as fh:
            print(fh.read())