From c6be2b2664782fa34444daedfd7541697d13cdd4 Mon Sep 17 00:00:00 2001 From: sinuhet Date: Thu, 20 Mar 2025 00:54:11 +0100 Subject: [PATCH] =?UTF-8?q?je=C5=A1t=C4=9B=20lep=C5=A1=C3=AD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- robovojtik.py | 204 ++++++++++++++++++++++++-------------------------- 1 file changed, 98 insertions(+), 106 deletions(-) diff --git a/robovojtik.py b/robovojtik.py index 6803166..11cd848 100644 --- a/robovojtik.py +++ b/robovojtik.py @@ -1,17 +1,16 @@ #!/usr/bin/env python3 """ -Robovojtík – Linuxový shell asistent s interaktivním rozhraním +Robovojtík – Linuxový shell asistent s vylepšeným 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". + • Rozhraní s barevným oddělením: + - Levá strana je rozdělena na hlavičku (s trvalými instrukcemi a spinnerem) a chat (historie). + - Pravá strana zobrazuje výstup příkazů. + - Mezi levým a pravým panelem je vertikální oddělovač. + - Dolní část je vstupní oblast (o dva řádky). + • Spinner se zobrazuje v hlavičce, uprostřed, vždy když čekáme na odpověď asistenta. + • Do chatu se zaznamenává i uživatelské potvrzení. + • Logování (při spuštění s --log) běží asynchronně a loguje vše na úrovni DEBUG do souboru "robovojtik.log". """ import openai @@ -30,13 +29,12 @@ import logging.handlers # === 1. Načtení konfigurace z config.ini === config = configparser.ConfigParser() config.read("config.ini") - openai.api_key = config["OpenAI"]["api_key"] ASSISTANT_ID = config["OpenAI"]["assistant_id"] # 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. +automode = False # Automatický režim +log_enabled = False # Zapnutí logování přes --log thread_id = None # ID vlákna konverzace s OpenAI # Asynchroní logování @@ -93,12 +91,11 @@ FUNCTIONS = [ } ] -# === 3. (System prompt není načítán ze zdrojového kódu, vše se spravuje externě) === +# === 3. (System prompt není načítán ze zdrojového kódu) === # === 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 @@ -106,23 +103,15 @@ def vytvor_nove_vlakno(): 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(): @@ -134,7 +123,6 @@ def volani_asistenta(prompt, spinner_func=None): 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: vytvor_nove_vlakno() @@ -159,10 +147,6 @@ def posli_dotaz_do_assistenta(prompt): return answer def posli_prikaz_do_assistenta(command): - """ - 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: @@ -205,23 +189,19 @@ def posli_prikaz_do_assistenta(command): return answer def is_command_response(response): - """Vrací True, pokud odpověď začíná prefixem indikujícím návrh příkazu.""" return response.strip().lower().startswith("navrhovaný příkaz:") -def extract_command(response): - """Odstraní prefix 'Navrhovaný příkaz:' a vrátí čistý shellový příkaz.""" - prefix = "navrhovaný příkaz:" - if response.strip().lower().startswith(prefix): - return response.strip()[len(prefix):].strip() - return response.strip() +def get_first_command_proposal(response): + """Vrátí první řádek, který začíná 'Navrhovaný příkaz:'; jinak None.""" + lines = response.splitlines() + for line in lines: + if line.strip().lower().startswith("navrhovaný příkaz:"): + return line + return None # === 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) @@ -230,9 +210,6 @@ def spust_prikaz(command): 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) @@ -243,11 +220,6 @@ def vytvor_skript(nazev, obsah): 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:"): @@ -257,46 +229,81 @@ def run_command_locally_and_report(command): answer = posli_dotaz_do_assistenta(report_text) return output, answer -# === 6. Interaktivní rozhraní s curses (vertikální rozdělení) === +# === 6. Vylepšené interaktivní rozhraní s curses === def main_curses(stdscr): + curses.start_color() + curses.use_default_colors() + curses.init_pair(1, curses.COLOR_WHITE, curses.COLOR_BLUE) # Hlavička + curses.init_pair(2, curses.COLOR_WHITE, -1) # Chat + curses.init_pair(3, curses.COLOR_GREEN, -1) # Výstup + curses.init_pair(4, curses.COLOR_YELLOW, -1) # Vstup + curses.init_pair(5, curses.COLOR_CYAN, -1) # Spinner + curses.init_pair(6, curses.COLOR_BLACK, curses.COLOR_WHITE) # Oddělovač + curses.curs_set(1) stdscr.nodelay(False) stdscr.clear() height, width = stdscr.getmaxyx() - # Rozdělíme levý panel (chat) a pravý panel (výstup). + header_height = 3 + prompt_height = 2 left_width = width // 2 - right_width = width - left_width + right_width = width - left_width - 1 # oddělovač má 1 sloupec + chat_height = height - header_height - prompt_height - # 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) + header_win = curses.newwin(header_height, left_width, 0, 0) + chat_win = curses.newwin(chat_height, left_width, header_height, 0) + prompt_win = curses.newwin(prompt_height, width, height - prompt_height, 0) + output_win = curses.newwin(height - prompt_height, right_width, 0, left_width + 1) + divider_win = curses.newwin(height - prompt_height, 1, 0, left_width) + + header_win.bkgd(' ', curses.color_pair(1)) + chat_win.bkgd(' ', curses.color_pair(2)) + prompt_win.bkgd(' ', curses.color_pair(4)) + output_win.bkgd(' ', curses.color_pair(3)) + divider_win.bkgd(' ', curses.color_pair(6)) + + d_height, _ = divider_win.getmaxyx() + for y in range(d_height): + try: + divider_win.addch(y, 0, curses.ACS_VLINE) + except curses.error: + pass + divider_win.refresh() + + header_text = [ + "Vítejte v Robovojtikovi!", + "Zadejte dotaz, příkaz (prefix 'cmd:'), 'automat' nebo 'skript: nazev; obsah'.", + "Pro ukončení zadejte 'vypni' nebo 'exit'." + ] + for idx, line in enumerate(header_text): + header_win.addstr(idx, 1, line) + header_win.refresh() chat_win.scrollok(True) output_win.scrollok(True) - + prompt_win.scrollok(True) output_win.box() output_win.refresh() - spinner_chars = ["|", "/", "-", "\\"] - + # Funkce pro spinner v hlavičce (uprostřed) def spinner_func(ch): - spinner_win.erase() - spinner_win.addstr(0, 0, f"Čekám na Robovojtíka... {ch}") - spinner_win.refresh() + mid_x = left_width // 2 - 5 + header_win.addstr(1, mid_x, f"Čekám... {ch}", curses.color_pair(5) | curses.A_BOLD) + header_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() + chat_history = [] + def add_to_chat(text): + chat_history.append(text) + chat_win.addstr(text + "\n") + chat_win.refresh() + + add_to_chat("Historie chatu:") while True: prompt_win.erase() - prompt_win.addstr(">> ") + prompt_win.addstr(">> ", curses.A_BOLD) prompt_win.refresh() curses.echo() try: @@ -308,9 +315,10 @@ def main_curses(stdscr): if not user_input: continue + add_to_chat("Ty: " + user_input) + if user_input.lower() in ["vypni", "exit"]: - chat_win.addstr("Ukončuji Robovojtíka...\n") - chat_win.refresh() + add_to_chat("Ukončuji Robovojtíka...") time.sleep(1) break @@ -318,8 +326,7 @@ def main_curses(stdscr): global automode automode = not automode stav = "zapnut" if automode else "vypnut" - chat_win.addstr(f"Automód byl nyní {stav}.\n") - chat_win.refresh() + add_to_chat(f"Automód byl nyní {stav}.") continue if user_input.lower().startswith("skript:"): @@ -329,75 +336,62 @@ def main_curses(stdscr): nazev = nazev.strip() obsah = obsah.strip() vysledek = vytvor_skript(nazev, obsah) - chat_win.addstr(vysledek + "\n") + add_to_chat(vysledek) except Exception as e: - chat_win.addstr(f"Chyba při vytváření skriptu: {str(e)}\n") - chat_win.refresh() + add_to_chat(f"Chyba při vytváření skriptu: {str(e)}") continue if user_input.startswith("cmd:"): command = user_input[4:].strip() - chat_win.addstr(f"Rozpoznán přímý příkaz: {command}\n") - chat_win.refresh() + add_to_chat(f"Rozpoznán přímý příkaz: {command}") if not automode: prompt_win.erase() - prompt_win.addstr("Spustit tento příkaz? (y/n): ") + prompt_win.addstr("Spustit tento příkaz? (y/n): ", curses.A_BOLD) 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() + add_to_chat("Ty: " + potvrzeni) if potvrzeni != "y": - chat_win.addstr("Příkaz nebyl spuštěn.\n") - chat_win.refresh() + add_to_chat("Příkaz nebyl spuštěn.") 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() + add_to_chat("Robovojtík odpovídá:\n" + response) continue - # 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() + add_to_chat("Robovojtík odpovídá:\n" + assistant_response) - 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() + proposal_line = get_first_command_proposal(assistant_response) + if proposal_line: + navrhovany_prikaz = proposal_line[len("Navrhovaný příkaz:"):].strip() + add_to_chat(f"Navrhovaný příkaz: {navrhovany_prikaz}") if not automode: prompt_win.erase() - prompt_win.addstr("Spustit tento příkaz? (y/n): ") + prompt_win.addstr("Spustit tento příkaz? (y/n): ", curses.A_BOLD) prompt_win.refresh() potvrzeni = prompt_win.getstr().decode("utf-8").strip().lower() - chat_win.addstr(f"Ty: {potvrzeni}\n") - chat_win.refresh() + add_to_chat("Ty: " + potvrzeni) if potvrzeni != "y": - chat_win.addstr("Příkaz nebyl spuštěn.\n") - chat_win.refresh() + add_to_chat("Příkaz nebyl spuštěn.") 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() + add_to_chat("Robovojtík odpovídá:\n" + response) + + prompt_win.erase() + prompt_win.refresh() def main(): - global log_enabled, log_handler, log_queue, listener + global log_enabled, listener, log_queue 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) @@ -409,9 +403,7 @@ def main(): listener = logging.handlers.QueueListener(log_queue, fh) listener.start() logger.debug("Logování zapnuto.") - curses.wrapper(main_curses) - if listener: listener.stop()