From 3affdd3b0fee506a6ebf6cf778f83e7d7780ed9a Mon Sep 17 00:00:00 2001 From: sinuhet Date: Thu, 20 Mar 2025 00:42:48 +0100 Subject: [PATCH] =?UTF-8?q?=C3=9Apln=C4=9B=20nov=C3=A1=20generace.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- robovojtik.py | 411 ++++++++++++++++++++++++++++++++++++++------------ 1 file changed, 315 insertions(+), 96 deletions(-) diff --git a/robovojtik.py b/robovojtik.py index cb9bfcb..6803166 100644 --- a/robovojtik.py +++ b/robovojtik.py @@ -1,9 +1,31 @@ +#!/usr/bin/env python3 +""" +Robovojtík – Linuxový shell asistent s interaktivním rozhraním + +Funkce: + • Rozhraní rozdělené vertikálně na dvě části: + - Levý panel: chat (dotazy, odpovědi, potvrzení, historie) + - Pravý panel: výpis příkazů a jejich výstup. + • Podpora automatického režimu (automód), kdy se příkazy spouštějí bez potvrzení. + • Při vykonání příkazu se jeho výstup zobrazí v pravém panelu a odešle se asistenti jako report, + přičemž do reportu se přidá indikace úspěšnosti (pokud nedošlo k chybě). + • Spinner se zobrazuje ve vyhrazeném okně pokaždé, když čekáme na odpověď z API (dotazy nebo reporty). + • V historii chatu se zaznamenává i potvrzení uživatele. + • Pokud je zapnuto logování (--log), loguje se vše na úrovni DEBUG asynchronně do souboru "robovojtik.log". +""" + import openai import subprocess -import time +import curses import configparser import json import sys +import logging +import threading +import time +import argparse +import queue +import logging.handlers # === 1. Načtení konfigurace z config.ini === config = configparser.ConfigParser() @@ -11,9 +33,19 @@ config.read("config.ini") openai.api_key = config["OpenAI"]["api_key"] ASSISTANT_ID = config["OpenAI"]["assistant_id"] -thread_id = None -# === 2. Definice funkcí, které asistent může volat === +# Globální proměnné +automode = False # Automatický režim: pokud True, příkaz se spustí ihned bez potvrzení. +log_enabled = False # Zapnutí logování – nastaví se příkazovou volbou. +thread_id = None # ID vlákna konverzace s OpenAI + +# Asynchroní logování +logger = logging.getLogger("robovojtik") +logger.setLevel(logging.DEBUG) +log_queue = None +listener = None + +# === 2. Definice funkcí pro volání OpenAI API === FUNCTIONS = [ { "name": "execute_shell_command", @@ -26,64 +58,121 @@ FUNCTIONS = [ "description": "Shellový příkaz k vykonání." } }, - "additionalProperties": False, - "required": ["command"] + "required": ["command"], + "additionalProperties": False + } + }, + { + "name": "get_system_load", + "description": "Získá aktuální zátěž systému (load average).", + "parameters": { + "type": "object", + "properties": {}, + "required": [], + "additionalProperties": False + } + }, + { + "name": "create_script", + "description": "Vytvoří skript s daným obsahem v souboru a nastaví ho na spustitelný.", + "parameters": { + "type": "object", + "properties": { + "file_name": { + "type": "string", + "description": "Název souboru (nebo cesta), do kterého se skript uloží." + }, + "content": { + "type": "string", + "description": "Obsah skriptu, který se má uložit." + } + }, + "required": ["file_name", "content"], + "additionalProperties": False } } ] -# === 3. Pomocné funkce === -def vytvor_nove_vlakno(): - """Vytvoří nové vlákno pro zachování historie konverzace.""" - thread = openai.beta.threads.create() - return thread.id +# === 3. (System prompt není načítán ze zdrojového kódu, vše se spravuje externě) === -def spust_prikaz(prikaz): - """Spustí shellový příkaz a vrátí jeho výstup.""" - try: - output = subprocess.check_output(prikaz, shell=True, stderr=subprocess.STDOUT, universal_newlines=True) - return output.strip() - except subprocess.CalledProcessError as e: - return f"Chyba při vykonávání příkazu:\n{e.output}" +# === 4. Pomocné funkce pro komunikaci s OpenAI API === + +def vytvor_nove_vlakno(): + """Vytvoří nové vlákno konverzace s asistentem a vrátí jeho ID.""" + global thread_id + thread = openai.beta.threads.create() + thread_id = thread.id + logger.debug(f"Vytvořeno nové vlákno: {thread_id}") + return thread_id + +def clean_command(command): + """Odstraní zpětné apostrofy a nežádoucí znaky z příkazu.""" + return command.replace("`", "").strip() + +def volani_asistenta(prompt, spinner_func=None): + """ + Spustí posli_dotaz_do_assistenta(prompt) v samostatném vlákně a během čekání volá spinner_func. + Vrací odpověď od asistenta. + """ + result_container = {} + + def worker(): + odpoved = posli_dotaz_do_assistenta(prompt) + result_container['answer'] = odpoved + + thread = threading.Thread(target=worker) + thread.start() + + idx = 0 + spinner = ["|", "/", "-", "\\"] + while thread.is_alive(): + if spinner_func: + spinner_func(spinner[idx % len(spinner)]) + time.sleep(0.2) + idx += 1 + thread.join() + return result_container.get('answer', "") def posli_dotaz_do_assistenta(prompt): """Odesílá dotaz v přirozeném jazyce do asistenta a vrací jeho odpověď.""" global thread_id if thread_id is None: - thread_id = vytvor_nove_vlakno() - + vytvor_nove_vlakno() + logger.debug(f"Odesílám dotaz: {prompt}") openai.beta.threads.messages.create( thread_id=thread_id, role="user", content=prompt ) - run = openai.beta.threads.runs.create( thread_id=thread_id, assistant_id=ASSISTANT_ID ) - while True: run_status = openai.beta.threads.runs.retrieve(thread_id=thread_id, run_id=run.id) if run_status.status == "completed": break time.sleep(1) - messages = openai.beta.threads.messages.list(thread_id=thread_id) - return messages.data[0].content[0].text.value + answer = messages.data[0].content[0].text.value + logger.debug(f"Asistent odpověděl: {answer}") + return answer def posli_prikaz_do_assistenta(command): - """Odesílá schválený příkaz k vykonání asistentovi a vrací výstup příkazu.""" + """ + Odesílá schválený příkaz k vykonání asistentovi a vrací jeho výstup. + Příkaz se před odesláním vyčistí. + """ + command = clean_command(command) global thread_id if thread_id is None: - thread_id = vytvor_nove_vlakno() - + vytvor_nove_vlakno() + logger.debug(f"Odesílám příkaz k vykonání: {command}") run = openai.beta.threads.runs.create( thread_id=thread_id, assistant_id=ASSISTANT_ID, instructions=f"Prosím, spusť tento příkaz: {command}" ) - while True: run_status = openai.beta.threads.runs.retrieve(thread_id=thread_id, run_id=run.id) if run_status.status == "requires_action": @@ -94,6 +183,8 @@ def posli_prikaz_do_assistenta(command): arguments = json.loads(tool_call.function.arguments) if tool_name == "execute_shell_command": vysledek = spust_prikaz(arguments["command"]) + elif tool_name == "create_script": + vysledek = vytvor_skript(arguments["file_name"], arguments["content"]) else: vysledek = "Neznámá funkce." tool_outputs.append({ @@ -108,9 +199,10 @@ def posli_prikaz_do_assistenta(command): elif run_status.status == "completed": break time.sleep(1) - messages = openai.beta.threads.messages.list(thread_id=thread_id) - return messages.data[0].content[0].text.value + answer = messages.data[0].content[0].text.value + logger.debug(f"Výsledek spuštěného příkazu: {answer}") + return answer def is_command_response(response): """Vrací True, pokud odpověď začíná prefixem indikujícím návrh příkazu.""" @@ -123,78 +215,205 @@ def extract_command(response): return response.strip()[len(prefix):].strip() return response.strip() -def provide_help(): - """Vrací text s nápovědou o funkcích Robovojtíka.""" - help_text = ( - "Nápověda k Robovojtikovi – Linuxový Shell Asistent:\n\n" - "Funkce:\n" - " • Převádí dotazy v přirozeném jazyce na návrhy shellových příkazů.\n" - " • Před spuštěním příkazu vždy čeká na potvrzení od uživatele.\n" - " • Přímé příkazy zadejte s prefixem 'cmd:' (např. 'cmd: ls -la').\n" - " • Pro ukončení Robovojtíka zadejte 'vypni' nebo 'exit'.\n\n" - "Pokud potřebujete další pomoc, zeptejte se!" - ) - return help_text +# === 5. Funkce pro spouštění příkazů a report === + +def spust_prikaz(command): + """ + Spustí příkaz synchronně a vrátí jeho výstup. + Používá se při volání funkce execute_shell_command. + """ + try: + logger.debug(f"Lokálně spouštím příkaz: {command}") + output = subprocess.check_output(command, shell=True, stderr=subprocess.STDOUT, universal_newlines=True) + return output.strip() + except subprocess.CalledProcessError as e: + return f"Chyba při vykonávání příkazu:\n{e.output}" + +def vytvor_skript(nazev, obsah): + """ + Zapíše obsah do souboru s daným názvem a nastaví spustitelnost. + """ + try: + with open(nazev, "w") as f: + f.write(obsah) + subprocess.call(f"chmod +x {nazev}", shell=True) + logger.debug(f"Skript {nazev} vytvořen a nastaven jako spustitelný.") + return f"Skript {nazev} byl úspěšně vytvořen." + except Exception as e: + return f"Nastala chyba při tvorbě skriptu: {str(e)}" + +def run_command_locally_and_report(command): + """ + Spustí příkaz lokálně, vyčistí ho a odešle jeho výstup jako report. + Pokud příkaz proběhne bez chyby, reportu přidá informaci, že příkaz proběhl úspěšně. + Vrací tuple (output, assistant_response). + """ + command = clean_command(command) + output = spust_prikaz(command) + if not output.startswith("Chyba při vykonávání příkazu:"): + report_text = f"Výstup příkazu (příkaz proběhl úspěšně):\n{output}" + else: + report_text = f"Výstup příkazu:\n{output}" + answer = posli_dotaz_do_assistenta(report_text) + return output, answer + +# === 6. Interaktivní rozhraní s curses (vertikální rozdělení) === + +def main_curses(stdscr): + curses.curs_set(1) + stdscr.nodelay(False) + stdscr.clear() + height, width = stdscr.getmaxyx() + + # Rozdělíme levý panel (chat) a pravý panel (výstup). + left_width = width // 2 + right_width = width - left_width + + # Vytvoříme chat_win s rezervovanou oblastí pro spinner. + chat_win = curses.newwin(height - 2, left_width, 0, 0) + spinner_win = curses.newwin(1, left_width, height - 2, 0) + prompt_win = curses.newwin(1, width, height - 1, 0) + output_win = curses.newwin(height - 1, right_width, 0, left_width) + + chat_win.scrollok(True) + output_win.scrollok(True) + + output_win.box() + output_win.refresh() + + spinner_chars = ["|", "/", "-", "\\"] + + def spinner_func(ch): + spinner_win.erase() + spinner_win.addstr(0, 0, f"Čekám na Robovojtíka... {ch}") + spinner_win.refresh() + + # Úvodní text v chatu + chat_win.addstr("Vítejte v Robovojtikovi!\n") + chat_win.addstr("Napište dotaz, příkaz ('cmd: ls'), 'automat' pro přepnutí automódu,\n") + chat_win.addstr("'skript: nazev; obsah' pro vytvoření skriptu, nebo 'vypni'/'exit' pro ukončení.\n") + chat_win.refresh() -# === 4. Hlavní interaktivní smyčka === -def main(): - print("Linuxový shell s OpenAI Assistant v2 API a funkcemi.") - print("Pro přímý příkaz použij prefix 'cmd:'") - print("Pro ukončení Robovojtíka zadej 'vypni' nebo 'exit'.") - print("Pro nápovědu zadej 'nápověda' nebo 'help'.") - while True: - user_input = input("\nZadej příkaz v přirozeném jazyce: ").strip() - - # Ukončení Robovojtíka - if user_input.lower() in ["vypni", "exit"]: - potvrzeni = input("Skutečně chceš ukončit Robovojtíka? (y/n): ").strip().lower() - if potvrzeni == "y": - print("Ukončuji Robovojtíka...") - sys.exit(0) - else: - print("Ukončení zrušeno. Pokračuji.") - continue - - # Zobrazení nápovědy - if user_input.lower() in ["nápověda", "help"]: - print("\n" + provide_help()) + prompt_win.erase() + prompt_win.addstr(">> ") + prompt_win.refresh() + curses.echo() + try: + user_input = prompt_win.getstr().decode("utf-8").strip() + except: continue - - # Přímý příkaz bez asistenta + curses.noecho() + + if not user_input: + continue + + if user_input.lower() in ["vypni", "exit"]: + chat_win.addstr("Ukončuji Robovojtíka...\n") + chat_win.refresh() + time.sleep(1) + break + + if user_input.lower() == "automat": + global automode + automode = not automode + stav = "zapnut" if automode else "vypnut" + chat_win.addstr(f"Automód byl nyní {stav}.\n") + chat_win.refresh() + continue + + if user_input.lower().startswith("skript:"): + try: + _, rest = user_input.split("skript:", 1) + nazev, obsah = rest.split(";", 1) + nazev = nazev.strip() + obsah = obsah.strip() + vysledek = vytvor_skript(nazev, obsah) + chat_win.addstr(vysledek + "\n") + except Exception as e: + chat_win.addstr(f"Chyba při vytváření skriptu: {str(e)}\n") + chat_win.refresh() + continue + if user_input.startswith("cmd:"): command = user_input[4:].strip() - print(f"Rozpoznán přímý příkaz: {command}") - potvrzeni = input("Spustit tento příkaz? (y/n): ").strip().lower() - if potvrzeni == "y": - vysledek = spust_prikaz(command) - print("\nVýstup příkazu:") - print(vysledek) - else: - print("Příkaz nebyl spuštěn.") + chat_win.addstr(f"Rozpoznán přímý příkaz: {command}\n") + chat_win.refresh() + if not automode: + prompt_win.erase() + prompt_win.addstr("Spustit tento příkaz? (y/n): ") + prompt_win.refresh() + potvrzeni = prompt_win.getstr().decode("utf-8").strip().lower() + # Zaznamenáme potvrzení do chatu + chat_win.addstr(f"Ty: {potvrzeni}\n") + chat_win.refresh() + if potvrzeni != "y": + chat_win.addstr("Příkaz nebyl spuštěn.\n") + chat_win.refresh() + continue + output, response = run_command_locally_and_report(command) + output_win.erase() + output_win.addstr(1, 1, output) + output_win.box() + output_win.refresh() + chat_win.addstr("Robovojtík odpovídá:\n" + response + "\n") + chat_win.refresh() continue - - # Dotaz pro asistenta – očekává návrh příkazu - assistant_response = posli_dotaz_do_assistenta(user_input) - - # Pokud odpověď nevypadá jako návrh příkazu, zobrazíme ji jako informativní odpověď - if not is_command_response(assistant_response): - print("\nRobovojtík odpovídá:") - print(assistant_response) - continue - - # Pokud odpověď obsahuje navržený příkaz, vyzveme uživatele k potvrzení - print("\nRobovojtík navrhuje příkaz:") - print(assistant_response) - - potvrzeni = input("Spustit tento příkaz? (y/n): ").strip().lower() - if potvrzeni == "y": - command_to_execute = extract_command(assistant_response) - execution_result = posli_prikaz_do_assistenta(command_to_execute) - print("\nVýstup příkazu:") - print(execution_result) - else: - print("Příkaz nebyl spuštěn. Čekám na další vstup.") + + # Pokud se jedná o dotaz v přirozeném jazyce + chat_win.addstr(f"Vy: {user_input}\n") + chat_win.refresh() + assistant_response = volani_asistenta(user_input, spinner_func=spinner_func) + spinner_win.erase() + spinner_win.refresh() + chat_win.addstr("Robovojtík odpovídá:\n" + assistant_response + "\n") + chat_win.refresh() + + if is_command_response(assistant_response): + navrhovany_prikaz = extract_command(assistant_response) + chat_win.addstr(f"Navrhovaný příkaz: {navrhovany_prikaz}\n") + chat_win.refresh() + if not automode: + prompt_win.erase() + prompt_win.addstr("Spustit tento příkaz? (y/n): ") + prompt_win.refresh() + potvrzeni = prompt_win.getstr().decode("utf-8").strip().lower() + chat_win.addstr(f"Ty: {potvrzeni}\n") + chat_win.refresh() + if potvrzeni != "y": + chat_win.addstr("Příkaz nebyl spuštěn.\n") + chat_win.refresh() + continue + output, response = run_command_locally_and_report(navrhovany_prikaz) + output_win.erase() + output_win.addstr(1, 1, output) + output_win.box() + output_win.refresh() + chat_win.addstr("Robovojtík odpovídá:\n" + response + "\n") + chat_win.refresh() + +def main(): + global log_enabled, log_handler, log_queue, listener + parser = argparse.ArgumentParser(description="Robovojtík – interaktivní shell asistent") + parser.add_argument("--log", action="store_true", help="Zapne logování do souboru robovojtik.log") + args = parser.parse_args() + + if args.log: + log_enabled = True + log_queue = queue.Queue(-1) + qh = logging.handlers.QueueHandler(log_queue) + logger.addHandler(qh) + logger.setLevel(logging.DEBUG) + fh = logging.FileHandler("robovojtik.log") + fh.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")) + listener = logging.handlers.QueueListener(log_queue, fh) + listener.start() + logger.debug("Logování zapnuto.") + + curses.wrapper(main_curses) + + if listener: + listener.stop() if __name__ == "__main__": - main( \ No newline at end of file + main()