py.lib.aw_log

Yohn Y. 2024-02-11 Parent:41b53fd5637e Child:9155a66edb31

3:4ab8c8590cab Go to Latest

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

. Оформление репозитория

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