Difficult-Rocket/libs/msdnicrosoft_logger/shenjack.py

"""
@author shenjackyuanjie
@contact 3695888@qq.com
"""
import atexit
import threading

from time import strftime
from typing import Optional, Union, Dict, Iterable, Tuple, List, Callable
from logging import NOTSET, DEBUG, INFO, WARNING, ERROR, FATAL

from Difficult_Rocket.utils.thread import ThreadLock

# 如果想要直接使用 logger 来 logging
# 直接调用 logger.debug() 即可
# 默认配置会有
# ----------
# 配置方式一
# 直接使用 logger.Logger()
# 将会创建一个空 logger
# 可以自行通过
# 配置方式二
#
#

color_reset_suffix = "\033[0m"
""" 只是用来重置颜色的后缀 """

"""
OFF > FATAL > ERROR > WARN > INFO > FINE > FINER > DEBUG > TRACE > ALL
logging.py
CRITICAL = 50
FATAL = CRITICAL
ERROR = 40
WARNING = 30
WARN = WARNING
INFO = 20
DEBUG = 10
NOTSET = 0
"""
ALL = NOTSET
TRACE = 5
FINE = 7


class LogFileCache:
    """日志文件缓存"""

    def __init__(self, file_name: str = 'logs//log.log', flush_time: Optional[Union[int, float]] = 1, log_cache_lens_max: int = 10):
        """

        @param file_name: 日志文件名称
        @param flush_time: 刷新日志缓存，写入文件的时长间隔
        @param log_cache_lens_max: 日志缓存在自动写入前的最大缓存长度
        """
        # 配置相关
        self._logfile_name = file_name  # log 文件名称
        self.flush_time = flush_time  # 缓存刷新时长
        self.cache_entries_num = log_cache_lens_max
        # 日志缓存表
        self._log_cache = []
        # 同步锁
        self.cache_lock = threading.Lock()  # 主锁
        self.write_lock = threading.Lock()  # 写入锁
        self.with_thread_lock = ThreadLock(self.cache_lock, time_out=1 / 60)  # 直接用于 with 的主锁
        self.threaded_write = threading.Timer(1, self._log_file_time_write)  # 基于 timer 的多线程

    def end_thread(self) -> None:
        """结束日志写入进程，顺手把目前的缓存写入"""
        self.cache_lock.acquire(blocking=True)
        self.threaded_write.cancel()
        self._log_file_time_write()

    def start_thread(self) -> None:
        self.threaded_write.start()
        atexit.register(self.end_thread)

    @property
    def logfile_name(self) -> str:
        return self._logfile_name

    @logfile_name.setter
    def logfile_name(self, value: str) -> None:
        with self.with_thread_lock:
            self._logfile_name = value

    @property
    def log_caches(self) -> list:
        return self._log_cache

    @log_caches.setter
    def log_caches(self, value: Union[str, Iterable[str]]):
        if type(value) == str:
            with self.with_thread_lock:
                self._log_cache.append(value)
                return
        elif isinstance(value, Iterable):
            with self.with_thread_lock:
                list(map(self._log_cache.append, value))
        ...

    def _log_file_time_write(self) -> None:
        """使用 threading.Timer 调用的定时写入日志文件的函数"""
        if self.log_caches:
            with self.with_thread_lock:
                if self.log_caches:
                    ...

        ...

    def write_logs(self, string: str, wait_for_cache: bool = True) -> None:
        if wait_for_cache:
            with self.with_thread_lock and open(file=self.logfile_name, encoding='utf-8', mode='a') as log_file:
                log_file.writelines(self._log_cache)
                log_file.write(string)
            ...
        else:
            ...


