sysom1/environment/1_sdk/clogger/clogger.py

# -*- coding: utf-8 -*- #
"""
Time                2023/4/4 18:08
Author:             mingfeng (SunnyQjm)
Email               mfeng@linux.alibaba.com
File                clogger.py
Description:
"""
import logging
import sys
from typing import Optional

DEFAULT_FORMAT = "%(asctime)s | %(levelname)s | %(message)s"

CLOGGER_CONST_FILENAME = "clogger_filename"
CLOGGER_CONST_FUNC_NAME = "clogger_func_name"
CLOGGER_CONST_LINENO = "clogger_lineno"


class LessThanFilter(logging.Filter):
    def __init__(self, exclusive_maximum, name=""):
        super(LessThanFilter, self).__init__(name)
        self.max_level = exclusive_maximum

    def filter(self, record):
        return 1 if record.levelno < self.max_level else 0


def find_caller_info(func):
    def wrapper(*args, **kwargs):
        filename = sys._getframe(1).f_code.co_filename
        func_name = sys._getframe(1).f_code.co_name
        lineno = sys._getframe(1).f_lineno
        kwargs[CLOGGER_CONST_FILENAME] = filename
        kwargs[CLOGGER_CONST_LINENO] = lineno
        kwargs[CLOGGER_CONST_FUNC_NAME] = func_name
        func(*args, **kwargs)

    return wrapper


class Clogger:
    def __init__(self, name: str = "root", level: str = "DEBUG",
                 file=None, log_format: str = DEFAULT_FORMAT):
        # Init logger
        self.logger = logging.getLogger(name)

        self.file_logger_handler: Optional[logging.FileHandler] = None
        self.stdout_logger_handler = logging.StreamHandler(stream=sys.stdout)
        self.stderr_logger_handler = logging.StreamHandler(stream=sys.stderr)

        if not self.logger.hasHandlers():
            # FileLoggerHandler
            if file:
                self.file_logger_handler = logging.FileHandler(file)
                self.logger.addHandler(self.file_logger_handler)

            # stdout logger
            self.stdout_logger_handler.addFilter(LessThanFilter(logging.WARNING))
            self.logger.addHandler(self.stdout_logger_handler)

            # stderr logger
            self.logger.addHandler(self.stderr_logger_handler)

        self.set_format(log_format)
        self.set_level(level)

    def set_format(self, log_format: str, date_fmt: str = "%Y-%m-%d %H:%M:%S"):
        fmt = logging.Formatter(log_format, datefmt=date_fmt)
        if self.file_logger_handler:
            self.file_logger_handler.setFormatter(fmt)
        if self.stdout_logger_handler:
            self.stdout_logger_handler.setFormatter(fmt)
        if self.stderr_logger_handler:
            self.stderr_logger_handler.setFormatter(fmt)

    def set_level(self, level: str):
        self.logger.setLevel(level)
        if self.file_logger_handler:
            self.file_logger_handler.setLevel(level)
        if self.stdout_logger_handler is not None and logging.getLevelName(
                level) < logging.WARNING:
            self.stdout_logger_handler.setLevel(level)
        if self.stderr_logger_handler is not None:
            if logging.getLevelName(level) > logging.WARNING:
                self.stderr_logger_handler.setLevel(level)
            else:
                self.stderr_logger_handler.setLevel(logging.WARNING)

    def _wrapper_msg(self, msg, kwargs):
        filename = kwargs.pop(CLOGGER_CONST_FILENAME, "")
        _ = kwargs.pop(CLOGGER_CONST_FUNC_NAME, "")
        lineno = kwargs.pop(CLOGGER_CONST_LINENO, "")
        return f"{filename}:{lineno} | {msg}"

    @find_caller_info
    def debug(self, msg, *args, **kwargs):
        """
        Log 'msg % args' with severity 'DEBUG'.

        To pass exception information, use the keyword argument exc_info with
        a true value, e.g.

        logger.debug("Houston, we have a %s", "thorny problem", exc_info=1)
        """
        self.logger.debug(self._wrapper_msg(msg, kwargs), *args, **kwargs)

    @find_caller_info
    def info(self, msg, *args, **kwargs):
        """
        Log 'msg % args' with severity 'INFO'.

        To pass exception information, use the keyword argument exc_info with
        a true value, e.g.

        logger.info("Houston, we have a %s", "interesting problem", exc_info=1)
        """
        self.logger.info(self._wrapper_msg(msg, kwargs), *args, **kwargs)

    @find_caller_info
    def warning(self, msg, *args, **kwargs):
        """
        Log 'msg % args' with severity 'WARNING'.

        To pass exception information, use the keyword argument exc_info with
        a true value, e.g.

        logger.warning("Houston, we have a %s", "bit of a problem", exc_info=1)
        """
        self.logger.warning(self._wrapper_msg(msg, kwargs), *args, **kwargs)

    @find_caller_info
    def error(self, msg, *args, **kwargs):
        """
        Log 'msg % args' with severity 'ERROR'.

        To pass exception information, use the keyword argument exc_info with
        a true value, e.g.

        logger.error("Houston, we have a %s", "major problem", exc_info=1)
        """
        self.logger.error(self._wrapper_msg(msg, kwargs), *args, **kwargs)

    def exception(self, msg, *args, exc_info=True, **kwargs):
        """
        Convenience method for logging an ERROR with exception information.
        """
        self.logger.exception(msg, *args, exc_info=exc_info, **kwargs)

    @find_caller_info
    def critical(self, msg, *args, **kwargs):
        """
        Log 'msg % args' with severity 'CRITICAL'.

        To pass exception information, use the keyword argument exc_info with
        a true value, e.g.

        logger.critical("Houston, we have a %s", "major disaster", exc_info=1)
        """
        self.logger.critical(self._wrapper_msg(msg, kwargs), *args, **kwargs)


logger = Clogger()