DR_Team/Difficult-Rocket
DR_Team/Difficult-Rocket
6
2
Fork 0
Code Issues Pull Requests Projects Releases 25 Wiki Activity

迁移了logger

Browse Source 
loggers

logger update

delete

delete
This commit is contained in:
shenjack 2022-09-06 21:47:46 +08:00
parent 25c26750dd
commit 7cc953a11c
11 changed files with 25949 additions and 266 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
configs/main.toml
View File

@ -8,8 +8,8 @@ fonts_folder = "libs/fonts"
[window]
style = "None"
width = 763
height = 448
width = 907
height = 570
visible = true
caption = "Difficult Rocket {version}"
resizable = true

207
libs/msdnicrosoft_logger/__init__.py
View File

@ -1,207 +0,0 @@
from time import strftime
__all__ = [
"Log"
]
class Log:
"""
参数:\n
Shell -- 是否输出消息 -> 控制台 (Default：True) | Boolean\n
LogFile -- 是否输出消息 -> 文件 (Default: False) | Boolean\n
FileName [已使用格式化时间] -- 输出文件名（仅 LogFile=True 时此参数有效）（Default：%Y-%m-%d_%H-%M.log） | String\n
FileName 请注意符合对应操作系统的文件命名要求\n
"""
def __init__(self,
Shell=True,
LogFile=False,
FileName="%Y-%m-%d_%H-%M.log"
):
self.Shell = Shell
self.LogFile = LogFile
self.FileName = FileName
def info(self, Message):
if self.Shell:
Console.info(Message)
if self.LogFile:
file = File(FileName=self.FileName)
file.info(Message)
def warning(self, Message):
if self.Shell:
Console.warning(Message)
if self.LogFile:
file = File(FileName=self.FileName)
file.warning(Message)
def error(self, Message):
if self.Shell:
Console.error(Message)
if self.LogFile:
file = File(FileName=self.FileName)
file.error(Message)
def fatal(self, Message):
if self.Shell:
Console.fatal(Message)
if self.LogFile:
file = File(FileName=self.FileName)
file.fatal(Message)
def debug(self, Message):
if self.Shell:
Console.debug(Message)
if self.LogFile:
file = File(FileName=self.FileName)
file.debug(Message)
class Console:
"""
This class can output colored messages to Shell.
"""
from colorama import init
sColorSuffix = "\033[0m"
init(autoreset=True)
@staticmethod
def info(Message):
print(
strftime(
f"[%H:%M:%S] [\033[32mINFO{Console.sColorSuffix}]: {Message}"
)
)
@staticmethod
def warning(Message):
print(
strftime(
f"[%H:%M:%S] [\033[33mWARN{Console.sColorSuffix}]: {Message}"
)
)
@staticmethod
def error(Message):
print(
strftime(
f"[%H:%M:%S] [\033[31mERROR{Console.sColorSuffix}]: {Message}"
)
)
@staticmethod
def fatal(Message):
print(
strftime(
f"[%H:%M:%S] [\033[1;31;47mFATAL{Console.sColorSuffix}]: {Message}"
)
)
@staticmethod
def debug(Message):
print(
strftime(
f"[%H:%M:%S] [\033[34mDEBUG{Console.sColorSuffix}]: {Message}"
)
)
class File:
"""
This class can output messages to file.
"""
def __init__(self, FileName):
self.File = open(strftime(FileName), "a", encoding="utf-8")
def __del__(self):
self.File.close()
def info(self, Message):
self.File.write(
strftime(
f"[%H:%M:%S] [INFO]: {Message}\n"
)
)
def warning(self, Message):
self.File.write(
strftime(
f"[%H:%M:%S] [WARN]: {Message}\n"
)
)
def error(self, Message):
self.File.write(
strftime(
f"[%H:%M:%S] [ERROR]: {Message}\n"
)
)
def fatal(self, Message):
self.File.write(
strftime(
f"[%H:%M:%S] [FATAL]: {Message}\n"
)
)
def debug(self, Message):
self.File.write(
strftime(
f"[%H:%M:%S] [DEBUG]: {Message}\n"
)
)
import atexit
from typing import Optional
color_reset_suffix = "\033[0m"
class LogFileCache:
"""日志文件缓存"""
def __init__(self, time_cache: Optional[int, float] = 1, log_cache: int = 10):
self.time_cache = time_cache
self.log_cache = log_cache
class Logger:
"""shenjack logger"""
def __init__(self, config: dict = None) -> None:
if config is None:
self.config = {}
else:
self.config = config
class GetLogger:
"""shenjack牌logger"""
def __init__(self):
self.configs = {}
def add_config(self, name: str, config: dict) -> dict:
self.configs[name] = config
return self.configs
def logger(self, name: str = 'root') -> Logger:
return Logger(config=self.configs)
if __name__ == '__main__':
from time import sleep
logger = Log(LogFile=True)
logger.info("这是一条正常消息")
sleep(1)
logger.warning("这是一条警告消息")
sleep(2)
logger.error("这是一条错误消息")
sleep(3)
logger.fatal("这是一条致命错误消息")
sleep(4)
logger.debug("这是一条调试消息")

