core/logger.py - master - HookDev-Arch/ServerMonitor

core/logger.py
157 строк6 KB
HookDev-Arch
upload files
09 окт 2025, 02:25
09 окт 2025, 02:258bd6873
"""
Система логирования для HookDev-Arch Dashboard v3.5
Структурированное логирование с ротацией файлов
"""
import logging
import logging.handlers
import json
import time
from pathlib import Path
from typing import Any, Dict, Optional
from core.config import settings


class StructuredFormatter(logging.Formatter):
    """Форматтер для структурированного логирования"""
    
    def format(self, record: logging.LogRecord) -> str:
        log_entry = {
            'timestamp': time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(record.created)),
            'level': record.levelname,
            'logger': record.name,
            'message': record.getMessage(),
            'module': record.module,
            'function': record.funcName,
            'line': record.lineno
        }
        
        # Добавляем исключения если есть
        if record.exc_info:
            log_entry['exception'] = self.formatException(record.exc_info)
        
        # Добавляем дополнительные поля
        if hasattr(record, 'user_id'):
            log_entry['user_id'] = record.user_id
        if hasattr(record, 'request_id'):
            log_entry['request_id'] = record.request_id
        if hasattr(record, 'duration'):
            log_entry['duration'] = record.duration
        
        return json.dumps(log_entry, ensure_ascii=False)


class SecurityFilter(logging.Filter):
    """Фильтр для скрытия чувствительной информации"""
    
    SENSITIVE_KEYS = ['password', 'token', 'secret', 'key', 'auth']
    
    def filter(self, record: logging.LogRecord) -> bool:
        # Скрываем чувствительные данные в сообщениях
        if hasattr(record, 'args') and record.args:
            args = list(record.args)
            for i, arg in enumerate(args):
                if isinstance(arg, dict):
                    args[i] = self._sanitize_dict(arg)
                elif isinstance(arg, str) and any(key in arg.lower() for key in self.SENSITIVE_KEYS):
                    args[i] = "***REDACTED***"
            record.args = tuple(args)
        
        return True
    
    def _sanitize_dict(self, data: Dict[str, Any]) -> Dict[str, Any]:
        """Рекурсивно очищает словарь от чувствительных данных"""
        sanitized = {}
        for key, value in data.items():
            if any(sensitive in key.lower() for sensitive in self.SENSITIVE_KEYS):
                sanitized[key] = "***REDACTED***"
            elif isinstance(value, dict):
                sanitized[key] = self._sanitize_dict(value)
            else:
                sanitized[key] = value
        return sanitized


def setup_logging() -> logging.Logger:
    """Настраивает систему логирования"""
    
    # Создаем логгер
    logger = logging.getLogger("hookdev_arch")
    logger.setLevel(getattr(logging, settings.log_level.upper()))
    
    # Очищаем существующие обработчики
    logger.handlers.clear()
    
    # Создаем директорию для логов
    log_file = Path(settings.log_file)
    log_file.parent.mkdir(parents=True, exist_ok=True)
    
    # Обработчик для файла с ротацией
    file_handler = logging.handlers.RotatingFileHandler(
        log_file,
        maxBytes=settings.log_max_size,
        backupCount=settings.log_backup_count,
        encoding='utf-8'
    )
    file_handler.setFormatter(StructuredFormatter())
    file_handler.addFilter(SecurityFilter())
    
    # Обработчик для консоли
    console_handler = logging.StreamHandler()
    console_formatter = logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )
    console_handler.setFormatter(console_formatter)
    console_handler.addFilter(SecurityFilter())
    
    # Добавляем обработчики
    logger.addHandler(file_handler)
    logger.addHandler(console_handler)
    
    # Настраиваем логирование для внешних библиотек
    logging.getLogger("uvicorn").setLevel(logging.WARNING)
    logging.getLogger("fastapi").setLevel(logging.WARNING)
    
    return logger


def get_logger(name: str = None) -> logging.Logger:
    """Получает логгер с заданным именем"""
    if name:
        return logging.getLogger(f"hookdev_arch.{name}")
    return logging.getLogger("hookdev_arch")


# Создаем основной логгер
logger = setup_logging()


class LoggerMixin:
    """Миксин для добавления логирования в классы"""
    
    @property
    def logger(self) -> logging.Logger:
        return get_logger(self.__class__.__name__)


def log_performance(func_name: str, duration: float, **kwargs):
    """Логирует производительность операций"""
    logger.info(
        f"Performance: {func_name}",
        extra={
            'duration': duration,
            'function': func_name,
            **kwargs
        }
    )


def log_security_event(event_type: str, user_id: str = None, **kwargs):
    """Логирует события безопасности"""
    logger.warning(
        f"Security event: {event_type}",
        extra={
            'event_type': event_type,
            'user_id': user_id,
            **kwargs
        }
    )