poyee-data-warehouse/dw_base/utils/log_utils.py

# -*- coding:utf-8 -*-

import logging
import os
import re
import sys
from logging import handlers

from dw_base import DO_RESET, NORM_CYN, NORM_GRN, NORM_RED
from dw_base import LOG_ROOT_DIR, IS_RUN_BY_RELEASE_USER, IS_RUN_IN_RELEASE_DIR
from dw_base.common.container import ValueContainer
from dw_base.utils.datetime_utils import formatted_now

CURRENT_LOG_FILE = ValueContainer()


class Logging(object):
    _logger_pool = {}
    _root_logger = logging.getLogger()
    _file_logger = logging.getLogger()
    _default_format = '%(asctime)s - %(module)s [line:%(lineno)d] - %(levelname)s: %(message)s'
    # 设置日志输出格式
    _log_format = logging.Formatter(_default_format)
    # 设置日志在控制台输出
    _stream_handler = logging.StreamHandler()
    # 设置控制台中输出日志格式
    _stream_handler.setFormatter(_log_format)

    @staticmethod
    def _new_logger(log_file_path: str):
        os.system(f'mkdir -p {os.path.dirname(log_file_path)}')
        # 设置日志输出到文件
        file_handler = handlers.TimedRotatingFileHandler(filename=log_file_path, when='d', encoding="utf-8")
        # 设置日志文件中的输出格式
        file_handler.setFormatter(Logging._log_format)
        # 设置日志输出文件
        logger = logging.getLogger(log_file_path)
        # 设置日志级别
        logger.setLevel(logging.INFO)
        # 将输出对象添加到logger中
        logger.addHandler(Logging._stream_handler)
        logger.addHandler(file_handler)
        return logger

    @staticmethod
    def logger():
        log_file_path = CURRENT_LOG_FILE.get()
        if not log_file_path:
            return Logging._root_logger
        if Logging._file_logger.name != log_file_path and not Logging._logger_pool.__contains__(log_file_path):
            Logging._logger_pool[log_file_path] = Logging._new_logger(log_file_path)
            Logging._file_logger = Logging._logger_pool[log_file_path]
        return Logging._file_logger


def pretty_print(log_content, silent: bool = False):
    if IS_RUN_IN_RELEASE_DIR and IS_RUN_BY_RELEASE_USER:
        # 以发布用户在发布目录下运行程序才会在文件中记录日志
        log_file = CURRENT_LOG_FILE.get()
        saved_out = sys.stdout
        if log_file:
            try:
                with open(log_file, 'a') as f:
                    sys.stdout = f
                    # 将颜色渲染转义设置去除
                    message = re.sub("\033\\[.+?m", '', str(log_content))
                    print(f'{formatted_now()} {message}')
            except Exception as e:
                print(e)
        sys.stdout = saved_out
    if not silent:
        print(f'{NORM_CYN}{formatted_now()}{DO_RESET} {str(log_content)}{DO_RESET}')


def get_log_file_path(module_name: str, dt: str, file_name: str) -> str:
    """
    根据模块名称、日期和文件名生成日志文件绝对路径
    Args:
        module_name: 模块名称
        dt: 日期
        file_name: 文件名

    Returns:日志文件绝对路径
    """
    log_file_dir = f'{LOG_ROOT_DIR}/{module_name}/{dt}'
    os.system(f'mkdir -p {log_file_dir}')
    log_file_path = f'{LOG_ROOT_DIR}/{module_name}/{dt}/{file_name}.log'
    return log_file_path


if __name__ == '__main__':
    CURRENT_LOG_FILE.set('x.log')
    x = f'{NORM_RED}This is an {NORM_GRN}apple{NORM_RED} and I like it{DO_RESET}'
    print(x)
    y = re.sub("\033\\[.+?m", '', x)
    print(y)
    # Logging.logger().info(f'{NORM_MGT}This is an {NORM_GRN}apple{NORM_MGT} and I like it{DO_RESET}')
    # Logging.logger().warning(f'{NORM_YEL}This is an {NORM_GRN}apple{NORM_YEL} and I like it{DO_RESET}')
    # Logging.logger().error(f'{NORM_RED}This is an {NORM_GRN}apple{NORM_RED} and I like it{DO_RESET}')
    # pretty_print(f'{NORM_MGT}This is an {NORM_GRN}apple{NORM_MGT} and I like it{DO_RESET}')
    # pretty_print(f'{NORM_YEL}This is an {NORM_GRN}apple{NORM_YEL} and I like it{DO_RESET}')
    # pretty_print(f'{NORM_RED}This is an {NORM_GRN}apple{NORM_RED} and I like it{DO_RESET}')