5
libs/utils/__init__.py Normal file
View File

@ -0,0 +1,5 @@
# -------------------------------
# Difficult Rocket
# Copyright © 2021-2022 by shenjackyuanjie 3695888@qq.com
# All rights reserved
# -------------------------------

12
libs/utils/build.ps1 Normal file
View File

@ -0,0 +1,12 @@
Write-Output ==========_win-py3.8_==========
python3.8.exe ./logger_setup.py build
Write-Output ==========_win-py3.10_==========
python3.10.exe ./logger_setup.py build
Write-Output ==========_wsl-py3.8_==========
wsl.exe python3 logger_setup.py build
Write-Output ==========_wsl-py3.10_==========
wsl.exe python3.10 logger_setup.py build

25112
libs/utils/logger.c Normal file
View File

File diff suppressed because it is too large Load Diff

49
libs/utils/logger.pyi Normal file
View File

@ -0,0 +1,49 @@
# -------------------------------
# Difficult Rocket
# Copyright © 2021-2022 by shenjackyuanjie 3695888@qq.com
# All rights reserved
# -------------------------------
import threading
from types import FrameType
from typing import Union, List, Optional, Iterable
class ThreadLock:
# with ThreadLock
def __init__(self, the_lock: threading.Lock, time_out: Union[float, int] = 1 / 60) -> None: ...
def __enter__(self) -> "ThreadLock": ...
def __exit__(self, exc_type, exc_val, exc_tb) -> None: ...
class ListCache:
"""一个线程安全的列表缓存"""
def __init__(self, lock: ThreadLock) -> None: ...
def __call__(self, *args, **kwargs) -> List[str]: ...
# ListCache()
def __iter__(self) -> "ListCache": ...
def __next__(self) -> str: ...
# support for x in ListCache
def __bool__(self) -> bool: ...
# True if have cache
def __getitem__(self, item: int) -> str: ...
# ListCache[int]
def append(self, value: Union[str, Iterable[str]]): ...
# ListCache.append('abc' | ['abc'])
@property
def cache(self) -> List[str]: ...
# ListCache.cache
def clear(self) -> None: ...
# ListCache.clear()
class LogFileCache:
...
class Logger:
...
def test_logger(the_logger: Logger) -> None:
...
def get_logger(name: str = 'root') -> Logger:
...

601
libs/utils/logger.pyx Normal file
View File

