log更新
This commit is contained in:
+76
-107
@@ -1,129 +1,98 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
跨平台日志工具模块
|
||||
功能：
|
||||
1. 自动适配不同操作系统的日志路径
|
||||
2. 支持中文等非ASCII字符
|
||||
3. 日志文件自动按日期分割
|
||||
4. 控制台与文件双输出
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import logging
|
||||
from logging.handlers import TimedRotatingFileHandler
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from loguru import logger
|
||||
import platform
|
||||
from datetime import datetime
|
||||
import zipfile
|
||||
|
||||
class CrossPlatformLog:
|
||||
"""跨平台日志系统（支持Linux/Windows/Mac）"""
|
||||
|
||||
class CrossPlatformLogger:
|
||||
def __init__(self, name="intelligence_system"):
|
||||
"""
|
||||
初始化跨平台日志系统
|
||||
def __init__(self):
|
||||
self.log_dir = self._get_log_dir()
|
||||
self._setup_logger()
|
||||
|
||||
参数：
|
||||
name: 日志名称（用于创建日志文件夹）
|
||||
"""
|
||||
self.system = platform.system().lower()
|
||||
self.logger = logging.getLogger(name)
|
||||
self.logger.setLevel(logging.INFO)
|
||||
def _get_log_dir(self):
|
||||
"""获取跨平台日志目录（相对路径）"""
|
||||
base_dir = Path(__file__).parent.parent # 项目根目录
|
||||
log_dir = base_dir / "logs"
|
||||
|
||||
# 确保日志目录存在
|
||||
self.log_dir = self._get_log_dir(name)
|
||||
os.makedirs(self.log_dir, exist_ok=True)
|
||||
# 自动创建日志目录
|
||||
log_dir.mkdir(exist_ok=True)
|
||||
|
||||
# 配置日志格式
|
||||
formatter = logging.Formatter(
|
||||
'%(asctime)s - %(name)s - %(levelname)s - %(message)s',
|
||||
datefmt='%Y-%m-%d %H:%M:%S'
|
||||
)
|
||||
|
||||
# 控制台处理器
|
||||
self._setup_console_handler(formatter)
|
||||
|
||||
# 文件处理器（按天分割）
|
||||
self._setup_file_handler(formatter)
|
||||
|
||||
# 处理未捕获的异常
|
||||
sys.excepthook = self.handle_uncaught_exception
|
||||
|
||||
def _get_log_dir(self, name: str) -> str:
|
||||
"""获取适合当前平台的日志目录路径"""
|
||||
if self.system == 'windows':
|
||||
base_dir = os.path.join(os.environ['APPDATA'], name)
|
||||
elif self.system == 'darwin': # macOS
|
||||
base_dir = os.path.expanduser(f"~/Library/Logs/{name}")
|
||||
else: # Linux及其他Unix-like系统
|
||||
base_dir = f"/var/log/{name}" if os.access("/var/log", os.W_OK) \
|
||||
else os.path.expanduser(f"~/.local/share/{name}")
|
||||
|
||||
return base_dir
|
||||
|
||||
def _setup_console_handler(self, formatter: logging.Formatter):
|
||||
"""配置控制台输出（兼容不同终端的编码）"""
|
||||
console = logging.StreamHandler()
|
||||
|
||||
# Windows终端特殊处理
|
||||
if self.system == 'windows' and not sys.stdout.isatty():
|
||||
import colorama
|
||||
# Windows特殊权限处理
|
||||
if platform.system() == "Windows":
|
||||
try:
|
||||
colorama.init()
|
||||
except ImportError:
|
||||
os.chmod(log_dir, 0o777) # 确保写入权限
|
||||
except:
|
||||
pass
|
||||
|
||||
# 解决Windows控制台编码问题
|
||||
if sys.stdout.encoding != 'utf-8':
|
||||
import io
|
||||
sys.stdout = io.TextIOWrapper(
|
||||
sys.stdout.buffer,
|
||||
encoding='utf-8',
|
||||
errors='replace'
|
||||
)
|
||||
return log_dir
|
||||
|
||||
console.setFormatter(formatter)
|
||||
self.logger.addHandler(console)
|
||||
def _setup_logger(self):
|
||||
"""配置跨平台日志处理器"""
|
||||
logger.remove() # 清除默认配置
|
||||
|
||||
def _setup_file_handler(self, formatter: logging.Formatter):
|
||||
"""配置日志文件输出（UTF-8编码）"""
|
||||
log_file = os.path.join(
|
||||
self.log_dir,
|
||||
f"{datetime.now().strftime('%Y%m%d')}.log"
|
||||
# 统一控制台输出格式
|
||||
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及以上级别
|
||||
)
|
||||
|
||||
# 使用TimedRotatingFileHandler实现日志分割
|
||||
file_handler = TimedRotatingFileHandler(
|
||||
filename=log_file,
|
||||
when='midnight', # 每天午夜分割
|
||||
encoding='utf-8',
|
||||
backupCount=30 # 保留30天日志
|
||||
)
|
||||
file_handler.setFormatter(formatter)
|
||||
self.logger.addHandler(file_handler)
|
||||
# 主日志文件（兼容所有平台路径）
|
||||
self._add_main_log()
|
||||
|
||||
def handle_uncaught_exception(self, exc_type, exc_value, exc_traceback):
|
||||
"""全局异常捕获"""
|
||||
self.logger.error(
|
||||
"未捕获的异常:",
|
||||
exc_info=(exc_type, exc_value, exc_traceback)
|
||||
# 错误日志单独存储
|
||||
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",
|
||||
format="{time:YYYY-MM-DD HH:mm:ss.SSS} | {level: <8} | {module}:{line} - {message}",
|
||||
retention="30 days",
|
||||
enqueue=True # 线程安全
|
||||
)
|
||||
|
||||
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}\n{exception}",
|
||||
rotation="10 MB",
|
||||
retention="90 days"
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def get_logger(name: str = None) -> logging.Logger:
|
||||
"""获取配置好的日志实例"""
|
||||
return CrossPlatformLogger(name).logger
|
||||
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 # 返回原文件路径继续使用
|
||||
|
||||
def setup_logging(name: str = "intelligence_system"):
|
||||
"""快速配置日志（兼容旧代码）"""
|
||||
return CrossPlatformLogger(name).logger
|
||||
@classmethod
|
||||
def get_logger(cls, module_name=None):
|
||||
"""获取模块专属日志器"""
|
||||
return logger.bind(module=module_name or "__main__")
|
||||
|
||||
|
||||
# 测试代码
|
||||
if __name__ == "__main__":
|
||||
logger = CrossPlatformLogger().logger
|
||||
logger.info("这是一条info日志（包含中文测试）")
|
||||
try:
|
||||
1 / 0
|
||||
except Exception as e:
|
||||
logger.error("除零错误示例", exc_info=True)
|
||||
# 初始化全局日志器
|
||||
log = CrossPlatformLog().get_logger()
|
||||
Reference in New Issue
Block a user