Crazy! Added support for ALT+\
This commit is contained in:
126
src/finish.py
126
src/finish.py
@@ -1,4 +1,4 @@
|
|||||||
#!/usr/bin/env python3
|
#!/home/mike/.venvs/cli/bin/python3
|
||||||
"""
|
"""
|
||||||
finish.py – AI shell completions that never leave your machine.
|
finish.py – AI shell completions that never leave your machine.
|
||||||
"""
|
"""
|
||||||
@@ -32,9 +32,9 @@ console = Console()
|
|||||||
# Config
|
# Config
|
||||||
# --------------------------------------------------------------------------- #
|
# --------------------------------------------------------------------------- #
|
||||||
DEFAULT_CFG = {
|
DEFAULT_CFG = {
|
||||||
"provider": "lmstudio",
|
"provider": "ollama",
|
||||||
"model": "darkidol-llama-3.1-8b-instruct-1.3-uncensored_gguf:2",
|
"model": "llama3:latest",
|
||||||
"endpoint": "http://plato.lan:1234/v1/chat/completions",
|
"endpoint": "http://localhost:11434/api/chat",
|
||||||
"temperature": 0.0,
|
"temperature": 0.0,
|
||||||
"api_prompt_cost": 0.0,
|
"api_prompt_cost": 0.0,
|
||||||
"api_completion_cost": 0.0,
|
"api_completion_cost": 0.0,
|
||||||
@@ -148,22 +148,48 @@ async def llm_complete(prompt: str) -> List[str]:
|
|||||||
else:
|
else:
|
||||||
raw = body["choices"][0]["message"]["content"]
|
raw = body["choices"][0]["message"]["content"]
|
||||||
|
|
||||||
|
# Debug logging
|
||||||
|
if os.getenv("FINISH_DEBUG"):
|
||||||
|
LOG_FILE.parent.mkdir(exist_ok=True)
|
||||||
|
with open(LOG_FILE, "a") as f:
|
||||||
|
f.write(f"\n=== {time.strftime('%Y-%m-%d %H:%M:%S')} ===\n")
|
||||||
|
f.write(f"Raw response:\n{raw}\n")
|
||||||
|
|
||||||
# try json first
|
# try json first
|
||||||
try:
|
try:
|
||||||
return json.loads(raw)["completions"]
|
result = json.loads(raw)["completions"]
|
||||||
except Exception:
|
if os.getenv("FINISH_DEBUG"):
|
||||||
|
with open(LOG_FILE, "a") as f:
|
||||||
|
f.write(f"Parsed completions: {result}\n")
|
||||||
|
return result
|
||||||
|
except Exception as e:
|
||||||
|
if os.getenv("FINISH_DEBUG"):
|
||||||
|
with open(LOG_FILE, "a") as f:
|
||||||
|
f.write(f"JSON parse failed: {e}\n")
|
||||||
# fallback: grep command-like lines
|
# fallback: grep command-like lines
|
||||||
return [ln for ln in raw.splitlines() if re.match(r"^(ls|cd|find|cat|grep|echo|mkdir|rm|cp|mv|pwd|chmod|chown)\b", ln)][:5]
|
fallback = [ln for ln in raw.splitlines() if re.match(r"^(ls|cd|find|cat|grep|echo|mkdir|rm|cp|mv|pwd|chmod|chown)\b", ln)][:5]
|
||||||
|
if os.getenv("FINISH_DEBUG"):
|
||||||
|
with open(LOG_FILE, "a") as f:
|
||||||
|
f.write(f"Fallback completions: {fallback}\n")
|
||||||
|
return fallback
|
||||||
|
|
||||||
# --------------------------------------------------------------------------- #
|
# --------------------------------------------------------------------------- #
|
||||||
# TUI picker
|
# TUI picker
|
||||||
# --------------------------------------------------------------------------- #
|
# --------------------------------------------------------------------------- #
|
||||||
def select_completion(completions: List[str]) -> Optional[str]:
|
async def select_completion(completions: List[str]) -> Optional[str]:
|
||||||
if not completions:
|
if not completions:
|
||||||
return None
|
return None
|
||||||
if len(completions) == 1:
|
if len(completions) == 1:
|
||||||
return completions[0]
|
return completions[0]
|
||||||
|
|
||||||
|
# If not in an interactive terminal, return first completion
|
||||||
|
if not sys.stdin.isatty():
|
||||||
|
return completions[0]
|
||||||
|
|
||||||
|
# Import here to avoid issues when not needed
|
||||||
|
from prompt_toolkit.input import create_input
|
||||||
|
from prompt_toolkit.output import create_output
|
||||||
|
|
||||||
kb = KeyBindings()
|
kb = KeyBindings()
|
||||||
current = 0
|
current = 0
|
||||||
|
|
||||||
@@ -195,13 +221,24 @@ def select_completion(completions: List[str]) -> Optional[str]:
|
|||||||
event.app.exit(result=None)
|
event.app.exit(result=None)
|
||||||
|
|
||||||
control = FormattedTextControl(get_text)
|
control = FormattedTextControl(get_text)
|
||||||
|
|
||||||
|
# Force output to /dev/tty to avoid interfering with bash command substitution
|
||||||
|
try:
|
||||||
|
output = create_output(stdout=open('/dev/tty', 'w'))
|
||||||
|
input_obj = create_input(stdin=open('/dev/tty', 'r'))
|
||||||
|
except Exception:
|
||||||
|
# Fallback to first completion if tty not available
|
||||||
|
return completions[0]
|
||||||
|
|
||||||
app = Application(
|
app = Application(
|
||||||
layout=Layout(HSplit([Window(control, height=len(completions) + 2)])),
|
layout=Layout(HSplit([Window(control, height=len(completions) + 2)])),
|
||||||
key_bindings=kb,
|
key_bindings=kb,
|
||||||
mouse_support=False,
|
mouse_support=False,
|
||||||
erase_when_done=True,
|
erase_when_done=True,
|
||||||
|
output=output,
|
||||||
|
input=input_obj,
|
||||||
)
|
)
|
||||||
return app.run()
|
return await app.run_async()
|
||||||
|
|
||||||
# --------------------------------------------------------------------------- #
|
# --------------------------------------------------------------------------- #
|
||||||
# Cache
|
# Cache
|
||||||
@@ -229,27 +266,67 @@ async def complete_line(line: str) -> Optional[str]:
|
|||||||
key = hashlib.md5(line.encode()).hexdigest()
|
key = hashlib.md5(line.encode()).hexdigest()
|
||||||
comps = cached(key)
|
comps = cached(key)
|
||||||
if comps is None:
|
if comps is None:
|
||||||
with console.status("[green]Thinking…[/]"):
|
# Write to /dev/tty if available to avoid interfering with command substitution
|
||||||
|
try:
|
||||||
|
status_console = Console(file=open('/dev/tty', 'w'), stderr=False)
|
||||||
|
except Exception:
|
||||||
|
status_console = console
|
||||||
|
|
||||||
|
with status_console.status("[green]Thinking…[/]"):
|
||||||
comps = await llm_complete(build_prompt(line))
|
comps = await llm_complete(build_prompt(line))
|
||||||
store_cache(key, comps)
|
store_cache(key, comps)
|
||||||
return select_completion(comps)
|
return await select_completion(comps)
|
||||||
|
|
||||||
# --------------------------------------------------------------------------- #
|
# --------------------------------------------------------------------------- #
|
||||||
# CLI
|
# CLI
|
||||||
# --------------------------------------------------------------------------- #
|
# --------------------------------------------------------------------------- #
|
||||||
def install_keybinding():
|
def install_keybinding():
|
||||||
r"""Inject Bash binding Alt+\ -> finish --accept-current-line"""
|
r"""Inject Bash binding Alt+\ -> finish"""
|
||||||
bind = r'"\e\\":" \C-u\C-kfinish --accept-current-line \C-m"'
|
|
||||||
rc = Path.home()/".bashrc"
|
rc = Path.home()/".bashrc"
|
||||||
marker = "# finish.py key-binding"
|
marker = "# finish.py key-binding"
|
||||||
snippet = f"{marker}\nbind '{bind}'"
|
|
||||||
|
# Use bind -x to allow modifying READLINE_LINE
|
||||||
|
snippet = f'''{marker}
|
||||||
|
_finish_complete() {{
|
||||||
|
local result
|
||||||
|
result=$(finish --readline-complete "$READLINE_LINE" 2>/dev/null)
|
||||||
|
if [[ -n "$result" ]]; then
|
||||||
|
READLINE_LINE="$result"
|
||||||
|
READLINE_POINT=${{#READLINE_LINE}}
|
||||||
|
fi
|
||||||
|
}}
|
||||||
|
bind -x '"\\e\\\\": _finish_complete'
|
||||||
|
'''
|
||||||
|
|
||||||
text = rc.read_text() if rc.exists() else ""
|
text = rc.read_text() if rc.exists() else ""
|
||||||
if marker in text:
|
if marker in text:
|
||||||
return
|
return
|
||||||
rc.write_text(text + "\n" + snippet + "\n")
|
rc.write_text(text + "\n" + snippet)
|
||||||
console.print("[green]Key-binding installed (Alt+\\)[/] – restart your shell.")
|
console.print("[green]Key-binding installed (Alt+\\)[/] – restart your shell.")
|
||||||
|
|
||||||
def main():
|
def main():
|
||||||
|
# Handle readline-complete flag before argparse (for bash bind -x)
|
||||||
|
if "--readline-complete" in sys.argv:
|
||||||
|
idx = sys.argv.index("--readline-complete")
|
||||||
|
line = sys.argv[idx + 1] if idx + 1 < len(sys.argv) else ""
|
||||||
|
if line.strip():
|
||||||
|
choice = asyncio.run(complete_line(line))
|
||||||
|
if choice:
|
||||||
|
print(choice)
|
||||||
|
return
|
||||||
|
|
||||||
|
# Legacy flag support
|
||||||
|
if sys.argv[-1] == "--accept-current-line":
|
||||||
|
line = os.environ.get("READLINE_LINE", "")
|
||||||
|
if line.strip():
|
||||||
|
choice = asyncio.run(complete_line(line))
|
||||||
|
if choice:
|
||||||
|
sys.stdout.write(f"\x1b]0;\a")
|
||||||
|
sys.stdout.write(f"\x1b[2K\r")
|
||||||
|
sys.stdout.write(choice)
|
||||||
|
sys.stdout.flush()
|
||||||
|
return
|
||||||
|
|
||||||
parser = argparse.ArgumentParser(prog="finish", description="AI shell completions")
|
parser = argparse.ArgumentParser(prog="finish", description="AI shell completions")
|
||||||
parser.add_argument("--version", action="version", version=VERSION)
|
parser.add_argument("--version", action="version", version=VERSION)
|
||||||
sub = parser.add_subparsers(dest="cmd")
|
sub = parser.add_subparsers(dest="cmd")
|
||||||
@@ -273,26 +350,17 @@ def main():
|
|||||||
return
|
return
|
||||||
choice = asyncio.run(complete_line(line))
|
choice = asyncio.run(complete_line(line))
|
||||||
if choice:
|
if choice:
|
||||||
|
# Check if we're in an interactive shell
|
||||||
|
if os.isatty(sys.stdout.fileno()):
|
||||||
|
console.print(f"[dim]Selected:[/] [cyan]{choice}[/]")
|
||||||
|
console.print(f"\n[yellow]Tip:[/] Use [bold]Alt+\\[/] keybinding for seamless completion!")
|
||||||
|
else:
|
||||||
print(choice)
|
print(choice)
|
||||||
return
|
return
|
||||||
if len(sys.argv) == 1:
|
if len(sys.argv) == 1:
|
||||||
parser.print_help()
|
parser.print_help()
|
||||||
return
|
return
|
||||||
|
|
||||||
# Hidden flag used by the key-binding:
|
|
||||||
if sys.argv[-1] == "--accept-current-line":
|
|
||||||
# Bash already put the current line into READLINE_LINE
|
|
||||||
line = os.environ.get("READLINE_LINE", "")
|
|
||||||
if not line.strip():
|
|
||||||
return
|
|
||||||
choice = asyncio.run(complete_line(line))
|
|
||||||
if choice:
|
|
||||||
# Replace current line with chosen completion
|
|
||||||
sys.stdout.write(f"\x1b]0;\a") # optional bell
|
|
||||||
sys.stdout.write(f"\x1b[2K\r") # clear line
|
|
||||||
sys.stdout.write(choice)
|
|
||||||
sys.stdout.flush()
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
try:
|
try:
|
||||||
main()
|
main()
|
||||||
|
|||||||
Reference in New Issue
Block a user