MVP claix

README.md

@@ -1,3 +1,3 @@
 # claix
 
-A CLI to enhance command crafting for other CLIs.
+A CLI translating user requests into specific CLI commands

claix/__init__.py

@@ -1,4 +1,4 @@
 import importlib.metadata
 
 
-__version__ = importlib.metadata.version(__package__ or __name__)
+__version__ = importlib.metadata.version(__package__ or __name__)

claix/__main__.py

@@ -1,2 +1,3 @@
 from .main import app
-app(prog_name="claix")
+
+app(prog_name="claix")

claix/bot.py

@@ -0,0 +1,91 @@
+from openai import OpenAI
+import time
+
+
+class Bot:
+    def __init__(self, assistant_id: str, thread_id: str = None):
+        self.assistant_id = assistant_id
+        self.thread_id = thread_id
+        self.client = OpenAI()
+
+    def __call__(self, prompt, thread_id = None):
+        if not thread_id:
+            thread_id = self.thread_id
+
+        self.create_thread_message(prompt, thread_id)
+
+        run = self.run_thread(thread_id)
+
+        run = self.wait_for_run(run)
+
+        return self.get_last_message(thread_id)
+
+    @staticmethod
+    def create_assistant(
+        name: str,
+        instructions: str,
+        model: str = "gpt-4-1106-preview",
+        tools: list[dict] = [{"type": "code_interpreter"}, {"type": "retrieval"}],
+    ):
+        client = OpenAI()
+        assistant = client.beta.assistants.create(
+            name=name,
+            instructions=instructions,
+            tools=tools,
+            model=model,
+        )
+        return assistant
+
+
+    @staticmethod
+    def create_thread():
+        client = OpenAI()
+        thread = client.beta.threads.create()
+        return thread
+
+    def create_thread_message(self, prompt: str, thread_id: str):
+        message = self.client.beta.threads.messages.create(
+            thread_id=thread_id,
+            role="user",
+            content=prompt,
+        )
+        return message
+
+    def run_thread(self, thread_id: str):
+        run = self.client.beta.threads.runs.create(
+            thread_id=thread_id,
+            assistant_id=self.assistant_id,
+        )
+        return run
+
+    def update_run(self, run):
+        run = self.client.beta.threads.runs.retrieve(
+            run_id=run.id,
+            thread_id=run.thread_id,
+        )
+        return run
+
+    def wait_for_run(self, run):
+        while run.status != "completed":
+            run = self.update_run(run)
+            time.sleep(0.1)
+        return run
+
+    def get_thread_messages(self, thread_id: str):
+        messages = self.client.beta.threads.messages.list(
+            thread_id=thread_id,
+        )
+        return messages
+
+    def get_last_message(self, thread_id: str):
+        messages = self.get_thread_messages(thread_id)
+        return list(messages)[0].content[0].text.value
+
+    def add_files_to_assistant(self, file_ids: str | list[str]):
+        if isinstance(file_ids, str):
+            file_ids = [file_ids]
+
+        updated_assistant = self.client.beta.assistants.update(
+            self.assistant_id,
+            file_ids=file_ids,
+        )

claix/main.py

@@ -1,11 +1,108 @@
 # src/main.py
+from claix.bot import Bot
+from claix.utils import get_or_create_default_assistant_id, get_or_create_default_thread_id, run_shell_command, simulate_clear
+from rich.panel import Panel
+from rich.text import Text
+from rich.console import Console
+import rich
 import typer
 
-app = typer.Typer()
+app = typer.Typer(name="claix", no_args_is_help=True, add_completion=False)
+console = Console()
+
+@app.command(help="Turns your instructions into CLI commands ready to execute.")
+def main(instructions: list[str] = typer.Argument(None, help="The instructions that you want to execute. Pass them as a list of strings.", )):
+    """
+    Claix is a command line assistant that helps you translate instructions into CLI commands.
+
+    Example: claix list all docker containers
+
+    You can give it a set of instructions, and it will attempt to generate
+    the appropriate command-line commands to execute.
+
+    Args:
+    instructions: Your instructions as text.
+    """
+    simulate_clear(console)
+    if not instructions:
+        # If no instructions are provided, display a helpful message and an example usage
+        error_message = Text("No instructions provided. Claix needs a set of instructions to generate commands.", style="white")
+        example_usage = Text("Example usage:\nclaix list all docker containers\nclaix show me active network interfaces", style="white")
+        rich.print(Panel(error_message, title="[bold]Error[/bold]", expand=False, border_style="red"))
+        rich.print(Panel(example_usage, title="[bold]Example Usage[/bold]", expand=False, border_style="green"))
+        return
+    instructions: str = " ".join(instructions)
+    rich.print(Panel(Text(instructions, style="white"), title="Instructions\U0001F4DD", expand=False, border_style="blue"))
+    assistant_id = get_or_create_default_assistant_id()
+    thread_id = get_or_create_default_thread_id()
+
+    bot = Bot(assistant_id, thread_id)
+
+    rich.print(Panel(Text("Thinking...", style="white"), title="Claix", expand=False, border_style="purple"))
+    proposed_command: str = bot(instructions)
+    if proposed_command == ".":
+        rich.print(Panel(Text("I don't know how to solve this problem, exiting", style="white"), title="Claix", expand=False, border_style="purple"))
+        return
+    rich.print(Panel(proposed_command, title="Command\U0001F4BB", expand=False, border_style="green", padding=(1, 2)))
+
+    # Ask if user wants Claix to run the command
+    prompt_text = Text("Run command? Press Enter to run or [Y/n]", style="white", end="")
+    rich.print(Panel(prompt_text, title="Action\u2757", expand=False, border_style="yellow"), end="")
+    run_command_input = input()
+    run_command = run_command_input.lower() in ["y", "yes", ""]
+    simulate_clear(console)
+
+    if not run_command:
+        return
+
+    result = run_shell_command(proposed_command)
+
+    error_iterations = 0
+    while result.returncode != 0:
+        if error_iterations > 2:
+            simulate_clear(console)
+            rich.print(Panel(Text("Too many errors, exiting", style="white"), title="Error", expand=False, border_style="red"))
+            break
+
+        rich.print(Panel(Text(instructions, style="white"), title="Instructions\U0001F4DD", expand=False, border_style="blue"))
+        rich.print(Panel(Text(f"Error iteration {error_iterations}", style="white"), title="Error iteration", expand=False, border_style="red"))
+        rich.print(Panel(result.stderr, title="Error", expand=False, border_style="red"))
+        error_prompt = \
+f"""I want to: '{instructions}'
+I tried '{proposed_command}'
+but got this error: '{result.stderr}'
+
+Having this error in mind, fix my original command of '{proposed_command}' or give me a new command to solve: '{instructions}'"""
+
+
+        rich.print(Panel(Text("Thinking...", style="white"), title="Claix", expand=False, border_style="purple"))
+        proposed_solution = bot(error_prompt)
+
+        if proposed_solution == ".":
+            rich.print(Panel(Text("I don't know how to solve this problem, exiting", style="white"), title="Claix", expand=False, border_style="purple"))
+            return
+
+        rich.print(Panel(proposed_solution, title="Command\U0001F4BB", expand=False, border_style="green", padding=(1, 2)))
+        # Ask if user wants Claix to run the command
+        prompt_text = Text("Run command? Press Enter to run or [Y/n]", style="white", end="")
+        rich.print(Panel(prompt_text, title="Action\u2757", expand=False, border_style="yellow"), end="")
+        run_command_input = input()
+        run_command = run_command_input.lower() in ["y", "yes", ""]
+        simulate_clear(console)
+
+        if not run_command:
+            return
+
+        result = run_shell_command(proposed_solution)
+        error_iterations += 1
+
+    else:
+        simulate_clear(console)
+        # success
+        if result.stdout:
+            rich.print(Panel(result.stdout, title="Output", expand=False, border_style="blue"))
+
 
