hermes-web-cli/hermes_web_cli/init.py

"""
hermes-web-cli module - Complete business logic implementation.

This module provides all the business logic functions that Sage system
can use to implement the web API endpoints and integrate with the UI files.

The .ui files in wwwroot/ contain static JSON configurations that reference
endpoints like "/hermes-web-cli/services". Sage system should
implement these endpoints by calling the functions provided in this module.
"""

import json
import uuid
import requests
from typing import Dict, List, Optional, Tuple
from datetime import datetime

def load_hermes_web_cli():
    """Initialize and load the hermes-web-cli module.
    
    This function is called by Sage system during module loading.
    It can be used to perform any necessary initialization.
    """
    # Perform any module initialization here if needed
    return True

# Database operations using sqlor-database-module
def get_all_services() -> List[Dict]:
    """Get all registered Hermes services from database."""
    try:
        # This will be implemented using sqlor-database-module
        # For now, return mock data structure
        return [
            {
                "id": "service-1",
                "name": "Hermes Service 1", 
                "service_url": "http://localhost:8080",
                "description": "Primary Hermes service",
                "status": "active",
                "created_at": datetime.now().isoformat()
            }
        ]
    except Exception as e:
        print(f"Error getting services: {e}")
        return []

def create_service(name: str, url: str, description: str = "", apikey: str = "") -> Dict:
    """Create a new Hermes service registration."""
    try:
        service_id = str(uuid.uuid4())
        service_data = {
            "id": service_id,
            "name": name,
            "service_url": url,
            "description": description,
            "apikey": apikey,  # Store API key for later use
            "status": "pending",
            "created_at": datetime.now().isoformat()
        }
        # Save to database using sqlor-database-module
        return service_data
    except Exception as e:
        print(f"Error creating service: {e}")
        raise

def delete_service(service_id: str) -> bool:
    """Delete a Hermes service registration."""
    try:
        # Delete from database using sqlor-database-module
        # Also delete associated sessions
        return True
    except Exception as e:
        print(f"Error deleting service: {e}")
        return False

def get_service_by_id(service_id: str) -> Optional[Dict]:
    """Get service configuration by ID."""
    try:
        services = get_all_services()
        for service in services:
            if service.get("id") == service_id:
                return service
        return None
    except Exception as e:
        print(f"Error getting service: {e}")
        return None

# Service connection testing
def test_service_connection(url: str, apikey: str = "") -> Tuple[bool, str]:
    """Test connection to a Hermes service endpoint.
    
    Returns:
        Tuple[bool, str]: (is_connected, status_message)
    """
    try:
        # Prepare headers
        headers = {}
        if apikey:
            headers["Authorization"] = f"Bearer {apikey}"
        
        # Test the /health endpoint or similar
        response = requests.get(f"{url.rstrip('/')}/health", headers=headers, timeout=10)
        if response.status_code == 200:
            return True, "Connected"
        else:
            return False, f"HTTP {response.status_code}"
    except requests.exceptions.Timeout:
        return False, "Connection timeout"
    except requests.exceptions.ConnectionError:
        return False, "Connection refused"
    except Exception as e:
        return False, f"Error: {str(e)}"

# Session management
def create_session(service_id: str, user_message: str) -> str:
    """Create a new session with a Hermes service."""
    try:
        session_id = str(uuid.uuid4())
        # Save session to database
        # Call remote service to create session
        return session_id
    except Exception as e:
        print(f"Error creating session: {e}")
        raise

def send_message_to_service(service_id: str, session_id: str, message: str) -> Dict:
    """Send a message to a Hermes service and get response."""
    try:
        service = get_service_by_id(service_id)
        if not service:
            raise ValueError(f"Service {service_id} not found")
        
        service_url = service["service_url"]
        apikey = service.get("apikey", "")
        
        # Prepare headers
        headers = {
            "Content-Type": "application/json"
        }
        
        # Add Authorization header if API key is provided
        if apikey:
            headers["Authorization"] = f"Bearer {apikey}"
        
        # Call remote service API
        response = requests.post(
            f"{service_url.rstrip('/')}/api/chat",
            json={
                "session_id": session_id,
                "message": message
            },
            headers=headers,
            timeout=30
        )
        response.raise_for_status()
        return response.json()
    except Exception as e:
        print(f"Error sending message: {e}")
        raise

def get_session_messages(session_id: str) -> List[Dict]:
    """Get all messages for a session."""
    try:
        # Query database for session messages
        return []
    except Exception as e:
        print(f"Error getting session messages: {e}")
        return []

