Python Loguru日志封装
#!/usr/bin/env python
# _*_ coding:utf-8 _*_
# @Time : 2023/6/25 15:49
# @Author : jingang.hou082613@gmail.com
# @Site :
# @File : operateLogs.py
# @Software: PyCharm
""" 日志处理 """
import inspect
import os
import sys
from functools import wraps
from time import strftime
from time import perf_counter
from base.singletonModel import Singleton
from loguru import logger
from utils.operateFile._ini import IniFile
class MyLogs(metaclass=Singleton):
LOG_DIR = os.path.join(os.path.abspath(os.path.dirname(__file__)), "../logs") # 存放日志
INI_DIR = os.path.join(os.path.abspath(os.path.dirname(__file__)), "../config/log_config.ini") # ini配置文件
ini_data = IniFile(INI_DIR).get_itemsAll() # 获取ini配置中的数据
logger_h = logger.opt(colors=True)
def __new__(cls, *args, **kwargs):
# hasattr是Python的一个内置函数，用于检查对象是否具有指定的属性或方法。
if not hasattr(cls, '_logger'):
cls._setup_logger()
return super().__new__(cls)
@classmethod
def _setup_logger(cls):
logger.remove()
# 设置日志文件路径和格式
filename = strftime("%Y%m%d-%H%M%S")
log_file_path = os.path.join(cls.LOG_DIR, f'{filename}.log')
log_format = "<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | " \
"<level>{level}</level> " \
"| <level>{message}</level>"
level_: str = MyLogs.ini_data["level"]
rotation_: str = MyLogs.ini_data["rotation"]
# 添加日志处理器：写入文件
cls.logger_h.add(log_file_path,
enqueue=True,
backtrace=True,
diagnose=True,
encoding="utf8",
rotation=rotation_
)
# 添加日志处理器：控制台输出
cls.logger_h.add(
sys.stderr,
format=log_format,
enqueue=True,
colorize=True,
backtrace=True,
diagnose=True,
level=level_,
# filter=cls._debug_filter # 使用自定义过滤器函数
)
# @staticmethod
# def _debug_filter(record):
# """自定义过滤器函数，仅输出 DEBUG 级别的日志"""
# if record["level"].name == MyLogs.ini_data["filter_level"]:
# return True
# return False
@classmethod
def log(cls, level: str, msg: str):
"""
···
:param level: 日志等级：info,debug,trace,error,warning,critical,exception
:param msg: 要输出的内容
:return: msg
# 栗子
MyLogs.log("info", "-----------分割线-----------")
"""
getattr(cls.logger_h, level)(msg)
@classmethod
def log_decorator(cls, msg: str):
"""
日志装饰器，记录函数的名称、参数、返回值、运行时间和异常信息
栗子：
@log.log_decorator("这里填写def功能")
def test_zero_division_error(a, b):
return a / b
"""
def decorator(func):
func_line = inspect.currentframe().f_back.f_lineno
@wraps(func)
def wrapper(*args, **kwargs):
cls.log("info", "\n")
cls.log("info", "<green>-----------分割线-----------</>")
cls.log("info", f"<white>{msg} ↓↓↓</>")
cls.log("debug", f'<red>{func.__qualname__}:{func.__name__}:{func_line} |</> <white> args: {args}, kwargs:{kwargs}</>')
start = perf_counter()
try:
result = func(*args, **kwargs)
end = perf_counter()
duration = end - start
cls.log("debug",
f"<red>{func.__qualname__}:{func.__name__}:{func_line} |</> <white> 返回结果：{result}, 耗时：{duration:4f}s</>")
return result
except Exception as e:
cls.log("exception", f"<red>{func.__qualname__}:{func.__name__}:{func_line} |</>: {msg}")
finally:
cls.log("info", "<green>-----------分割线-----------</>")
return wrapper
return decorator
MyLogs()
if __name__ == '__main__':
MyLogs.log("debug", "Executing step 3 of the algorithm")
MyLogs.log("info", "Server started on port")
MyLogs.log("warning", "Invalid input provided, using default values")
MyLogs.log("error", "Invalid user input detected, unable to proceed")
MyLogs.log("critical", "Database connection lost, terminating the application")
@MyLogs.log_decorator("1111111111111111111")
def A(a, b):
a / b
A(1, 0)
输出结果：
