From c481eb125e193d657e6a45cbebcfa7d857fc93f4 Mon Sep 17 00:00:00 2001 From: 7000pctAUTO Date: Sun, 1 Feb 2026 21:45:45 +0000 Subject: [PATCH] Initial upload: Devtoolbelt v1.0.0 - unified CLI toolkit for developers --- devtoolbelt/interactive.py | 261 +++++++++++++++++++++++++++++++++++++ 1 file changed, 261 insertions(+) create mode 100644 devtoolbelt/interactive.py diff --git a/devtoolbelt/interactive.py b/devtoolbelt/interactive.py new file mode 100644 index 0000000..9576f36 --- /dev/null +++ b/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 ", "Connect to a database") + help_table.add_row("db tables ", "List tables in a database") + help_table.add_row("api list", "List configured API endpoints") + help_table.add_row("api get ", "Make GET request") + help_table.add_row("api post ", "Make POST request") + help_table.add_row("uuid", "Generate a UUID") + help_table.add_row("hash ", "Hash a string") + help_table.add_row("json ", "Format/validate JSON") + help_table.add_row("base64 ", "Encode/decode Base64") + help_table.add_row("url ", "Encode/decode URL") + help_table.add_row("password", "Generate a password") + help_table.add_row("timer ", "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 [/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 [/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 [/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 [/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 [/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} [/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()