cli.py

import typer
import asyncio
import inspect
from typing import List, Dict, Any
from rich.console import Console
from modules.loader import ModuleLoader

app = typer.Typer(
    help="AWS VM Deployment CLI Tool", no_args_is_help=True, add_completion=False
)
console = Console()
module_loader = ModuleLoader()


def parse_params(params: List[str]) -> Dict[str, str]:
    """Parse key=value pairs into a dictionary"""
    if not params:
        return {}

    result = {}
    for param in params:
        try:
            key, value = param.split("=", 1)
            result[key.strip()] = value.strip()
        except ValueError:
            console.print(
                f"[red]Invalid parameter format: {param}. Use key=value format.[/red]"
            )
            raise typer.Exit(code=1)
    return result


def print_help() -> None:
    """Print help information"""
    console.print("\n[bold green]AWS VM Deployment Tool[/bold green]")
    for service_class in module_loader.list_services():
        console.print(f"\n[bold blue]{service_class.__name__}[/bold blue]")
        service_instance = service_class()
        asyncio.run(service_instance.print_help())


@app.command(name="")
def main(
    service_type: str = typer.Argument(help="Service type to execute (vm, sqs ...)"),
    command: str = typer.Argument(help="Command to execute"),
    params: List[str] = typer.Argument(
        None, help="Additional parameters in key=value format"
    ),
) -> None:
    """Execute a command for a specific service type"""
    try:
        param_dict = parse_params(params or [])
        manager_class = module_loader.get_manager(service_type)
        manager = manager_class()

        commands = get_service_commands(service_type)
        if command not in commands:
            available_commands = ", ".join(commands.keys())
            console.print(f"[red]Unknown command: {command}[/red]")
            console.print(f"Available commands: {available_commands}")
            raise typer.Exit(code=1)

        asyncio.run(getattr(manager, command)(**param_dict))
    except Exception as e:
        console.print(f"[red]Error executing command: {str(e)}[/red]")
        if service_type:
            try:
                manager_class = module_loader.get_manager(service_type)
                manager = manager_class()
                asyncio.run(manager.print_help())
            except:
                print_help()
        raise typer.Exit(code=1)


def get_service_commands(service_type: str) -> Dict[str, Any]:
    """Get all available commands for a service type"""
    manager_class = module_loader.get_manager(service_type)
    methods = inspect.getmembers(manager_class, predicate=inspect.isfunction)
    return {
        name: method
        for name, method in methods
        if not name.startswith("_") and name != "print_help"
    }


if __name__ == "__main__":
    app()