-
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathsize_rotating.py
More file actions
executable file
·117 lines (104 loc) · 4.41 KB
/
size_rotating.py
File metadata and controls
executable file
·117 lines (104 loc) · 4.41 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
import logging.handlers
import os
import re
from pathlib import Path
from pythonlogs.core.constants import MB_TO_BYTES
from pythonlogs.core.log_utils import (
RotatingLogMixin,
check_directory_permissions,
check_filename_instance,
get_level,
get_log_path,
get_logger_and_formatter,
get_stream_handler,
gzip_file_with_sufix,
remove_old_logs,
write_stderr,
)
from pythonlogs.core.memory_utils import register_logger_weakref
from pythonlogs.core.settings import get_log_settings
from pythonlogs.core.thread_safety import auto_thread_safe
@auto_thread_safe(["init"])
class SizeRotatingLog(RotatingLogMixin):
"""Size-based rotating logger with context manager support for automatic resource cleanup."""
def __init__(
self,
level: str | None = None,
name: str | None = None,
directory: str | None = None,
filenames: list | tuple | None = None,
maxmbytes: int | None = None,
daystokeep: int | None = None,
encoding: str | None = None,
datefmt: str | None = None,
timezone: str | None = None,
streamhandler: bool | None = None,
showlocation: bool | None = None,
):
_settings = get_log_settings()
self.level = get_level(level or _settings.level)
self.appname = name or _settings.appname
self.directory = directory or _settings.directory
self.filenames = filenames or (_settings.filename,)
self.maxmbytes = maxmbytes or _settings.max_file_size_mb
self.daystokeep = daystokeep or _settings.days_to_keep
self.encoding = encoding or _settings.encoding
self.datefmt = datefmt or _settings.date_format
self.timezone = timezone or _settings.timezone
self.streamhandler = streamhandler or _settings.stream_handler
self.showlocation = showlocation or _settings.show_location
self.logger = None
def init(self):
check_filename_instance(self.filenames)
check_directory_permissions(self.directory)
logger, formatter = get_logger_and_formatter(self.appname, self.datefmt, self.showlocation, self.timezone)
if logger.level != self.level:
logger.setLevel(self.level)
for file in self.filenames:
log_file_path = get_log_path(self.directory, file)
file_handler = logging.handlers.RotatingFileHandler(
filename=log_file_path,
mode="a",
maxBytes=self.maxmbytes * MB_TO_BYTES,
backupCount=self.daystokeep,
encoding=self.encoding,
delay=False,
errors=None,
)
file_handler.rotator = GZipRotatorSize(self.directory, self.daystokeep)
file_handler.setFormatter(formatter)
file_handler.setLevel(self.level)
logger.addHandler(file_handler)
if self.streamhandler:
stream_hdlr = get_stream_handler(self.level, formatter)
logger.addHandler(stream_hdlr)
self.logger = logger
# Register weak reference for memory tracking
register_logger_weakref(logger)
return logger
class GZipRotatorSize:
def __init__(self, dir_logs: str, daystokeep: int):
self.directory = dir_logs
self.daystokeep = daystokeep
def __call__(self, source: str, dest: str) -> None:
remove_old_logs(self.directory, self.daystokeep)
if os.path.isfile(source) and os.stat(source).st_size > 0:
source_filename, _ = os.path.basename(source).split(".")
new_file_number = self._get_new_file_number(self.directory, source_filename)
if os.path.isfile(source):
gzip_file_with_sufix(source, str(new_file_number))
@staticmethod
def _get_new_file_number(directory: str, source_filename: str) -> int:
pattern = re.compile(rf"{re.escape(source_filename)}_(\d+)\.log\.gz$")
max_num = 0
try:
# Use pathlib for better performance with large directories
dir_path = Path(directory)
for file_path in dir_path.iterdir():
if file_path.is_file():
match = pattern.match(file_path.name)
if match:
max_num = max(max_num, int(match.group(1)))
except OSError as e:
write_stderr(f"Unable to get previous gz log file number | {type(e).__name__}: {e}")
return max_num + 1