""" 日志工具模块 提供统一的日志配置和管理功能 """ import logging import sys from pathlib import Path from datetime import datetime from logging.handlers import RotatingFileHandler, TimedRotatingFileHandler from typing import Optional class ColoredFormatter(logging.Formatter): """ 彩色日志格式化器 为不同级别的日志添加颜色 """ # ANSI 颜色代码 COLORS = { 'DEBUG': '\033[36m', # 青色 'INFO': '\033[32m', # 绿色 'WARNING': '\033[33m', # 黄色 'ERROR': '\033[31m', # 红色 'CRITICAL': '\033[35m', # 紫色 } RESET = '\033[0m' def format(self, record): # 添加颜色 if record.levelname in self.COLORS: record.levelname = f"{self.COLORS[record.levelname]}{record.levelname}{self.RESET}" return super().format(record) class LoggerManager: """ 日志管理器 负责配置和管理应用程序日志 """ def __init__( self, name: str = "CutThenThink", log_dir: Optional[Path] = None, level: str = "INFO", console_output: bool = True, file_output: bool = True, colored_console: bool = True ): """ 初始化日志管理器 Args: name: 日志器名称 log_dir: 日志文件目录 level: 日志级别 (DEBUG, INFO, WARNING, ERROR, CRITICAL) console_output: 是否输出到控制台 file_output: 是否输出到文件 colored_console: 控制台是否使用彩色输出 """ self.name = name self.log_dir = log_dir self.level = getattr(logging, level.upper(), logging.INFO) self.console_output = console_output self.file_output = file_output self.colored_console = colored_console self.logger: Optional[logging.Logger] = None def setup(self) -> logging.Logger: """ 设置日志系统 Returns: 配置好的 Logger 对象 """ if self.logger is not None: return self.logger # 创建日志器 self.logger = logging.getLogger(self.name) self.logger.setLevel(self.level) self.logger.handlers.clear() # 清除已有的处理器 # 日志格式 log_format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s' date_format = '%Y-%m-%d %H:%M:%S' # 控制台处理器 if self.console_output: console_handler = logging.StreamHandler(sys.stdout) console_handler.setLevel(self.level) if self.colored_console: console_formatter = ColoredFormatter(log_format, datefmt=date_format) else: console_formatter = logging.Formatter(log_format, datefmt=date_format) console_handler.setFormatter(console_formatter) self.logger.addHandler(console_handler) # 文件处理器 if self.file_output and self.log_dir: # 确保日志目录存在 self.log_dir.mkdir(parents=True, exist_ok=True) # 主日志文件（按大小轮转） log_file = self.log_dir / f"{self.name}.log" file_handler = RotatingFileHandler( log_file, maxBytes=10 * 1024 * 1024, # 10MB backupCount=5, encoding='utf-8' ) file_handler.setLevel(self.level) file_formatter = logging.Formatter(log_format, datefmt=date_format) file_handler.setFormatter(file_formatter) self.logger.addHandler(file_handler) # 错误日志文件（单独记录错误和严重错误） error_file = self.log_dir / f"{self.name}_error.log" error_handler = RotatingFileHandler( error_file, maxBytes=10 * 1024 * 1024, # 10MB backupCount=5, encoding='utf-8' ) error_handler.setLevel(logging.ERROR) error_formatter = logging.Formatter(log_format, datefmt=date_format) error_handler.setFormatter(error_formatter) self.logger.addHandler(error_handler) return self.logger def get_logger(self) -> logging.Logger: """ 获取日志器 Returns: Logger 对象 """ if self.logger is None: return self.setup() return self.logger def set_level(self, level: str): """ 动态设置日志级别 Args: level: 日志级别字符串 """ log_level = getattr(logging, level.upper(), logging.INFO) self.logger.setLevel(log_level) for handler in self.logger.handlers: handler.setLevel(log_level) # 全局日志管理器 _global_logger_manager: Optional[LoggerManager] = None def init_logger( log_dir: Optional[Path] = None, level: str = "INFO", console_output: bool = True, file_output: bool = True, colored_console: bool = True ) -> logging.Logger: """ 初始化全局日志系统 Args: log_dir: 日志目录 level: 日志级别 console_output: 是否输出到控制台 file_output: 是否输出到文件 colored_console: 控制台是否彩色 Returns: Logger 对象 """ global _global_logger_manager # 默认日志目录 if log_dir is None: project_root = Path(__file__).parent.parent.parent log_dir = project_root / "logs" _global_logger_manager = LoggerManager( name="CutThenThink", log_dir=log_dir, level=level, console_output=console_output, file_output=file_output, colored_console=colored_console ) return _global_logger_manager.setup() def get_logger(name: Optional[str] = None) -> logging.Logger: """ 获取日志器 Args: name: 日志器名称，如果为 None 则返回全局日志器 Returns: Logger 对象 """ if _global_logger_manager is None: init_logger() if name is None: return _global_logger_manager.get_logger() # 返回指定名称的子日志器 return logging.getLogger(f"CutThenThink.{name}") class LogCapture: """ 日志捕获器 用于捕获日志并显示在 GUI 中 """ def __init__(self, max_entries: int = 1000): """ 初始化日志捕获器 Args: max_entries: 最大保存条目数 """ self.max_entries = max_entries self.entries = [] self.callbacks = [] def add_entry(self, level: str, message: str, timestamp: Optional[datetime] = None): """ 添加日志条目 Args: level: 日志级别 message: 日志消息 timestamp: 时间戳 """ if timestamp is None: timestamp = datetime.now() entry = { 'timestamp': timestamp, 'level': level, 'message': message } self.entries.append(entry) # 限制条目数量 if len(self.entries) > self.max_entries: self.entries = self.entries[-self.max_entries:] # 触发回调 for callback in self.callbacks: callback(entry) def register_callback(self, callback): """ 注册回调函数 Args: callback: 回调函数，接收 entry 参数 """ self.callbacks.append(callback) def clear(self): """清空日志""" self.entries.clear() def get_entries(self, level: Optional[str] = None, limit: Optional[int] = None) -> list: """ 获取日志条目 Args: level: 过滤级别，None 表示不过滤 limit: 限制数量 Returns: 日志条目列表 """ entries = self.entries if level is not None: entries = [e for e in entries if e['level'] == level] if limit is not None: entries = entries[-limit:] return entries def get_latest(self, count: int = 10) -> list: """ 获取最新的 N 条日志 Args: count: 数量 Returns: 日志条目列表 """ return self.entries[-count:] # 全局日志捕获器 _log_capture: Optional[LogCapture] = None def get_log_capture() -> LogCapture: """ 获取全局日志捕获器 Returns: LogCapture 对象 """ global _log_capture if _log_capture is None: _log_capture = LogCapture() return _log_capture class LogHandler(logging.Handler): """ 自定义日志处理器 将日志发送到 LogCapture """ def __init__(self, capture: LogCapture): super().__init__() self.capture = capture def emit(self, record): """ 发出日志记录 Args: record: 日志记录 """ try: message = self.format(record) level = record.levelname timestamp = datetime.fromtimestamp(record.created) self.capture.add_entry(level, message, timestamp) except Exception: self.handleError(record) def setup_gui_logging(capture: Optional[LogCapture] = None): """ 设置 GUI 日志捕获 Args: capture: 日志捕获器，如果为 None 则使用全局捕获器 """ if capture is None: capture = get_log_capture() # 创建处理器 handler = LogHandler(capture) handler.setLevel(logging.INFO) handler.setFormatter(logging.Formatter('%(message)s')) # 添加到根日志器 logging.getLogger("CutThenThink").addHandler(handler) # 便捷函数 def log_debug(message: str): """记录 DEBUG 日志""" get_logger().debug(message) def log_info(message: str): """记录 INFO 日志""" get_logger().info(message) def log_warning(message: str): """记录 WARNING 日志""" get_logger().warning(message) def log_error(message: str, exc_info: bool = False): """记录 ERROR 日志""" get_logger().error(message, exc_info=exc_info) def log_critical(message: str, exc_info: bool = False): """记录 CRITICAL 日志""" get_logger().critical(message, exc_info=exc_info)