|
import platform |
|
import os |
|
import inspect |
|
import logging |
|
import logging.handlers |
|
from datetime import datetime |
|
from typing import List, Optional |
|
|
|
# Дополнительная проверка для ОС Windows |
|
if platform.system() == "Windows": |
|
os.system('color') |
|
|
|
class CustomFormatter(logging.Formatter): |
|
DEFAULT_FORMAT = "%(asctime)s - %(name)s - %(levelname)s - %(message)s (%(filename)s:%(lineno)d)" |
|
|
|
COLORS = { |
|
"black": "\x1b[30m", |
|
"red": "\x1b[31m", |
|
"green": "\x1b[32m", |
|
"yellow": "\x1b[33m", |
|
"blue": "\x1b[34m", |
|
"magenta": "\x1b[35m", |
|
"cyan": "\x1b[36m", |
|
"white": "\x1b[37m", |
|
"reset": "\x1b[0m" |
|
} |
|
|
|
LEVEL_COLORS = { |
|
logging.DEBUG: COLORS["green"], |
|
logging.INFO: COLORS["magenta"], |
|
logging.WARNING: COLORS["yellow"], |
|
logging.ERROR: COLORS["red"], |
|
logging.CRITICAL: COLORS["cyan"] |
|
} |
|
|
|
CONSOLE_FORMATS = { |
|
logging.DEBUG: LEVEL_COLORS[logging.DEBUG] + DEFAULT_FORMAT + COLORS["reset"], |
|
logging.INFO: LEVEL_COLORS[logging.INFO] + DEFAULT_FORMAT + COLORS["reset"], |
|
logging.WARNING: LEVEL_COLORS[logging.WARNING] + DEFAULT_FORMAT + COLORS["reset"], |
|
logging.ERROR: LEVEL_COLORS[logging.ERROR] + DEFAULT_FORMAT + COLORS["reset"], |
|
logging.CRITICAL: LEVEL_COLORS[logging.CRITICAL] + DEFAULT_FORMAT + COLORS["reset"] |
|
} |
|
|
|
FILE_FORMATS = { |
|
logging.DEBUG: DEFAULT_FORMAT, |
|
logging.INFO: DEFAULT_FORMAT, |
|
logging.WARNING: DEFAULT_FORMAT, |
|
logging.ERROR: DEFAULT_FORMAT, |
|
logging.CRITICAL: DEFAULT_FORMAT |
|
} |
|
|
|
def __init__(self, is_file=False): |
|
super().__init__() |
|
self.is_file = is_file |
|
|
|
def format(self, record: logging.LogRecord) -> str: |
|
if self.is_file: |
|
log_fmt = self.FILE_FORMATS.get(record.levelno, self.DEFAULT_FORMAT) |
|
else: |
|
log_fmt = self.CONSOLE_FORMATS.get(record.levelno, self.DEFAULT_FORMAT) |
|
self._style._fmt = log_fmt |
|
return super().format(record) |
|
|
|
@classmethod |
|
def set_file_format(cls, level: str, format_string: str): |
|
"""Устанавливает форматирование для заданного уровня логирования для файлового вывода.""" |
|
log_level = getattr(logging, level.upper(), None) |
|
if log_level is not None: |
|
cls.FILE_FORMATS[log_level] = format_string |
|
else: |
|
raise ValueError(f"Некорректный уровень логирования: {level}") |
|
|
|
@classmethod |
|
def set_console_format(cls, level: str, format_string: str): |
|
"""Устанавливает форматирование для заданного уровня логирования.""" |
|
log_level = getattr(logging, level.upper(), None) |
|
if log_level is not None: |
|
cls.CONSOLE_FORMATS[log_level] = format_string |
|
else: |
|
raise ValueError(f"Некорректный уровень логирования: {level}") |
|
|
|
@classmethod |
|
def set_color(cls, level: str, color: str): |
|
"""Устанавливает цвет для заданного уровня логирования.""" |
|
log_level = getattr(logging, level.upper(), None) |
|
if log_level in cls.LEVEL_COLORS: |
|
cls.LEVEL_COLORS[log_level] = cls.COLORS[color] |
|
cls.CONSOLE_FORMATS[log_level] = cls.LEVEL_COLORS[log_level] + cls.DEFAULT_FORMAT + cls.COLORS["reset"] |
|
else: |
|
raise ValueError(f"Некорректный уровень логирования: {level}") |
|
|
|
|
|
class MultiLevelFilter(logging.Filter): |
|
def __init__(self, levels: List[int]): |
|
super().__init__() |
|
self.levels = levels |
|
|
|
def filter(self, record: logging.LogRecord) -> bool: |
|
return record.levelno in self.levels |
|
|
|
|
|
class LevelFilter(logging.Filter): |
|
def __init__(self, level: int): |
|
super().__init__() |
|
self.level = level |
|
|
|
def filter(self, record: logging.LogRecord) -> bool: |
|
return record.levelno == self.level |
|
|
|
|
|
class LoggerManager: |
|
LEVEL_MAPPING = { |
|
'DEBUG': logging.DEBUG, |
|
'INFO': logging.INFO, |
|
'WARNING': logging.WARNING, |
|
'ERROR': logging.ERROR, |
|
'CRITICAL': logging.CRITICAL |
|
} |
|
|
|
def __init__(self, *, name: str, level: int = logging.DEBUG, log_to_file: bool = False, file_name: str = 'app.log', |
|
log_dir: str = 'logs', max_bytes: int = 1024 * 1024, backup_count: int = 5): |
|
if not isinstance(name, str) or not name.strip(): |
|
print(f"Некорректное имя логгера: '{name}'. Устанавливается имя по умолчанию 'DefaultLogger'.") |
|
name = "DefaultLogger" |
|
self.max_bytes: int = max_bytes |
|
self.backup_count: int = backup_count |
|
self.logger = logging.getLogger(name) |
|
self.logger.setLevel(level) |
|
self.console_handler: Optional[logging.StreamHandler] = None |
|
self.file_handler: Optional[logging.FileHandler] = None |
|
self.log_filter: Optional[logging.Filter] = None |
|
self.setup_console_handler(level) |
|
if log_to_file: |
|
self.enable_file_logging(file_name, log_dir) |
|
|
|
def setup_console_handler(self, level: int): |
|
"""Настройка консольного обработчика и форматтера""" |
|
self.console_handler = logging.StreamHandler() |
|
self.console_handler.setLevel(level) |
|
self.console_handler.setFormatter(CustomFormatter()) |
|
self.logger.addHandler(self.console_handler) |
|
|
|
def enable_console_logging(self): |
|
"""Включение логирования в консоль""" |
|
if self.console_handler: |
|
self.console_handler.setLevel(self.logger.level) |
|
self.logger.info("Логирование в консоль включено") |
|
|
|
def disable_console_logging(self): |
|
"""Отключение логирования в консоль""" |
|
if self.console_handler: |
|
self.console_handler.setLevel( |
|
logging.CRITICAL + 1) # Уровень выше CRITICAL, чтобы игнорировать все сообщения |
|
self.logger.info("Логирование в консоль отключено") |
|
else: |
|
self.logger.warning("Обработчик консольного логирования уже отключен или не был инициализирован") |
|
|
|
def enable_logging(self): |
|
"""Включение логирования""" |
|
self.logger.disabled = False |
|
self.logger.info("Логирование включено") |
|
|
|
def disable_logging(self): |
|
"""Отключение логирования""" |
|
self.logger.disabled = True |
|
self.logger.info("Логирование отключено") |
|
|
|
def enable_file_logging(self, file_name: str = 'app.log', log_dir: str = 'logs'): |
|
try: |
|
if not os.path.exists(log_dir): |
|
os.makedirs(log_dir) |
|
except Exception as e: |
|
self.logger.error(f"Ошибка при создании директории для логов: {e}") |
|
raise |
|
|
|
log_path = os.path.join(log_dir, file_name) |
|
|
|
if self.file_handler: |
|
self.logger.removeHandler(self.file_handler) |
|
self.file_handler.close() |
|
|
|
try: |
|
self.file_handler = logging.handlers.RotatingFileHandler( |
|
log_path, |
|
maxBytes=self.max_bytes, # Максимальный размер файла |
|
backupCount=self.backup_count # Количество ротируемых файлов |
|
) |
|
self.file_handler.setLevel(self.logger.level) |
|
file_formatter = CustomFormatter(is_file=True) |
|
self.file_handler.setFormatter(file_formatter) |
|
self.logger.addHandler(self.file_handler) |
|
self.logger.info(f"Логирование в файл '{log_path}' включено") |
|
except Exception as e: |
|
self.logger.error(f"Ошибка при настройке файлового логгера с ротацией: {e}") |
|
raise |
|
|
|
def disable_file_logging(self): |
|
"""Отключение логирования в файл""" |
|
if self.file_handler: |
|
self.logger.removeHandler(self.file_handler) |
|
self.file_handler.close() |
|
self.file_handler = None |
|
self.logger.info("Логирование в файл отключено") |
|
else: |
|
self.logger.warning("Обработчик файла логирования уже отключен или не был инициализирован") |
|
|
|
def set_file_handler_params(self, max_bytes: int, backup_count: int): |
|
"""Установка параметров для RotatingFileHandler.""" |
|
self.max_bytes = max_bytes |
|
self.backup_count = backup_count |
|
if self.file_handler: |
|
self.file_handler.maxBytes = max_bytes |
|
self.file_handler.backupCount = backup_count |
|
self.logger.info( |
|
f"Изменены параметры RotatingFileHandler: maxBytes={max_bytes}, backupCount={backup_count}") |
|
else: |
|
self.logger.warning( |
|
"Файловый обработчик не инициализирован. Новые параметры будут применены при следующем включении логирования в файл.") |
|
|
|
def set_name(self, name: str): |
|
"""Установка нового имени логгера""" |
|
if not isinstance(name, str) or not name.strip(): |
|
self.logger.error(f"Некорректное новое имя логгера: '{name}'. Имя остается без изменений.") |
|
return |
|
|
|
self.logger.name = name |
|
self.logger.info(f"Имя логгера изменено на '{name}'") |
|
|
|
def set_level(self, level: str): |
|
"""Установка уровня логирования""" |
|
log_level = self.LEVEL_MAPPING.get(level) |
|
if log_level is not None: |
|
self.logger.setLevel(log_level) |
|
if self.console_handler: |
|
self.console_handler.setLevel(log_level) |
|
if self.file_handler: |
|
self.file_handler.setLevel(log_level) |
|
self.logger.info(f"Уровень логирования установлен на {level}") |
|
else: |
|
self.logger.error( |
|
f"Некорректный уровень логирования: {level}. Используйте один из: {list(self.LEVEL_MAPPING.keys())}") |
|
|
|
def set_filter(self, level: str): |
|
"""Установка фильтра для логирования""" |
|
log_level = self.LEVEL_MAPPING.get(level) |
|
if log_level is not None: |
|
self.clear_filter() |
|
self.log_filter = LevelFilter(log_level) |
|
self.logger.addFilter(self.log_filter) |
|
self.logger.info(f"Фильтр уровня логирования установлен на {level}") |
|
else: |
|
self.logger.error( |
|
f"Некорректный уровень фильтра: {level}. Используйте один из: {list(self.LEVEL_MAPPING.keys())}") |
|
|
|
def set_filter_list(self, levels: List[str]): |
|
"""Установка списка фильтров для логирования""" |
|
if not levels: |
|
self.logger.error("Список уровней фильтра не должен быть пустым.") |
|
return |
|
|
|
valid_levels = [self.LEVEL_MAPPING[level] for level in levels if level in self.LEVEL_MAPPING] |
|
|
|
if not valid_levels: |
|
self.logger.error( |
|
f"Некорректные уровни фильтра: {levels}. Используйте один из: {list(self.LEVEL_MAPPING.keys())}") |
|
return |
|
|
|
self.clear_filter() |
|
self.log_filter = MultiLevelFilter(valid_levels) |
|
self.logger.addFilter(self.log_filter) |
|
self.logger.info(f"Фильтр уровней логирования установлен на {levels}") |
|
|
|
def reset_level(self): |
|
"""Сброс уровня логирования до уровня DEBUG""" |
|
self.set_level('DEBUG') |
|
|
|
def clear_filter(self): |
|
"""Очистка установленного фильтра""" |
|
if self.log_filter: |
|
self.logger.removeFilter(self.log_filter) |
|
self.log_filter = None |
|
self.logger.info("Фильтр логирования очищен") |
|
|
|
def time_execution(self): |
|
"""Контекстный менеджер для логирования времени выполнения блока кода""" |
|
return TimeExecutionLogger(self.logger) |
|
|
|
|
|
class TimeExecutionLogger: |
|
def __init__(self, with_logger, level=logging.INFO): |
|
self.logger = with_logger |
|
self.level = level |
|
self.start_time = datetime.now() |
|
caller_frame = inspect.stack()[ |
|
2] # Получаем информацию о вызывающем коде (2 - это индекс фрейма с вызывающим кодом) |
|
self.caller_file = caller_frame.filename |
|
self.caller_line = caller_frame.lineno |
|
self.logger.log(self.level, |
|
f"Начало выполнения блока кода в {self.start_time.strftime('%Y-%m-%d %H:%M:%S.%f')}. Вызвано из файла '{self.caller_file}', строка {self.caller_line}") |
|
|
|
def __enter__(self): |
|
return self |
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb): |
|
end_time = datetime.now() |
|
execution_time = (end_time - self.start_time).total_seconds() |
|
if exc_type is None: |
|
self.logger.log(self.level, |
|
f"Завершение выполнения блока кода в {end_time.strftime('%Y-%m-%d %H:%M:%S.%f')}. Время выполнения: {execution_time:.6f} секунд. Вызвано из файла '{self.caller_file}', строка {self.caller_line}") |
|
else: |
|
self.logger.log(self.level, |
|
f"Завершение выполнения блока кода с ошибкой {exc_val} в {end_time.strftime('%Y-%m-%d %H:%M:%S.%f')}. Время выполнения: {execution_time:.6f} секунд. Вызвано из файла '{self.caller_file}', строка {self.caller_line}") |
|
return False # Возвращаем False, чтобы исключение продолжило подъём |
|
|
|
|
|
if __name__ == "__main__": |
|
import time |
|
|
|
logger_manager = LoggerManager(name="TestLogger") |
|
# Установим форматы для файлового логирования |
|
CustomFormatter.set_file_format('DEBUG', "%(asctime)s - !DEBUG!: %(message)s (%(filename)s:%(lineno)d)") |
|
CustomFormatter.set_file_format('INFO', "%(asctime)s - !INFO!: %(message)s") |
|
CustomFormatter.set_file_format('WARNING', "%(asctime)s - !WARNING!: %(message)s") |
|
CustomFormatter.set_file_format('ERROR', "%(asctime)s - !ERROR!: %(message)s") |
|
# CustomFormatter.set_file_format('CRITICAL', "%(asctime)s - !CRITICAL!: %(message)s") |
|
# CustomFormatter.set_console_format('INFO', f"%(asctime)s - %(levelname)s - %(message)s (%(filename)s:%(lineno)d)") # Задать свой формат |
|
# CustomFormatter.set_color('DEBUG', 'blue') # Задать цвет из доступных |
|
|
|
# logger_manager.set_file_handler_params(max_bytes = 2048*1024, backup_count = 10) # Максимальный размер файла и количество ротируемых файлов |
|
# logger_manager.enable_console_logging() # Включение логирования в консоль |
|
# logger_manager.disable_console_logging() # Отключение логирования в консоль |
|
# logger_manager.disable_logging() # Отключение логирования |
|
# logger_manager.enable_logging() # Включение логирования |
|
# logger_manager.set_level('INFO') # Установка уровня логирования на INFO |
|
# logger_manager.set_filter_list(['DEBUG' ,'WARNING', 'ERROR' ]) # Установка списка фильтров для логирования |
|
# logger_manager.set_filter('INFO') # Фильтрация сообщений только уровня INFO |
|
# logger_manager.clear_filter() # Очистка установленного фильтра |
|
# logger_manager.reset_level() # Сброс уровня логирования до уровня DEBUG |
|
# logger_manager.set_name("NewNameLogger") # Установка нового имени логгера |
|
with logger_manager.time_execution(): |
|
for i in range(6): |
|
# Включение и отключение логирования в файл |
|
logger_manager.enable_file_logging(f"{datetime.now().strftime('%Y-%m-%d %H-%M')}-{i}.log") |
|
# logger_manager.disable_file_logging() |
|
|
|
# Логирование сообщений |
|
logger = logger_manager.logger |
|
logger.debug(f"debug message {i}") |
|
logger.info(f"info message {i}") |
|
logger.warning(f"warning message {i}") |
|
logger.error(f"error message {i}") |
|
logger.critical(f"critical message {i}") |
|
time.sleep(1) |