class Logger:
    """shenjack logger"""

    def __init__(self, **kwargs) -> None:
        """
        配置模式: 使用 kwargs 配置
        @param config: 字典格式的配置
        @param kwargs: key word 格式的配置
        """
        self.name = 'root'
        self.level = DEBUG
        self.colors = None
        if kwargs is not None:  # 使用 kwargs 尝试配置
            if name := kwargs.pop('name', False):  # 顺手把获取到的配置填入临时变量 如果成功获取再填入 self
                self.name = name
            if level := kwargs.pop('level', False):
                self.level = level
        else:

            ...

        self.file_cache = LogFileCache()
        self.warn = self.warning

    def make_log(self, *values: object,
                 level: int,
                 sep: Optional[str] = ' ',
                 end: Optional[str] = '\n',
                 flush: Optional[bool] = False) -> None:
        if level < self.level:
            return None
        print(level, values, sep, end, flush, sep='|')
        write_text = sep.join(i if type(i) is str else str(i) for i in values).__add__(end)
        print(write_text, end='')
        ...

    def trace(self, *values: object,
              sep: Optional[str] = ' ',
              end: Optional[str] = '\n',
              flush: Optional[bool] = False) -> None:
        return self.make_log(*values, level=TRACE, sep=sep, end=end, flush=flush)

    def fine(self, *values: object,
             sep: Optional[str] = ' ',
             end: Optional[str] = '\n',
             flush: Optional[bool] = False) -> None:
        return self.make_log(*values, level=FINE, sep=sep, end=end, flush=flush)

    def debug(self,
              *values: object,
              sep: Optional[str] = ' ',
              end: Optional[str] = '\n',
              flush: Optional[bool] = False) -> None:
        return self.make_log(*values, level=DEBUG, sep=sep, end=end, flush=flush)

    def info(self,
             *values: object,
             sep: Optional[str] = ' ',
             end: Optional[str] = '\n',
             flush: Optional[bool] = False) -> None:
        return self.make_log(*values, level=INFO, sep=sep, end=end, flush=flush)

    def warning(self,
                *values: object,
                sep: Optional[str] = ' ',
                end: Optional[str] = '\n',
                flush: Optional[bool] = False) -> None:
        return self.make_log(*values, level=WARNING, sep=sep, end=end, flush=flush)

    def error(self,
              *values: object,
              sep: Optional[str] = ' ',
              end: Optional[str] = '\n',
              flush: Optional[bool] = False) -> None:
        return self.make_log(*values, level=ERROR, sep=sep, end=end, flush=flush)

    def fatal(self,
              *values: object,
              sep: Optional[str] = ' ',
              end: Optional[str] = '\n',
              flush: Optional[bool] = False) -> None:
        return self.make_log(*values, level=FATAL, sep=sep, end=end, flush=flush)


def color_in_033(*args) -> str:
    color_text = ';'.join(args)
    color_text = f'\033[{color_text}m'
    return color_text


def rgb(r: int, g: int, b: int) -> Tuple[int, int, int]:
    return r, g, b


def logging_color() -> Dict:
    ...
    return {'info': ..., 'message': ...}


logger_configs = {
    'Logger': {
        'root': {
            'level': TRACE,
            'color': {
                DEBUG: '\033[0m'
            },
            'file': 'main_log_file',
        },
    },
    'Color': {
        TRACE: {'info': '\033[34;40m', 'message': '\033[48;2;40;40;40m'},
        FINE: {'info': '', 'message': '\033[35m'},
        DEBUG: {'info': '', 'message': '\033[38;2;133;138;149m'},
        INFO: {'info': '\033[32;40m', 'message': ''},
        WARNING: {'info': '', 'message': '\033[33m'},
        ERROR: {'info': '', 'message': '\033[31m'},
        FATAL: {'info': '', 'message': '\033[33;41'}
    },
    'File': {
        'main_log_file': {
            'mode': 'a',
            'encoding': 'utf-8',
            'level': DEBUG,
            'file_name': '{file_time}_logs.md'
        },
    },
    'Formatter': {
        'file_time': {'strftime': '%Y-%m-%d %H-%M'},
        'main_time': {'strftime': '%Y-%m-%d %H-%M-%S'},
        'version': 'game.version',
        'level': 'level',
        'encoding': 'utf-8',
        ...: ...
    }
}


def add_dict_config_to_global(some_dict: Union[dict, list, str], name: str) -> dict:
    """
    提前声明，这个函数很有可能搞坏 config
    请使用 add_kwargs_to_global 来修改配置
    如果你不知道你在改什么，请**务必不要**用这个函数来修改配置
    @param some_dict: 一个你丢进来的 logger 设置
    @param name: 这个 logger 设置的名称
    @return: 修改过的 logger 配置
    """
    logger_configs[name] = some_dict
    return logger_configs  # 修改过的 logger 配置


def add_kwargs_to_global(**kwargs) -> dict:
    """

    @param kwargs: 你要改的 logger配置
    @return: 修改过的 logger 配置
    """
    ...


def get_logger(name: str = 'name') -> Logger:
    """
    此函数用于从 global_config 中取出对应的配置建立一个相应的 logger
    @param name: logger的名称
    @return: 创建好的 logger
    """
    ...


if __name__ == "__main__":
    # 在这里可以使用 add_kwargs_to_global
    some_logger = Logger(name='aaa')
    some_logger.level = ALL
    some_logger.warn('aaaa', 'aaaa')
    ...