#!/usr/bin/env python3 # -*- coding: utf-8 -*- import logging import logging.handlers import os import asyncio from pathlib import Path from typing import Dict, List, Callable import json from datetime import datetime logger = logging.getLogger(__name__) class LogService: """日志服务""" def __init__(self, config: Dict): self.config = config self.log_consumers: List[Callable] = [] self.log_buffer: List[Dict] = [] self.buffer_size = 100 self.log_dir = Path("logs") self.is_initialized = False self._in_emit = False # 文件数量限制 self.max_log_files_per_folder = self.config['logging']['max_log_files'] # 每个文件夹文件数量上限 # 立即初始化日志系统 self._setup_logging_sync() def _setup_logging_sync(self): """同步设置日志系统 - 增强版""" try: # 创建日志目录 self.log_dir.mkdir(exist_ok=True) (self.log_dir / "debug").mkdir(exist_ok=True) (self.log_dir / "runtime").mkdir(exist_ok=True) # 清理旧日志文件（在创建新文件之前） self._cleanup_old_log_files() # 生成基于时间戳和会话ID的日志文件名 timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') session_id = os.urandom(4).hex() # 生成8位随机会话ID runtime_log_file = f"framework_{timestamp}_{session_id}.log" debug_log_file = f"debug_{timestamp}_{session_id}.log" # 保存当前会话的日志文件名（用于后续引用） self.current_session_logs = { 'runtime': runtime_log_file, 'debug': debug_log_file, 'timestamp': timestamp, 'session_id': session_id } # 配置根日志记录器 root_logger = logging.getLogger() root_logger.setLevel(logging.DEBUG) # 清除现有处理器 for handler in root_logger.handlers[:]: root_logger.removeHandler(handler) # 创建过滤器实例 shared_filter = self.SafeLogFilter(self) # 控制台处理器 console_handler = logging.StreamHandler() console_handler.setLevel(getattr(logging, self.config['logging']['level'], logging.INFO)) console_formatter = logging.Formatter( '%(asctime)s [%(levelname)-8s] %(name)s: %(message)s', datefmt='%H:%M:%S' ) console_handler.setFormatter(console_formatter) console_handler.addFilter(shared_filter) root_logger.addHandler(console_handler) # 文件处理器 if self.config['logging'].get('enable_file_logging', True): file_formatter = logging.Formatter( '%(asctime)s [%(levelname)-8s] %(name)s:%(lineno)d - %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) # 运行时文件处理器 runtime_handler = logging.handlers.RotatingFileHandler( self.log_dir / "runtime" / runtime_log_file, maxBytes=self._parse_size(self.config['logging'].get('max_file_size', '10MB')), backupCount=self.config['logging'].get('max_log_files', 3) ) runtime_handler.setLevel(getattr(logging, self.config['logging']['level'], logging.INFO)) runtime_handler.setFormatter(file_formatter) runtime_handler.addFilter(shared_filter) root_logger.addHandler(runtime_handler) # Debug文件处理器 if self.config['logging'].get('debug_level_file', True): debug_handler = logging.handlers.RotatingFileHandler( self.log_dir / "debug" / debug_log_file, maxBytes=self._parse_size(self.config['logging'].get('max_file_size', '10MB')), backupCount=self.config['logging'].get('max_log_files', 3) ) debug_handler.setLevel(logging.DEBUG) debug_handler.setFormatter(file_formatter) debug_handler.addFilter(shared_filter) root_logger.addHandler(debug_handler) self.is_initialized = True logger.info(f"✅ 日志系统初始化完成 - 会话ID: {session_id}") logger.info(f"📝 运行时日志: logs/runtime/{runtime_log_file}") logger.info(f"🐛 调试日志: logs/debug/{debug_log_file}") except Exception as e: logger.error(f"日志系统设置失败: {e}") # 回退到基础配置 logging.basicConfig(level=logging.INFO, format='%(levelname)s: %(message)s') self.is_initialized = True def _cleanup_old_log_files(self): """清理旧的日志文件，保持每个文件夹文件上限""" try: logger.debug("开始清理旧日志文件...") # 清理 runtime 文件夹 runtime_dir = self.log_dir / "runtime" if runtime_dir.exists(): runtime_files = list(runtime_dir.glob("*.log")) self._remove_old_files(runtime_files, "runtime") # 清理 debug 文件夹 debug_dir = self.log_dir / "debug" if debug_dir.exists(): debug_files = list(debug_dir.glob("*.log")) self._remove_old_files(debug_files, "debug") except Exception as e: logger.error(f"清理旧日志文件时出错: {e}") def _remove_old_files(self, files: List[Path], folder_name: str): """删除最旧的文件，直到文件数量不超过限制""" try: if len(files) <= self.max_log_files_per_folder: logger.debug(f"{folder_name} 文件夹文件数量正常: {len(files)}/{self.max_log_files_per_folder}") return # 按修改时间排序（最旧的在前） files_sorted = sorted(files, key=lambda x: x.stat().st_mtime) # 计算需要删除的文件数量 files_to_remove = len(files_sorted) - self.max_log_files_per_folder if files_to_remove > 0: logger.info(f"清理 {folder_name} 文件夹: 删除 {files_to_remove} 个旧日志文件") for i in range(files_to_remove): old_file = files_sorted[i] try: old_file.unlink() logger.debug(f"删除旧日志文件: {old_file.name}") except Exception as e: logger.error(f"删除文件失败 {old_file}: {e}") logger.info(f"{folder_name} 文件夹清理完成: {self.max_log_files_per_folder} 个文件") except Exception as e: logger.error(f"删除 {folder_name} 文件夹旧文件时出错: {e}") def cleanup_log_files(self): """手动清理日志文件（可以定期调用）""" try: logger.info("开始手动清理日志文件...") self._cleanup_old_log_files() logger.info("日志文件清理完成") except Exception as e: logger.error(f"手动清理日志文件时出错: {e}") def get_log_file_counts(self) -> Dict[str, int]: """获取当前日志文件数量统计""" try: runtime_count = len(list((self.log_dir / "runtime").glob("*.log"))) debug_count = len(list((self.log_dir / "debug").glob("*.log"))) return { "runtime": runtime_count, "debug": debug_count, "max_limit": self.max_log_files_per_folder } except Exception as e: logger.error(f"获取日志文件统计时出错: {e}") return {"runtime": 0, "debug": 0, "max_limit": self.max_log_files_per_folder} def get_current_session_info(self) -> Dict: """获取当前会话的日志信息""" return getattr(self, 'current_session_logs', {}) def _parse_size(self, size_str: str) -> int: """解析文件大小字符串""" try: units = {'B': 1, 'KB': 1024, 'MB': 1024**2, 'GB': 1024**3} number = ''.join(filter(str.isdigit, size_str)) unit = ''.join(filter(str.isalpha, size_str)).upper() return int(number) * units.get(unit, 1) except Exception: return 10 * 1024 * 1024 def add_log_consumer(self, callback: Callable): """添加日志消费者""" if callback not in self.log_consumers: self.log_consumers.append(callback) logger.debug(f"添加日志消费者，总数: {len(self.log_consumers)}") def emit_log(self, log_record: Dict): """发射日志到消费者""" if self._in_emit: return self._in_emit = True try: # 格式化日志记录 timestamp = log_record['timestamp'] level = log_record['level'] name = log_record['name'] message = log_record['message'] # 转换时间戳 if isinstance(timestamp, (int, float)): timestamp_str = datetime.fromtimestamp(timestamp).strftime('%H:%M:%S') else: timestamp_str = str(timestamp) formatted_record = { 'timestamp': timestamp, 'timestamp_str': timestamp_str, 'level': level, 'name': name, 'message': message, 'module': log_record.get('module', ''), 'line': log_record.get('line', 0), 'formatted_message': f"{timestamp_str} [{level:8}] {name}: {message}", 'simple_message': f"{timestamp_str} [{level:8}] {message}", 'original_message': message } # 添加到缓冲区 self.log_buffer.append(formatted_record) if len(self.log_buffer) > self.buffer_size: self.log_buffer.pop(0) # 发送给消费者 for consumer in self.log_consumers: try: # 检查TUI级别过滤 if hasattr(consumer, '_tui_level_filter'): tui_level = consumer._tui_level_filter if self._should_display_for_tui(level, tui_level): consumer(formatted_record) else: # 默认发送所有日志 consumer(formatted_record) except Exception as e: logger.error(f"日志消费者处理出错: {e}") except Exception as e: logger.error(f"发射日志时出错: {e}") finally: self._in_emit = False def _should_display_for_tui(self, log_level: str, tui_level: str) -> bool: """检查日志是否应该显示在TUI中""" level_priority = { 'DEBUG': 10, 'INFO': 20, 'WARNING': 30, 'ERROR': 40, 'CRITICAL': 50 } log_priority = level_priority.get(log_level, 0) tui_priority = level_priority.get(tui_level, 0) return log_priority >= tui_priority def add_tui_log_consumer(self, callback: Callable, tui_level: str = "INFO"): """专门为TUI添加日志消费者""" callback._tui_level_filter = tui_level if callback not in self.log_consumers: self.log_consumers.append(callback) logger.debug(f"添加TUI日志消费者，级别: {tui_level}") def get_recent_logs(self, count: int = 50) -> List[Dict]: """获取最近的日志""" return self.log_buffer[-count:] def shutdown(self): """关闭日志服务""" self.log_consumers.clear() logging.shutdown() logger.info("日志服务已关闭") class SafeLogFilter(logging.Filter): """安全的日志过滤器""" def __init__(self, log_service): super().__init__() self.log_service = log_service self._in_filter = False def filter(self, record): """过滤日志记录""" if self._in_filter: return True self._in_filter = True try: # 创建日志记录 log_record = { 'timestamp': record.created, 'name': record.name, 'level': record.levelname, 'message': record.getMessage(), 'module': record.module, 'line': record.lineno } # 发射日志记录 self.log_service.emit_log(log_record) except Exception: pass finally: self._in_filter = False return True