mcp_server.py

"""
MCP Server для интеграции симулятора заказов с ИИ агентами
Предоставляет инструменты для автоматизированной работы с симулятором
"""

import json
import pandas as pd
from typing import Dict, List, Any, Optional, Union
from dataclasses import asdict
import tempfile
import os

from config_manager import ConfigManager, SimulatorConfig, create_config_from_ui_state, apply_config_to_ui

class MergeSimulatorMCPServer:
    """MCP сервер для симулятора генерации заказов"""
    
    def __init__(self):
        self.config_manager = ConfigManager()
        self.current_simulation_results = None
        self.current_stats_report = ""
    
    # ===============================================================
    # CONFIG MANAGEMENT TOOLS
    # ===============================================================
    
    def mcp_save_simulator_config(
        self,
        name: str,
        description: str,
        chain_data: List[Dict[str, Any]] = None,
        energy_rewards_data: List[Dict[str, Any]] = None,
        item_rewards_data: List[Dict[str, Any]] = None,
        max_history_orders: int = 5,
        increment_difficulty: int = 2,
        energy_chance: int = 90,
        requirement_weights: str = "70,30",
        reduction_factor: int = 3,
        increase_factor: int = 5,
        iteration_count: int = 100,
        initial_energy: int = 10000
    ) -> Dict[str, Any]:
        """
        Сохраняет конфигурацию симулятора для последующего использования
        
        Args:
            name: Название конфигурации
            description: Описание конфигурации
            chain_data: Данные цепочек merge-предметов
            energy_rewards_data: Данные энергетических наград
            item_rewards_data: Данные предметных наград
            max_history_orders: Максимальное количество заказов в истории
            increment_difficulty: Инкремент сложности
            energy_chance: Шанс выпадения энергетической награды (%)
            requirement_weights: Веса требований (строка, например "70,30")
            reduction_factor: Фактор уменьшения
            increase_factor: Фактор увеличения
            iteration_count: Количество итераций симуляции
            initial_energy: Начальное количество энергии
            
        Returns:
            Словарь с результатом операции и путем к сохраненному файлу
        """
        try:
            config = SimulatorConfig(
                name=name,
                description=description,
                created_at=pd.Timestamp.now().isoformat(),
                max_history_orders=max_history_orders,
                increment_difficulty=increment_difficulty,
                energy_chance=energy_chance,
                requirement_weights=requirement_weights,
                reduction_factor=reduction_factor,
                increase_factor=increase_factor,
                iteration_count=iteration_count,
                initial_energy=initial_energy,
                chain_data=chain_data or [],
                energy_rewards_data=energy_rewards_data or [],
                item_rewards_data=item_rewards_data or []
            )
            
            filepath = self.config_manager.save_config(config)
            
            return {
                "success": True,
                "message": f"Конфигурация '{name}' успешно сохранена",
                "filepath": filepath,
                "config_name": name
            }
        except Exception as e:
            return {
                "success": False,
                "error": str(e),
                "message": f"Ошибка при сохранении конфигурации: {str(e)}"
            }
    
    def mcp_load_simulator_config(self, config_name_or_path: str) -> Dict[str, Any]:
        """
        Загружает конфигурацию симулятора
        
        Args:
            config_name_or_path: Название конфигурации или путь к файлу
            
        Returns:
            Словарь с данными конфигурации
        """
        try:
            # Если передан путь к файлу
            if config_name_or_path.endswith('.json') and os.path.exists(config_name_or_path):
                config = self.config_manager.load_config(config_name_or_path)
            else:
                # Ищем конфигурацию по имени
                configs = self.config_manager.list_configs()
                matching_config = next((c for c in configs if c['name'] == config_name_or_path), None)
                
                if not matching_config:
                    return {
                        "success": False,
                        "error": f"Конфигурация '{config_name_or_path}' не найдена",
                        "available_configs": [c['name'] for c in configs]
                    }
                
                config = self.config_manager.load_config(matching_config['filepath'])
            
            return {
                "success": True,
                "config": config.to_dict(),
                "message": f"Конфигурация '{config.name}' успешно загружена"
            }
        except Exception as e:
            return {
                "success": False,
                "error": str(e),
                "message": f"Ошибка при загрузке конфигурации: {str(e)}"
            }
    
    def mcp_list_simulator_configs(self) -> Dict[str, Any]:
        """
        Возвращает список доступных конфигураций симулятора
        
        Returns:
            Словарь со списком конфигураций
        """
        try:
            configs = self.config_manager.list_configs()
            return {
                "success": True,
                "configs": configs,
                "count": len(configs),
                "message": f"Найдено {len(configs)} конфигураций"
            }
        except Exception as e:
            return {
                "success": False,
                "error": str(e),
                "message": f"Ошибка при получении списка конфигураций: {str(e)}"
            }
    
    def mcp_delete_simulator_config(self, config_name_or_path: str) -> Dict[str, Any]:
        """
        Удаляет конфигурацию симулятора
        
        Args:
            config_name_or_path: Название конфигурации или путь к файлу
            
        Returns:
            Словарь с результатом операции
        """
        try:
            # Если передан путь к файлу
            if config_name_or_path.endswith('.json') and os.path.exists(config_name_or_path):
                filepath = config_name_or_path
            else:
                # Ищем конфигурацию по имени
                configs = self.config_manager.list_configs()
                matching_config = next((c for c in configs if c['name'] == config_name_or_path), None)
                
                if not matching_config:
                    return {
                        "success": False,
                        "error": f"Конфигурация '{config_name_or_path}' не найдена"
                    }
                
                filepath = matching_config['filepath']
            
            success = self.config_manager.delete_config(filepath)
            
            if success:
                return {
                    "success": True,
                    "message": f"Конфигурация успешно удалена"
                }
            else:
                return {
                    "success": False,
                    "error": "Не удалось удалить конфигурацию"
                }
        except Exception as e:
            return {
                "success": False,
                "error": str(e),
                "message": f"Ошибка при удалении конфигурации: {str(e)}"
            }
    
    # ===============================================================
    # SIMULATION TOOLS
    # ===============================================================
    
    def mcp_run_simulation(
        self,
        config_name_or_data: Union[str, Dict[str, Any]] = None,
        chain_data: List[Dict[str, Any]] = None,
        energy_rewards_data: List[Dict[str, Any]] = None,
        item_rewards_data: List[Dict[str, Any]] = None,
        max_history_orders: int = None,
        increment_difficulty: int = None,
        energy_chance: int = None,
        requirement_weights: str = None,
        reduction_factor: int = None,
        increase_factor: int = None,
        iteration_count: int = None,
        initial_energy: int = None,
        return_detailed_results: bool = True
    ) -> Dict[str, Any]:
        """
        Запускает симуляцию генерации заказов
        
        Args:
            config_name_or_data: Название конфигурации или данные конфигурации
            chain_data: Данные цепочек (переопределяет данные из конфигурации)
            energy_rewards_data: Данные энергетических наград
            item_rewards_data: Данные предметных наград
            max_history_orders: Максимальное количество заказов в истории
            increment_difficulty: Инкремент сложности
            energy_chance: Шанс выпадения энергетической награды (%)
            requirement_weights: Веса требований
            reduction_factor: Фактор уменьшения
            increase_factor: Фактор увеличения
            iteration_count: Количество итераций симуляции
            initial_energy: Начальное количество энергии
            return_detailed_results: Возвращать ли детальные результаты
            
        Returns:
            Словарь с результатами симуляции
        """
        try:
            # Импортируем функцию симуляции
            from app import run_simulation_interface
            
            # Загружаем конфигурацию если указано имя
            if isinstance(config_name_or_data, str):
                config_result = self.mcp_load_simulator_config(config_name_or_data)
                if not config_result["success"]:
                    return config_result
                config_data = config_result["config"]
            elif isinstance(config_name_or_data, dict):
                config_data = config_name_or_data
            else:
                config_data = {}
            
            # Применяем параметры (переданные параметры имеют приоритет над конфигурацией)
            params = {
                'max_history': max_history_orders or config_data.get('max_history_orders', 5),
                'increment_diff': increment_difficulty or config_data.get('increment_difficulty', 2),
                'energy_chance': energy_chance or config_data.get('energy_chance', 90),
                'req_weights': requirement_weights or config_data.get('requirement_weights', "70,30"),
                'reduction_factor': reduction_factor or config_data.get('reduction_factor', 3),
                'increase_factor': increase_factor or config_data.get('increase_factor', 5),
                'iteration_count': iteration_count or config_data.get('iteration_count', 100),
                'initial_energy': initial_energy or config_data.get('initial_energy', 10000)
            }
            
            # Подготавливаем данные
            chain_df = pd.DataFrame(chain_data or config_data.get('chain_data', []))
            energy_rewards_df = pd.DataFrame(energy_rewards_data or config_data.get('energy_rewards_data', []))
            item_rewards_df = pd.DataFrame(item_rewards_data or config_data.get('item_rewards_data', []))
            
            # Запускаем симуляцию
            results_df, stats_report, _, _ = run_simulation_interface(
                chain_df, energy_rewards_df, item_rewards_df,
                params['max_history'], params['increment_diff'], params['energy_chance'],
                params['req_weights'], params['reduction_factor'], params['increase_factor'],
                params['iteration_count'], params['initial_energy']
            )
            
            # Сохраняем результаты
            self.current_simulation_results = results_df
            self.current_stats_report = stats_report
            
            response = {
                "success": True,
                "message": "Симуляция завершена успешно",
                "stats_report": stats_report,
                "total_orders": len(results_df),
                "simulation_parameters": params
            }
            
            if return_detailed_results:
                response["detailed_results"] = results_df.to_dict('records')
            
            return response
            
        except Exception as e:
            return {
                "success": False,
                "error": str(e),
                "message": f"Ошибка при выполнении симуляции: {str(e)}"
            }
    
    def mcp_get_simulation_results(self, format: str = "summary") -> Dict[str, Any]:
        """
        Получает результаты последней симуляции
        
        Args:
            format: Формат результатов ("summary", "detailed", "csv")
            
        Returns:
            Словарь с результатами симуляции
        """
        try:
            if self.current_simulation_results is None:
                return {
                    "success": False,
                    "error": "Нет доступных результатов симуляции",
                    "message": "Сначала запустите симуляцию"
                }
            
            if format == "summary":
                return {
                    "success": True,
                    "stats_report": self.current_stats_report,
                    "total_orders": len(self.current_simulation_results),
                    "avg_difficulty": self.current_simulation_results['Total_Difficulty'].mean(),
                    "final_energy": self.current_simulation_results['MEnergy_Amount'].iloc[-1] if len(self.current_simulation_results) > 0 else 0
                }
            elif format == "detailed":
                return {
                    "success": True,
                    "stats_report": self.current_stats_report,
                    "detailed_results": self.current_simulation_results.to_dict('records')
                }
            elif format == "csv":
                # Создаем временный CSV файл
                with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.csv', encoding='utf-8') as tmp_file:
                    self.current_simulation_results.to_csv(tmp_file.name, index=False)
                    csv_path = tmp_file.name
                
                return {
                    "success": True,
                    "csv_file_path": csv_path,
                    "stats_report": self.current_stats_report
                }
            else:
                return {
                    "success": False,
                    "error": f"Неизвестный формат: {format}",
                    "available_formats": ["summary", "detailed", "csv"]
                }
                
        except Exception as e:
            return {
                "success": False,
                "error": str(e),
                "message": f"Ошибка при получении результатов: {str(e)}"
            }
    
    def mcp_analyze_simulation_results(self, analysis_type: str = "basic") -> Dict[str, Any]:
        """
        Анализирует результаты симуляции
        
        Args:
            analysis_type: Тип анализа ("basic", "detailed", "chains", "rewards")
            
        Returns:
            Словарь с результатами анализа
        """
        try:
            if self.current_simulation_results is None:
                return {
                    "success": False,
                    "error": "Нет доступных результатов симуляции",
                    "message": "Сначала запустите симуляцию"
                }
            
            df = self.current_simulation_results
            
            if analysis_type == "basic":
                return {
                    "success": True,
                    "analysis": {
                        "total_orders": len(df),
                        "avg_difficulty": float(df['Total_Difficulty'].mean()),
                        "min_difficulty": int(df['Total_Difficulty'].min()),
                        "max_difficulty": int(df['Total_Difficulty'].max()),
                        "avg_energy_cost": float(df['MergeEnergyPrice'].mean()),
                        "total_energy_spent": int(df['MergeEnergyPrice'].sum()),
                        "final_energy": int(df['MEnergy_Amount'].iloc[-1]) if len(df) > 0 else 0,
                        "energy_rewards_count": int((df['ExpeditionEnergyReward'] > 0).sum()),
                        "item_rewards_count": int((df['MergeItemReward'] != "").sum())
                    }
                }
            elif analysis_type == "chains":
                # Анализ по цепочкам
                chain_analysis = {}
                for col in df.columns:
                    if col.startswith('ChainId_'):
                        chain_counts = df[col].value_counts()
                        chain_analysis.update(chain_counts.to_dict())
                
                return {
                    "success": True,
                    "chain_usage": chain_analysis,
                    "most_used_chain": max(chain_analysis.items(), key=lambda x: x[1]) if chain_analysis else None
                }
            elif analysis_type == "rewards":
                # Анализ наград
                energy_rewards = df[df['ExpeditionEnergyReward'] > 0]
                item_rewards = df[df['MergeItemReward'] != ""]
                
                return {
                    "success": True,
                    "rewards_analysis": {
                        "energy_rewards": {
                            "count": len(energy_rewards),
                            "total_amount": int(energy_rewards['ExpeditionEnergyReward'].sum()),
                            "avg_amount": float(energy_rewards['ExpeditionEnergyReward'].mean()) if len(energy_rewards) > 0 else 0
                        },
                        "item_rewards": {
                            "count": len(item_rewards),
                            "unique_items": item_rewards['MergeItemReward'].nunique(),
                            "most_common_item": item_rewards['MergeItemReward'].mode().iloc[0] if len(item_rewards) > 0 else None
                        }
                    }
                }
            else:
                return {
                    "success": False,
                    "error": f"Неизвестный тип анализа: {analysis_type}",
                    "available_types": ["basic", "chains", "rewards"]
                }
                
        except Exception as e:
            return {
                "success": False,
                "error": str(e),
                "message": f"Ошибка при анализе результатов: {str(e)}"
            }
    
    # ===============================================================
    # UTILITY TOOLS
    # ===============================================================
    
    def mcp_create_chain_data_template(self) -> Dict[str, Any]:
        """
        Создает шаблон данных для цепочек merge-предметов
        
        Returns:
            Словарь с шаблоном данных цепочек
        """
        template = [
            {
                "ChainId": "example_chain",
                "MergeItemId": "item_level_1",
                "RequirementWeight": 100,
                "RewardDifficulty": 10
            },
            {
                "ChainId": "example_chain",
                "MergeItemId": "item_level_2",
                "RequirementWeight": 80,
                "RewardDifficulty": 25
            },
            {
                "ChainId": "example_chain",
                "MergeItemId": "item_level_3",
                "RequirementWeight": 60,
                "RewardDifficulty": 50
            }
        ]
        
        return {
            "success": True,
            "template": template,
            "description": "Шаблон данных цепочек. ChainId - идентификатор цепочки, MergeItemId - идентификатор предмета, RequirementWeight - вес для генерации требований, RewardDifficulty - сложность для расчета наград"
        }
    
    def mcp_create_rewards_data_template(self) -> Dict[str, Any]:
        """
        Создает шаблоны данных для наград
        
        Returns:
            Словарь с шаблонами данных наград
        """
        energy_template = [
            {"DifficultyScore": 100, "Amount": 1},
            {"DifficultyScore": 500, "Amount": 3},
            {"DifficultyScore": 1000, "Amount": 5}
        ]
        
        item_template = [
            {
                "DifficultyScore": 200,
                "Amount": 1,
                "MergeItemId": "energy_1",
                "RewardWeight": 80,
                "ReductionFactor": 5
            },
            {
                "DifficultyScore": 500,
                "Amount": 1,
                "MergeItemId": "coins_1",
                "RewardWeight": 20,
                "ReductionFactor": 0
            }
        ]
        
        return {
            "success": True,
            "energy_rewards_template": energy_template,
            "item_rewards_template": item_template,
            "description": "Шаблоны данных наград. DifficultyScore - порог сложности, Amount - количество награды, MergeItemId - ID предмета для предметных наград, RewardWeight - вес награды, ReductionFactor - фактор уменьшения"
        }
    
    def mcp_validate_config_data(self, config_data: Dict[str, Any]) -> Dict[str, Any]:
        """
        Валидирует данные конфигурации симулятора
        
        Args:
            config_data: Данные конфигурации для валидации
            
        Returns:
            Словарь с результатами валидации
        """
        try:
            errors = []
            warnings = []
            
            # Проверяем обязательные поля
            required_fields = ['name', 'description']
            for field in required_fields:
                if not config_data.get(field):
                    errors.append(f"Отсутствует обязательное поле: {field}")
            
            # Проверяем числовые параметры
            numeric_params = {
                'max_history_orders': (1, 20),
                'increment_difficulty': (0, 10),
                'energy_chance': (0, 100),
                'reduction_factor': (0, 100),
                'increase_factor': (0, 100),
                'iteration_count': (1, 10000),
                'initial_energy': (1, 1000000)
            }
            
            for param, (min_val, max_val) in numeric_params.items():
                value = config_data.get(param)
                if value is not None:
                    if not isinstance(value, (int, float)) or value < min_val or value > max_val:
                        errors.append(f"Параметр {param} должен быть числом от {min_val} до {max_val}")
            
            # Проверяем данные цепочек
            chain_data = config_data.get('chain_data', [])
            if chain_data:
                for i, chain_item in enumerate(chain_data):
                    if not chain_item.get('ChainId'):
                        errors.append(f"Цепочка {i+1}: отсутствует ChainId")
                    if not chain_item.get('MergeItemId'):
                        errors.append(f"Цепочка {i+1}: отсутствует MergeItemId")
                    if not isinstance(chain_item.get('RequirementWeight', 0), (int, float)):
                        errors.append(f"Цепочка {i+1}: RequirementWeight должен быть числом")
                    if not isinstance(chain_item.get('RewardDifficulty', 0), (int, float)):
                        errors.append(f"Цепочка {i+1}: RewardDifficulty должен быть числом")
            else:
                warnings.append("Нет данных о цепочках - симуляция может не работать корректно")
            
            # Проверяем веса требований
            req_weights = config_data.get('requirement_weights', "70,30")
            try:
                weights = [int(w.strip()) for w in req_weights.split(',')]
                if len(weights) != 2:
                    errors.append("Веса требований должны содержать ровно 2 значения")
                elif any(w < 0 for w in weights):
                    errors.append("Веса требований должны быть положительными числами")
            except:
                errors.append("Неверный формат весов требований (ожидается 'число,число')")
            
            is_valid = len(errors) == 0
            
            return {
                "success": True,
                "is_valid": is_valid,
                "errors": errors,
                "warnings": warnings,
                "message": "Конфигурация валидна" if is_valid else f"Найдено {len(errors)} ошибок"
            }
            
        except Exception as e:
            return {
                "success": False,
                "error": str(e),
                "message": f"Ошибка при валидации: {str(e)}"
            }

# ===============================================================
# MCP SERVER INSTANCE
# ===============================================================

# Глобальный экземпляр MCP сервера
mcp_server = MergeSimulatorMCPServer()

# Экспорт функций для использования в других модулях
def get_mcp_server() -> MergeSimulatorMCPServer:
    """Возвращает экземпляр MCP сервера"""
    return mcp_server

# Список всех доступных MCP функций
MCP_FUNCTIONS = {
    "mcp_save_simulator_config": mcp_server.mcp_save_simulator_config,
    "mcp_load_simulator_config": mcp_server.mcp_load_simulator_config,
    "mcp_list_simulator_configs": mcp_server.mcp_list_simulator_configs,
    "mcp_delete_simulator_config": mcp_server.mcp_delete_simulator_config,
    "mcp_run_simulation": mcp_server.mcp_run_simulation,
    "mcp_get_simulation_results": mcp_server.mcp_get_simulation_results,
    "mcp_analyze_simulation_results": mcp_server.mcp_analyze_simulation_results,
    "mcp_create_chain_data_template": mcp_server.mcp_create_chain_data_template,
    "mcp_create_rewards_data_template": mcp_server.mcp_create_rewards_data_template,
    "mcp_validate_config_data": mcp_server.mcp_validate_config_data
}