@ -0,0 +1,601 @@
"""
@author shenjackyuanjie
@contact 3695888@qq.com
"""
# -------------------------------
# Difficult Rocket
# Copyright © 2021-2022 by shenjackyuanjie 3695888@qq.com
# All rights reserved
# -------------------------------
# cython.language_level = 3
import re
import os
import time
import cython
import atexit
import inspect
import threading
from time import strftime
from objprint import op
from logging import NOTSET, DEBUG, INFO, WARNING, ERROR, FATAL
from types import FrameType
from typing import Optional, Union, Dict, Iterable, Tuple, Any, List
os.system('')
# print(os.path.abspath(os.curdir))
# 如果想要直接使用 logger 来 logging
# 直接调用 logger.debug() 即可
# 默认配置会有
# ----------
# 配置方式一
# 直接使用 logger.Logger()
# 将会创建一个空 logger
# 可以自行通过
# 配置方式二
cdef str color_reset_suffix = "\033[0m"
""" 只是用来重置颜色的后缀 """
cdef str re_find_color_code = r'\033\[[^\f\n\r\t\vm]*m'
re_color_code = re.compile(re_find_color_code)
"""
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
"""
cdef int ALL = NOTSET
cdef int TRACE = 5
cdef int FINE = 7
cdef dict level_name_map = {
ALL: 'ALL', # NOTSET
TRACE: 'TRACE',
FINE: 'FINE',
DEBUG: 'DEBUG',
INFO: 'INFO',
WARNING: 'WARNING', # WARN
ERROR: 'ERROR',
FATAL: 'FATAL'
}
cdef dict name_level_map = {
'NOTSET': ALL,
'ALL': ALL,
'TRACE': TRACE,
'FINE': FINE,
'DEBUG': DEBUG,
'INFO': INFO,
'WARNING': WARNING,
'WARN': WARNING,
'ERROR': ERROR,
'CRITICAL': FATAL,
'FATAL': FATAL
}
cdef dict logger_configs = {
'Logger': {
'root': {
'level': TRACE,
'color': 'main_color',
'file': 'main_log_file',
},
'client': {
'level': TRACE,
'color': 'main_color',
# 'file': 'main_log_file',
},
'server': {
'level': TRACE,
'color': 'DiGua_color',
'file': 'main_log_file',
},
},
'Color': {
'main_color': {
'file_time': '\033[38;2;201;222;56m',
'main_time': '\033[38;2;201;222;56m',
'file_name': '\033[38;2;0;255;180m',
'code_line': '\033[38;2;0;255;180m',
'logger': '\033[0m',
TRACE: {'info': '\033[38;2;138;173;244m', 'message': '\033[38;2;138;173;244m'},
FINE: {'info': '\033[35;48;2;44;44;54m', 'message': '\033[35m'},
DEBUG: {'info': '\033[38;2;133;138;149m', 'message': '\033[38;2;133;138;149m'},
INFO: {'info': '\033[0m', 'message': '\033[0m'},
WARNING: {'info': '\033[33m', 'message': '\033[33m'},
ERROR: {'info': '\033[31m', 'message': '\033[31m'},
FATAL: {'info': '\033[38;2;255;255;0;48;2;120;10;10m', 'message': '\033[38;2;255;255;0;48;2;120;10;10m'}
},
'DiGua_color': {
# catppuccin Macchiato
'file_time': '\033[38;2;238;212;159m',
'main_time': '\033[38;2;202;211;245m',
'file_name': '\033[38;2;139;213;202m',
'code_line': '\033[38;2;166;218;149m',
'logger': '\033[0m',
TRACE: {'info': '\033[38;2;138;173;244m', 'message': '\033[38;2;138;173;244m'},
FINE: {'info': '\033[38;2;198;160;246m', 'message': '\033[38;2;198;160;246m'},
DEBUG: {'info': '\033[38;2;133;138;149m', 'message': '\033[38;2;133;138;149m'},
INFO: {'info': '\033[0m', 'message': '\033[0m'},
WARNING: {'info': '\033[38;2;245;169;127m', 'message': '\033[38;2;245;169;127m'},
ERROR: {'info': '\033[38;2;237;135;150m', 'message': '\033[38;2;237;135;150m'},
FATAL: {'info': '\033[38;2;255;255;0;48;2;120;10;10m', 'message': '\033[38;2;255;255;0;48;2;120;10;10m', 'logger': '\033[38;2;245;189;230m'}
}
},
'File': {
'main_log_file': {
'mode': 'a',
'encoding': 'utf-8',
'level': TRACE,
'file_name': './logs/{file_time}_logs.md',
'cache_len': 10,
'cache_time': 1
},
},
'Formatter': {
'MESSAGE': {
'format': '[{main_time}] [{logger_name}] {level} | {file_name}:{code_line} | {message}'
},
'file_name': 'no frame',
'code_line': 'no frame',
'file_time': {'strftime': '%Y-%m-%d %H-%M'},
'main_time': {'strftime': '%Y-%m-%d %H-%M-%S:%%S'}, # %%S 三位毫秒
...: ...
}
}
ctypedef fused union_int_float:
int
float
class ThreadLock:
"""一个用来 with 的线程锁"""
def __init__(self, the_lock: threading.Lock, time_out: Union[float, int] = 1 / 60) -> None:
self.lock = the_lock
self.time_out = time_out
def __enter__(self):
self.lock.acquire(timeout=self.time_out)
if not self.lock.locked():
raise RuntimeError(f'Lock time Out with {self.time_out}')
return self
def __exit__(self, exc_type, exc_val, exc_tb):
if self.lock.locked():
self.lock.release()
class ListCache:
"""一个线程安全的列表缓存"""
def __init__(self, lock: ThreadLock):
self._cache = []
self.with_thread_lock = lock
def append(self, value: Union[str, Iterable[str]]):
with self.with_thread_lock:
if isinstance(value, str):
self._cache.append(value)
elif isinstance(value, Iterable):
self._cache.append(*value)
else:
raise TypeError(f"cache must be string or Iterable. not a {type(value)}")
def __getitem__(self, item) -> str:
assert isinstance(item, int)
with self.with_thread_lock:
try:
return self._cache[item]
except IndexError as exp:
print(f'cache:{self.cache}')
raise IndexError(f'there is no cache at {item}!\ncache:{self.cache}\n{exp}')
def __call__(self, *args, **kwargs) -> List[str]:
return self.cache
def __iter__(self):
self._iter_len = len(self.cache)
return self
def __next__(self):
if self._iter_len == -1:
del self._iter_len
raise StopIteration('there is no more cache')
returns = self.cache[-self._iter_len]
self._iter_len -= 1
return returns
def __bool__(self):
with self.with_thread_lock:
return True if len(self.cache) > 0 else False
@property
def cache(self):
return self._cache
def clear(self):
with self.with_thread_lock:
self.cache.clear()
class LogFileCache:
"""日志文件缓存"""
def __init__(self, dict file_conf):
"""
@param file_conf: 日志文件配置
"""
# 配置相关
self._logfile_name = os.path.abspath(format_str(file_conf['file_name'])) # log 文件名称
self.level = get_key_from_dict(file_conf, 'level', DEBUG)
self.file_conf = file_conf
self.flush_time = file_conf['cache_time'] # 缓存刷新时长
self.cache_entries_num = file_conf['cache_len']
self.started = True
self.running = False
# 同步锁
self.cache_lock = threading.Lock() # 主锁
self.time_limit_lock = ThreadLock(self.cache_lock, time_out=1 / 60) # 直接用于 with 的主锁
self.threaded_write = threading.Timer(1, self._log_file_time_write, kwargs={'thread': True}) # 基于 timer 的多线程
# 日志缓存表
self.log_cache = ListCache(self.time_limit_lock)
self.file_setup()
def file_setup(self):
cdef int cache_time = 0
cdef str file_type = self.logfile_name[self.logfile_name.rfind('.'):]
cdef str file_pure_name = self.logfile_name[:self.logfile_name.rfind('.')]
while os.path.isfile(self.logfile_name):
cache_time += 1
self.logfile_name = f'{file_pure_name}-{cache_time}{file_type}'
def end_thread(self) -> None:
"""结束日志写入进程，顺手把目前的缓存写入"""
self.cache_lock.acquire(blocking=True)
if self.running:
self.threaded_write.cancel()
self.running = False
self.started = False
self._log_file_time_write()
atexit.unregister(self.end_thread)
def start_thread(self) -> None:
self.threaded_write.start()
self.started = True
self.running = True
atexit.register(self.end_thread)
@property
def logfile_name(self) -> str:
self._logfile_name: str
return self._logfile_name
@logfile_name.setter
def logfile_name(self, str value) -> None:
with self.time_limit_lock:
self._logfile_name = value
def _log_file_time_write(self, bint thread = False) -> None:
"""使用 threading.Timer 调用的定时写入日志文件的函数"""
if self.log_cache:
with self.time_limit_lock:
if self.log_cache:
with open(file=self.logfile_name,
encoding=get_key_from_dict(self.file_conf, 'encoding', 'utf-8'),
mode=get_key_from_dict(self.file_conf, 'mode', 'a')) as log_file:
log_file.writelines(self.log_cache.cache.copy())
self.log_cache.clear()
if thread:
self.running = False
def write_logs(self, str string, bint flush = False) -> None:
self.log_cache.append(string)
if len(self.log_cache.cache) >= 10:
self._log_file_time_write()
return None
if flush:
self._log_file_time_write()
if self.started and not self.running:
self.threaded_write = threading.Timer(1, self._log_file_time_write, kwargs={'thread': True}) # 基于 timer 的多线程
self.threaded_write.start()
self.running = True
class Logger:
"""shenjack logger"""
def __init__(self,
str name = 'root',
int level = DEBUG,
file_conf: List[LogFileCache] = None,
colors: Dict[Union[int, str], Dict[str, str]] = None,
formats=None) -> None:
"""
配置模式: 使用 kwargs 配置
@param name: logger 名称 默认为 root
@param level: logging 输出等级 默认为 DEBUG(10)
@param file_conf: logger 的文件处理配置
@param colors: dict 颜色配置
@param formats: 格式化配置
"""
self.name = name
self.level = level
self.colors = colors or logger_configs['Color']['main_color']
self.formats = formats or logger_configs['Formatter'].copy()
self.min_level = self.level
if file_conf:
self.file_cache = file_conf
self.min_level = min(*[file.level for file in file_conf], self.level)
else:
self.file_cache = []
self.warn = self.warning
def add_file(self, handler: LogFileCache):
self.file_cache.append(handler)
self.min_level = min(*[file.level for file in self.file_cache], self.level)
def remove_file(self, handler: LogFileCache):
self.file_cache.pop(self.file_cache.index(handler))
self.min_level = min(*[file.level for file in self.file_cache], self.level)
def make_log(self, *values: object,
int level,
str sep = ' ',
str end = '\n',
bint flush = False,
frame: Optional[FrameType] = None) -> None:
if level < self.min_level:
return None
if frame is None:
_frame = inspect.currentframe()
if _frame is not None:
frame = _frame if _frame.f_back is None else _frame.f_back
# op(frame, indent=3, depth=1, honor_existing=False)
# op(_frame, indent=3, depth=1, honor_existing=False)
# text = sep.join(i if type(i) is str else str(i) for i in values)
cdef str text = f"{self.colors[level]['message']}{sep.join(i if type(i) is str else str(i) for i in values)}{color_reset_suffix}"
cdef str print_text = self.format_text(level=level, text=text, frame=frame)
if level >= self.level:
print(print_text, end=end)
for file in self.file_cache:
if level < file.level:
continue
file.write_logs(f"{re.sub(re_find_color_code, '', print_text)}{end}", flush=flush)
return None
def format_text(self, level: int, text: str, frame: Optional[FrameType]) -> str:
cdef str level_with_color = f"[{get_key_from_dict(self.colors[level], 'info')}{level_name_map[level]}{color_reset_suffix}]"
level_with_color = f"{level_with_color}{' ' * (9 - len_without_color_maker(level_with_color))}"
cdef dict formats = self.formats.copy()
if frame is not None:
formats['file_name'] = f"{get_key_from_dict(self.colors[level], 'file_name', self.colors['file_name'])}{os.path.split(frame.f_code.co_filename)[-1]}{color_reset_suffix}"
formats['code_line'] = f"{get_key_from_dict(self.colors[level], 'code_line', self.colors['code_line'])}{frame.f_lineno}{color_reset_suffix}"
formats['logger_name'] = f'{get_key_from_dict(self.colors[level], "logger", self.colors["logger"])}{self.name}{color_reset_suffix}'
cdef str now_time = str(time.time())
for key, value in formats.items():
if isinstance(value, dict):
if 'strftime' in value:
value['strftime']: str
formats[key] = f"{get_key_from_dict(self.colors[level], key, self.colors[key])}{strftime(value['strftime'].replace('%%S', now_time[now_time.find('.') + 1:now_time.find('.') + 4]))}{color_reset_suffix}"
return self.formats['MESSAGE']['format'].format(level_with_color=level_with_color,
level=level_with_color, message=text,
**formats)
def trace(self, *values: object,
str sep = ' ',
str end = '\n',
bint flush = False,
frame: Optional[FrameType] = None) -> None:
return self.make_log(*values, level=TRACE, sep=sep, end=end, flush=flush, frame=frame)
def fine(self, *values: object,
str sep = ' ',
str end = '\n',
bint flush = False,
frame: Optional[FrameType] = None) -> None:
return self.make_log(*values, level=FINE, sep=sep, end=end, flush=flush, frame=frame)
def debug(self, *values: object,
str sep = ' ',
str end = '\n',
bint flush = False,
frame: Optional[FrameType] = None) -> None:
return self.make_log(*values, level=DEBUG, sep=sep, end=end, flush=flush, frame=frame)
def info(self, *values: object,
str sep = ' ',
str end = '\n',
bint flush = False,
frame: Optional[FrameType] = None) -> None:
return self.make_log(*values, level=INFO, sep=sep, end=end, flush=flush, frame=frame)
def warning(self, *values: object,
str sep = ' ',
str end = '\n',
bint flush = False,
frame: Optional[FrameType] = None) -> None:
return self.make_log(*values, level=WARNING, sep=sep, end=end, flush=flush, frame=frame)
def error(self, *values: object,
str sep = ' ',
str end = '\n',
bint flush = False,
frame: Optional[FrameType] = None) -> None:
return self.make_log(*values, level=ERROR, sep=sep, end=end, flush=flush, frame=frame)
def fatal(self, *values: object,
str sep = ' ',
str end = '\n',
bint flush = False,
frame: Optional[FrameType] = None) -> None:
return self.make_log(*values, level=FATAL, sep=sep, end=end, flush=flush, frame=frame)
def get_key_from_dict(a_dict: Dict, key: Any, default: Any = None) -> Optional[Any]:
if default is None:
return a_dict[key]
try:
return a_dict[key]
except KeyError:
return default
cdef str format_str(str text: str):
cdef dict formats = logger_configs['Formatter'].copy()
cdef str now_time = str(time.time())
for key, value in formats.items():
if isinstance(value, dict):
if 'strftime' in value:
value['strftime']: str
formats[key] = strftime(value['strftime'].replace('%%S', now_time[now_time.find('.') + 1:now_time.find('.') + 4]))
return text.format(**formats)
cdef int len_without_color_maker(text: str):
with_out_text = re.sub(re_find_color_code, '', text)
return len(with_out_text)
def gen_file_conf(str file_name,
int file_level = DEBUG,
str file_mode = 'a',
str file_encoding = 'utf-8',
int file_cache_len = 10,
union_int_float file_cache_time = 1) -> dict:
"""
生成一个文件配置
@param file_name: 日志文件名
@param file_level: 日志文件记录级别
@param file_mode: 文件模式
@param file_encoding: 文件编码
@param file_cache_len: 文件缓存长度
@param file_cache_time: 文件缓存时间
@return: 生成的配置
"""
return {'file_name': file_name,
'level': file_level,
'mode': file_mode,
'encoding': file_encoding,
'cache_len': file_cache_len,
'cache_time': file_cache_time}
def gen_color_conf(color_name: str = None, **colors) -> dict:
default_color = logger_configs['Color']['main_color' if color_name is None else color_name].copy()
default_color.update(colors)
return default_color
def logger_with_default_settings(str name not None,
int level = DEBUG,
dict file_conf = None,
dict colors = None,
formats: dict = None) -> Logger:
return Logger(name=name,
level=level,
file_conf=[LogFileCache(gen_file_conf(**file_conf))],
colors=gen_color_conf(**colors),
formats=logger_configs['Formatter'].copy().update(formats))
cdef void add_file_config(str conf_name,
str file_name,
int file_level = DEBUG,
str file_mode = 'a',
str file_encoding = 'utf-8',
int file_cache_len = 10,
union_int_float file_cache_time = 1):
"""
向 logger config 里添加一个文件配置
@param conf_name: 文件配置名称
@param file_name: 日志文件名
@param file_level: 日志文件记录级别
@param file_mode: 文件模式
@param file_encoding: 文件编码
@param file_cache_len: 文件缓存长度
@param file_cache_time: 文件缓存时间
@return: None
"""
logger_configs['File'][conf_name] = {'file_name': file_name,
'level': file_level,
'mode': file_mode,
'encoding': file_encoding,
'cache_len': file_cache_len,
'cache_time': file_cache_time}
def add_dict_config_to_global(some_dict: Union[dict, list, str],
str name) -> 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 get_logger(str name = 'root') -> Logger:
"""
此函数用于从 global_config 中取出对应的配置建立一个相应的 logger
@param name: logger的名称
@return: 创建好的 logger
"""
if name in logger_configs['Logger']:
the_config = logger_configs['Logger'][name]
else:
the_config = logger_configs['Logger']['root']
file_handler = None
if 'file' in the_config:
file_handler = [LogFileCache(logger_configs['File'][the_config['file']])]
return Logger(name=name,
level=the_config['level'],
file_conf=file_handler,
colors=logger_configs['Color'][get_key_from_dict(the_config, 'color', 'main_color')],
formats=logger_configs['Formatter'].copy())
def test_logger(the_logger: Logger):
the_logger.trace('tracing')
the_logger.fine('some fine!')
the_logger.debug('debugging')
the_logger.info("Hello World!!")
the_logger.warn('warning')
the_logger.error('error haaaa')
the_logger.fatal('oh no')
if __name__ == "__main__":
os.chdir('../../')
logger = get_logger('server')
logger.info('my name is:', logger.name)
a_logger = get_logger('client')
a_logger.trace('tracing')
a_logger.fine('some fine!')
a_logger.debug('debugging')
a_logger.info("Hello World!!")
a_logger.warn('warning')
a_logger.error('error haaaa')
a_logger.fatal('oh no')
logger.info('my name is:', logger.name)
for x in range(5):
test_logger(logger)
test_logger(a_logger)

