diff --git a/api_interface.py b/api_interface.py new file mode 100644 index 0000000..eba9721 --- /dev/null +++ b/api_interface.py @@ -0,0 +1,137 @@ +""" +Module: api_interface.py +-------------------------- +Obsahuje funkce pro komunikaci s OpenAI API a správu konverzačního vlákna. +""" + +import openai +import json +import time +import threading +import logging +from configparser import ConfigParser + +logger = logging.getLogger("robovojtik.api_interface") +config = ConfigParser() +config.read("config.ini") +try: + openai.api_key = config["OpenAI"]["api_key"] +except KeyError: + raise ValueError("API key not found in config.ini. Please set the 'api_key' under [OpenAI] section or set OPENAI_API_KEY environment variable.") + +ASSISTANT_ID = config["OpenAI"].get("assistant_id", None) +if not ASSISTANT_ID: + raise ValueError("assistant_id not found in config.ini. Please set the 'assistant_id' under [OpenAI] section.") + +# Globální proměnná pro uchování ID konverzačního vlákna +thread_id = None + +def vytvor_nove_vlakno(): + """ + Vytvoří nové vlákno konverzace s asistentem pomocí OpenAI API. + """ + 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 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() + 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) + 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í do asistenta a vrací jeho odpověď. + Před odesláním se příkaz vyčistí. + """ + from shell_functions import spust_prikaz, vytvor_skript # Importujeme zde, abychom předešli cyklickým závislostem. + global thread_id + if thread_id is None: + 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": + tool_calls = run_status.required_action.submit_tool_outputs.tool_calls + tool_outputs = [] + for tool_call in tool_calls: + tool_name = tool_call.function.name + 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({ + "tool_call_id": tool_call.id, + "output": vysledek + }) + openai.beta.threads.runs.submit_tool_outputs( + thread_id=thread_id, + run_id=run.id, + tool_outputs=tool_outputs + ) + elif run_status.status == "completed": + break + time.sleep(1) + messages = openai.beta.threads.messages.list(thread_id=thread_id) + answer = messages.data[0].content[0].text.value + logger.debug(f"Výsledek spuštěného příkazu: {answer}") + return answer + +def volani_asistenta(prompt, spinner_func=None): + """ + Spustí dotaz do asistenta v samostatném vlákně a během čekání volá spinner_func. + Vrací odpověď asistenta. Pokud dojde k chybě, vrací prázdný řetězec. + """ + result_container = {} + + def worker(): + try: + odpoved = posli_dotaz_do_assistenta(prompt) + result_container['answer'] = odpoved + except Exception as ex: + result_container['answer'] = f"Chyba při volání asistenta: {str(ex)}" + logger.exception("Chyba v worker funkci volani_asistenta.") + + 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', "") diff --git a/config.ini b/config.ini index 09f035f..9c1f1aa 100644 --- a/config.ini +++ b/config.ini @@ -1,3 +1,3 @@ [OpenAI] api_key = -assistant_id = \ No newline at end of file +assistant_id = diff --git a/main.py b/main.py new file mode 100644 index 0000000..0a21ff4 --- /dev/null +++ b/main.py @@ -0,0 +1,35 @@ +""" +Module: main.py +--------------- +Hlavní spouštěcí skript pro Robovojtíka. +Importuje moduly pro API komunikaci, shellové funkce a uživatelské rozhraní. +""" + +import argparse +import queue +import logging +import logging.handlers +import ui + +def main(): + 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() + listener = None + if args.log: + log_queue = queue.Queue(-1) + logger = logging.getLogger("robovojtik") + logger.setLevel(logging.DEBUG) + qh = logging.handlers.QueueHandler(log_queue) + logger.addHandler(qh) + 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.") + ui.main_ui() + if listener: + listener.stop() + +if __name__ == "__main__": + main() diff --git a/shell_functions.py b/shell_functions.py new file mode 100644 index 0000000..40c6701 --- /dev/null +++ b/shell_functions.py @@ -0,0 +1,55 @@ +""" +Module: shell_functions.py +---------------------------- +Obsahuje funkce pro vykonávání lokálních shellových příkazů a tvorbu skriptů. +""" + +import subprocess +import logging + +logger = logging.getLogger("robovojtik.shell_functions") + +def clean_command(command): + """ + Odstraní backticky a nepotřebné mezery z příkazu. + """ + return command.replace("`", "").strip() + +def spust_prikaz(command): + """ + Spustí příkaz lokálně a vrátí jeho výstup. + V případě chyby vrací chybovou hlášku. + """ + 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í jej jako spustitelný. + """ + 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, api_interface): + """ + Spustí příkaz lokálně, odešle jeho výstup jako report (s prefixem) + do asistenta a vrátí 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 = api_interface.posli_dotaz_do_assistenta(report_text) + return output, answer diff --git a/ui.py b/ui.py new file mode 100644 index 0000000..0f3905f --- /dev/null +++ b/ui.py @@ -0,0 +1,190 @@ +""" +Module: ui.py +------------- +Obsahuje interaktivní rozhraní Robovojtíka založené na knihovně curses. +""" + +import curses +import time +import threading +import logging + +import api_interface +import shell_functions + +logger = logging.getLogger("robovojtik.ui") + +# Definujeme globální proměnnou automode zde +automode = False + +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() + + header_height = 3 + prompt_height = 2 + left_width = width // 2 + right_width = width - left_width - 1 # oddělovač = 1 sloupec + chat_height = height - header_height - prompt_height + + 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 napiš 'napiš mi skript, ...' pro generování skriptu.", + "Pro ukončení zadejte 'vypni' nebo 'exit'." + ] + max_chars = max(0, left_width - 2) + for idx, line in enumerate(header_text): + try: + header_win.addnstr(idx, 1, line, max_chars) + except curses.error: + pass + header_win.refresh() + + chat_win.scrollok(True) + output_win.scrollok(True) + prompt_win.scrollok(True) + output_win.box() + output_win.refresh() + + spinner_chars = ["|", "/", "-", "\\"] + + def spinner_func(ch): + mid_x = left_width // 2 - 5 + header_win.addnstr(1, mid_x, f"Čekám... {ch}", left_width - mid_x - 1, curses.color_pair(5) | curses.A_BOLD) + header_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(">> ", curses.A_BOLD) + prompt_win.refresh() + curses.echo() + try: + user_input = prompt_win.getstr().decode("utf-8").strip() + except Exception: + continue + curses.noecho() + + if not user_input: + continue + + add_to_chat("Ty: " + user_input) + + if user_input.lower() in ["vypni", "exit"]: + add_to_chat("Ukončuji Robovojtíka...") + time.sleep(1) + break + + if user_input.lower() == "automat": + global automode + automode = not automode + stav = "zapnut" if automode else "vypnut" + add_to_chat(f"Automód byl nyní {stav}.") + continue + + if user_input.lower().startswith("skript:"): + try: + _, rest = user_input.split("skript:", 1) + parts = rest.split(";", 1) + if len(parts) < 2: + add_to_chat("Pro vytvoření skriptu uveďte název a popis oddělené středníkem.") + continue + file_name = parts[0].strip() + description = parts[1].strip() + add_to_chat(f"Vytvářím skript '{file_name}' na základě popisu: {description}") + generated_content = api_interface.posli_dotaz_do_assistenta("Vygeneruj skript podle popisu: " + description) + add_to_chat("Generovaný obsah skriptu:\n" + generated_content) + result = shell_functions.vytvor_skript(file_name, generated_content) + add_to_chat(result) + except Exception as e: + add_to_chat(f"Chyba při vytváření skriptu: {str(e)}") + continue + + if user_input.startswith("cmd:"): + command = user_input[4:].strip() + 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): ", curses.A_BOLD) + prompt_win.refresh() + potvrzeni = prompt_win.getstr().decode("utf-8").strip().lower() + add_to_chat("Ty: " + potvrzeni) + if potvrzeni != "y": + add_to_chat("Příkaz nebyl spuštěn.") + continue + output, response = shell_functions.run_command_locally_and_report(command, api_interface) + output_win.erase() + output_win.addstr(1, 1, output) + output_win.box() + output_win.refresh() + add_to_chat("Robovojtík odpovídá:\n" + response) + continue + + assistant_response = api_interface.volani_asistenta(user_input, spinner_func=spinner_func) + add_to_chat("Robovojtík odpovídá:\n" + assistant_response) + + if assistant_response.strip().lower().startswith("navrhovaný příkaz:"): + lines = assistant_response.splitlines() + proposal_line = lines[0] + 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): ", curses.A_BOLD) + prompt_win.refresh() + potvrzeni = prompt_win.getstr().decode("utf-8").strip().lower() + add_to_chat("Ty: " + potvrzeni) + if potvrzeni != "y": + add_to_chat("Příkaz nebyl spuštěn.") + continue + output, response = shell_functions.run_command_locally_and_report(navrhovany_prikaz, api_interface) + output_win.erase() + output_win.addstr(1, 1, output) + output_win.box() + output_win.refresh() + add_to_chat("Robovojtík odpovídá:\n" + response) + + prompt_win.erase() + prompt_win.refresh() + +def main_ui(): + curses.wrapper(main_curses)