86 lines
2.7 KiB
Python
86 lines
2.7 KiB
Python
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
|