py.lib.aw_log

Yohn Y. 2024-10-31 Parent:9155a66edb31

7:5db1541311ef Go to Latest

py.lib.aw_log/src/aw_log/file.py

. Исправлена ссылка на документацию. Сделал её прямой

Download raw file
View source Diff to previous Annotate
History
1 # coding: utf-8
2
3 from typing import Any
4 from time import monotonic, ctime
5 from datetime import date
6 from typing import TextIO, Optional
7 from threading import RLock
8 from os.path import join as p_join, abspath, split as p_split
9
10 from . import AbstractLogBase
11
12 TIME_TO_FLUSH: int = 120 # Время по умолчанию с момента прошлого сброса лога,
13 # после которого, выполняется принудительный сброс буферов
14
15
16 class AbstractFileLog(AbstractLogBase):
17 def __init__(self, prefix: str = 'main'):
18 super().__init__(prefix)
19 self.fd: Optional[TextIO] = None
20 self.lock = RLock()
21
22 def _write(self, mark: str, msg: Any):
23 if self.fd is None:
24 raise ValueError(f'Не определён файл логирования')
25
26 self.lock.acquire()
27 try:
28
29 tm = ctime()
30 for l in super()._write_helper(mark, msg):
31 self.fd.write(f'{tm} | {l}')
32
33 finally:
34 self.lock.release()
35
36
37 class FileLog(AbstractFileLog):
38 @staticmethod
39 def _open_file(file_name: str):
40 return open(file_name, 'a', encoding='utf-8')
41
42 def __init__(self, prefix: str = 'main',
43 file_name: Optional[str] = None,
44 file_obj: Optional[TextIO] = None,
45 time_to_flush: int = TIME_TO_FLUSH
46 ):
47
48 super().__init__(prefix=prefix)
49 self.fd: Optional[TextIO] = None
50
51 self.flush_time = monotonic()
52 self.time_to_flush = time_to_flush # Время с момента прошлого сброса лога,
53 # после которого, выполняется принудительный сброс буферов
54
55 if file_name is not None:
56 self.fd = self._open_file(file_name)
57
58 else:
59 self.fd = file_obj
60
61 if self.fd is None:
62 raise ValueError('Не задан файл для записи журналов')
63
64 def flush(self, time_mark: float = None):
65 if time_mark is None:
66 time_mark = monotonic()
67
68 self.flush_time = time_mark
69 self.fd.flush()
70
71 def close(self):
72 self.fd.flush()
73 self.fd.close()
74 self.fd = None
75
76 def __del__(self):
77 if self.fd is not None:
78 self.fd.flush()
79 self.fd.close()
80 self.fd = None
81
82 def _flush_event(self):
83 t = monotonic()
84 if t - self.flush_time >= self.time_to_flush:
85 self.flush(t)
86
87 def _write(self, mark: str, msg: Any):
88 super()._write(mark=mark, msg=msg)
89
90 self._flush_event()
91
92 def sub_log(self, name: str):
93 if self.fd is None:
94 raise ValueError('Попытка использовать закрытый файл журнала')
95
96 return self.__class__(f'{self.prefix}/{name}', file_obj=self.fd, time_to_flush=self.time_to_flush)
97
98
99 class LogrotateFile(FileLog):
100 def __init__(self,
101 file_base: str,
102 prefix: str = 'main',
103 file_obj: Optional[TextIO] = None,
104 time_to_flush: int = TIME_TO_FLUSH):
105
106 d = date.today()
107 if file_obj is None:
108 super().__init__(prefix=prefix, file_name=f'{file_base}-{d}.log', time_to_flush=time_to_flush)
109 else:
110 super().__init__(prefix=prefix, file_obj=file_obj, time_to_flush=time_to_flush)
111
112 self.logrotate_base = file_base
113 self.logrotate_date = d
114
115 @classmethod
116 def make(cls, directory: str = '.', prefix: str = 'main', time_to_flush: int = TIME_TO_FLUSH):
117 if len(p_split(prefix)) > 1:
118 raise ValueError(f'Префикс журналирования не должен быть похож '
119 f'на пусть файловой системы, убери знак разделения директорий: {prefix}')
120
121 directory = abspath(directory)
122 file_base = p_join(directory, prefix)
123 return cls(prefix=prefix, file_base=file_base, time_to_flush=time_to_flush)
124
125 def flush(self, time_mark: float = None, no_rotate: bool = False):
126 if not no_rotate:
127 d = date.today()
128 if self.logrotate_date != d:
129 self.logrotate_rotate()
130 return
131
132 super().flush(time_mark=time_mark)
133
134 def logrotate_rotate(self):
135 d = date.today()
136 file_name = f'{self.logrotate_base}-{d}.log'
137 _fd = self.fd
138 self.fd = self._open_file(file_name)
139 self.logrotate_date = d
140 self.flush(no_rotate=True)
141 _fd.flush()
142 _fd.close()
143
144 def sub_log(self, name: str):
145 return self.__class__(
146 file_base=self.logrotate_base,
147 file_obj=self.fd, prefix=f'{self.prefix}/{name}',
148 time_to_flush=self.time_to_flush
149 )