Initial upload: Devtoolbelt v1.0.0 - unified CLI toolkit for developers

devtoolbelt/interactive.py

@@ -0,0 +1,261 @@
+"""Interactive mode for Devtoolbelt."""
+
+import cmd
+import shlex
+from typing import Dict, List, Optional
+
+from rich import print as rprint
+from rich.table import Table
+
+from devtoolbelt.utils import console
+from .commands.database import connect_db, list_tables
+from .commands.api import api_get, api_post
+
+
+class DevtoolbeltInteractive(cmd.Cmd):
+    """Interactive shell for Devtoolbelt."""
+
+    prompt = "[devtoolbelt] "
+    intro = """[bold green]Welcome to Devtoolbelt![/bold green]
+Type 'help' for available commands, 'exit' to quit.
+"""
+
+    def __init__(self, config_path: Optional[str] = None):
+        """Initialize interactive mode.
+
+        Args:
+            config_path: Optional path to configuration file.
+        """
+        super().__init__()
+        self.config_path = config_path
+        self._histories: Dict[str, List[str]] = {
+            "queries": [],
+            "requests": []
+        }
+
+    def do_exit(self, arg: str) -> bool:
+        """Exit the interactive shell."""
+        rprint("[bold]Goodbye![/bold]")
+        return True
+
+    def do_quit(self, arg: str) -> bool:
+        """Exit the interactive shell."""
+        return self.do_exit(arg)
+
+    def do_help(self, arg: str):
+        """Show help information."""
+        if arg:
+            self._show_command_help(arg)
+        else:
+            self._show_general_help()
+
+    def _show_general_help(self):
+        """Show general help information."""
+        help_table = Table(title="Available Commands")
+        help_table.add_column("Command", style="cyan")
+        help_table.add_column("Description", style="green")
+
+        help_table.add_row("db list", "List configured databases")
+        help_table.add_row("db connect <name>", "Connect to a database")
+        help_table.add_row("db tables <name>", "List tables in a database")
+        help_table.add_row("api list", "List configured API endpoints")
+        help_table.add_row("api get <name|url>", "Make GET request")
+        help_table.add_row("api post <name|url>", "Make POST request")
+        help_table.add_row("uuid", "Generate a UUID")
+        help_table.add_row("hash <string>", "Hash a string")
+        help_table.add_row("json <string>", "Format/validate JSON")
+        help_table.add_row("base64 <string>", "Encode/decode Base64")
+        help_table.add_row("url <string>", "Encode/decode URL")
+        help_table.add_row("password", "Generate a password")
+        help_table.add_row("timer <seconds>", "Countdown timer")
+        help_table.add_row("stopwatch", "Stopwatch")
+        help_table.add_row("help [command]", "Show help")
+        help_table.add_row("exit", "Exit")
+
+        console.print(help_table)
+
+    def _show_command_help(self, command: str):
+        """Show help for a specific command."""
+        rprint(f"Help for '{command}' command")
+
+    def do_db(self, arg: str):
+        """Database commands."""
+        self._run_command("database", arg)
+
+    def do_api(self, arg: str):
+        """API commands."""
+        self._run_command("api", arg)
+
+    def do_uuid(self, arg: str):
+        """Generate UUID."""
+        import uuid
+        rprint(str(uuid.uuid4()))
+
+    def do_hash(self, arg: str):
+        """Hash a string."""
+        if not arg:
+            rprint("[yellow]Usage: hash <string>[/yellow]")
+            return
+        import hashlib
+        hasher = hashlib.sha256()
+        hasher.update(arg.encode('utf-8'))
+        rprint(f"[green]{hasher.hexdigest()}[/green]")
+
+    def do_json(self, arg: str):
+        """Format JSON."""
+        if not arg:
+            rprint("[yellow]Usage: json <json_string>[/yellow]")
+            return
+        import json
+        try:
+            data = json.loads(arg)
+            rprint(json.dumps(data, indent=2))
+        except json.JSONDecodeError as e:
+            rprint(f"[red]Invalid JSON: {e}[/red]")
+
+    def do_base64(self, arg: str):
+        """Base64 encode/decode."""
+        if not arg:
+            rprint("[yellow]Usage: base64 <string>[/yellow]")
+            return
+        import base64
+        try:
+            result = base64.b64encode(arg.encode('utf-8')).decode('utf-8')
+            rprint(result)
+        except Exception as e:
+            rprint(f"[red]Error: {e}[/red]")
+
+    def do_url(self, arg: str):
+        """URL encode/decode."""
+        if not arg:
+            rprint("[yellow]Usage: url <string>[/yellow]")
+            return
+        from urllib.parse import quote
+        rprint(quote(arg, safe=''))
+
+    def do_password(self, arg: str):
+        """Generate password."""
+        import random
+        import string
+        chars = string.ascii_letters + string.digits + string.punctuation
+        password = ''.join(random.choice(chars) for _ in range(16))
+        rprint(password)
+
+    def do_timer(self, arg: str):
+        """Countdown timer."""
+        import time
+        try:
+            seconds = int(arg)
+        except ValueError:
+            rprint("[yellow]Usage: timer <seconds>[/yellow]")
+            return
+
+        rprint(f"[bold]Timer for {seconds} seconds...[/bold]")
+        for remaining in range(seconds, 0, -1):
+            mins, secs = divmod(remaining, 60)
+            rprint(f"\r[bold]{mins:02d}:{secs:02d}[/bold] ", end="", flush=True)
+            time.sleep(1)
+        rprint("\n[bold green]Time's up![/bold green]")
+
+    def do_stopwatch(self, arg: str):
+        """Stopwatch."""
+        import time
+        rprint("[bold]Press Enter to start...[/bold]")
+        input()
+        start = time.time()
+        rprint("[bold]Running... Press Enter to stop.[/bold]")
+        input()
+        elapsed = time.time() - start
+        rprint(f"[bold green]Elapsed: {elapsed:.2f}s[/bold green]")
+
+    def _run_command(self, category: str, arg: str):
+        """Run a command from a category."""
+        if not arg:
+            rprint(f"[yellow]Usage: {category} <subcommand>[/yellow]")
+            return
+
+        parts = shlex.split(arg)
+        if not parts:
+            return
+
+        subcommand = parts[0]
+        args = parts[1:]
+
+        if category == "database":
+            if subcommand == "list":
+                self._cmd_list_databases()
+            elif subcommand == "connect" and args:
+                connect_db(args[0], self.config_path)
+            elif subcommand == "tables" and args:
+                list_tables(args[0], self.config_path, False)
+            else:
+                rprint(f"[yellow]Unknown db command: {subcommand}[/yellow]")
+        elif category == "api":
+            if subcommand == "list":
+                self._cmd_list_apis()
+            elif subcommand == "get" and args:
+                api_get(args[0], (), None, 30, self.config_path, False)
+            elif subcommand == "post" and args:
+                api_post(args[0], None, None, (), None, 30, self.config_path, False)
+            else:
+                rprint(f"[yellow]Unknown api command: {subcommand}[/yellow]")
+
+    def _cmd_list_databases(self):
+        """List configured databases."""
+        from devtoolbelt.config import get_config
+        cfg = get_config(self.config_path)
+        databases = cfg.get_database_configs()
+
+        if not databases:
+            rprint("[yellow]No databases configured.[/yellow]")
+            return
+
+        table = Table(title="Databases")
+        table.add_column("Name", style="cyan")
+        table.add_column("Type", style="green")
+
+        for name, config in databases.items():
+            table.add_row(name, config.get("type", "unknown"))
+
+        console.print(table)
+
+    def _cmd_list_apis(self):
+        """List configured APIs."""
+        from devtoolbelt.config import get_config
+        cfg = get_config(self.config_path)
+        endpoints = cfg.get_api_endpoints()
+
+        if not endpoints:
+            rprint("[yellow]No APIs configured.[/yellow]")
+            return
+
+        table = Table(title="API Endpoints")
+        table.add_column("Name", style="cyan")
+        table.add_column("URL", style="green")
+
+        for name, url in endpoints.items():
+            table.add_row(name, url)
+
+        console.print(table)
+
+    def default(self, line: str):
+        """Handle unknown commands."""
+        rprint(f"[red]Unknown command: {line}[/red]")
+        rprint("Type 'help' for available commands.")
+
+    def emptyline(self) -> bool:
+        """Handle empty lines."""
+        return False
+
+
+def start_interactive(config_path: Optional[str] = None):
+    """Start interactive mode.
+
+    Args:
+        config_path: Optional path to configuration file.
+    """
+    from rich.console import Console
+    console = Console()
+    with console.screen(style=""):
+        shell = DevtoolbeltInteractive(config_path)
+        shell.cmdloop()