AI Intervention Agent

""" 增强的日志系统模块 本模块提供企业级日志功能，解决 MCP 服务器环境中的日志问题，包括重复输出、级别错误、 安全风险和性能问题。 核心功能 -------- 1. **单例日志管理器**: 防止 logger 重复初始化导致的日志重复输出 2. **多流输出策略**: 基于日志级别智能路由到不同输出流（全部输出到 stderr） 3. **日志脱敏处理**: 自动检测和脱敏密码、API 密钥等敏感信息 4. **注入攻击防护**: 转义危险字符，防止日志分割和注入攻击 5. **日志去重机制**: 在时间窗口内去除重复日志，减少噪声 6. **智能级别映射**: 根据消息内容自动调整日志级别 主要组件 -------- - SingletonLogManager: 单例日志管理器，确保每个 logger 只初始化一次 - LevelBasedStreamHandler: 基于级别的多流输出处理器（所有输出到 stderr） - LogSanitizer: 日志脱敏处理器，保护敏感信息 - SecureLogFormatter: 安全日志格式化器，集成脱敏功能 - AntiInjectionFilter: 防注入过滤器，转义危险字符 - LogDeduplicator: 日志去重器，在时间窗口内去除重复日志 - EnhancedLogger: 增强日志记录器，集成所有功能的高级接口 设计原则 -------- 1. **MCP 友好**: 所有日志输出到 stderr，避免污染 MCP stdio 通信通道 2. **线程安全**: 使用锁保护共享状态，支持多线程环境 3. **性能优化**: 去重缓存自动清理，限制最大缓存大小 4. **安全第一**: 默认启用脱敏和注入防护 5. **易用性**: 提供与标准 logging 兼容的 API 使用场景 -------- - MCP 服务器日志记录（避免污染 stdio） - 多线程环境的安全日志 - 需要脱敏的敏感信息日志 - 高频日志场景（需要去重） - 防止日志注入攻击的安全日志 典型用法 -------- 创建增强日志记录器并使用标准 API。 脱敏规则 -------- 自动脱敏以下类型的敏感信息： - 密码字段（password、passwd） - 密钥字段（secret_key、private_key） - OpenAI API key（sk-xxx） - Slack Bot Token（xoxb-xxx） - GitHub Personal Access Token（ghp_xxx） 去重规则 -------- - 时间窗口: 5 秒（可配置） - 缓存大小: 1000 条（可配置） - 重复计数: 自动附加 "(重复 N 次)" 信息 线程安全 -------- - SingletonLogManager 使用双重检查锁实现线程安全的单例 - LogDeduplicator 使用线程锁保护缓存操作 - 所有共享状态都使用 threading.Lock 保护 注意事项 -------- - 日志级别映射基于消息内容关键词匹配（可扩展） - 脱敏处理使用正则表达式，可能影响性能（已优化） - 去重缓存会占用内存，自动清理旧条目 - 所有日志输出到 stderr，确保 stdout 用于 MCP 通信 依赖 ---- - logging: Python 标准库日志模块 - threading: 线程安全保护 - re: 正则表达式（用于脱敏和注入防护） - 【性能优化】使用内置 hash() 代替 hashlib.md5，无需额外依赖 """ import json # noqa: F401 import logging import os # noqa: F401 import re import sys import threading import time from typing import Any, Dict, Optional, Set, Tuple # noqa: F401 class SingletonLogManager: """ 单例日志管理器 功能概述 -------- 确保每个 logger 只被初始化一次，防止重复注册 handler 导致的日志重复输出问题。 使用双重检查锁实现线程安全的单例模式。 核心特性 -------- 1. **单例模式**: 全局唯一实例，防止多次初始化 2. **Logger 去重**: 跟踪已初始化的 logger 名称 3. **自动配置**: 为每个 logger 自动配置多流输出和安全过滤器 4. **线程安全**: 使用锁保护初始化过程 内部状态 -------- - _instance: 单例实例（类变量） - _lock: 线程锁（类变量） - _initialized_loggers: 已初始化的 logger 名称集合（类变量） 单例实现 ---------- 使用双重检查锁（Double-Checked Locking）： - 第一次检查: 快速路径，避免不必要的锁竞争 - 加锁: 确保只有一个线程创建实例 - 第二次检查: 防止多个线程同时通过第一次检查 使用场景 -------- - 创建应用级日志记录器 - 确保日志不重复输出 - 多模块共享日志配置 注意事项 -------- - 一旦 logger 初始化，无法更改其配置 - 所有 logger 共享相同的 handler 配置 - 默认日志级别为 WARNING """ _instance = None _lock = threading.Lock() _initialized_loggers: Set[str] = set() def __new__(cls): """ 创建或返回单例实例（双重检查锁） 返回 ---- SingletonLogManager 全局唯一的管理器实例 线程安全 -------- 使用双重检查锁确保多线程环境下只创建一个实例 """ if cls._instance is None: with cls._lock: if cls._instance is None: cls._instance = super().__new__(cls) return cls._instance def setup_logger(self, name: str, level=logging.WARNING): """ 设置并返回已配置的 logger 实例 参数 ---- name : str logger 名称（通常使用 __name__） level : int, optional 日志级别，默认 logging.WARNING 返回 ---- logging.Logger 配置好的 logger 实例 功能 ---- 1. 检查 logger 是否已初始化（快速路径） 2. 如果未初始化，加锁并配置： - 清除现有 handler（避免重复） - 创建并附加 LevelBasedStreamHandler - 设置日志级别 - 禁用传播到父 logger（防止重复） - 标记为已初始化 3. 返回 logger 实例 线程安全 -------- 使用 _lock 保护初始化过程，确保每个 logger 只被配置一次。 始终在锁内检查并返回，避免快速路径的竞态条件。 注意事项 -------- - 重复调用返回相同的 logger 实例（已配置） - handler 配置无法更改（一次性初始化） - propagate=False 防止日志向上传播到 root logger - 移除快速路径，所有访问都在锁内进行，确保线程安全 """ # 始终加锁检查，避免快速路径的竞态条件 # 原逻辑的快速路径可能导致返回未完全初始化的 logger with self._lock: if name not in self._initialized_loggers: logger = logging.getLogger(name) # 清除现有处理器 logger.handlers.clear() # 使用多流输出策略 stream_handler = LevelBasedStreamHandler() stream_handler.attach_to_logger(logger) logger.setLevel(level) logger.propagate = False # 防止向父logger传播 self._initialized_loggers.add(name) return logging.getLogger(name) class LevelBasedStreamHandler: """ 基于日志级别的多流输出处理器 功能概述 -------- 创建两个 StreamHandler，根据日志级别将日志路由到不同的 handler： - DEBUG/INFO: 通过第一个 handler - WARNING/ERROR/CRITICAL: 通过第二个 handler 所有 handler 都输出到 stderr，避免污染 MCP 的 stdio 通信通道。 设计原因 -------- 在 MCP 环境中，stdout 用于协议通信，必须保持纯净。所有日志输出 （包括 INFO 级别）都应该输出到 stderr。 内部结构 ---------- - stdout_handler: 处理 DEBUG 和 INFO 级别（实际输出到 stderr） - stderr_handler: 处理 WARNING、ERROR、CRITICAL 级别（输出到 stderr） - formatter: SecureLogFormatter（集成脱敏功能） - anti_injection_filter: AntiInjectionFilter（防注入防护） 日志格式 -------- "%(asctime)s - %(name)s - %(levelname)s - %(message)s" 安全特性 -------- - 所有 handler 都添加了 AntiInjectionFilter（防注入攻击） - 使用 SecureLogFormatter（自动脱敏） 使用场景 -------- - 被 SingletonLogManager 自动创建和配置 - 不应该手动创建或配置 注意事项 -------- - 虽然名为 stdout_handler，但实际输出到 stderr - 两个 handler 的日志会按时间顺序交织输出 - 所有输出流都是 sys.stderr """ def __init__(self): """ 初始化多流输出处理器 初始化流程 ---------- 1. 创建两个 StreamHandler（都输出到 stderr） 2. 设置日志级别和过滤器 3. 配置 SecureLogFormatter（脱敏功能） 4. 添加 AntiInjectionFilter（防注入功能） """ self.stdout_handler = logging.StreamHandler(sys.stderr) self.stdout_handler.setLevel(logging.DEBUG) self.stdout_handler.addFilter(self._stdout_filter) # WARNING和ERROR使用stderr self.stderr_handler = logging.StreamHandler(sys.stderr) self.stderr_handler.setLevel(logging.WARNING) # 设置安全格式化器 formatter = SecureLogFormatter( "%(asctime)s - %(name)s - %(levelname)s - %(message)s" ) self.stdout_handler.setFormatter(formatter) self.stderr_handler.setFormatter(formatter) # 添加注入防护过滤器 anti_injection_filter = AntiInjectionFilter() self.stdout_handler.addFilter(anti_injection_filter) self.stderr_handler.addFilter(anti_injection_filter) def _stdout_filter(self, record): """ 过滤器：只允许 DEBUG 和 INFO 级别通过 参数 ---- record : logging.LogRecord 日志记录对象 返回 ---- bool True: 允许通过（DEBUG 或 INFO） False: 拒绝通过（WARNING 或更高级别） 功能 ---- 确保此 handler 只处理 DEBUG 和 INFO 级别的日志， WARNING 及以上级别由 stderr_handler 处理。 """ return record.levelno <= logging.INFO def attach_to_logger(self, logger): """ 将两个处理器附加到指定 logger 参数 ---- logger : logging.Logger 要配置的 logger 实例 功能 ---- 将 stdout_handler 和 stderr_handler 同时添加到 logger， 实现基于级别的多流输出。 注意事项 -------- - 两个 handler 会同时处理所有日志 - 过滤器确保每条日志只被一个 handler 输出 """ logger.addHandler(self.stdout_handler) logger.addHandler(self.stderr_handler) class LogSanitizer: """ 日志脱敏处理器 功能概述 -------- 自动检测并脱敏日志中的敏感信息，只处理真正的密码、密钥和 API Token， 避免过度脱敏导致日志可读性下降。 支持的敏感信息类型 ------------------ 1. **密码字段**: password, passwd（至少 6 字符） 2. **密钥字段**: secret_key, private_key（至少 16 字符） 3. **知名 API Token**: - OpenAI API key (sk-xxx，至少 32 字符) - Slack Bot Token (xoxb-xxx，至少 50 字符) - GitHub Personal Access Token (ghp_xxx，36 字符) 脱敏规则 -------- - 匹配的敏感信息替换为 "***REDACTED***" - 使用长度限制减少误判（如 password 至少 6 字符） - 使用边界匹配（\b）确保精确匹配 Token 格式 实现方式 -------- 使用预编译的正则表达式列表，在初始化时编译所有模式，提高性能。 设计原则 -------- - **精确匹配**: 避免过度脱敏（如 "timeout" 不会被误判为 "password"） - **性能优化**: 编译正则表达式，避免重复编译 - **可扩展**: 可通过添加新模式支持更多敏感信息类型 使用场景 -------- - 被 SecureLogFormatter 自动调用 - 也可独立使用：sanitizer = LogSanitizer(); sanitizer.sanitize(message) 注意事项 -------- - 脱敏是不可逆的 - 可能无法检测复杂编码或混淆的敏感信息 - 正则匹配可能影响高频日志性能 """ def __init__(self): """ 初始化脱敏处理器 初始化流程 ---------- 预编译所有正则表达式模式，提高后续匹配性能。 编译的模式 ---------- - 密码字段（至少 6 字符） - 密钥字段（至少 16 字符） - OpenAI、Slack、GitHub 等知名服务的 API Token 性能优化 -------- 使用 re.compile() 预编译正则表达式，避免每次匹配都重新编译。 """ # 只保护真正的密码和密钥，避免过度脱敏 self.sensitive_patterns = [ # 明确的密码字段 re.compile(r'password["\']?\s*[:=]\s*["\']?[^\s"\']{6,}["\']?'), re.compile(r'passwd["\']?\s*[:=]\s*["\']?[^\s"\']{6,}["\']?'), # 明确的密钥字段 re.compile( r'secret[_-]?key["\']?\s*[:=]\s*["\']?[A-Za-z0-9._-]{16,}["\']?' ), re.compile( r'private[_-]?key["\']?\s*[:=]\s*["\']?[A-Za-z0-9._-]{16,}["\']?' ), # 知名API密钥格式（精确匹配） re.compile(r"\bsk-[A-Za-z0-9]{32,}\b"), # OpenAI API key re.compile(r"\bxoxb-[A-Za-z0-9-]{50,}\b"), # Slack Bot Token re.compile(r"\bghp_[A-Za-z0-9]{36}\b"), # GitHub Personal Access Token ] def sanitize(self, message: str) -> str: """ 脱敏处理日志消息 参数 ---- message : str 原始日志消息 返回 ---- str 脱敏后的日志消息 处理流程 -------- 1. 遍历所有预编译的正则模式 2. 对每个模式执行替换 3. 返回脱敏后的消息 替换策略 -------- 所有匹配的敏感信息替换为 "***REDACTED***" 性能 ---- - 时间复杂度: O(n * m)，n 为消息长度，m 为模式数量 - 使用预编译正则表达式，减少编译开销 注意事项 -------- - 不会修改原始消息（返回新字符串） - 脱敏是不可逆的 - 可能存在误判或漏判 """ for pattern in self.sensitive_patterns: message = pattern.sub("***REDACTED***", message) return message class SecureLogFormatter(logging.Formatter): """ 安全的日志格式化器 功能概述 -------- 继承自标准 logging.Formatter，在格式化后自动脱敏敏感信息。 工作流程 -------- 1. 使用父类的 format() 方法进行标准格式化 2. 调用 LogSanitizer 脱敏敏感信息 3. 返回脱敏后的日志字符串 使用场景 -------- - 被 LevelBasedStreamHandler 自动使用 - 可用于任何需要脱敏的 logging.Handler 优势 ---- - 无缝集成到标准 logging 系统 - 对用户代码透明（自动脱敏） - 不影响日志的其他功能（级别、时间戳、模块名等） 注意事项 -------- - 脱敏发生在格式化之后，不影响日志记录的原始数据 - 脱敏会略微影响性能（正则匹配） """ def __init__(self, *args, **kwargs): """ 初始化安全日志格式化器 参数 ---- *args, **kwargs 传递给父类 logging.Formatter 的参数 初始化流程 ---------- 1. 调用父类构造函数 2. 创建 LogSanitizer 实例 """ super().__init__(*args, **kwargs) self.sanitizer = LogSanitizer() def format(self, record): """ 格式化并脱敏日志记录 参数 ---- record : logging.LogRecord 日志记录对象 返回 ---- str 格式化并脱敏后的日志字符串 处理流程 -------- 1. 调用父类的 format() 进行标准格式化 2. 调用 sanitizer.sanitize() 脱敏 3. 返回脱敏后的字符串 性能 ---- - 增加的开销主要来自脱敏正则匹配 - 对于不含敏感信息的日志，影响较小 """ # 先进行标准格式化 formatted = super().format(record) # 然后进行脱敏处理 return self.sanitizer.sanitize(formatted) class AntiInjectionFilter(logging.Filter): """ 防止日志注入攻击的过滤器 功能概述 -------- 转义日志消息中的危险字符，防止日志分割攻击和日志伪造攻击。 攻击场景 -------- 攻击者可能通过在输入中插入换行符或控制字符，伪造日志条目或分割日志： - 插入 "\\n" 可以伪造多行日志 - 插入 "\\x00"（空字节）可能导致日志处理器异常 - 插入 "\\r" 可能覆盖同一行的日志 防护策略 -------- - 转义空字节（\\x00）为 "\\\\x00" - 转义换行符（\\n）为 "\\\\n" - 转义回车符（\\r）为 "\\\\r" - 不转义 HTML 字符（保持可读性） 处理范围 -------- - record.msg: 日志消息模板 - record.args: 日志消息参数（仅字符串类型） 设计原则 -------- - **安全优先**: 转义所有潜在危险字符 - **最小影响**: 只转义必要的字符，保持可读性 - **不阻止日志**: 始终返回 True，允许日志记录 使用场景 -------- - 被 LevelBasedStreamHandler 自动添加 - 适用于所有 logging.Handler 注意事项 -------- - 转义后日志可读性略有下降（"\\n" 显示为 "\\\\n"） - 不处理非字符串类型的参数 - 不修改日志记录的其他属性 """ def filter(self, record): """ 过滤并转义日志记录中的危险字符 参数 ---- record : logging.LogRecord 日志记录对象 返回 ---- bool 始终返回 True（允许日志通过） 处理流程 -------- 1. 转义 record.msg 中的危险字符（如果是字符串） 2. 转义 record.args 中所有字符串参数的危险字符 3. 返回 True 转义规则 -------- - \\x00 → \\\\x00（空字节） - \\n → \\\\n（换行符） - \\r → \\\\r（回车符） 副作用 ------ 直接修改 record.msg 和 record.args，不会创建新对象。 性能 ---- - 只处理字符串类型 - 只转义 3 种字符，性能影响很小 线程安全 -------- 每个 LogRecord 只被一个线程处理，无需加锁。 注意事项（修复） -------------- record.msg和record.args都需要转义危险字符，确保一致性 """ # 转义record.msg中的危险字符（换行符、回车符、空字节） if hasattr(record, "msg") and isinstance(record.msg, str): record.msg = ( record.msg.replace("\x00", "\\x00") # 空字节 .replace("\n", "\\n") # 换行符 .replace("\r", "\\r") # 回车符 ) # 转义 record.args 中的危险字符 if hasattr(record, "args"): escaped_args = [] for arg in record.args: if isinstance(arg, str): # 转义换行符、回车符和空字节，保持可读性 escaped_arg = ( arg.replace("\n", "\\n") .replace("\r", "\\r") .replace("\x00", "\\x00") ) escaped_args.append(escaped_arg) else: escaped_args.append(arg) record.args = tuple(escaped_args) return True class LogDeduplicator: """ 日志去重器 功能概述 -------- 在指定时间窗口内去除重复日志，减少日志噪声，提高日志可读性。 去重策略 -------- - **时间窗口**: 默认 5 秒，在此期间的重复日志被抑制 - **哈希匹配**: 【性能优化】使用 Python 内置 hash() 函数判断消息是否重复（比 MD5 快 5-10 倍） - **计数累加**: 重复日志的计数会累加，可附加到最终日志 缓存管理 ---------- - **过期清理**: 超出时间窗口的缓存条目自动删除 - **大小限制**: 缓存超过 max_cache_size 时，删除最旧的 25% 条目 - **线程安全**: 使用 threading.Lock 保护缓存操作 内部结构 ---------- - cache: {log_hash: (timestamp, count)} - log_hash: 消息的 MD5 哈希 - timestamp: 最后一次出现的时间戳 - count: 累计重复次数 使用场景 -------- - 高频日志场景（如轮询、心跳） - 错误重试场景（避免相同错误刷屏） - EnhancedLogger 自动集成 性能考虑 -------- - MD5 哈希计算成本较低 - 缓存查找为 O(1) - 定期清理防止内存泄漏 注意事项 -------- - 不同的日志内容（如时间戳）会导致哈希不同 - 去重可能隐藏重要信息（需谨慎配置时间窗口） - 缓存会占用内存（由 max_cache_size 限制） """ def __init__(self, time_window=5.0, max_cache_size=1000): """ 初始化日志去重器 参数 ---- time_window : float, optional 时间窗口（秒），默认 5.0 在此时间内的重复日志将被抑制 max_cache_size : int, optional 最大缓存大小，默认 1000 超过此大小时自动清理最旧的 25% 条目 初始化流程 ---------- 1. 设置时间窗口和缓存大小 2. 初始化空缓存字典 3. 创建线程锁 内部状态 -------- - time_window: 时间窗口（秒） - max_cache_size: 最大缓存大小 - cache: 日志哈希 -> (时间戳, 计数) 的映射 - lock: 线程锁 """ self.time_window = time_window # 时间窗口（秒） self.max_cache_size = max_cache_size # 使用内置 hash(message)（int）作为 key self.cache: Dict[int, Tuple[float, int]] = {} # {msg_hash: (timestamp, count)} self.lock = threading.Lock() def should_log(self, message: str) -> Tuple[bool, Optional[str]]: """ 检查是否应该记录日志 参数 ---- message : str 日志消息 返回 ---- Tuple[bool, Optional[str]] - bool: 是否应该记录日志 - Optional[str]: 重复信息（如 "重复 3 次"），无重复则为 None 处理流程 -------- 1. 生成消息的 MD5 哈希 2. 检查哈希是否在缓存中： - 存在且在时间窗口内: 增加计数，不记录，返回 (False, "重复 N 次") - 存在但超出时间窗口: 重置计数，记录，返回 (True, None) - 不存在: 添加到缓存，记录，返回 (True, None) 3. 定期清理过期缓存 线程安全 -------- 使用 self.lock 保护整个操作，确保缓存一致性。 性能 ---- - 时间复杂度: O(1)（哈希查找） - 【性能优化】使用 Python 内置 hash()，比 MD5 快 5-10 倍 - 清理操作: O(m)（m 为缓存大小） 注意事项 -------- - 相同内容的消息会被去重 - 不同时间戳的消息会被视为不同（需在格式化前去重） - 使用内置 hash() 而非加密哈希，因为日志去重不需要加密安全性 """ with self.lock: current_time = time.time() # 【性能优化】使用 Python 内置 hash()，比 MD5 快 5-10 倍 # 对于日志去重场景，不需要加密安全性，只需要高效的哈希区分 msg_hash = hash(message) if msg_hash in self.cache: last_time, count = self.cache[msg_hash] if current_time - last_time <= self.time_window: # 在时间窗口内，增加计数但不记录 self.cache[msg_hash] = (current_time, count + 1) return False, f"重复 {count + 1} 次" else: # 超出时间窗口，重新记录 self.cache[msg_hash] = (current_time, 1) return True, None else: # 新消息，记录 self.cache[msg_hash] = (current_time, 1) self._cleanup_cache(current_time) return True, None def _cleanup_cache(self, current_time: float): """ 清理过期缓存 参数 ---- current_time : float 当前时间戳 清理策略 -------- 1. **过期清理**: 删除超出时间窗口的所有条目 2. **大小限制**: 如果缓存超过 max_cache_size，删除最旧的 25% 条目 处理流程 -------- 1. 找出所有过期的键（超出时间窗口） 2. 删除过期键 3. 检查缓存大小，如果超过限制： - 按时间戳排序 - 删除最旧的 25% 条目 性能 ---- - 过期清理: O(m)（m 为缓存大小） - 大小限制: O(m log m)（排序） 线程安全 -------- 调用者（should_log）已持有 self.lock，无需再次加锁。 注意事项 -------- - 清理是惰性的（不是定时任务） - 只在添加新消息时触发 - 删除 25% 的策略是启发式的，可根据需要调整 """ expired_keys = [ key for key, (timestamp, _) in self.cache.items() if current_time - timestamp > self.time_window ] for key in expired_keys: del self.cache[key] # 限制缓存大小 if len(self.cache) > self.max_cache_size: # 删除最旧的条目 sorted_items = sorted(self.cache.items(), key=lambda x: x[1][0]) for key, _ in sorted_items[: len(sorted_items) // 4]: del self.cache[key] class EnhancedLogger: """ 增强的日志记录器 功能概述 -------- 提供企业级日志功能的高级接口，集成所有底层优化： - 单例管理（防止日志重复） - 日志去重（减少噪声） - 自动脱敏（保护敏感信息） - 注入防护（防止日志攻击） - 智能级别映射（动态调整日志级别） 核心特性 -------- 1. **单例 Logger**: 通过 SingletonLogManager 确保 logger 不重复初始化 2. **自动去重**: 5 秒时间窗口内的重复日志自动抑制 3. **透明脱敏**: 自动检测并脱敏密码、API key 等敏感信息 4. **防注入攻击**: 自动转义换行符和控制字符 5. **智能级别**: 根据消息内容关键词自动调整日志级别 使用场景 -------- - 替代标准 logging.Logger 使用 - 需要高级日志功能的场景 - MCP 服务器日志记录 API 兼容性 ---------- 提供与标准 logging.Logger 兼容的方法： - debug(message, *args, **kwargs) - info(message, *args, **kwargs) - warning(message, *args, **kwargs) - error(message, *args, **kwargs) 级别映射规则 ------------ 根据消息中的关键词自动调整日志级别（可扩展）： - "收到反馈请求", "Web UI 配置加载成功" → DEBUG - "等待用户反馈", "收到用户反馈" → INFO - "服务启动失败", "配置加载失败" → ERROR 性能 ---- - 去重增加少量开销（MD5 哈希 + 缓存查找） - 级别映射使用简单字符串匹配（O(m)，m 为映射数量） - 脱敏和注入防护在底层 Handler 处理 注意事项 -------- - 去重可能隐藏重要信息（时间窗口 5 秒） - 级别映射基于字符串匹配，可能误判 - 不支持动态修改 level_mapping（需重新创建实例） """ def __init__(self, name: str): """ 初始化增强日志记录器 参数 ---- name : str logger 名称（通常使用 __name__） 初始化流程 ---------- 1. 创建 SingletonLogManager 并获取 logger 2. 创建 LogDeduplicator（5 秒窗口，1000 条缓存） 3. 配置级别映射规则 内部组件 -------- - log_manager: SingletonLogManager 实例 - logger: 配置好的 logging.Logger - deduplicator: LogDeduplicator 实例 - level_mapping: 消息关键词 -> 日志级别的映射 """ self.log_manager = SingletonLogManager() self.logger = self.log_manager.setup_logger(name) self.deduplicator = LogDeduplicator( time_window=5.0, max_cache_size=1000, ) self.level_mapping = { "收到反馈请求": logging.DEBUG, "Web UI 配置加载成功": logging.DEBUG, "启动反馈界面": logging.DEBUG, "Web 服务已在运行": logging.DEBUG, "内容已更新": logging.INFO, "等待用户反馈": logging.INFO, "收到用户反馈": logging.INFO, "服务启动失败": logging.ERROR, "配置加载失败": logging.ERROR, } def _get_effective_level(self, message: str, default_level: int) -> int: """ 根据消息内容获取有效的日志级别 参数 ---- message : str 日志消息 default_level : int 默认日志级别（如果没有匹配的映射） 返回 ---- int 有效的日志级别 匹配规则 -------- 遍历 level_mapping，如果消息中包含关键词，返回对应级别。 如果没有匹配，返回 default_level。 性能 ---- - 时间复杂度: O(m * n) - m: level_mapping 的大小 - n: 消息长度 - 使用简单字符串匹配（str.contains） 扩展性 ------ 可通过修改 level_mapping 添加新的映射规则，或在初始化后动态修改。 注意事项 -------- - 匹配是子串匹配，可能存在误判 - 第一个匹配的规则生效（顺序敏感） """ for pattern, level in self.level_mapping.items(): if pattern in message: return level return default_level def log(self, level: int, message: str, *args, **kwargs): """ 记录日志（带去重和级别优化） 参数 ---- level : int 日志级别（DEBUG, INFO, WARNING, ERROR, CRITICAL） message : str 日志消息 *args 传递给 logger.log() 的额外参数 **kwargs 传递给 logger.log() 的关键字参数 处理流程 -------- 1. 根据消息内容获取有效日志级别（可能与 level 不同） 2. 检查是否应该记录（去重） 3. 如果应该记录： - 如果有重复信息，附加到消息末尾 - 调用底层 logger.log() 去重机制 -------- 在 5 秒时间窗口内，相同消息只记录一次，重复计数附加到消息。 级别映射 -------- 即使调用 debug()，如果消息匹配到 ERROR 映射，也会以 ERROR 级别记录。 注意事项 -------- - 去重基于消息内容（不包括参数） - 级别映射可能覆盖传入的 level - 底层 handler 会自动脱敏和防注入 """ effective_level = self._get_effective_level(message, level) should_log, duplicate_info = self.deduplicator.should_log(message) if should_log: if duplicate_info: message += f" ({duplicate_info})" self.logger.log(effective_level, message, *args, **kwargs) def setLevel(self, level: int) -> None: """兼容标准 logging.Logger API：设置底层 logger 的级别。""" self.logger.setLevel(level) def debug(self, message: str, *args, **kwargs): """ 记录 DEBUG 级别日志 参数 ---- message : str 日志消息 *args 额外参数 **kwargs 关键字参数 功能 ---- 调用 self.log(logging.DEBUG, message, *args, **kwargs) 注意 ---- 实际日志级别可能被 level_mapping 覆盖 """ self.log(logging.DEBUG, message, *args, **kwargs) def info(self, message: str, *args, **kwargs): """ 记录 INFO 级别日志 参数 ---- message : str 日志消息 *args 额外参数 **kwargs 关键字参数 功能 ---- 调用 self.log(logging.INFO, message, *args, **kwargs) 注意 ---- 实际日志级别可能被 level_mapping 覆盖 """ self.log(logging.INFO, message, *args, **kwargs) def warning(self, message: str, *args, **kwargs): """ 记录 WARNING 级别日志 参数 ---- message : str 日志消息 *args 额外参数 **kwargs 关键字参数 功能 ---- 调用 self.log(logging.WARNING, message, *args, **kwargs) 注意 ---- 实际日志级别可能被 level_mapping 覆盖 """ self.log(logging.WARNING, message, *args, **kwargs) def error(self, message: str, *args, **kwargs): """ 记录 ERROR 级别日志 参数 ---- message : str 日志消息 *args 额外参数 **kwargs 关键字参数 功能 ---- 调用 self.log(logging.ERROR, message, *args, **kwargs) 注意 ---- 实际日志级别可能被 level_mapping 覆盖 """ self.log(logging.ERROR, message, *args, **kwargs) enhanced_logger = EnhancedLogger(__name__) # ======================================================================== # 日志级别配置工具 # ======================================================================== # 日志级别映射 LOG_LEVEL_MAP = { "DEBUG": logging.DEBUG, "INFO": logging.INFO, "WARNING": logging.WARNING, "ERROR": logging.ERROR, "CRITICAL": logging.CRITICAL, } # 有效的日志级别名称 VALID_LOG_LEVELS = tuple(LOG_LEVEL_MAP.keys()) def get_log_level_from_config() -> int: """ 从配置文件读取日志级别 返回 ---- int logging 模块的日志级别常量 处理逻辑 -------- 1. 尝试从 config_manager 读取 web_ui.log_level 配置 2. 如果配置无效或读取失败，使用默认级别 WARNING 3. 忽略大小写（如 "warning" 等同于 "WARNING"） 示例 ---- >>> # config.jsonc: {"web_ui": {"log_level": "DEBUG"}} >>> get_log_level_from_config() 10 # logging.DEBUG """ try: from config_manager import config_manager web_ui_config = config_manager.get("web_ui", {}) log_level_str = web_ui_config.get("log_level", "WARNING") # 标准化为大写 log_level_upper = str(log_level_str).upper() if log_level_upper in LOG_LEVEL_MAP: return LOG_LEVEL_MAP[log_level_upper] else: logging.warning( f"无效的日志级别 '{log_level_str}'，" f"有效值: {VALID_LOG_LEVELS}，使用默认值 WARNING" ) return logging.WARNING except Exception as e: # 配置读取失败时使用默认级别 logging.debug(f"读取日志级别配置失败: {e}，使用默认值 WARNING") return logging.WARNING def configure_logging_from_config() -> None: """ 根据配置文件设置全局日志级别 功能 ---- 1. 从配置读取日志级别 2. 设置 root logger 级别 3. 更新所有已存在的 handler 级别 使用场景 -------- 在应用启动时调用，确保日志级别与配置一致 示例 ---- >>> configure_logging_from_config() >>> # 现在所有日志都使用配置中的级别 """ log_level = get_log_level_from_config() # 设置 root logger root_logger = logging.getLogger() root_logger.setLevel(log_level) # 更新所有 handler for handler in root_logger.handlers: handler.setLevel(log_level) logging.info(f"日志级别已设置为: {logging.getLevelName(log_level)}")