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 registers all module functions with the ServerEnv instance
    so they can be called directly from .ui and .dspy files.
    """
    from ahserver.serverenv import ServerEnv
    
    # Get the ServerEnv instance
    env = ServerEnv()
    
    # Register all module functions with ServerEnv
    env.get_setting = get_setting
    env.save_setting = save_setting
    env.get_all_services = get_all_services
    env.create_service = create_service
    env.delete_service = delete_service
    env.get_service_by_id = get_service_by_id
    env.test_service_connection = test_service_connection
    env.create_session = create_session
    env.send_message_to_service = send_message_to_service
    env.get_session_messages = get_session_messages
    env.get_active_sessions = get_active_sessions
    env.get_recent_sessions = get_recent_sessions
    env.get_session_by_id = get_session_by_id
    env.validate_service_url = validate_service_url
    env.generate_session_id = generate_session_id
    
    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:
        # Get service configuration
        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 to create session
        response = requests.post(
            f"{service_url.rstrip('/')}/api/v1/sessions",
            json={
                "user_id": "web-cli-user",  # This could be made configurable
                "initial_message": user_message if user_message else None
            },
            headers=headers,
            timeout=30
        )
        response.raise_for_status()
        result = response.json()
        
        # Return the session ID from the remote service
        return result.get("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'
]