"""Interactive mode implementation.""" import os import shutil from collections.abc import Generator from prompt_toolkit import PromptSession from prompt_toolkit.completion import Completer, Completion from prompt_toolkit.document import Document from prompt_toolkit.history import FileHistory from prompt_toolkit.key_binding import KeyBindings, KeyPressEvent from prompt_toolkit.keys import Keys from shell_speak.config import ensure_data_dir, get_data_dir from shell_speak.history import get_history_manager from shell_speak.library import get_loader from shell_speak.matcher import get_matcher from shell_speak.models import CommandMatch from shell_speak.output import ( console, display_command, display_error, display_help_header, display_history, ) class ShellSpeakCompleter(Completer): """Auto-completion for shell-speak.""" def __init__(self) -> None: self._loader = get_loader() self._history_manager = get_history_manager() def get_completions( self, document: Document, complete_event: object ) -> Generator[Completion, None, None]: text = document.text_before_cursor last_word = text.split()[-1] if text.split() else "" history = self._history_manager.get_recent(50) for entry in reversed(history): if entry.query.lower().startswith(last_word.lower()): yield Completion( entry.query, start_position=-len(last_word), style="fg:cyan", ) patterns = self._loader.get_patterns() for pattern in patterns: for ptn in pattern.patterns: if ptn.lower().startswith(last_word.lower()): yield Completion( ptn, start_position=-len(last_word), style="fg:green", ) def create_key_bindings() -> KeyBindings: """Create key bindings for interactive mode.""" kb = KeyBindings() @kb.add(Keys.ControlC) def _(event: KeyPressEvent) -> None: event.app.exit() @kb.add(Keys.ControlL) def _(event: KeyPressEvent) -> None: os.system("clear" if os.name == "posix" else "cls") return kb def get_terminal_size() -> tuple[int, int]: """Get terminal size.""" return shutil.get_terminal_size() def run_interactive_mode() -> None: # noqa: C901 """Run the interactive shell mode.""" ensure_data_dir() display_help_header() history_file = get_data_dir() / ".history" session: PromptSession[str] = PromptSession( history=FileHistory(str(history_file)), completer=ShellSpeakCompleter(), key_bindings=create_key_bindings(), complete_while_typing=True, enable_history_search=True, ) history_manager = get_history_manager() history_manager.load() loader = get_loader() loader.load_library() console.print("\n[info]Interactive mode started. Type 'help' for commands, 'exit' to quit.[/]\n") while True: try: user_input = session.prompt( "[shell-speak]>> ", multiline=False, ).strip() except KeyboardInterrupt: console.print("\n[info]Use 'exit' to quit.[/]") continue except EOFError: break if not user_input: continue if user_input.lower() in ("exit", "quit", "q"): break if user_input.lower() == "help": _show_interactive_help() continue if user_input.lower() == "clear": os.system("clear" if os.name == "posix" else "cls") continue if user_input.lower() == "history": entries = history_manager.get_recent(50) display_history(entries) continue if user_input.startswith("learn "): parts = user_input[6:].split("::") if len(parts) >= 2: query, command = parts[0].strip(), parts[1].strip() tool = parts[2].strip() if len(parts) > 2 else "custom" loader.add_correction(query, command, tool) console.print(f"[success]Learned: {query} -> {command}[/]") else: console.print("[error]Usage: learn ::::[/]") continue if user_input.startswith("forget "): query = user_input[7:].strip() tool = "custom" if loader.remove_correction(query, tool): console.print(f"[success]Forgot: {query}[/]") else: console.print(f"[warning]Pattern not found: {query}[/]") continue if user_input.startswith("repeat"): parts = user_input.split() if len(parts) > 1: try: idx = int(parts[1]) entries = history_manager.get_recent(100) if 1 <= idx <= len(entries): entry = entries[-idx] console.print(f"[info]Repeating command {idx} entries ago:[/]") _process_query(entry.query, entry.tool) else: console.print("[error]Invalid history index[/]") except ValueError: console.print("[error]Invalid index[/]") continue detected_tool: str | None = _detect_tool(user_input) match = _process_query(user_input, detected_tool) if match: history_manager.add(user_input, match.command, match.pattern.tool, match.explanation) console.print("\n[info]Goodbye![/]") def _detect_tool(query: str) -> str | None: """Detect which tool the query is about.""" query_lower = query.lower() docker_keywords = ["docker", "container", "image", "run", "build", "pull", "push", "ps", "logs"] kubectl_keywords = ["kubectl", "k8s", "kubernetes", "pod", "deploy", "service", "namespace", "apply"] git_keywords = ["git", "commit", "push", "pull", "branch", "merge", "checkout", "clone"] for kw in docker_keywords: if kw in query_lower: return "docker" for kw in kubectl_keywords: if kw in query_lower: return "kubectl" for kw in git_keywords: if kw in query_lower: return "git" return None def _process_query(query: str, tool: str | None) -> CommandMatch | None: """Process a user query and display the result.""" matcher = get_matcher() match = matcher.match(query, tool) if match and match.confidence >= 0.3: display_command(match, explain=False) return match else: display_error(f"Could not find a matching command for: '{query}'") console.print("[info]Try rephrasing or use 'learn' to teach me a new command.[/]") return None def _show_interactive_help() -> None: """Show help for interactive mode.""" help_text = """ [bold]Shell Speak - Interactive Help[/bold] [bold]Commands:[/bold] help Show this help message clear Clear the screen history Show command history repeat Repeat the nth command from history (1 = most recent) learn :::: Learn a new command pattern forget Forget a learned pattern exit Exit interactive mode [bold]Examples:[/bold] show running containers commit changes with message "fix bug" list files in current directory apply kubernetes config [bold]Tips:[/bold] - Use up/down arrows to navigate history - Tab to autocomplete from history - Corrections are saved automatically """ console.print(help_text)