-@app.command()
-def main():
-    print("Hello")
 
 if __name__ == "__main__":
-    app()
+    app()

claix/utils.py

@@ -0,0 +1,88 @@
+import shelve
+import subprocess
+
+import rich
+from claix.bot import Bot
+
+DEFAULT_INSTRUCTIONS = """Claix exclusively provides Linux CLI command translations in plain text, with no code blocks or additional formatting. When a user's input aligns with Linux CLI commands, Claix responds with the exact command in simple text. If the input is unrelated to Linux CLI commands, Claix replies with a single '.' to maintain focus on its primary role.
+
+This GPT avoids any execution or simulation of CLI commands and does not engage in discussions beyond Linux CLI command translation. Claix's responses are brief and to the point, delivering Linux CLI commands in an unembellished, clear format, ensuring users receive direct and unformatted command syntax for their Linux-related inquiries."""
+
+
+
+
+def run_shell_command(command):
+    result = subprocess.run(command, 
+                            stdout=subprocess.PIPE, 
+                            stderr=subprocess.PIPE, 
+                            text=True, 
+                            shell=True)
+    return result
+
+
+
+def get_assistant_id(assistant: str = "default"):
+    with shelve.open("db") as db:
+        try:
+            return db["assistants"][assistant]["id"]
+        except KeyError:
+            return None
+
+
+def set_assistant_id(assistant_id, assistant="default"):
+    with shelve.open("db") as db:
+        assistants = db.get("assistants", {})
+        default = assistants.get(assistant)
+        if default:
+            db["assistants"][assistant]["id"] = assistant_id
+        else:
+            db["assistants"] = {assistant: {"id": assistant_id}}
+    return assistant_id
+
+
+def get_or_create_default_assistant_id():
+    assistant_id = get_assistant_id()
+    if not assistant_id:
+        assistant = Bot.create_assistant(
+            name="default",
+            instructions=DEFAULT_INSTRUCTIONS,
+        )
+        assistant_id = set_assistant_id(assistant.id, assistant="default")
+    return assistant_id
+
+
+
+def get_thread_id(thread: str = "default"):
+    with shelve.open("db") as db:
+        try:
+            return db["threads"][thread]["id"]
+        except KeyError:
+            return None
+
+def set_thread_id(thread_id, thread="default"):
+    with shelve.open("db") as db:
+        threads = db.get("threads", {})
+        default = threads.get(thread)
+        if default:
+            db["threads"][thread]["id"] = thread_id
+        else:
+            db["threads"] = {thread: {"id": thread_id}}
+    return thread_id
+
+def get_or_create_default_thread_id():
+    thread_id = get_thread_id()
+    if not thread_id:
+        thread = Bot.create_thread()
+        thread_id = set_thread_id(thread.id, thread="default")
+    return thread_id
+
+
+
+def simulate_clear(console: rich.console.Console):
+    """
+    Simulates clearing the console by printing enough new lines to push old content out of view,
+    then moves the cursor back to the top of the console window.
+    """
+    height = console.size.height
+    print("\n" * height, end='')  # Print newlines to push content out of view
+    print(f"\033[{height}A", end='')  # Move the cursor back up to the top of the console window

pyproject.toml

@@ -1,7 +1,7 @@
 [tool.poetry]
 name = "claix"
-version = "0.1.0"
-description = "A CLI to enhance command crafting for other CLIs."
+version = "0.1.1"
+description = "A CLI translating user requests into specific CLI commands"
 authors = ["Facundo Goiriz <[email protected]>"]
 readme = "README.md"