nosys_libs/app/common/logging.py

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")