#!/usr/bin/env python3 """ Hotel MCP Server - Script de Diagnóstico Genera un reporte completo del estado del sistema. Uso: uv run python scripts/diagnostic.py """ # flake8: noqa: E402,E226 import json import os import platform import sys import time from pathlib import Path from typing import Any, Dict # Agregar src al path para imports sys.path.insert(0, str(Path(__file__).parent.parent / "src")) # Cargar variables de entorno from dotenv import load_dotenv load_dotenv() def print_section(title: str): """Imprimir título de sección.""" print(f"\n{'='*60}") print(f" {title}") print(f"{'='*60}") def print_check(name: str, status: bool, details: str = ""): """Imprimir resultado de verificación.""" icon = "✅" if status else "❌" print(f"{icon} {name}") if details: print(f" {details}") def check_system_info() -> Dict[str, Any]: """Verificar información del sistema.""" print_section("INFORMACIÓN DEL SISTEMA") info = { "platform": platform.platform(), "python_version": f"{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}", "architecture": platform.architecture()[0], "processor": platform.processor(), "hostname": platform.node(), } print(f"Sistema Operativo: {info['platform']}") print(f"Python: {info['python_version']}") print(f"Arquitectura: {info['architecture']}") print(f"Procesador: {info['processor']}") print(f"Hostname: {info['hostname']}") return info def check_dependencies() -> Dict[str, Any]: """Verificar dependencias instaladas.""" print_section("DEPENDENCIAS") dependencies = {} required_packages = [ "fastmcp", "supabase", "dotenv", # python-dotenv se importa como 'dotenv' "pytest", ] for package in required_packages: try: __import__(package.replace("-", "_")) dependencies[package] = {"installed": True, "version": "unknown"} print_check(f"{package}", True, "Instalado") except ImportError: dependencies[package] = {"installed": False, "version": None} print_check(f"{package}", False, "No instalado") # Verificar uv uv_available = os.system("which uv > /dev/null 2>&1") == 0 dependencies["uv"] = {"installed": uv_available} print_check("uv package manager", uv_available) return dependencies def check_environment_variables() -> Dict[str, Any]: """Verificar variables de entorno.""" print_section("VARIABLES DE ENTORNO") required_vars = ["SUPABASE_URL", "SUPABASE_ANON_KEY", "DEFAULT_SITE_ID"] optional_vars = ["DEFAULT_LANGUAGE", "MEDIA_BASE_URL", "DEBUG", "LOG_LEVEL"] env_status = {} print("Variables requeridas:") for var in required_vars: value = os.getenv(var) is_set = value is not None and value != "" env_status[var] = {"set": is_set, "value": value if is_set else None} if var == "SUPABASE_ANON_KEY" and is_set: masked_value = value[:10] + "..." + value[-4:] if len(value) > 14 else "***" print_check(f"{var}", is_set, f"Configurada ({masked_value})") else: print_check(f"{var}", is_set, value if is_set else "No configurada") print("\nVariables opcionales:") for var in optional_vars: value = os.getenv(var) is_set = value is not None and value != "" env_status[var] = {"set": is_set, "value": value if is_set else None} print_check(f"{var}", is_set, value if is_set else "No configurada") return env_status def check_file_structure() -> Dict[str, Any]: """Verificar estructura de archivos.""" print_section("ESTRUCTURA DE ARCHIVOS") required_files = [ "hotel_mcp.py", "pyproject.toml", ".env.example", "src/database.py", "src/translations.py", "src/media.py", "src/resources.py", "src/tools/__init__.py", "src/tools/rooms.py", "src/tools/activities.py", "src/tools/facilities.py", "src/tools/menu.py", "src/tools/galleries.py", ] file_status = {} for file_path in required_files: exists = os.path.exists(file_path) file_status[file_path] = {"exists": exists} print_check(file_path, exists) # Verificar directorio de tests test_files = list(Path("tests").glob("test_*.py")) if Path("tests").exists() else [] file_status["test_files"] = { "count": len(test_files), "files": [str(f) for f in test_files], } print_check( "Archivos de test", len(test_files) > 0, f"{len(test_files)} archivos encontrados", ) return file_status def check_database_connection() -> Dict[str, Any]: """Verificar conexión a base de datos.""" print_section("CONEXIÓN A BASE DE DATOS") db_status = {"connected": False, "error": None, "site_data": {}} try: from src.database import SupabaseClient db_client = SupabaseClient() site_id = os.getenv("DEFAULT_SITE_ID") if not site_id: db_status["error"] = "DEFAULT_SITE_ID no configurado" print_check("Conexión a Supabase", False, "DEFAULT_SITE_ID no configurado") return db_status # Probar conexión con consulta simple start_time = time.time() suites = db_client.get_suites(site_id=site_id, limit=1, offset=0) response_time = time.time() - start_time db_status["connected"] = True db_status["response_time"] = response_time db_status["site_data"]["suites_count"] = len(suites) print_check("Conexión a Supabase", True, f"Conectado en {response_time:.3f}s") print_check( "Datos de habitaciones", len(suites) > 0, f"{len(suites)} habitaciones encontradas", ) # Probar otras tablas try: activities = db_client.get_activities(site_id=site_id, limit=1, offset=0) db_status["site_data"]["activities_count"] = len(activities) print_check( "Datos de actividades", len(activities) > 0, f"{len(activities)} actividades encontradas", ) except Exception as e: print_check("Datos de actividades", False, f"Error: {e}") except Exception as e: db_status["error"] = str(e) print_check("Conexión a Supabase", False, f"Error: {e}") return db_status def check_mcp_server() -> Dict[str, Any]: """Verificar servidor MCP.""" print_section("SERVIDOR MCP") mcp_status = { "importable": False, "tools_count": 0, "resources_count": 0, "error": None, } try: # Intentar importar el servidor from hotel_mcp import mcp mcp_status["importable"] = True print_check("Importación del servidor", True) # Verificar que es una instancia de FastMCP from fastmcp import FastMCP if isinstance(mcp, FastMCP): print_check("Tipo de servidor", True, "FastMCP") mcp_status["server_type"] = "FastMCP" mcp_status["server_name"] = mcp.name print_check("Nombre del servidor", True, mcp.name) else: print_check("Tipo de servidor", False, f"Tipo inesperado: {type(mcp)}") # Nota: FastMCP usa métodos async para obtener herramientas/recursos # Por simplicidad, solo verificamos que se puede importar print_check("Servidor MCP funcional", True, "Importación exitosa") except Exception as e: mcp_status["error"] = str(e) print_check("Importación del servidor", False, f"Error: {e}") return mcp_status def run_basic_tests() -> Dict[str, Any]: """Ejecutar tests básicos.""" print_section("TESTS BÁSICOS") test_status = {"passed": 0, "failed": 0, "total": 0, "details": []} try: import subprocess # Ejecutar pytest con output capturado result = subprocess.run( ["uv", "run", "pytest", "tests/", "-v", "--tb=short"], capture_output=True, text=True, timeout=120, ) # Parsear output output_lines = result.stdout.split("\n") for line in output_lines: if " PASSED" in line: test_status["passed"] += 1 elif " FAILED" in line: test_status["failed"] += 1 test_status["details"].append(line.strip()) test_status["total"] = test_status["passed"] + test_status["failed"] success = result.returncode == 0 print_check( "Tests básicos", success, f"{test_status['passed']}/{test_status['total']} tests pasaron", ) if test_status["failed"] > 0: print("\nTests fallidos:") for detail in test_status["details"][:5]: # Mostrar solo los primeros 5 print(f" - {detail}") except Exception as e: print_check("Tests básicos", False, f"Error ejecutando tests: {e}") test_status["error"] = str(e) return test_status def generate_report() -> Dict[str, Any]: """Generar reporte completo de diagnóstico.""" print("🏨 Hotel MCP Server - Diagnóstico del Sistema") print("=" * 60) report = { "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"), "system_info": check_system_info(), "dependencies": check_dependencies(), "environment": check_environment_variables(), "file_structure": check_file_structure(), "database": check_database_connection(), "mcp_server": check_mcp_server(), "tests": run_basic_tests(), } # Resumen final print_section("RESUMEN") # Calcular estado general critical_checks = [ report["dependencies"]["fastmcp"]["installed"], report["dependencies"]["supabase"]["installed"], report["environment"]["SUPABASE_URL"]["set"], report["environment"]["SUPABASE_ANON_KEY"]["set"], report["environment"]["DEFAULT_SITE_ID"]["set"], report["file_structure"]["hotel_mcp.py"]["exists"], report["mcp_server"]["importable"], ] critical_passed = sum(critical_checks) total_critical = len(critical_checks) print(f"Verificaciones críticas: {critical_passed}/{total_critical}") print( f"Tests básicos: {report['tests'].get('passed', 0)}/{report['tests'].get('total', 0)}" ) print(f"Conexión a BD: {'✅' if report['database']['connected'] else '❌'}") print(f"Herramientas MCP: {report['mcp_server']['tools_count']}") print(f"Recursos MCP: {report['mcp_server']['resources_count']}") if critical_passed == total_critical and report["database"]["connected"]: print("\n🎉 ¡Sistema completamente funcional!") elif critical_passed >= total_critical - 1: print("\n⚠️ Sistema mayormente funcional con problemas menores") else: print("\n❌ Sistema requiere configuración adicional") return report def save_report(report: Dict[str, Any]): """Guardar reporte en archivo JSON.""" report_file = f"diagnostic_report_{int(time.time())}.json" try: with open(report_file, "w", encoding="utf-8") as f: json.dump(report, f, indent=2, ensure_ascii=False) print(f"\n📄 Reporte guardado en: {report_file}") except Exception as e: print(f"\n❌ Error guardando reporte: {e}") if __name__ == "__main__": try: report = generate_report() save_report(report) except KeyboardInterrupt: print("\n\n⏹️ Diagnóstico interrumpido por el usuario") except Exception as e: print(f"\n❌ Error durante el diagnóstico: {e}") sys.exit(1)