40
libs/utils/logger_setup.py Normal file
View File

@ -0,0 +1,40 @@
# -------------------------------
# Difficult Rocket
# Copyright © 2021-2022 by shenjackyuanjie 3695888@qq.com
# All rights reserved
# -------------------------------
import os
import shutil
from setuptools import setup
from Cython.Build import cythonize
setup(name='logger',
zip_safe=False,
ext_modules=cythonize('logger.pyx',
language_level=3,
annotate=True))
def dir_not_temp(path: str):
return path.find('temp') and os.path.isdir(os.path.join('build', path))
build_dir = os.listdir('./build')
build_dir = list(filter(dir_not_temp, build_dir))
# 用列表推导式把 是文件夹 且 名字里没有 temp 的抽出来
print(build_dir)
os.chdir('build')
# 把运行路径迁移过去，方便copy
for build in build_dir:
copy = os.listdir(build)
if not os.path.isdir(build):
for file in copy:
shutil.copy(os.path.join(build, file), './')
else:
for dirs in copy:
i = os.listdir(os.path.join(build, dirs))
for j in i:
shutil.copy(os.path.join(build, dirs, j), './')

41
libs/utils/logger_test.py Normal file
View File

@ -0,0 +1,41 @@
# -------------------------------
# Difficult Rocket
# Copyright © 2021-2022 by shenjackyuanjie 3695888@qq.com
# All rights reserved
# -------------------------------
import inspect
from viztracer import VizTracer
import py_logger
from build import logger as c_logger
A_py_logger = py_logger.get_logger(name='client')
A_c_logger = c_logger.get_logger(name='client')
A_py_logger.info('aaaaa', frame=inspect.currentframe())
A_c_logger.info('aaaaa', frame=inspect.currentframe())
a_frame = inspect.currentframe()
def a_test(log: py_logger.Logger):
log.info('123', frame=a_frame)
print('py_py')
py_logger.test_logger(A_py_logger)
print('py_c')
py_logger.test_logger(A_c_logger)
print('c_py')
c_logger.test_logger(A_py_logger)
print('c_c')
c_logger.test_logger(A_c_logger)
a_test(A_c_logger)
with VizTracer(output_file='loggers.json', file_info=True) as _:
print('py_py')
py_logger.test_logger(A_py_logger)
print('py_c')
py_logger.test_logger(A_c_logger)
print('c_py')
c_logger.test_logger(A_py_logger)
print('c_c')
c_logger.test_logger(A_c_logger)

