Odstranit robovojtik.py

robovojtik.py

@@ -1,411 +0,0 @@
-#!/usr/bin/env python3
-"""
-Robovojtík – Linuxový shell asistent s vylepšeným interaktivním rozhraním
-
-Funkce:
-  • 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
-import subprocess
-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()
-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
-log_enabled = False    # Zapnutí logování přes --log
-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",
-        "description": "Spustí shellový příkaz a vrátí jeho výstup.",
-        "parameters": {
-            "type": "object",
-            "properties": {
-                "command": {
-                    "type": "string",
-                    "description": "Shellový příkaz k vykonání."
-                }
-            },
-            "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. (System prompt není načítán ze zdrojového kódu) ===
-
-# === 4. Pomocné funkce pro komunikaci s OpenAI API ===
-
-def vytvor_nove_vlakno():
-    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):
-    return command.replace("`", "").strip()
-
-def volani_asistenta(prompt, spinner_func=None):
-    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):
-    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):
-    command = clean_command(command)
-    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 is_command_response(response):
-    return response.strip().lower().startswith("navrhovaný příkaz:")
-
-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):
-    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):
-    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):
-    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. 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()
-
-    header_height = 3
-    prompt_height = 2
-    left_width = width // 2
-    right_width = width - left_width - 1   # oddělovač má 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 '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()
-
-    # Funkce pro spinner v hlavičce (uprostřed)
-    def spinner_func(ch):
-        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()
-
-    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:
-            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)
-                nazev, obsah = rest.split(";", 1)
-                nazev = nazev.strip()
-                obsah = obsah.strip()
-                vysledek = vytvor_skript(nazev, obsah)
-                add_to_chat(vysledek)
-            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 = run_command_locally_and_report(command)
-            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 = volani_asistenta(user_input, spinner_func=spinner_func)
-        add_to_chat("Robovojtík odpovídá:\n" + assistant_response)
-
-        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): ", 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 = run_command_locally_and_report(navrhovany_prikaz)
-            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():
-    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)
-        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()