From 91d459227291057f4749812604b3a325cc2a7e98 Mon Sep 17 00:00:00 2001 From: mike Date: Thu, 11 Dec 2025 12:34:09 +0100 Subject: [PATCH] -init py- --- requirements.txt | 4 + src/finish.py | 300 +++++++++++++++++++++++++++++++++++++++++++++++ src/finish.sh | 1 + 3 files changed, 305 insertions(+) create mode 100644 requirements.txt create mode 100755 src/finish.py diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..3d7b41a --- /dev/null +++ b/requirements.txt @@ -0,0 +1,4 @@ +httpx +prompt_toolkit +rich +hashlib \ No newline at end of file diff --git a/src/finish.py b/src/finish.py new file mode 100755 index 0000000..bba3464 --- /dev/null +++ b/src/finish.py @@ -0,0 +1,300 @@ +#!/usr/bin/env python3 +""" +finish.py – AI shell completions that never leave your machine. +""" +from __future__ import annotations +import hashlib + +import argparse, asyncio, json, os, re, shlex, subprocess, sys, time +from pathlib import Path +from typing import List, Dict, Optional + +import httpx +from prompt_toolkit import ANSI, Application +from prompt_toolkit.buffer import Buffer +from prompt_toolkit.key_binding import KeyBindings +from prompt_toolkit.layout import ( + ConditionalContainer, FormattedTextControl, HSplit, Layout, Window +) +from prompt_toolkit.widgets import Label +from rich.console import Console +from rich.spinner import Spinner + +VERSION = "0.5.0" +CFG_DIR = Path.home()/".finish" +CFG_FILE = CFG_DIR/"finish.json" +CACHE_DIR = CFG_DIR/"cache" +LOG_FILE = CFG_DIR/"finish.log" + +console = Console() + +# --------------------------------------------------------------------------- # +# Config +# --------------------------------------------------------------------------- # +DEFAULT_CFG = { + "provider": "lmstudio", + "model": "darkidol-llama-3.1-8b-instruct-1.3-uncensored_gguf:2", + "endpoint": "http://plato.lan:1234/v1/chat/completions", + "temperature": 0.0, + "api_prompt_cost": 0.0, + "api_completion_cost": 0.0, + "max_history_commands": 20, + "max_recent_files": 20, + "cache_size": 100, +} + +def cfg() -> dict: + if not CFG_FILE.exists(): + CFG_DIR.mkdir(exist_ok=True) + CFG_FILE.write_text(json.dumps(DEFAULT_CFG, indent=2)) + return json.loads(CFG_FILE.read_text()) + +# --------------------------------------------------------------------------- # +# Context builders +# --------------------------------------------------------------------------- # +def _sanitise_history() -> str: + hist = subprocess.check_output(["bash", "-ic", "history"]).decode() + # scrub tokens / hashes + for pat in [ + r"\b[0-9a-f]{32,40}\b", # long hex + r"\b[A-Za-z0-9-]{36}\b", # uuid + r"\b[A-Za-z0-9]{16,40}\b", # api-keyish + ]: + hist = re.sub(pat, "REDACTED", hist) + return "\n".join(hist.splitlines()[-cfg()["max_history_commands"]:]) + +def _recent_files() -> str: + try: + files = subprocess.check_output( + ["find", ".", "-maxdepth", "1", "-type", "f", "-printf", "%T@ %p\n"], + stderr=subprocess.DEVNULL, + ).decode() + return "\n".join(sorted(files.splitlines(), reverse=True)[: cfg()["max_recent_files"]]) + except Exception: + return "" + +def _help_for(cmd: str) -> str: + try: + return subprocess.check_output([cmd, "--help"], stderr=subprocess.STDOUT, timeout=2).decode() + except Exception: + return f"{cmd} --help not available" + +def build_prompt(user_input: str) -> str: + c = cfg() + term_info = f"""User: {os.getenv("USER")} +PWD: {os.getcwd()} +HOME: {os.getenv("HOME")} +HOST: {os.getenv("HOSTNAME")} +SHELL: bash""" + prompt = f"""You are a helpful bash-completion script. +Generate 2–5 concise, valid bash commands that complete the user’s intent. + +Reply **only** JSON: {{"completions":["cmd1","cmd2",...]}} + +User command: {user_input} + +Terminal context: +{term_info} + +History: +{_sanitise_history()} + +Recent files: +{_recent_files()} + +Help: +{_help_for(user_input.split()[0])}""" + return prompt + +# --------------------------------------------------------------------------- # +# LLM call +# --------------------------------------------------------------------------- # +async def llm_complete(prompt: str) -> List[str]: + c = cfg() + payload = { + "model": c["model"], + "temperature": c["temperature"], + "messages": [ + {"role": "system", "content": "You are a helpful bash completion assistant."}, + {"role": "user", "content": prompt}, + ], + "response_format": {"type": "text"}, + } + if c["provider"] == "ollama": + payload["format"] = "json" + payload["stream"] = False + + headers = {"Content-Type": "application/json"} + if c.get("api_key"): + headers["Authorization"] = f"Bearer {c['api_key']}" + + try: + async with httpx.AsyncClient(timeout=30) as client: + resp = await client.post(c["endpoint"], headers=headers, json=payload) + resp.raise_for_status() + body = resp.json() + except httpx.ConnectError as e: + console.print(f"[red]Error:[/] Cannot connect to {c['endpoint']}") + console.print(f"[yellow]Check that your LLM server is running and endpoint is correct.[/]") + console.print(f"[yellow]Run 'finish config' to see current config.[/]") + return [] + except Exception as e: + console.print(f"[red]Error:[/] {e}") + return [] + + # extract content + if c["provider"] == "ollama": + raw = body["message"]["content"] + else: + raw = body["choices"][0]["message"]["content"] + + # try json first + try: + return json.loads(raw)["completions"] + except Exception: + # 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] + +# --------------------------------------------------------------------------- # +# TUI picker +# --------------------------------------------------------------------------- # +def select_completion(completions: List[str]) -> Optional[str]: + if not completions: + return None + if len(completions) == 1: + return completions[0] + + kb = KeyBindings() + current = 0 + + def get_text(): + return [ + ("", "Select completion (↑↓ navigate, Enter accept, Esc cancel)\n\n"), + *[ + ("[SetCursorPosition]" if i == current else "", f"{c}\n") + for i, c in enumerate(completions) + ], + ] + + @kb.add("up") + def _up(event): + nonlocal current + current = (current - 1) % len(completions) + + @kb.add("down") + def _down(event): + nonlocal current + current = (current + 1) % len(completions) + + @kb.add("enter") + def _accept(event): + event.app.exit(result=completions[current]) + + @kb.add("escape") + def _cancel(event): + event.app.exit(result=None) + + control = FormattedTextControl(get_text) + app = Application( + layout=Layout(HSplit([Window(control, height=len(completions) + 2)])), + key_bindings=kb, + mouse_support=False, + erase_when_done=True, + ) + return app.run() + +# --------------------------------------------------------------------------- # +# Cache +# --------------------------------------------------------------------------- # +def cached(key: str) -> Optional[List[str]]: + if not CACHE_DIR.exists(): + CACHE_DIR.mkdir(parents=True) + f = CACHE_DIR / f"{key}.json" + if f.exists(): + return json.loads(f.read_text()) + return None + +def store_cache(key: str, completions: List[str]) -> None: + f = CACHE_DIR / f"{key}.json" + f.write_text(json.dumps(completions)) + # LRU eviction + all_files = sorted(CACHE_DIR.glob("*.json"), key=lambda p: p.stat().st_mtime) + for old in all_files[: -cfg()["cache_size"]]: + old.unlink(missing_ok=True) + +# --------------------------------------------------------------------------- # +# Main entry +# --------------------------------------------------------------------------- # +async def complete_line(line: str) -> Optional[str]: + key = hashlib.md5(line.encode()).hexdigest() + comps = cached(key) + if comps is None: + with console.status("[green]Thinking…[/]"): + comps = await llm_complete(build_prompt(line)) + store_cache(key, comps) + return select_completion(comps) + +# --------------------------------------------------------------------------- # +# CLI +# --------------------------------------------------------------------------- # +def install_keybinding(): + r"""Inject Bash binding Alt+\ -> finish --accept-current-line""" + bind = r'"\e\\":" \C-u\C-kfinish --accept-current-line \C-m"' + rc = Path.home()/".bashrc" + marker = "# finish.py key-binding" + snippet = f"{marker}\nbind '{bind}'" + text = rc.read_text() if rc.exists() else "" + if marker in text: + return + rc.write_text(text + "\n" + snippet + "\n") + console.print("[green]Key-binding installed (Alt+\\)[/] – restart your shell.") + +def main(): + parser = argparse.ArgumentParser(prog="finish", description="AI shell completions") + parser.add_argument("--version", action="version", version=VERSION) + sub = parser.add_subparsers(dest="cmd") + sub.add_parser("install", help="add Alt+\\ key-binding to ~/.bashrc") + sub.add_parser("config", help="show current config") + p = sub.add_parser("command", help="simulate double-tab") + p.add_argument("words", nargs="*", help="partial command") + p.add_argument("--dry-run", action="store_true", help="show prompt only") + args = parser.parse_args() + + if args.cmd == "install": + install_keybinding() + return + if args.cmd == "config": + console.print_json(json.dumps(cfg())) + return + if args.cmd == "command": + line = " ".join(args.words) + if args.dry_run: + console.print(build_prompt(line)) + return + choice = asyncio.run(complete_line(line)) + if choice: + print(choice) + return + if len(sys.argv) == 1: + parser.print_help() + 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__": + try: + main() + except KeyboardInterrupt: + pass \ No newline at end of file diff --git a/src/finish.sh b/src/finish.sh index 9f8bb00..3b83d27 100755 --- a/src/finish.sh +++ b/src/finish.sh @@ -253,6 +253,7 @@ log_request() { created=$(echo "$response_body" | jq -r ".created // $created") api_cost=$(echo "$prompt_tokens_int * $ACSH_API_PROMPT_COST + $completion_tokens_int * $ACSH_API_COMPLETION_COST" | bc) log_file=${ACSH_LOG_FILE:-"$HOME/.finish/finish.log"} + mkdir -p "$(dirname "$log_file")" 2>/dev/null echo "$created,$user_input_hash,$prompt_tokens_int,$completion_tokens_int,$api_cost" >> "$log_file" }