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