app.py

"""
Gradio приложение для мультиагентной системы анализа логов.
Оркестрирует работу трёх агентов и предоставляет веб-интерфейс.
"""

import gradio as gr
import json
import os
from typing import Tuple, Dict, Any

from agents import LogParserAgent, AnomalyDetectionAgent, RootCauseAgent


class LogAnalysisOrchestrator:
    """Оркестратор для последовательного выполнения агентов."""
    
    def __init__(self):
        """Инициализация оркестратора и агентов."""
        self.parser_agent = LogParserAgent()
        self.anomaly_agent = AnomalyDetectionAgent()
        self.rca_agent = RootCauseAgent()
    
    def analyze(self, raw_logs: str) -> Tuple[str, str, str]:
        """
        Выполняет полный цикл анализа логов через всех агентов.
        
        Args:
            raw_logs: Сырые логи в виде строки
            
        Returns:
            Кортеж из трёх строк:
            - JSON с распарсенными логами
            - JSON с обнаруженными аномалиями
            - Markdown с рекомендациями
        """
        try:
            # Валидация входных данных
            if not raw_logs or not raw_logs.strip():
                empty_json = json.dumps({"error": "Логи не предоставлены"}, ensure_ascii=False, indent=2)
                return empty_json, empty_json, "# Ошибка\n\nЛоги не предоставлены для анализа."
            
            # Agent 1: Парсинг логов
            try:
                structured_data = self.parser_agent.parse(raw_logs)
                parsed_json = json.dumps(structured_data, ensure_ascii=False, indent=2)
            except Exception as e:
                error_msg = {"error": f"Ошибка парсинга логов: {str(e)}"}
                parsed_json = json.dumps(error_msg, ensure_ascii=False, indent=2)
                return parsed_json, parsed_json, f"# Ошибка\n\nОшибка на этапе парсинга: {str(e)}"
            
            # Agent 2: Обнаружение аномалий
            try:
                anomaly_report = self.anomaly_agent.detect(structured_data)
                anomalies_json = json.dumps(anomaly_report, ensure_ascii=False, indent=2)
            except Exception as e:
                error_msg = {"error": f"Ошибка обнаружения аномалий: {str(e)}"}
                anomalies_json = json.dumps(error_msg, ensure_ascii=False, indent=2)
                return parsed_json, anomalies_json, f"# Ошибка\n\nОшибка на этапе обнаружения аномалий: {str(e)}"
            
            # Agent 3: Анализ первопричин и рекомендации
            try:
                recommendations_md = self.rca_agent.analyze(anomaly_report)
            except Exception as e:
                recommendations_md = f"# Ошибка\n\nОшибка на этапе анализа первопричин: {str(e)}"
            
            return parsed_json, anomalies_json, recommendations_md
            
        except Exception as e:
            error_json = json.dumps({"error": f"Критическая ошибка: {str(e)}"}, ensure_ascii=False, indent=2)
            return error_json, error_json, f"# Критическая ошибка\n\n{str(e)}"


# Глобальный экземпляр оркестратора
orchestrator = LogAnalysisOrchestrator()


def analyze_logs(raw_logs: str) -> Tuple[str, str, str]:
    """
    Обёртка для Gradio интерфейса.
    
    Args:
        raw_logs: Сырые логи из интерфейса
        
    Returns:
        Кортеж результатов для отображения
    """
    return orchestrator.analyze(raw_logs)


