import os import logging import inspect from collections import deque from typing import Deque from io import StringIO from pathlib import Path from .paths import ROOT_DIR, LOGS_DIR class CustomLoggingFormatter(logging.Formatter): """ Custom logging formatter with ANSI colors for console output and standard format for file output. """ grey = "\x1b[0;37m" green = "\x1b[1;32m" yellow = "\x1b[1;33m" red = "\x1b[1;31m" purple = "\x1b[1;35m" blue = "\x1b[1;34m" light_blue = "\x1b[1;36m" bold_red = "\x1b[31;1m" reset = "\x1b[0m" prefix = light_blue + '%(asctime)s' + reset + ' |' colored_level = '%(levelname)-8s' message = '| %(message)s' suffix = purple + ' (%(name)s %(filename)s:%(lineno)d)' + reset FORMATS = { logging.DEBUG: prefix + grey + colored_level + reset + message + suffix, logging.INFO: prefix + blue + colored_level + reset + message + suffix, logging.WARNING: prefix + yellow + colored_level + reset + message + suffix, logging.ERROR: prefix + red + colored_level + reset + message + suffix, logging.CRITICAL: prefix + bold_red + colored_level + reset + message + suffix, } file_format = '%(asctime)s | %(levelname)-8s | %(message)s | %(name)s (%(filename)s:%(lineno)d)' def format(self, record): log_fmt = self.FORMATS.get(record.levelno, self.file_format) formatter = logging.Formatter(log_fmt) return formatter.format(record) def add_logging_console_handler(logger: logging.Logger, logging_level=logging.DEBUG): """ Add console handler with colored formatter to a logger. """ sh = logging.StreamHandler() sh.setLevel(logging_level) sh.setFormatter(CustomLoggingFormatter()) logger.addHandler(sh) def add_logging_file_handler(logger: logging.Logger, directory: str, file_name: str, logging_level=logging.DEBUG): """ Add file handler with plain formatter to a logger. """ file_path = LOGS_DIR / directory / f"{file_name}.log" file_path.parent.mkdir(parents=True, exist_ok=True) fh = logging.FileHandler(file_path, encoding="utf-8") fh.setLevel(logging_level) fh.setFormatter(logging.Formatter(CustomLoggingFormatter().file_format)) logger.addHandler(fh) class BufferHandler(logging.Handler): """ Logging handler that stores log records in a limited deque. Allows multiple listeners to be notified when a new log line is added. """ def __init__(self, maxlen: int = 200): super().__init__() self.buffer: Deque[str] = deque(maxlen=maxlen) self.formatter = CustomLoggingFormatter() self._listeners: list[callable] = [] def emit(self, record: logging.LogRecord): for listener in self._listeners: try: listener(record) except Exception as e: print(f"BufferHandler listener error: {e}") def get_logs(self): return list(self.buffer) def clear(self): self.buffer.clear() def register_listener(self, listener: callable): """ Register a listener callback that will be called with each new log line. """ if listener not in self._listeners: self._listeners.append(listener) def unregister_listener(self, listener: callable): """Remove a previously registered listener.""" if listener in self._listeners: self._listeners.remove(listener) _LOGGER_BUFFERS: dict[str, BufferHandler] = {} def add_logging_buffer_handler(logger: logging.Logger, logging_level=logging.DEBUG, maxlen: int = 300) -> BufferHandler: """ Add buffer handler with plain formatter to a logger. Returns the buffer so it can be accessed later. """ bh = BufferHandler(maxlen=maxlen) logger.addHandler(bh) _LOGGER_BUFFERS[logger.name] = bh def get_logger(name: str = "", buffer: bool = False) -> logging.Logger: """ Returns a logger instance with console, file, and optional buffer handlers. Logger name is automatically derived from caller module path. """ file_path = os.path.normpath(inspect.stack()[1].filename) head = os.path.split(file_path)[0] # Determine library name if head == os.path.normpath(ROOT_DIR): lib = "root" else: after_lib = head.replace(os.path.join(ROOT_DIR, "libs"), "") lib = after_lib.split(os.sep)[1] if len(after_lib.split(os.sep)) > 1 else "unknown" # Build logger name if name == "root": logger_name = name elif name: logger_name = f"{lib}.{name}" else: logger_name = lib file_name = name if name else lib logger = logging.getLogger(logger_name) logger.propagate = False if not logger.hasHandlers(): logger.setLevel(logging.DEBUG) add_logging_console_handler(logger=logger, logging_level=logging.DEBUG) add_logging_file_handler(logger=logger, directory=lib, file_name=file_name, logging_level=logging.DEBUG) if buffer and logger_name not in _LOGGER_BUFFERS: add_logging_buffer_handler(logger, logging_level=logging.DEBUG) return logger def get_logger_buffer(logger_name: str): """ Returns the buffer of a previously configured logger. Raises KeyError if the logger does not have a buffer handler configured or does not exist. """ if logger_name in _LOGGER_BUFFERS: return _LOGGER_BUFFERS[logger_name] raise KeyError(f"Logger '{logger_name}' does not have a buffer handler configured or does not exist.") def list_loggers() -> list[str]: """ Returns a list of all currently registered logger names. """ return sorted([ name for name, obj in logging.Logger.manager.loggerDict.items() if isinstance(obj, logging.Logger) ]) def config_root_logger() -> logging.Logger: """ Configure and return the root logger. """ return get_logger(name="root")