ymj_data_collect/clog.py
2025-02-25 10:15:21 +08:00

93 lines
2.8 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

import logging
import sys
from fileinput import filename
from colorlog import ColoredFormatter
# 创建一个 logger
logger = logging.getLogger('my_logger')
logger.setLevel(logging.DEBUG)
# 创建一个控制台处理器
console_handler = logging.StreamHandler()
# 定义 SUCCESS 级别的数值 (INFO=20, WARNING=30所以 SUCCESS 可以设为 25)
SUCCESS = 25
logging.addLevelName(SUCCESS, "SUCCESS")
# 定义一个 success 方法
def success(self, message, *args, **kwargs):
if self.isEnabledFor(SUCCESS):
self._log(SUCCESS, message, args, **kwargs)
# 绑定到 Logger 类上
logging.Logger.success = success
# 创建带颜色的格式化器(用于控制台输出)
console_formatter = ColoredFormatter(
"%(log_color)s%(asctime)s - %(filename)s:%(lineno)d - [%(levelname)s]:%(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
reset=True,
log_colors={
'SUCCESS': 'green', # 这里定义 SUCCESS 级别的颜色
'DEBUG': 'cyan',
'INFO': 'white',
'WARNING': 'yellow',
'ERROR': 'red',
'CRITICAL': 'red,bg_black',
},
secondary_log_colors={
'message': {
'SUCCESS': 'green', # 这里定义 SUCCESS 级别的消息颜色
'DEBUG': 'cyan',
'INFO': 'white',
'WARNING': 'yellow',
'ERROR': 'red',
'CRITICAL': 'red,bg_black',
}
},
style='%'
)
# 创建普通格式的格式化器(用于日志文件)
file_formatter = logging.Formatter(
"%(asctime)s - %(filename)s:%(lineno)d - %(levelname)s:%(name)s:%(message)s",
datefmt="%Y-%m-%d %H:%M:%S"
)
# 设置格式化器
console_handler.setFormatter(console_formatter)
# 将处理器添加到 logger
logger.addHandler(console_handler)
def set_logger_file_handler(p):
# 创建一个文件处理器,保存日志到本地文件
file_handler = logging.FileHandler(f'{p}/run.log', encoding='utf-8')
file_handler.setFormatter(file_formatter)
logger.addHandler(file_handler)
# 重定向 stdout 和 stderr 到日志系统
class LoggerWriter:
def __init__(self, level):
self.level = level
def write(self, message):
if message.strip(): # 过滤掉空行
self.level(message.strip()) # 去掉首尾空白字符
def flush(self):
pass
sys.stdout = LoggerWriter(logger.info)
sys.stderr = LoggerWriter(logger.error)
# 测试日志
# logger.success("This is a debug message")
# logger.debug("This is a debug message")
# logger.info("This is an info message")
# logger.warning("This is a warning message")
# logger.error("This is an error message")
# logger.critical("This is a critical message")