Coverage for logger / logger.py: 33%
81 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-03-29 02:55 +0800
« prev ^ index » next coverage.py v7.13.5, created at 2026-03-29 02:55 +0800
1"""
2日志系统模块
4特性:
5- 文件日志:按会话 ID 分文件存储,按天轮转,保留指定天数
6- 控制台日志:带颜色的 Rich 格式输出
7- 敏感信息过滤:自动脱敏 API Key 等
8- 灵活配置:通过环境变量控制日志级别和输出方式
9- 统一日志目录:~/.qrclaw/logs/
10"""
12import logging
13import re
14from logging.handlers import TimedRotatingFileHandler
15from pathlib import Path
16from typing import Optional
18from rich.console import Console
19from rich.logging import RichHandler
22class SensitiveInfoFilter(logging.Filter):
23 """敏感信息过滤器,自动脱敏 API Key 等"""
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 ]
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)
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 )
58 return True
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
67class QRClawLogger:
68 """QRClaw 日志管理器"""
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
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 设置日志系统
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
108 # 日志目录:优先用传入的,否则用默认
109 if log_dir is None:
110 log_dir = Path.home() / ".qrclaw" / "logs"
111 log_dir.mkdir(parents=True, exist_ok=True)
113 # 获取 root logger
114 self._logger = logging.getLogger("qrclaw")
115 self._logger.setLevel(logging.DEBUG) # 设置为最低级别,由 handler 控制实际输出
117 # 关闭并清除现有 handlers,避免文件句柄泄漏
118 for handler in self._logger.handlers[:]:
119 handler.close()
120 self._logger.removeHandler(handler)
122 # 创建 Rich Console
123 self._console = Console()
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 )
131 # 控制台日志格式(Rich 已经自带时间戳,这里只保留关键信息)
132 console_format = logging.Formatter('%(message)s')
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)
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)
163 self._initialized = True
164 self._current_session_id = session_id
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)
177 def get_logger(self, name: str = "qrclaw") -> logging.Logger:
178 """
179 获取 logger 实例。
181 直接返回 logging 注册表中的 logger,不触发任何初始化。
182 handler 由 setup() 统一管理,调用方无需关心。
183 """
184 return logging.getLogger(name)
186 @property
187 def console(self) -> Console:
188 """获取 Rich Console 实例"""
189 if self._console is None:
190 self._console = Console()
191 return self._console
194# 全局日志管理器实例
195_logger_manager = QRClawLogger()
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 设置日志系统(全局函数)
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 )
230def get_logger(name: str = "qrclaw") -> logging.Logger:
231 """
232 获取 logger 实例(全局函数)
234 Args:
235 name: logger 名称,默认为 'qrclaw'
237 Returns:
238 logging.Logger: logger 实例
239 """
240 return _logger_manager.get_logger(name)
243def get_console() -> Console:
244 """获取 Rich Console 实例"""
245 return _logger_manager.console