Coverage for logger / logger.py: 33%

81 statements  

« prev     ^ index     » next       coverage.py v7.13.5, created at 2026-03-29 02:55 +0800

1""" 

2日志系统模块 

3 

4特性: 

5- 文件日志:按会话 ID 分文件存储,按天轮转,保留指定天数 

6- 控制台日志:带颜色的 Rich 格式输出 

7- 敏感信息过滤:自动脱敏 API Key 等 

8- 灵活配置:通过环境变量控制日志级别和输出方式 

9- 统一日志目录:~/.qrclaw/logs/ 

10""" 

11 

12import logging 

13import re 

14from logging.handlers import TimedRotatingFileHandler 

15from pathlib import Path 

16from typing import Optional 

17 

18from rich.console import Console 

19from rich.logging import RichHandler 

20 

21 

22class SensitiveInfoFilter(logging.Filter): 

23 """敏感信息过滤器,自动脱敏 API Key 等""" 

24 

25 # 需要过滤的敏感信息模式 

26 SENSITIVE_PATTERNS = [ 

27 # OpenAI API Key 

28 (r'sk-[a-zA-Z0-9]{48,}', 'sk-***REDACTED***'), 

29 # Generic API Keys (key=value format) 

30 (r'(api[_-]?key\s*=\s*)[\w\-]{20,}', r'\1***REDACTED***'), 

31 # Bearer tokens 

32 (r'(Bearer\s+)[\w\-\.]{20,}', r'\1***REDACTED***'), 

33 # Passwords 

34 (r'(password\s*=\s*)\S+', r'\1***REDACTED***'), 

35 # Secrets 

36 (r'(secret\s*=\s*)\S+', r'\1***REDACTED***'), 

37 ] 

38 

39 def filter(self, record: logging.LogRecord) -> bool: 

40 """过滤日志记录中的敏感信息""" 

41 if hasattr(record, 'msg') and isinstance(record.msg, str): 

42 for pattern, replacement in self.SENSITIVE_PATTERNS: 

43 record.msg = re.sub(pattern, replacement, record.msg, flags=re.IGNORECASE) 

44 

45 # 同时检查 args 中的字符串参数 

46 if hasattr(record, 'args') and record.args: 

47 if isinstance(record.args, dict): 

48 record.args = { 

49 k: self._filter_value(v) if isinstance(v, str) else v 

50 for k, v in record.args.items() 

51 } 

52 elif isinstance(record.args, tuple): 

53 record.args = tuple( 

54 self._filter_value(arg) if isinstance(arg, str) else arg 

55 for arg in record.args 

56 ) 

57 

58 return True 

59 

60 def _filter_value(self, value: str) -> str: 

61 """过滤字符串值中的敏感信息""" 

62 for pattern, replacement in self.SENSITIVE_PATTERNS: 

63 value = re.sub(pattern, replacement, value, flags=re.IGNORECASE) 

64 return value 

65 

66 

67class QRClawLogger: 

68 """QRClaw 日志管理器""" 

69 

70 def __init__(self): 

71 self._initialized = False 

72 self._logger: Optional[logging.Logger] = None 

73 self._console: Optional[Console] = None 

74 self._current_session_id: Optional[str] = None 

75 

76 def setup( 

77 self, 

78 session_id: str = "default", 

79 log_level: str = "INFO", 

80 log_to_file: bool = True, 

81 log_to_console: bool = True, 

82 log_max_days: int = 30, 

83 console_level: str = "WARNING", 

84 log_dir: Path = None, 

85 ): 

86 """ 

87 设置日志系统 

88 

89 Args: 

90 session_id: 会话 ID,用于区分不同会话的日志文件 

91 log_level: 文件日志级别 

92 log_to_file: 是否输出到文件 

93 log_to_console: 是否输出到控制台 

94 log_max_days: 日志文件保留天数 

95 console_level: 控制台日志级别 

96 log_dir: 日志目录(由 Workspace 提供,不传则用默认路径) 

97 """ 

98 # session_id 未变则跳过,但必须确认 logger 已挂载 handlers 

99 # (防止 switch A→B→A 时 handler 实际指向 B 的文件) 

100 if ( 

101 self._initialized 

102 and self._current_session_id == session_id 

103 and self._logger is not None 

104 and self._logger.handlers 

105 ): 

106 return 

107 

108 # 日志目录:优先用传入的,否则用默认 

109 if log_dir is None: 

110 log_dir = Path.home() / ".qrclaw" / "logs" 

111 log_dir.mkdir(parents=True, exist_ok=True) 

112 

113 # 获取 root logger 

114 self._logger = logging.getLogger("qrclaw") 

115 self._logger.setLevel(logging.DEBUG) # 设置为最低级别,由 handler 控制实际输出 

116 

