| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121 |
- import logging
- from datetime import datetime
- from pathlib import Path
- import atexit
- from typing import Optional
- from Config.Config import Config
- class Logger:
- def __init__(self, log_path: Optional[str] = None, log_name: Optional[str] = None):
- """
- 初始化日志记录器
- Args:
- log_path: 日志存储路径,默认为项目根目录下的Log文件夹
- log_name: 日志文件名,默认为当前日期时间
- """
- # 1. 处理日志路径(默认项目根目录/Logs,不存在则创建)
- if log_path is None:
- project_root = self._find_project_root()
- self.log_path = project_root / "Logs"
- else:
- self.log_path = Path(log_path)
- self.log_path.mkdir(parents=True, exist_ok=True) # 递归创建目录
- # 2. 处理日志文件名(默认格式:log_年月日_时分秒.log)
- if log_name is None:
- log_name = f"log_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log"
- self.log_file = self.log_path / log_name
- # 3. 初始化日志记录器(避免多实例冲突)
- self.logger = logging.getLogger(f"{__name__}_{id(self)}")
- self.logger.setLevel(Config.log_level)
- self.logger.handlers.clear() # 清除默认处理器,防止重复记录
- # 4. 创建文件日志处理器(核心功能,UTF-8编码避免中文乱码)
- self.file_handler = None
- try:
- self.file_handler = logging.FileHandler(self.log_file, encoding='utf-8')
- self.file_handler.setLevel(logging.DEBUG) # 文件记录所有级别日志
- # 应用配置的日志格式
- formatter = logging.Formatter(Config.log_format)
- self.file_handler.setFormatter(formatter)
- self.logger.addHandler(self.file_handler)
- except Exception as e:
- # 日志文件创建失败时,仅简单打印(无控制台日志,故用原始print)
- print(f"创建日志文件失败: {e}")
- # 5. 注册退出清理函数(程序结束时关闭日志文件)
- atexit.register(self.cleanup)
- # 6. 记录初始化完成日志
- if self._is_logging_available():
- self.logger.info(f"日志系统初始化完成,日志文件: {self.log_file}")
- def _find_project_root(self):
- """查找项目根目录(基于常见项目标记文件)"""
- current_dir = Path(__file__).resolve().parent
- # 向上遍历目录,直到找到项目根目录标记
- for parent in [current_dir] + list(current_dir.parents):
- if any((parent / marker).exists() for marker in
- ['.git', '.project', 'setup.py', 'requirements.txt', 'pyproject.toml']):
- return parent
- # 未找到标记时,使用当前工作目录
- return Path.cwd()
- def _is_logging_available(self):
- """检查日志系统是否可用"""
- return self.file_handler is not None and self.file_handler in self.logger.handlers
- def cleanup(self):
- """清理资源:关闭日志文件处理器"""
- if self.file_handler:
- try:
- self.file_handler.close()
- self.logger.removeHandler(self.file_handler)
- except Exception as e:
- print(f"关闭日志文件时出错: {e}")
- # ------------------------------ 核心日志方法------------------------------
- def debug(self, msg):
- if self._is_logging_available():
- self.logger.debug(msg)
- def info(self, msg):
- if self._is_logging_available():
- self.logger.info(msg)
- def warning(self, msg):
- if self._is_logging_available():
- self.logger.warning(msg)
- def error(self, msg):
- if self._is_logging_available():
- self.logger.error(msg)
- def critical(self, msg):
- if self._is_logging_available():
- self.logger.critical(msg)
- def exception(self, msg, exc_info=True):
- """记录异常(简化版:仅文件日志,含堆栈)"""
- if self._is_logging_available():
- self.logger.exception(msg, exc_info=exc_info)
- def log_exception(self, e, context=""):
- """记录异常及上下文(简化版:仅文件日志)"""
- if self._is_logging_available():
- error_msg = f"{context}: {type(e).__name__}: {str(e)}"
- self.error(error_msg)
- # 手动添加堆栈跟踪
- import traceback
- self.error("堆栈跟踪:\n" + "".join(traceback.format_tb(e.__traceback__)))
- # ------------------------------ 快捷日志创建方法------------------------------
- def log_for_api():
- return Logger(log_path="Logs/API", log_name="api_test.log")
- def log_for_ui():
- return Logger(log_path="Logs/UI", log_name="ui_test.log")
|