Coverage for main/logger.py: 95%

1import sys 

2from datetime import datetime 

3import logging 

4import os 

5import gzip 

6import shutil 

7from logging.handlers import BaseRotatingHandler 

8from pathlib import Path 

9 

10from utils import get_app_dir 

11 

12 

13class CompressedRotatingFileHandler(BaseRotatingHandler): 

14 """ 

15 A rotating log file handler that compresses rotated files. 

16 

17 This handler behaves like a standard rotating file handler, but when a 

18 rollover occurs, the old log file is compressed into a `.gz` archive. 

19 It also limits the number of backups retained. 

20 

21 Attributes 

22 ---------- 

23 maxBytes : int 

24 Maximum size in bytes before a rollover is triggered. 

25 backupCount : int 

26 Maximum number of compressed backup files to retain. 

27 log_dir : Path 

28 Directory where log files are stored. 

29 """ 

30 

31 def __init__(self, filename, mode="a", maxBytes=10 * 1024 * 1024, 

32 backupCount=5, encoding=None, delay=False): 

33 """ 

34 Initialize the handler and ensure the log directory exists. 

35 

36 Parameters 

37 ---------- 

38 filename : str or Path 

39 Path to the main log file. 

40 mode : str, optional 

41 File mode (default is "a"). 

42 maxBytes : int, optional 

43 Maximum size in bytes before triggering a rollover. 

44 backupCount : int, optional 

45 Number of compressed backups to retain. 

46 encoding : str, optional 

47 Encoding for the log file. 

48 delay : bool, optional 

49 If True, file opening is deferred until the first write. 

50 """ 

51 self.maxBytes = maxBytes 

52 self.backupCount = backupCount 

53 self.log_dir = Path(filename).parent 

54 

55 # Ensure log directory exists 

56 self.log_dir.mkdir(parents=True, exist_ok=True) 

57 

58 super().__init__(filename, mode, encoding=encoding, delay=delay) 

59 

60 def shouldRollover(self, record): 

61 """ 

62 Determine whether the log file should be rotated. 

63 

64 Checks if the size of the log file plus the next log message 

65 would exceed the `maxBytes` limit. 

66 

67 Parameters 

68 ---------- 

69 record : logging.LogRecord 

70 The log record about to be written. 

71 

72 Returns 

73 ------- 

74 bool 

75 True if rollover should occur, False otherwise. 

76 """ 

77 if self.stream is None: 

78 self.stream = self._open() 

79 

80 if self.maxBytes > 0: 

81 msg = f"{self.format(record)}\n" 

82 self.stream.seek(0, os.SEEK_END) 

83 if self.stream.tell() + len(msg.encode(self.encoding or "utf-8")) >= self.maxBytes: 

84 return True 

85 return False 

86 

87 def doRollover(self): 

88 """ 

89 Perform the rollover, compress the old file, and maintain only recent backups. 

90 

91 The current log file is closed, compressed into `.gz`, and deleted. 

92 Then only the most recent `backupCount` compressed files are kept. 

93 """ 

94 if self.stream: 

95 self.stream.close() 

96 self.stream = None 

97 

98 # Create timestamped filename 

99 ts = datetime.now().strftime("%Y-%m-%d_%H-%M-%S") 

100 rotated_file = self.log_dir / f"{Path(self.baseFilename).name}.{ts}.gz" 

101 

102 # Compress current file 

103 if os.path.exists(self.baseFilename): 

104 with open(self.baseFilename, "rb") as f_in, gzip.open(rotated_file, "wb") as f_out: 

105 shutil.copyfileobj(f_in, f_out) 

106 os.remove(self.baseFilename) 

107 

108 # Keep only the latest N compressed backups 

109 files = sorted( 

110 (f for f in self.log_dir.glob("*.gz") if f.is_file()), 

111 key=lambda f: f.stat().st_mtime 

112 ) 

113 if len(files) > self.backupCount: 

114 for old_file in files[:-self.backupCount]: 

115 old_file.unlink() 

116 

117 # Reopen new empty log file 

118 if not self.delay: 

119 self.stream = self._open() 

120 

121 

122def setup_logger(console, 

123 logger_name="GliAAns-UI", 

124 logfile=None, 

125 level=logging.ERROR, 

126 maxBytes=10 * 1024 * 1024, 

127 backupCount=5): 

128 """ 

129 Configure and return a logger with optional console output. 

130 

131 Parameters 

132 ---------- 

133 console : bool 

134 If True, also log messages to the console (stdout). 

135 logger_name : str, optional 

136 Name of the logger (default: "GliAAns-UI"). 

137 logfile : str or Path, optional 

138 Path to the log file. If None, defaults to `~/.log/log.txt` inside 

139 the app directory. 

140 level : int, optional 

141 Logging level (default: logging.ERROR). 

142 maxBytes : int, optional 

143 Maximum log file size before rotation. 

144 backupCount : int, optional 

145 Number of compressed backups to retain. 

146 

147 Returns 

148 ------- 

149 logging.Logger 

150 The configured logger instance. 

151 """ 

152 # Determine log directory under the application path 

153 app_dir = get_app_dir() 

154 log_dir = app_dir / ".log" 

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

156 

157 if logfile is None: 

158 logfile = log_dir / "log.txt" 

159 

160 logger = logging.getLogger(logger_name) 

161 logger.setLevel(level) 

162 

163 # Remove existing handlers to prevent duplicates 

164 if logger.hasHandlers(): 

165 logger.handlers.clear() 

166 

167 # File handler with compression and rotation 

168 handler = CompressedRotatingFileHandler( 

169 logfile, 

170 maxBytes=maxBytes, 

171 backupCount=backupCount 

172 ) 

173 file_formatter = logging.Formatter( 

174 "%(asctime)s\t%(levelname)s\t%(filename)s:%(lineno)d\t%(funcName)s\t%(message)s" 

175 ) 

176 handler.setFormatter(file_formatter) 

177 logger.addHandler(handler) 

178 

179 # Optional console handler for interactive debugging 

180 if console: 

181 console_handler = logging.StreamHandler(sys.stdout) 

182 console_formatter = logging.Formatter( 

183 "%(levelname)s\t%(filename)s:%(lineno)d\t%(funcName)s\t%(message)s" 

184 ) 

185 console_handler.setLevel(logging.DEBUG) 

186 console_handler.setFormatter(console_formatter) 

187 logger.addHandler(console_handler) 

188 

189 return logger 

190 

191 

192def get_logger(logger_name="GliAAns-UI"): 

193 """ 

194 Retrieve an existing logger instance by name. 

195 

196 Parameters 

197 ---------- 

198 logger_name : str, optional 

199 The logger name (default: "GliAAns-UI"). 

200 

201 Returns 

202 ------- 

203 logging.Logger 

204 The requested logger instance. 

205 """ 

206 return logging.getLogger(logger_name) 

207 

208 

209def set_log_level(level, logger_name="GliAAns-UI"): 

210 """ 

211 Set the log level for an existing logger. 

212 

213 Parameters 

214 ---------- 

215 level : int 

216 Logging level (e.g., logging.DEBUG, logging.INFO). 

217 logger_name : str, optional 

218 Name of the logger (default: "GliAAns-UI"). 

219 """ 

220 logging.getLogger(logger_name).setLevel(level)