129 lines
4.5 KiB
Python
129 lines
4.5 KiB
Python
import os
|
||
import sys
|
||
from pathlib import Path
|
||
from loguru import logger
|
||
import platform
|
||
from datetime import datetime
|
||
import zipfile
|
||
|
||
|
||
class CrossPlatformLog:
|
||
"""跨平台日志系统（支持Linux/Windows/Mac）"""
|
||
|
||
def __init__(self):
|
||
self.log_dir = self._get_log_dir()
|
||
self._setup_logger()
|
||
|
||
def _get_log_dir(self):
|
||
"""获取跨平台日志目录（相对路径）"""
|
||
base_dir = Path(__file__).parent.parent # 项目根目录
|
||
log_dir = base_dir / "logs"
|
||
|
||
# 自动创建日志目录
|
||
log_dir.mkdir(exist_ok=True)
|
||
|
||
# Windows特殊权限处理
|
||
if platform.system() == "Windows":
|
||
try:
|
||
os.chmod(log_dir, 0o777) # 确保写入权限
|
||
except:
|
||
pass
|
||
|
||
return log_dir
|
||
|
||
def _setup_logger(self):
|
||
"""配置跨平台日志处理器"""
|
||
logger.remove() # 清除默认配置
|
||
|
||
|
||
# 统一控制台输出格式
|
||
logger.add(
|
||
sys.stdout,
|
||
level="INFO",
|
||
format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{module}</cyan> - <level>{message}</level>",
|
||
filter=lambda record: record["level"].no >= 20 # INFO及以上级别
|
||
)
|
||
|
||
# 主日志文件（兼容所有平台路径）
|
||
self._add_main_log()
|
||
|
||
# 错误日志单独存储
|
||
self._add_error_log()
|
||
|
||
def _add_main_log(self):
|
||
"""主日志文件配置"""
|
||
main_log = self.log_dir / "application.log"
|
||
logger.add(
|
||
str(main_log),
|
||
rotation="20 MB",
|
||
compression=self._compress_log,
|
||
encoding="utf-8",
|
||
level="DEBUG",
|
||
# 👇 增加 {extra} 输出，并美化结构
|
||
# format="{time:YYYY-MM-DD HH:mm:ss.SSS} | {level: <8} | {module}:{line} - {message}{extra_output}",
|
||
retention="30 days",
|
||
enqueue=True,
|
||
# 👇 动态处理 extra 字段为可读格式
|
||
format=self._format_with_extra, # 使用自定义格式函数
|
||
)
|
||
|
||
def _format_with_extra(self, record):
|
||
# 构造 extra 的可读字符串
|
||
extra_str = ""
|
||
if record["extra"]:
|
||
extra_items = []
|
||
for key, value in record["extra"].items():
|
||
if key == "extra_output": # 跳过自己，避免递归
|
||
continue
|
||
value_repr = repr(value)
|
||
# 对于错误信息，增加截断长度限制，避免丢失重要信息
|
||
if key in ["error", "error_message", "sql", "params"]:
|
||
if len(value_repr) > 500:
|
||
value_repr = value_repr[:497] + "..."
|
||
elif len(value_repr) > 200:
|
||
value_repr = value_repr[:197] + "..."
|
||
extra_items.append(f"\n → {key}: {value_repr}")
|
||
extra_str = "".join(extra_items)
|
||
|
||
# 👉 直接将 extra_str 写入 message 或附加字段
|
||
record["extra"]["extra_output"] = extra_str
|
||
|
||
# ✅ 关键：返回的 format 字符串不再引用 {extra_output}，而是使用 {extra[extra_output]}
|
||
return "{time:YYYY-MM-DD HH:mm:ss.SSS} | {level: <8} | {module}:{line} - {message}{extra[extra_output]}\n"
|
||
def _add_error_log(self):
|
||
"""错误日志专用配置"""
|
||
error_log = self.log_dir / "errors.log"
|
||
logger.add(
|
||
str(error_log),
|
||
level="ERROR",
|
||
format="{time:YYYY-MM-DD HH:mm:ss.SSS} | ERROR | {module}:{line} - {message}{extra[extra_output]}\n{exception}",
|
||
rotation="10 MB",
|
||
retention="90 days",
|
||
enqueue=True
|
||
)
|
||
|
||
@staticmethod
|
||
def _compress_log(log_path):
|
||
"""通用日志压缩方法（兼容所有平台）"""
|
||
if not os.path.exists(log_path):
|
||
return
|
||
|
||
try:
|
||
zip_path = f"{log_path}.{datetime.now().strftime('%Y%m%d')}.zip"
|
||
with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
|
||
zipf.write(log_path, arcname=os.path.basename(log_path))
|
||
os.remove(log_path)
|
||
return zip_path
|
||
except Exception as e:
|
||
print(f"日志压缩失败: {str(e)}")
|
||
return log_path # 返回原文件路径继续使用
|
||
|
||
@classmethod
|
||
def get_logger(cls, module_name=None):
|
||
"""获取模块专属日志器"""
|
||
return logger.bind(module=module_name or "__main__")
|
||
|
||
|
||
# 初始化全局日志器
|
||
log = CrossPlatformLog().get_logger()
|