def create_interface():
    """Создаёт и настраивает Gradio интерфейс."""
    
    # Описание интерфейса
    description = """
    # 🔍 Мультиагентная система анализа логов
    
    Система использует трёх независимых агентов для анализа логов:
    1. **Log Parser Agent** - парсит и структурирует логи
    2. **Anomaly Detection Agent** - обнаруживает аномалии и паттерны
    3. **Root Cause Agent** - анализирует первопричины и генерирует рекомендации
    
    Вставьте логи в поле ниже или загрузите лог-файл, затем нажмите "Анализировать".
    """
    
    # Создание интерфейса
    with gr.Blocks(title="Multi-Agent Log Analysis") as app:
        gr.Markdown(description)
        
        with gr.Row():
            with gr.Column(scale=1):
                log_input = gr.Textbox(
                    label="Логи для анализа",
                    placeholder="Вставьте логи здесь или используйте кнопку загрузки ниже...",
                    lines=15,
                    max_lines=30
                )
                
                upload_btn = gr.UploadButton(
                    "📁 Загрузить лог-файл",
                    file_types=[".log", ".txt"],
                    file_count="single"
                )
                
                analyze_btn = gr.Button("🔍 Анализировать", variant="primary", size="lg")
                
                # Обработчик загрузки файла
                def load_file(file):
                    if file is None:
                        return ""
                    try:
                        with open(file.name, 'r', encoding='utf-8') as f:
                            content = f.read()
                        return content
                    except Exception as e:
                        return f"Ошибка чтения файла: {str(e)}"
                
                upload_btn.upload(load_file, inputs=upload_btn, outputs=log_input)
        
        with gr.Row():
            with gr.Tabs():
                with gr.Tab("📊 Распарсенные логи (JSON)"):
                    parsed_output = gr.JSON(
                        label="Структурированные данные"
                    )
                
                with gr.Tab("⚠️ Обнаруженные аномалии (JSON)"):
                    anomalies_output = gr.JSON(
                        label="Отчёт об аномалиях"
                    )
                
                with gr.Tab("💡 Анализ и рекомендации (Markdown)"):
                    recommendations_output = gr.Markdown(
                        label="Рекомендации по устранению проблем"
                    )
        
        # Примеры логов для тестирования
        gr.Markdown("### 📝 Примеры логов для тестирования")
        with gr.Row():
            example_logs = [
                """2024-01-15 10:00:00 INFO Application started
2024-01-15 10:00:05 INFO Database connection established
2024-01-15 10:01:00 ERROR Connection timeout to external API
2024-01-15 10:01:05 ERROR Connection timeout to external API
2024-01-15 10:01:10 ERROR Connection timeout to external API
2024-01-15 10:01:15 WARNING High memory usage detected: 85%
2024-01-15 10:02:00 CRITICAL System crash detected
2024-01-15 10:02:01 INFO Application shutdown""",
                
                """[2024-01-15 14:30:00] INFO User authentication successful
[2024-01-15 14:30:01] DEBUG Request received: GET /api/users
[2024-01-15 14:30:02] ERROR Database query failed: connection lost
[2024-01-15 14:30:03] ERROR Database query failed: connection lost
[2024-01-15 14:30:04] ERROR Database query failed: connection lost
[2024-01-15 14:30:05] ERROR Database query failed: connection lost
[2024-01-15 14:30:06] WARNING Retrying database connection
[2024-01-15 14:30:10] INFO Database connection restored"""
            ]
            
            example_btn1 = gr.Button("Загрузить пример 1", size="sm")
            example_btn2 = gr.Button("Загрузить пример 2", size="sm")
            
            example_btn1.click(
                lambda: example_logs[0],
                outputs=log_input
            )
            
            example_btn2.click(
                lambda: example_logs[1],
                outputs=log_input
            )
        
        # Связывание кнопки анализа с обработчиком
        analyze_btn.click(
            fn=analyze_logs,
            inputs=log_input,
            outputs=[parsed_output, anomalies_output, recommendations_output]
        )
        
        # Информация о системе
        gr.Markdown("""
        ---
        ### ℹ️ Информация о системе
        
        - **Архитектура:** Мультиагентная система (3 независимых агента)
        - **Платформа:** Hugging Face Spaces
        - **Интерфейс:** Gradio
        - **Поддержка:** До 10,000 строк логов
        """)
    
    return app

def env_flag(name: str, default: bool = False) -> bool:
    """
    Безопасно читает булевы переменные окружения (1/0, true/false и т.д.).
    """
    raw = os.getenv(name)
    if raw is None:
        return default
    return raw.lower() in ("1", "true", "yes", "on")

# Создание приложения
demo = create_interface()

# Запуск приложения
if __name__ == "__main__":
    in_hf_space = bool(os.getenv("SPACE_ID") or os.getenv("HF_SPACE"))
    share_flag = False if in_hf_space else env_flag("GRADIO_SHARE", default=False)
    host = os.getenv("GRADIO_HOST") or os.getenv("GRADIO_SERVER_NAME") or "127.0.0.1"
    port = int(os.getenv("PORT") or os.getenv("GRADIO_SERVER_PORT") or 7860)
    demo.queue(api_open=False).launch(
        server_name=host,
        server_port=port,
        share=share_flag,
        show_api=False,
    )