117 # 关闭并清除现有 handlers,避免文件句柄泄漏 

118 for handler in self._logger.handlers[:]: 

119 handler.close() 

120 self._logger.removeHandler(handler) 

121 

122 # 创建 Rich Console 

123 self._console = Console() 

124 

125 # 文件日志格式 

126 file_format = logging.Formatter( 

127 '%(asctime)s | %(levelname)-8s | %(name)s:%(lineno)d | %(message)s', 

128 datefmt='%Y-%m-%d %H:%M:%S' 

129 ) 

130 

131 # 控制台日志格式(Rich 已经自带时间戳,这里只保留关键信息) 

132 console_format = logging.Formatter('%(message)s') 

133 

134 # 文件日志 handler(按会话 ID 分文件) 

135 if log_to_file: 

136 log_file = log_dir / f"qrclaw-{session_id}.log" 

137 file_handler = TimedRotatingFileHandler( 

138 filename=str(log_file), 

139 when='midnight', 

140 interval=1, 

141 backupCount=log_max_days, 

142 encoding='utf-8' 

143 ) 

144 file_handler.setLevel(getattr(logging, log_level.upper())) 

145 file_handler.setFormatter(file_format) 

146 file_handler.addFilter(SensitiveInfoFilter()) 

147 self._logger.addHandler(file_handler) 

148 

149 # 控制台日志 handler(使用 Rich) 

150 if log_to_console: 

151 console_handler = RichHandler( 

152 console=self._console, 

153 show_path=True, 

154 show_time=True, 

155 rich_tracebacks=True, 

156 tracebacks_show_locals=True 

157 ) 

158 console_handler.setLevel(getattr(logging, console_level.upper())) 

159 console_handler.setFormatter(console_format) 

160 console_handler.addFilter(SensitiveInfoFilter()) 

161 self._logger.addHandler(console_handler) 

162 

163 self._initialized = True 

164 self._current_session_id = session_id 

165 

166 # 记录初始化日志 

167 self._logger.info("=" * 60) 

168 self._logger.info(f"QRClaw 日志系统初始化完成 (会话: {session_id})") 

169 self._logger.info(f"日志级别: {log_level} (文件) / {console_level} (控制台)") 

170 self._logger.info(f"日志目录: {log_dir}") 

171 self._logger.info(f"日志文件: qrclaw-{session_id}.log") 

172 self._logger.info(f"文件日志: {'启用' if log_to_file else '禁用'}") 

173 self._logger.info(f"控制台日志: {'启用' if log_to_console else '禁用'}") 

174 self._logger.info(f"日志保留: {log_max_days}") 

175 self._logger.info("=" * 60) 

176 

177 def get_logger(self, name: str = "qrclaw") -> logging.Logger: 

178 """ 

179 获取 logger 实例。 

180 

181 直接返回 logging 注册表中的 logger,不触发任何初始化。 

182 handler 由 setup() 统一管理,调用方无需关心。 

183 """ 

184 return logging.getLogger(name) 

185 

186 @property 

187 def console(self) -> Console: 

188 """获取 Rich Console 实例""" 

189 if self._console is None: 

190 self._console = Console() 

191 return self._console 

192 

193 

194# 全局日志管理器实例 

195_logger_manager = QRClawLogger() 

196 

197 

198def setup_logger( 

199 session_id: str = "default", 

200 log_level: str = "INFO", 

201 log_to_file: bool = True, 

202 log_to_console: bool = True, 

203 log_max_days: int = 30, 

204 console_level: str = "WARNING", 

205 log_dir: Path = None, 

206): 

207 """ 

208 设置日志系统(全局函数) 

209 

210 Args: 

211 session_id: 会话 ID,用于区分不同会话的日志文件 

212 log_level: 文件日志级别 

213 log_to_file: 是否输出到文件 

214 log_to_console: 是否输出到控制台 

215 log_max_days: 日志文件保留天数 

216 console_level: 控制台日志级别 

217 log_dir: 日志目录(由 Workspace 提供) 

218 """ 

219 _logger_manager.setup( 

220 session_id=session_id, 

221 log_level=log_level, 

222 log_to_file=log_to_file, 

223 log_to_console=log_to_console, 

224 log_max_days=log_max_days, 

225 console_level=console_level, 

226 log_dir=log_dir, 

227 ) 

228 

229 

230def get_logger(name: str = "qrclaw") -> logging.Logger: 

231 """ 

232 获取 logger 实例(全局函数) 

233 

234 Args: 

235 name: logger 名称,默认为 'qrclaw' 

236 

237 Returns: 

238 logging.Logger: logger 实例 

239 """ 

240 return _logger_manager.get_logger(name) 

241 

242 

243def get_console() -> Console: 

244 """获取 Rich Console 实例""" 

245 return _logger_manager.console