ИИ Анализ Логов: Интеллектуальное Обнаружение Ошибок и Анализ Первопричин

from sklearn.ensemble import IsolationForest

class ДетекторАномалийЛогов:
    def __init__(self):
        self.модель = IsolationForest(contamination=0.01)
        self.векторизатор = TfidfVectorizer(max_features=100)

    def обучить(self, нормальные_логи):
        """Обучить на нормальных операционных логах"""
        векторы_логов = self.векторизатор.fit_transform(нормальные_логи)
        self.модель.fit(векторы_логов.toarray())

    def обнаружить_аномалии(self, логи):
        """Обнаружить аномальные записи логов"""
        векторы_логов = self.векторизатор.transform(логи)
        предсказания = self.модель.predict(векторы_логов.toarray())

        аномалии = [
            {'лог': лог, 'оценка_аномалии': оценка}
            for лог, пред, оценка in zip(логи, предсказания, self.модель.score_samples(векторы_логов.toarray()))
            if пред == -1
        ]

        return аномалии

from sklearn.cluster import DBSCAN
from sentence_transformers import SentenceTransformer

class КластеризаторЛогов:
    def __init__(self):
        self.модель = SentenceTransformer('all-MiniLM-L6-v2')

    def кластеризовать_логи(self, логи_ошибок):
        """Группировать похожие сообщения об ошибках"""
        эмбеддинги = self.модель.encode(логи_ошибок)
        кластеризация = DBSCAN(eps=0.5, min_samples=2).fit(эмбеддинги)

        кластеры = {}
        for индекс, метка in enumerate(кластеризация.labels_):
            if метка not in кластеры:
                кластеры[метка] = []
            кластеры[метка].append(логи_ошибок[индекс])

        return {
            'количество_кластеров': len(set(кластеризация.labels_)) - (1 if -1 in кластеризация.labels_ else 0),
            'кластеры': кластеры,
            'шум': кластеры.get(-1, [])
        }

import networkx as nx

class АнализаторПервопричин:
    def __init__(self):
        self.граф_зависимостей = nx.DiGraph()

    def анализировать_сбой(self, упавший_сервис, логи_ошибок, окно_времени_минуты=5):
        """Проследить сбой до первопричины"""
        вышестоящие = list(nx.ancestors(self.граф_зависимостей, упавший_сервис))
        время_сбоя = логи_ошибок[упавший_сервис]['timestamp']
        первопричины = []

        for вышестоящий_сервис in вышестоящие:
            if вышестоящий_сервис in логи_ошибок:
                время_ошибки_вышестоящего = логи_ошибок[вышестоящий_сервис]['timestamp']
                разница_времени = (время_сбоя - время_ошибки_вышестоящего).total_seconds() / 60

                if 0 < разница_времени <= окно_времени_минуты:
                    первопричины.append({
                        'сервис': вышестоящий_сервис,
                        'время_до_сбоя': разница_времени,
                        'ошибка': логи_ошибок[вышестоящий_сервис]['сообщение']
                    })

        первопричины.sort(key=lambda x: x['время_до_сбоя'], reverse=True)

        return {
            'упавший_сервис': упавший_сервис,
            'вероятная_первопричина': первопричины[0] if первопричины else None,
            'способствующие_факторы': первопричины[1:] if len(первопричины) > 1 else []
        }

from transformers import pipeline

class СемантическийПарсерЛогов:
    def __init__(self):
        self.классификатор = pipeline("zero-shot-classification")
        self.категории = [
            "ошибка_аутентификации",
            "таймаут_базы_данных",
            "сбой_сети",
            "ошибка_памяти"
        ]

    def распарсить_лог(self, сообщение_лога):
        """Извлечь структурированную информацию из лога"""
        результат = self.классификатор(сообщение_лога, self.категории)

        return {
            'категория': результат['labels'][0],
            'уверенность': результат['scores'][0],
            'сырое_сообщение': сообщение_лога
        }

class МониторЛоговРеальногоВремени:
    def мониторить(self):
        """Мониторить логи в реальном времени"""
        буфер_логов = []

        for сообщение in self.потребитель:
            запись_лога = сообщение.value

            if self.детектор_аномалий.аномальный(запись_лога['сообщение']):
                self.отправить_алерт({
                    'серьёзность': 'ВЫСОКАЯ',
                    'тип': 'аномалия_обнаружена',
                    'лог': запись_лога
                })

            if запись_лога['уровень'] == 'ERROR':
                буфер_логов.append(запись_лога['сообщение'])

            if len(буфер_логов) >= 100:
                кластеры = self.кластеризатор.кластеризовать_логи(буфер_логов)

                for ид_кластера, логи in кластеры['кластеры'].items():
                    if ид_кластера != -1 and len(логи) > 10:
                        self.отправить_алерт({
                            'серьёзность': 'СРЕДНЯЯ',
                            'тип': 'всплеск_ошибок',
                            'ид_кластера': ид_кластера,
                            'количество': len(логи)
                        })

                буфер_логов = []