Coverage for src / osiris_cli / logger.py: 0%

1""" 

2Logging framework for Osiris CLI 

3 

4Provides structured logging with file rotation, different log levels, 

5and optional Sentry integration for error tracking. 

6""" 

7 

8import logging 

9import sys 

10from pathlib import Path 

11from logging.handlers import RotatingFileHandler 

12from typing import Optional 

13import os 

14from rich.console import Console 

15 

16# Shared Rich Console for high-end TUI rendering 

17console = Console(file=sys.__stdout__, force_terminal=None, color_system="auto", legacy_windows=False) 

18 

19# Log directory 

20LOG_DIR = Path.home() / ".osiris" / "logs" 

21LOG_DIR.mkdir(parents=True, exist_ok=True) 

22 

23# Log file paths 

24MAIN_LOG = LOG_DIR / "osiris.log" 

25ERROR_LOG = LOG_DIR / "errors.log" 

26DEBUG_LOG = LOG_DIR / "debug.log" 

27 

28 

29class OsirisLogger: 

30 """ 

31 Centralized logging system for Osiris CLI. 

32  

33 Features: 

34 - File rotation (10MB max, 5 backups) 

35 - Separate error log 

36 - Optional Sentry integration 

37 - Performance metrics logging 

38 - Colored console output 

39 """ 

40 

41 def __init__(self, name: str = "osiris", verbose: bool = False, debug: bool = False): 

42 self.logger = logging.getLogger(name) 

43 self.verbose = verbose 

44 self._debug_enabled = debug # Renamed to avoid shadowing debug() method 

45 

46 # Set level based on flags 

47 if debug: 

48 level = logging.DEBUG 

49 elif verbose: 

50 level = logging.INFO 

51 else: 

52 level = logging.WARNING 

53 

54 self.logger.setLevel(level) 

55 

56 # Remove existing handlers to avoid duplicates 

57 self.logger.handlers = [] 

58 

59 # Create formatters 

60 detailed_formatter = logging.Formatter( 

61 '%(asctime)s - %(name)s - %(levelname)s - %(funcName)s:%(lineno)d - %(message)s', 

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

63 ) 

64 

65 simple_formatter = logging.Formatter( 

66 '%(asctime)s - %(levelname)s - %(message)s', 

67 datefmt='%H:%M:%S' 

68 ) 

69 

70 console_formatter = logging.Formatter( 

71 '%(levelname)s: %(message)s' 

72 ) 

73 

74 # Main log file (all messages) 

75 main_handler = RotatingFileHandler( 

76 MAIN_LOG, 

77 maxBytes=10 * 1024 * 1024, # 10MB 

78 backupCount=5 

79 ) 

80 main_handler.setLevel(logging.DEBUG) 

81 main_handler.setFormatter(detailed_formatter) 

82 self.logger.addHandler(main_handler) 

83 

84 # Error log file (ERROR and above) 

85 error_handler = RotatingFileHandler( 

86 ERROR_LOG, 

87 maxBytes=10 * 1024 * 1024, # 10MB 

88 backupCount=5 

89 ) 

90 error_handler.setLevel(logging.ERROR) 

91 error_handler.setFormatter(detailed_formatter) 

92 self.logger.addHandler(error_handler) 

93 

94 # Debug log file (DEBUG and above) - only if debug mode 

95 if debug: 

96 debug_handler = RotatingFileHandler( 

97 DEBUG_LOG, 

98 maxBytes=10 * 1024 * 1024, # 10MB 

99 backupCount=3 

100 ) 

101 debug_handler.setLevel(logging.DEBUG) 

102 debug_handler.setFormatter(detailed_formatter) 

103 self.logger.addHandler(debug_handler) 

104 

105 # Console handler (respect verbosity) 

106 console_handler = logging.StreamHandler(sys.stdout) 

107 console_handler.setLevel(level) 

108 console_handler.setFormatter(console_formatter) 

109 self.logger.addHandler(console_handler) 

110 

111 # Sentry integration (optional) 

112 self._setup_sentry() 

113 

114 def _setup_sentry(self): 

115 """Set up Sentry for error tracking if SENTRY_DSN is set""" 

116 sentry_dsn = os.getenv("SENTRY_DSN") 

117 if sentry_dsn: 

118 try: 

119 import sentry_sdk 

120 sentry_sdk.init( 

121 dsn=sentry_dsn, 

122 # Set traces_sample_rate to 1.0 to capture 100% 

123 # of transactions for performance monitoring. 

124 traces_sample_rate=0.1, 

125 # Set profiles_sample_rate to 1.0 to profile 100% 

126 # of sampled transactions. 

127 profiles_sample_rate=0.1, 

128 environment=os.getenv("ENVIRONMENT", "production") 

129 ) 

