238 lines
7.3 KiB
Python
238 lines
7.3 KiB
Python
"""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)
|