111 lines
4.6 KiB
Python
111 lines
4.6 KiB
Python
from ruamel.yaml import YAML
|
||
from os.path import join, exists
|
||
from loguru import logger
|
||
from os import makedirs
|
||
import os, sys
|
||
import logging
|
||
from pathlib import Path
|
||
|
||
from utils.general import gradient_text
|
||
|
||
# 获取项目根目录
|
||
PROJECT_ROOT = Path(__file__).parent.parent.absolute()
|
||
|
||
# 自定义格式化函数
|
||
def relative_path_formatter(record):
|
||
"""将绝对路径转换为相对路径"""
|
||
try:
|
||
# 获取文件的绝对路径
|
||
abs_path = Path(record["file"].path)
|
||
# 转换为相对于项目根目录的路径
|
||
rel_path = abs_path.relative_to(PROJECT_ROOT)
|
||
record["extra"]["relative_path"] = str(rel_path)
|
||
except (ValueError, AttributeError):
|
||
# 如果转换失败，使用原文件名
|
||
record["extra"]["relative_path"] = record["file"].name
|
||
return record
|
||
|
||
# 定义一个日志处理器，将标准日志转发到loguru
|
||
class LoguruHandler(logging.Handler):
|
||
def emit(self, record):
|
||
# 获取日志记录的级别和消息
|
||
level = record.levelname
|
||
message = self.format(record)
|
||
|
||
# 获取原始日志记录的位置信息
|
||
frame_data = {
|
||
"file": record.filename,
|
||
"line": record.lineno,
|
||
"function": record.funcName
|
||
}
|
||
|
||
# 将标准日志转发到loguru，并使用原始位置信息
|
||
# logger.patch(lambda record: record.update(frame_data)).log(level, f'[后台消息] - {message}')
|
||
|
||
def getConfigs(path, read_type="yml"):
|
||
"""
|
||
读取配置文件并返回解析后的配置信息
|
||
|
||
:param path: 配置文件路径
|
||
:param read_type: 配置文件类型，默认为'yml'，可选'json'或'yml'
|
||
:return: 解析后的配置信息，JSON格式返回字典，YML格式返回对应的数据结构
|
||
:raises Exception: 当无法获取配置信息时抛出异常
|
||
"""
|
||
path = path.replace("\\", "/")
|
||
yaml = YAML(typ="safe", pure=True)
|
||
with open(path, "r", encoding="utf-8") as f:
|
||
return yaml.load(f)
|
||
# 如果未成功读取配置信息，则抛出异常
|
||
raise Exception("路径: %s未获取配置信息" % path)
|
||
def init_logger(base_dir: str):
|
||
log_config = getConfigs(join(base_dir, "webIOs/configs/th_agenter_config_logger.yml"))
|
||
# 判断日志文件是否存在，不存在创建
|
||
base_path = join(base_dir, log_config.get("base_path"))
|
||
if not exists(base_path):
|
||
makedirs(base_path)
|
||
# 移除日志设置
|
||
logger.remove(handler_id=None)
|
||
# 打印日志到文件
|
||
if bool(log_config.get("enable_file_log")):
|
||
logger.add(
|
||
join(base_path, log_config.get("log_name")),
|
||
rotation=log_config.get("rotation"),
|
||
retention=log_config.get("retention"),
|
||
format=log_config.get("log_fmt"),
|
||
level=log_config.get("level"),
|
||
enqueue=True,
|
||
encoding=log_config.get("encoding"),
|
||
filter=relative_path_formatter,
|
||
diagnose=True
|
||
)
|
||
# 控制台输出
|
||
if bool(log_config.get("enable_stderr")):
|
||
logger.add(
|
||
sys.stderr,
|
||
format=log_config.get("log_fmt"),
|
||
level=log_config.get("level"),
|
||
enqueue=True,
|
||
filter=relative_path_formatter,
|
||
diagnose=True
|
||
)
|
||
|
||
log_level = log_config.get("level", "INFO")
|
||
logging.basicConfig(handlers=[LoguruHandler()], level=log_level)
|
||
|
||
# 特别配置uvicorn和fastapi的日志记录器
|
||
for logger_name in ["uvicorn", "uvicorn.access", "uvicorn.error", "fastapi"]:
|
||
uvicorn_logger = logging.getLogger(logger_name)
|
||
uvicorn_logger.handlers = [LoguruHandler()]
|
||
uvicorn_logger.propagate = False
|
||
|
||
|
||
print("\n\n\n")
|
||
logger.info(f"----=========== {gradient_text('日志配置初始化完成, 开始新的日志记录')} ===========----")
|
||
logger.info("🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀")
|
||
logger.info("🚀 DDDD RRRR GGGG RRRR AAA PPPPP H H ")
|
||
logger.info("🚀 D D R R G R R A A P P H H ")
|
||
logger.info("🚀 D D RRRR GGGGG RRRR AAAAA PPPPP HHHHH ")
|
||
logger.info("🚀 D D R R G G R R A A P H H ")
|
||
logger.info("🚀 DDDD R R GGGGG R R A A P H H ")
|
||
logger.info("🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀🚀")
|