130 self.logger.info("Sentry error tracking enabled") 

131 except ImportError: 

132 self.logger.warning("Sentry SDK not installed, error tracking disabled") 

133 

134 def debug(self, message: str, **kwargs): 

135 """Log debug message""" 

136 self.logger.debug(message, extra=kwargs) 

137 

138 def info(self, message: str, **kwargs): 

139 """Log info message""" 

140 self.logger.info(message, extra=kwargs) 

141 

142 def warning(self, message: str, **kwargs): 

143 """Log warning message""" 

144 self.logger.warning(message, extra=kwargs) 

145 

146 def error(self, message: str, exc_info: bool = False, **kwargs): 

147 """Log error message""" 

148 self.logger.error(message, exc_info=exc_info, extra=kwargs) 

149 

150 def critical(self, message: str, exc_info: bool = True, **kwargs): 

151 """Log critical message""" 

152 self.logger.critical(message, exc_info=exc_info, extra=kwargs) 

153 

154 def exception(self, message: str, **kwargs): 

155 """Log exception with traceback""" 

156 self.logger.exception(message, extra=kwargs) 

157 

158 def performance(self, operation: str, duration_ms: float, **kwargs): 

159 """Log performance metric""" 

160 self.logger.info( 

161 f"PERF: {operation} took {duration_ms:.2f}ms", 

162 extra={"operation": operation, "duration_ms": duration_ms, **kwargs} 

163 ) 

164 

165 def api_call(self, provider: str, model: str, tokens: int, cost: float, duration_ms: float): 

166 """Log API call metrics""" 

167 self.logger.info( 

168 f"API: {provider}/{model} - {tokens} tokens, ${cost:.4f}, {duration_ms:.0f}ms", 

169 extra={ 

170 "type": "api_call", 

171 "provider": provider, 

172 "model": model, 

173 "tokens": tokens, 

174 "cost": cost, 

175 "duration_ms": duration_ms 

176 } 

177 ) 

178 

179 def user_action(self, action: str, **kwargs): 

180 """Log user action for analytics""" 

181 self.logger.info( 

182 f"USER: {action}", 

183 extra={"type": "user_action", "action": action, **kwargs} 

184 ) 

185 

186 

187# Global logger instance 

188_logger: Optional[OsirisLogger] = None 

189 

190 

191def setup_logger(verbose: bool = False, debug: bool = False) -> OsirisLogger: 

192 """ 

193 Set up global logger instance. 

194  

195 Args: 

196 verbose: Enable INFO level logging 

197 debug: Enable DEBUG level logging 

198  

199 Returns: 

200 Configured OsirisLogger instance 

201 """ 

202 global _logger 

203 _logger = OsirisLogger(name="osiris", verbose=verbose, debug=debug) 

204 return _logger 

205 

206 

207def get_logger() -> OsirisLogger: 

208 """ 

209 Get global logger instance. 

210  

211 Returns: 

212 OsirisLogger instance (creates default if not initialized) 

213 """ 

214 global _logger 

215 if _logger is None: 

216 _logger = OsirisLogger() 

217 return _logger 

218 

219 

220# Convenience functions 

221def debug(message: str, **kwargs): 

222 """Log debug message""" 

223 get_logger().debug(message, **kwargs) 

224 

225 

226def info(message: str, **kwargs): 

227 """Log info message""" 

228 get_logger().info(message, **kwargs) 

229 

230 

231def warning(message: str, **kwargs): 

232 """Log warning message""" 

233 get_logger().warning(message, **kwargs) 

234 

235 

236def error(message: str, exc_info: bool = False, **kwargs): 

237 """Log error message""" 

238 get_logger().error(message, exc_info=exc_info, **kwargs) 

239 

240 

241def critical(message: str, exc_info: bool = True, **kwargs): 

242 """Log critical message""" 

243 get_logger().critical(message, exc_info=exc_info, **kwargs) 

244 

245 

246def exception(message: str, **kwargs): 

247 """Log exception with traceback""" 

248 get_logger().exception(message, **kwargs) 

249 

250 

251def performance(operation: str, duration_ms: float, **kwargs): 

252 """Log performance metric""" 

253 get_logger().performance(operation, duration_ms, **kwargs) 

254 

255 

256def api_call(provider: str, model: str, tokens: int, cost: float, duration_ms: float): 

257 """Log API call metrics""" 

258 get_logger().api_call(provider, model, tokens, cost, duration_ms) 

259 

260 

261def user_action(action: str, **kwargs): 

262 """Log user action""" 

263 get_logger().user_action(action, **kwargs)