Initial upload: shell-speak CLI tool with natural language to shell command conversion
This commit is contained in:
237
shell_speak/interactive.py
Normal file
237
shell_speak/interactive.py
Normal file
@@ -0,0 +1,237 @@
|
||||
"""Interactive mode implementation."""
|
||||
|
||||
import os
|
||||
import shutil
|
||||
from typing import Callable, Optional
|
||||
|
||||
from prompt_toolkit import PromptSession
|
||||
from prompt_toolkit.completion import Completer, Completion
|
||||
from prompt_toolkit.history import FileHistory
|
||||
from prompt_toolkit.keys import Keys
|
||||
from prompt_toolkit.key_binding import KeyBindings
|
||||
|
||||
from shell_speak.config import ensure_data_dir, get_data_dir
|
||||
from shell_speak.library import get_loader
|
||||
from shell_speak.matcher import get_matcher
|
||||
from shell_speak.history import get_history_manager
|
||||
from shell_speak.output import (
|
||||
console,
|
||||
display_command,
|
||||
display_error,
|
||||
display_help_header,
|
||||
display_history,
|
||||
)
|
||||
from shell_speak.models import CommandMatch
|
||||
|
||||
|
||||
class ShellSpeakCompleter(Completer):
|
||||
"""Auto-completion for shell-speak."""
|
||||
|
||||
def __init__(self):
|
||||
self._loader = get_loader()
|
||||
self._history_manager = get_history_manager()
|
||||
|
||||
def get_completions(self, document, complete_event):
|
||||
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):
|
||||
event.app.exit()
|
||||
|
||||
@kb.add(Keys.ControlL)
|
||||
def _(event):
|
||||
os.system("clear" if os.name == "posix" else "cls")
|
||||
|
||||
return kb
|
||||
|
||||
|
||||
def get_terminal_size() -> tuple:
|
||||
"""Get terminal size."""
|
||||
return shutil.get_terminal_size()
|
||||
|
||||
|
||||
def run_interactive_mode():
|
||||
"""Run the interactive shell mode."""
|
||||
ensure_data_dir()
|
||||
|
||||
display_help_header()
|
||||
|
||||
history_file = get_data_dir() / ".history"
|
||||
session = 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 <query>::<command>::<tool>[/]")
|
||||
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
|
||||
|
||||
tool = _detect_tool(user_input)
|
||||
match = _process_query(user_input, 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) -> Optional[str]:
|
||||
"""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: Optional[str]) -> Optional[CommandMatch]:
|
||||
"""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():
|
||||
"""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 <n> Repeat the nth command from history (1 = most recent)
|
||||
learn <q>::<c>::<t> Learn a new command pattern
|
||||
forget <q> 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)
|
||||
Reference in New Issue
Block a user