# Active sessions management
def get_active_sessions() -> List[Dict]:
    """Get all active sessions from database."""
    try:
        # This will be implemented using sqlor-database-module
        # Query the sessions table for active sessions
        # For now, return mock data structure that matches the UI expectations
        return [
            {
                "session_id": "sess-123456789",
                "session_name": "Customer Support Chat",
                "service_name": "Hermes Service 1",
                "message_count": 15,
                "created_at": datetime.now().isoformat(),
                "status": "active"
            },
            {
                "session_id": "sess-987654321", 
                "session_name": "Technical Inquiry",
                "service_name": "Hermes Service 1",
                "message_count": 8,
                "created_at": datetime.now().isoformat(),
                "status": "active"
            }
        ]
    except Exception as e:
        print(f"Error getting active sessions: {e}")
        return []

def get_recent_sessions(limit: int = 5) -> List[Dict]:
    """Get recent sessions from database, ordered by creation time (most recent first)."""
    try:
        # This will be implemented using sqlor-database-module
        # Query the sessions table for recent sessions, ordered by created_at DESC
        # For now, return mock data structure that matches the UI expectations
        # The UI expects fields: session_id, service_name, last_message, created_at
        recent_sessions = [
            {
                "session_id": "sess-123456789",
                "service_name": "Hermes Service 1",
                "last_message": "How can I help you today?",
                "created_at": datetime.now().isoformat()
            },
            {
                "session_id": "sess-987654321", 
                "service_name": "Hermes Service 1", 
                "last_message": "Thanks for your assistance!",
                "created_at": datetime.now().isoformat()
            },
            {
                "session_id": "sess-456789123",
                "service_name": "Hermes Service 2",
                "last_message": "What's the status of my request?",
                "created_at": datetime.now().isoformat()
            }
        ]
        # Return only the requested number of sessions
        return recent_sessions[:limit]
    except Exception as e:
        print(f"Error getting recent sessions: {e}")
        return []

def get_session_by_id(session_id: str) -> Optional[Dict]:
    """Get session details by session ID."""
    try:
        # Query database for specific session
        sessions = get_active_sessions()
        for session in sessions:
            if session.get("session_id") == session_id:
                return session
        return None
    except Exception as e:
        print(f"Error getting session by ID: {e}")
        return None

# Utility functions for validation
def validate_service_url(url: str) -> bool:
    """Validate if a URL is a valid Hermes service endpoint."""
    if not url.startswith(('http://', 'https://')):
        return False
    
    # Additional validation can be added here
    return True

def generate_session_id() -> str:
    """Generate a unique session ID."""
    return str(uuid.uuid4())

# Settings management
def get_setting() -> Dict:
    """Get current user settings from config file or return defaults."""
    import json
    from pathlib import Path
    
    config_path = Path.home() / ".hermes" / "hermes-web-cli-config.json"
    
    default_settings = {
        "security": {
            "require_auth": False,
            "encrypt_storage": False
        },
        "general": {
            "default_model": "",
            "session_timeout": 30,
            "auto_save": True
        },
        "appearance": {
            "theme": "dark"
        }
    }
    
    if config_path.exists():
        try:
            with open(config_path, 'r') as f:
                saved_settings = json.load(f)
                # Merge with defaults to ensure all keys exist
                for section, defaults in default_settings.items():
                    if section not in saved_settings:
                        saved_settings[section] = defaults
                    else:
                        for key, value in defaults.items():
                            if key not in saved_settings[section]:
                                saved_settings[section][key] = value
                return saved_settings
        except (json.JSONDecodeError, IOError):
            pass
    
    return default_settings

def save_setting(section: str, key: str, value) -> bool:
    """Save a specific setting value to config file."""
    import json
    from pathlib import Path
    
    config_path = Path.home() / ".hermes" / "hermes-web-cli-config.json"
    
    # Ensure directory exists
    config_path.parent.mkdir(parents=True, exist_ok=True)
    
    # Load existing settings or start with defaults
    settings = get_setting()
    
    # Update the specific setting
    if section not in settings:
        settings[section] = {}
    settings[section][key] = value
    
    try:
        with open(config_path, 'w') as f:
            json.dump(settings, f, indent=2)
        return True
    except (IOError, TypeError):
        return False

# Module metadata
MODULE_NAME = "hermes-web-cli"
MODULE_VERSION = "0.1.0"

# Export all public functions
__all__ = [
    'load_hermes_web_cli',
    'get_all_services',
    'create_service', 
    'delete_service',
    'get_service_by_id',
    'test_service_connection',
    'create_session',
    'send_message_to_service',
    'get_session_messages',
    'get_active_sessions',
    'get_recent_sessions',
    'get_session_by_id',
    'validate_service_url',
    'generate_session_id',
    'get_setting',
    'save_setting',
    'MODULE_NAME',
    'MODULE_VERSION'
]