87 lines
2.9 KiB
Python
87 lines
2.9 KiB
Python
from collections import deque
|
|
from logging import handlers
|
|
import logging
|
|
from pathlib import Path
|
|
import sys
|
|
import time
|
|
|
|
|
|
class ConsoleColor:
|
|
"""Simple shortcut to use colors in console"""
|
|
HEADER = '\033[95m'
|
|
BLUE = '\033[94m'
|
|
GREEN = '\033[92m'
|
|
ORANGE = '\033[93m'
|
|
RED = '\033[91m'
|
|
ENDCOLOR = '\033[0m'
|
|
BOLD = '\033[1m'
|
|
UNDERLINE = '\033[4m'
|
|
|
|
|
|
class ColoredStrFormatStyle(logging.StrFormatStyle):
|
|
def __init__(self, fmt, *, defaults=None):
|
|
self._fmt = fmt or self.default_format
|
|
self._defaults = defaults
|
|
self._color = {
|
|
logging.CRITICAL: ConsoleColor.RED,
|
|
logging.ERROR: ConsoleColor.RED,
|
|
logging.WARNING: ConsoleColor.ORANGE,
|
|
logging.INFO: ConsoleColor.GREEN,
|
|
logging.DEBUG: ConsoleColor.BLUE,
|
|
}
|
|
self._endcolor = ConsoleColor.ENDCOLOR
|
|
|
|
def _format(self, record):
|
|
return self._fmt.format(levelcolor=self._color[record.levelno], endcolor=self._endcolor,
|
|
**(self._defaults | record.__dict__ if self._defaults else record.__dict__))
|
|
|
|
|
|
class MemoryHandler(logging.Handler):
|
|
def __init__(self, capacity):
|
|
"""
|
|
Initialize the handler with the buffer size.
|
|
"""
|
|
logging.Handler.__init__(self)
|
|
self.buffer: deque[logging.LogRecord] = deque(maxlen=capacity)
|
|
self.last_update = time.time()
|
|
|
|
def emit(self, record: logging.LogRecord):
|
|
self.buffer.append(record)
|
|
self.last_update = time.time()
|
|
|
|
def close(self):
|
|
self.buffer.clear()
|
|
logging.Handler.close(self)
|
|
|
|
|
|
def create_logger(name: str, level: int, buffer_capacity: int, log_dir: Path | None = None,
|
|
stdout=False) -> tuple[logging.Logger, MemoryHandler]:
|
|
logger = logging.getLogger(name)
|
|
logger.setLevel(level)
|
|
|
|
log_formatter = logging.Formatter('{asctime} {levelname} : {message}', style='{')
|
|
buffer_handler = MemoryHandler(buffer_capacity)
|
|
buffer_handler.setFormatter(log_formatter)
|
|
buffer_handler.setLevel(level)
|
|
logger.addHandler(buffer_handler)
|
|
|
|
if log_dir is not None:
|
|
log_dir.mkdir(parents=True, exist_ok=True)
|
|
file_log_handler = handlers.RotatingFileHandler(
|
|
log_dir / f'{name}.log',
|
|
maxBytes=500000,
|
|
backupCount=5)
|
|
file_log_handler.setLevel(level)
|
|
file_log_handler.setFormatter(log_formatter)
|
|
logger.addHandler(file_log_handler)
|
|
|
|
if stdout:
|
|
terminal_log_handler = logging.StreamHandler(sys.stdout)
|
|
terminal_log_handler.setLevel(level)
|
|
fmt = '{asctime} {levelcolor}{levelname}{endcolor} : {message}'
|
|
colored_log_formatter = logging.Formatter(fmt, style='{')
|
|
colored_log_formatter._style = ColoredStrFormatStyle(fmt) # noqa: SLF001
|
|
terminal_log_handler.setFormatter(colored_log_formatter)
|
|
logger.addHandler(terminal_log_handler)
|
|
|
|
return logger, buffer_handler
|