Add main application logic, UI components, and logging setup

core/logger.py

@@ -0,0 +1,168 @@
+import logging
+import os
+from datetime import datetime
+from logging.handlers import TimedRotatingFileHandler
+
+from colorlog import ColoredFormatter
+
+
+LOGGER_NAME = "DDev"
+
+
+class LevelAwareColoredFormatter(ColoredFormatter):
+    def __init__(self, normal_fmt, error_fmt, *args, **kwargs):
+        super().__init__(normal_fmt, *args, **kwargs)
+        self.normal_fmt = normal_fmt
+        self.error_fmt = error_fmt
+
+    def format(self, record):
+        original_fmt = self._style._fmt
+        try:
+            self._style._fmt = (
+                self.error_fmt if record.levelno >= logging.ERROR else self.normal_fmt
+            )
+            return super().format(record)
+        finally:
+            self._style._fmt = original_fmt
+
+
+class DailyFileHandler(TimedRotatingFileHandler):
+    def __init__(self, base_dir, backupCount=30, encoding="utf-8", delay=True):
+        self.base_dir = base_dir
+        os.makedirs(self.base_dir, exist_ok=True)
+        filename = self._build_filename()
+        super().__init__(
+            filename=filename,
+            when="midnight",
+            interval=1,
+            backupCount=backupCount,
+            encoding=encoding,
+            delay=delay,
+        )
+
+    def _build_filename(self, dt=None):
+        dt = dt or datetime.now()
+        return os.path.join(self.base_dir, f"app_{dt:%Y_%m_%d}.log")
+
+    def doRollover(self):
+        if self.stream:
+            self.stream.close()
+            self.stream = None
+
+        self.baseFilename = os.path.abspath(self._build_filename())
+        self.rolloverAt = self.computeRollover(int(datetime.now().timestamp()))
+
+        if not self.delay:
+            self.stream = self._open()
+
+
+def get_logger(level=logging.DEBUG):
+    logger = logging.getLogger(LOGGER_NAME)
+    logger.setLevel(level)
+    logger.propagate = False
+    return logger
+
+
+def setup_console_logger(level=logging.DEBUG):
+    logger = get_logger(level)
+
+    has_console = any(getattr(h, "_ddev_console", False) for h in logger.handlers)
+    if has_console:
+        return logger
+
+    normal_fmt = "%(log_color)s%(levelname)-8s%(reset)s %(log_color)s%(message)s"
+    error_fmt = (
+        "%(log_color)s%(levelname)-8s%(reset)s %(log_color)s%(message)s "
+        "[%(pathname)s:%(lineno)d | %(funcName)s | Thread: %(threadName)s | PID: %(process)d]"
+    )
+
+    formatter = LevelAwareColoredFormatter(
+        normal_fmt=normal_fmt,
+        error_fmt=error_fmt,
+        datefmt="%Y/%m/%d %H:%M:%S",
+        reset=True,
+        log_colors={
+            "DEBUG": "cyan",
+            "INFO": "green",
+            "WARNING": "yellow",
+            "ERROR": "red",
+            "CRITICAL": "bold_red",
+        },
+    )
+
+    handler = logging.StreamHandler()
+    handler.setFormatter(formatter)
+    handler._ddev_console = True
+
+    logger.addHandler(handler)
+    return logger
+
+
+def setup_file_logger(log_dir, level=logging.DEBUG):
+    logger = get_logger(level)
+
+    has_file = any(getattr(h, "_ddev_file", False) for h in logger.handlers)
+    if has_file:
+        return logger
+
+    formatter = logging.Formatter(
+        "%(asctime)s | %(levelname)s | %(name)s | %(message)s",
+        datefmt="%Y/%m/%d %H:%M:%S",
+    )
+
+    handler = DailyFileHandler(base_dir=log_dir, backupCount=30)
+    handler.setFormatter(formatter)
+    handler._ddev_file = True
+
+    logger.addHandler(handler)
+    return logger
+
+
+def remove_file_handlers():
+    logger = get_logger()
+    for h in logger.handlers[:]:
+        if getattr(h, "_ddev_file", False):
+            logger.removeHandler(h)
+            h.close()
+
+
+def start_logger(data_dir, subdir, level=logging.DEBUG, console=True, file=True):
+    log_dir = os.path.join(data_dir, subdir)
+
+    logger = get_logger(level)
+
+    if console:
+        setup_console_logger(level)
+
+    if file:
+        remove_file_handlers()
+        setup_file_logger(log_dir, level)
+
+    return logger
+
+
+g_logger = setup_console_logger()
+
+
+def debug(msg, *args, **kwargs):
+    g_logger.debug(msg, *args, stacklevel=2, **kwargs)
+
+
+def info(msg, *args, **kwargs):
+    g_logger.info(msg, *args, stacklevel=2, **kwargs)
+
+
+def warning(msg, *args, **kwargs):
+    g_logger.warning(msg, *args, stacklevel=2, **kwargs)
+
+
+def error(msg, *args, exc_info=False, **kwargs):
+    g_logger.error(msg, *args, exc_info=exc_info, stacklevel=2, **kwargs)
+
+
+def critical(msg, *args, exc_info=False, **kwargs):
+    g_logger.critical(msg, *args, exc_info=exc_info, stacklevel=2, **kwargs)
+
+
+def set_level(level):
+    get_logger().setLevel(level)