import logging import pickle import re 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 # 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.basicConfig(filename="server_error.log", level=logging.DEBUG) logging.error(f"LogReceiver-Fehler: {e}", exc_info=True) class OptionSelectList(dict): def __init__(self): super().__init__(self) # self.__default_value__ = True self.__selection_regex__ = "" def __sorted_keys__(self): rv = list(self.keys()) rv.sort() return rv def SetSelectionRegEx(self, regex: str) -> None: self.__selection_regex__ = regex def AddEntry(self, entry) -> None: if entry not in self: self[entry] = self.__default_value__ def Toggle(self, entry_num): if entry_num < 0: self.__default_value__ = entry_num == -1 for key in self: self[key] = self.__default_value__ elif entry_num > len(self): raise ValueError(f"The Entry '{entry} is not in the list") else: entry = self.__sorted_keys__()[entry_num] self[entry] = not self[entry] def GetSelectList(self): rv = [] if len(self) > 2: rv.append(('All', -1)) rv.append(('None', -2)) for index, key in enumerate(self.__sorted_keys__()): try: match = len(re.findall(self.__selection_regex__, key)) > 0 except re.error: match = True # No valid regular expression if match: prefix = "\\[X] " if self[key] else "\\[-] " rv.append((prefix + key, index)) return rv def IsSelected(self, entry): return self.get(entry, False) class MySelect(Select): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.__new_options__ = [] def set_options(self, options): self.__new_options__ = options def _on_enter(self, event): super().set_options(self.__new_options__) return super()._on_enter(event) 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") ] def __init__(self): super().__init__() # self.all_logs = [] self.__module_select_list__ = OptionSelectList() self.__module_selection__ = MySelect((), prompt="Module", id="module_filter") self.__level_select_list__ = OptionSelectList() self.__level_selection__ = MySelect((), 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_select_list__.AddEntry(record.name) self.__module_selection__.set_options(self.__module_select_list__.GetSelectList()) self.__level_select_list__.AddEntry(record.levelname) self.__level_selection__.set_options(self.__level_select_list__.GetSelectList()) self.all_logs.append(record) 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_select_list__.IsSelected(record.name) level_match = self.__level_select_list__.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()}" ) self.log_display.write(message) def action_clear_screen(self): self.all_logs = [] 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_select_list__.SetSelectionRegEx(message.value) self.__module_selection__.set_options(self.__module_select_list__.GetSelectList()) self.__module_selection__.prompt = "Full" if not message.value else "Filtered" def on_select_changed(self, message: Select.Changed) -> None: if message.select.id == 'module_filter': if type(message.value) is int: self.__module_select_list__.Toggle(message.value) self.__module_selection__.clear() self.__module_selection__.set_options(self.__module_select_list__.GetSelectList()) elif message.select.id == 'level_filter': if type(message.value) is int: self.__level_select_list__.Toggle(message.value) self.__level_selection__.clear() self.__level_selection__.set_options(self.__level_select_list__.GetSelectList()) self._update_display() if __name__ == "__main__": app = LogViewerApp() app.run()