from enum import Enum, auto from logging import Formatter, StreamHandler, Handler from logging.handlers import TimedRotatingFileHandler from pathlib import Path from typing import List class NamingCache: def __init__(self): self._cache = {} def __getitem__(self, key): if key not in self._cache: self._cache[key] = self.shorten(key) return self._cache[key] @staticmethod def shorten(logger_name: str) -> str: target_length = 18 if len(logger_name) > target_length: parts = logger_name.split('.') part = 0 while len(logger_name) > target_length: if part == len(parts) - 1: logger_name = f'...{logger_name[-(target_length - 3):]}' break parts[part] = parts[part][0] logger_name = '.'.join(parts) part += 1 return logger_name.ljust(target_length) class ApplicationFormatter(Formatter): def __init__(self, handler_prefix: str = ''): super().__init__() self._logger_names = NamingCache() self._handler_prefix = handler_prefix def format(self, record): from datetime import datetime timestamp = datetime.fromtimestamp(record.created).isoformat(sep=' ', timespec='milliseconds') level = record.levelname.replace('WARNING', 'WARN').rjust(5) thread_name = record.threadName.replace(' (', '#').replace(')', '').rjust(16)[-16:] # TODO: NamingCache? logger_name = self._logger_names[f"{self._handler_prefix}{record.name}"] message = record.getMessage() formatted = f"{timestamp} {level} [{thread_name}] {logger_name} : {message}" if record.exc_info: formatted += "\n" + self.formatException(record.exc_info) return formatted class HandlerFactory: class Target(Enum): CONSOLE = auto() FILE = auto() ALL = auto() @staticmethod def create(target: Target, handler_prefix: str = '', file_path: Path = None) -> List[Handler]: def console_handler(prefix: str = ''): handler = StreamHandler() handler.setFormatter(ApplicationFormatter(prefix)) handler.setLevel('INFO') return handler def file_handler(prefix: str = ''): handler = TimedRotatingFileHandler(file_path, when='midnight', backupCount=30) handler.setFormatter(ApplicationFormatter(prefix)) handler.setLevel('TRACE') return handler handlers = [] match target: case HandlerFactory.Target.CONSOLE: handlers.append(console_handler(handler_prefix)) case HandlerFactory.Target.FILE: handlers.append(file_handler(handler_prefix)) case HandlerFactory.Target.ALL: handlers.append(file_handler(handler_prefix)) handlers.append(console_handler(handler_prefix)) case _: raise ValueError(f"Unknown target: {target}") return handlers