import logging import sys from logging.config import dictConfig class UTF8StreamHandler(logging.StreamHandler): def __init__(self, stream=None): super().__init__(stream or sys.stdout) self.setStream(stream or sys.stdout) def setStream(self, stream): super().setStream(stream) if hasattr(stream, 'reconfigure'): stream.reconfigure(encoding='utf-8') class FilterByMessage(logging.Filter): def filter(self, record): # Фильтруем сообщения, содержащие 'Received 1 new updates' return 'Received ' not in record.getMessage() class LogManager: def __init__(self): self.setup_logging() def setup_logging(self): dictConfig({ 'version': 1, 'disable_existing_loggers': False, 'formatters': { 'default': { 'format': '[%(asctime)s] %(levelname)s %(module)s: %(message)s', }, 'debug': { 'format': '[%(asctime)s] %(levelname)s %(module)s [%(funcName)s:%(lineno)d]: %(message)s' } }, 'filters': { 'filter_by_message': { '()': FilterByMessage, } }, 'handlers': { 'console': { '()': UTF8StreamHandler, 'stream': 'ext://sys.stdout', 'formatter': 'default', 'filters': ['filter_by_message'], }, }, 'root': { 'level': 'WARNING', 'handlers': ['console'] }, 'loggers': { 'flask': { 'level': 'DEBUG', 'handlers': ['console'], 'propagate': False, }, 'telebot': { 'level': 'INFO', 'handlers': ['console'], 'propagate': False, }, 'werkzeug': { 'level': 'DEBUG', 'handlers': ['console'], 'propagate': False, }, 'flask_ldap3_login': { 'level': 'DEBUG', 'handlers': ['console'], 'propagate': False, }, 'flask_login': { 'level': 'DEBUG', 'handlers': ['console'], 'propagate': False, }, 'pyzabbix': { 'level': 'DEBUG', 'handlers': ['console'], 'propagate': False, }, 'app': { 'level': 'DEBUG', 'handlers': ['console'], 'propagate': False, }, 'pika': { 'level': 'INFO', 'handlers': ['console'], 'propagate': False, }, 'users_service': { 'level': 'DEBUG', 'handlers': ['console'], 'propagate': False, }, 'regions_service': { 'level': 'DEBUG', 'handlers': ['console'], 'propagate': False, }, 'debug': { 'level': 'DEBUG', 'handlers': ['console'], 'propagate': False, }, } }) def change_log_level(self, component, level): """Changes the log level of a specified component.""" if level not in ['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL']: return False, 'Invalid log level' log_level = getattr(logging, level, logging.DEBUG) if component in self.get_all_loggers(): logger = logging.getLogger(component) logger.setLevel(log_level) for handler in logger.handlers: handler.setLevel(log_level) return True, f'Log level for {component} changed to {level}' else: return False, 'Invalid component' def get_all_loggers(self): """Returns a list of all configured loggers.""" return list(logging.Logger.manager.loggerDict.keys())