295 lines
8.8 KiB
Python
295 lines
8.8 KiB
Python
"""
|
|
@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')
|
|
...
|