138
Difficult_Rocket/utils/logger.py → libs/utils/py_logger.py
View File

@ -2,6 +2,11 @@
@author shenjackyuanjie
@contact 3695888@qq.com
"""
# -------------------------------
# Difficult Rocket
# Copyright © 2021-2022 by shenjackyuanjie 3695888@qq.com
# All rights reserved
# -------------------------------
import re
import os
import time
@ -9,6 +14,7 @@ import atexit
import inspect
import threading
from objprint import op
from time import strftime
from logging import NOTSET, DEBUG, INFO, WARNING, ERROR, FATAL
from types import FrameType
@ -32,6 +38,7 @@ color_reset_suffix = "\033[0m"
""" 只是用来重置颜色的后缀 """
re_find_color_code = r'\033\[[^\f\n\r\t\vm]*m'
re_color_code = re.compile(re_find_color_code)
"""
OFF > FATAL > ERROR > WARN > INFO > FINE > FINER > DEBUG > TRACE > ALL
@ -84,7 +91,7 @@ logger_configs = {
'client': {
'level': TRACE,
'color': 'main_color',
'file': 'main_log_file',
# 'file': 'main_log_file',
},
'server': {
'level': TRACE,
@ -147,6 +154,7 @@ logger_configs = {
class ThreadLock:
"""一个用来 with 的线程锁"""
def __init__(self, the_lock: threading.Lock, time_out: Union[float, int] = 1 / 60) -> None:
self.lock = the_lock
@ -170,7 +178,7 @@ class ListCache:
self._cache = []
self.with_thread_lock = lock
def append(self, value: Union[str, Iterable]):
def append(self, value: Union[str, Iterable[str]]):
with self.with_thread_lock:
if isinstance(value, str):
self._cache.append(value)
@ -179,7 +187,7 @@ class ListCache:
else:
raise TypeError(f"cache must be string or Iterable. not a {type(value)}")
def __getitem__(self, item):
def __getitem__(self, item) -> str:
assert isinstance(item, int)
with self.with_thread_lock:
try:
@ -188,7 +196,7 @@ class ListCache:
print(f'cache:{self.cache}')
raise IndexError(f'there is no cache at {item}!\ncache:{self.cache}\n{exp}')
def __call__(self, *args, **kwargs):
def __call__(self, *args, **kwargs) -> List[str]:
return self.cache
def __iter__(self):
@ -304,8 +312,8 @@ class Logger:
"""shenjack logger"""
def __init__(self,
name: str = None,
level: int = None,
name: str = 'root',
level: int = DEBUG,
file_conf: List[LogFileCache] = None,
colors: Dict[Union[int, str], Dict[str, str]] = None,
formats=None) -> None:
@ -317,8 +325,8 @@ class Logger:
@param colors: dict 颜色配置
@param formats: 格式化配置
"""
self.name = name or 'root'
self.level = level if level is not None else DEBUG
self.name = name
self.level = level
self.colors = colors or logger_configs['Color']['main_color']
self.formats = formats or logger_configs['Formatter'].copy()
self.min_level = self.level
@ -345,13 +353,12 @@ class Logger:
frame: Optional[FrameType] = None) -> None:
if level < self.min_level:
return None
if not frame:
frame = inspect.currentframe()
frame = frame.f_back.f_back
elif (frame := inspect.currentframe()) is not None:
frame = frame.f_back
text = sep.join(i if type(i) is str else str(i) for i in values)
text = f"{self.colors[level]['message']}{text}{color_reset_suffix}"
if frame is None:
if (frame := inspect.currentframe()) is not None:
# print(frame.f_code.co_filename, frame.f_back.f_code.co_filename, frame.f_back.f_back.f_code.co_filename)
frame = frame if frame.f_back is None else frame.f_back if frame.f_back.f_back is None else frame.f_back.f_back
# text = sep.join(i if type(i) is str else str(i) for i in values)
text = f"{self.colors[level]['message']}{sep.join(i if type(i) is str else str(i) for i in values)}{color_reset_suffix}"
print_text = self.format_text(level=level, text=text, frame=frame)
if level >= self.level:
print(print_text, end=end)
@ -384,49 +391,56 @@ class Logger:
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)
flush: Optional[bool] = False,
frame: Optional[FrameType] = None) -> None:
return self.make_log(*values, level=TRACE, sep=sep, end=end, flush=flush, frame=frame)
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)
flush: Optional[bool] = False,
frame: Optional[FrameType] = None) -> None:
return self.make_log(*values, level=FINE, sep=sep, end=end, flush=flush, frame=frame)
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)
flush: Optional[bool] = False,
frame: Optional[FrameType] = None) -> None:
return self.make_log(*values, level=DEBUG, sep=sep, end=end, flush=flush, frame=frame)
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)
flush: Optional[bool] = False,
frame: Optional[FrameType] = None) -> None:
return self.make_log(*values, level=INFO, sep=sep, end=end, flush=flush, frame=frame)
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)
flush: Optional[bool] = False,
frame: Optional[FrameType] = None) -> None:
return self.make_log(*values, level=WARNING, sep=sep, end=end, flush=flush, frame=frame)
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)
flush: Optional[bool] = False,
frame: Optional[FrameType] = None) -> None:
return self.make_log(*values, level=ERROR, sep=sep, end=end, flush=flush, frame=frame)
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)
flush: Optional[bool] = False,
frame: Optional[FrameType] = None) -> None:
return self.make_log(*values, level=FATAL, sep=sep, end=end, flush=flush, frame=frame)
def get_key_from_dict(a_dict: Dict, key: Any, default: Any = None) -> Optional[Any]:
@ -449,27 +463,27 @@ def format_str(text: str) -> str:
return text.format(**formats)
def color_in_033(*args) -> str:
color_text = ';'.join(args)
color_text = f'\033[{color_text}m'
return color_text
def len_without_color_maker(text: str) -> int:
with_out_text = re.sub(re_find_color_code, '', text)
return len(with_out_text)
def rgb(r: int, g: int, b: int) -> Tuple[int, int, int]:
return r, g, b
def gen_file_conf(file_name: str,
file_level: int = DEBUG,
file_mode: str = 'a',
file_encoding: str = 'utf-8',
file_cache_len: int = 10,
file_cache_time: Union[int, float] = 1) -> dict:
"""
生成一个文件配置
@param file_name: 日志文件名
@param file_level: 日志文件记录级别
@param file_mode: 文件模式
@param file_encoding: 文件编码
@param file_cache_len: 文件缓存长度
@param file_cache_time: 文件缓存时间
@return: 生成的配置
"""
return {'file_name': file_name,
'level': file_level,
'mode': file_mode,
@ -479,7 +493,7 @@ def gen_file_conf(file_name: str,
def gen_color_conf(color_name: str = None, **colors) -> dict:
default_color = logger_configs['Color']['main_color'].copy()
default_color = logger_configs['Color']['main_color' if color_name is None else color_name].copy()
default_color.update(colors)
return default_color
@ -496,6 +510,32 @@ def logger_with_default_settings(name: str,
formats=logger_configs['Formatter'].copy().update(formats))
def add_file_config(conf_name: str,
file_name: str,
file_level: int = DEBUG,
file_mode: str = 'a',
file_encoding: str = 'utf-8',
file_cache_len: int = 10,
file_cache_time: Union[int, float] = 1) -> None:
"""
向 logger config 里添加一个文件配置
@param conf_name: 文件配置名称
@param file_name: 日志文件名
@param file_level: 日志文件记录级别
@param file_mode: 文件模式
@param file_encoding: 文件编码
@param file_cache_len: 文件缓存长度
@param file_cache_time: 文件缓存时间
@return: None
"""
logger_configs['File'][conf_name] = {'file_name': file_name,
'level': file_level,
'mode': file_mode,
'encoding': file_encoding,
'cache_len': file_cache_len,
'cache_time': file_cache_time}
def add_dict_config_to_global(some_dict: Union[dict, list, str], name: str) -> dict:
"""
提前声明，这个函数很有可能搞坏 config
@ -529,20 +569,6 @@ def get_logger(name: str = 'root') -> Logger:
formats=logger_configs['Formatter'].copy())
def _format_colors(colors: dict) -> None:
for key, value in colors.items():
if not isinstance(key, str):
continue
for level, value_ in colors.items():
if not isinstance(level, int):
continue
colors[level][key] = get_key_from_dict(colors[level], key, colors[key])
# for color in logger_configs['Color']:
# format_colors(logger_configs['Color'][color])
def test_logger(the_logger: Logger):
the_logger.trace('tracing')
the_logger.fine('some fine!')
@ -554,17 +580,15 @@ def test_logger(the_logger: Logger):
if __name__ == "__main__":
os.chdir('D:/githubs/DR')
os.chdir('../../')
logger = get_logger('server')
logger.info('my name is:', logger.name)
a_logger = get_logger('client')
a_logger.trace('tracing')
# time.sleep(1.1)
a_logger.fine('some fine!')
a_logger.debug('debugging')
# time.sleep(1.1)
a_logger.info("Hello World!!")
a_logger.warn('warning')
a_logger.error('error haaaa')

6
requirement.txt
View File

@ -2,3 +2,9 @@ psutil
pillow
objprint
# selenium
toml
setuptools
Cython
colorama
viztracer