420 lines
15 KiB
Python
420 lines
15 KiB
Python
#!/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 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: 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",
|
||
"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, vše se spravuje externě) ===
|
||
|
||
# === 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:
|
||
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í 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:
|
||
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):
|
||
"""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()
|
||
|
||
# === 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()
|
||
|
||
while True:
|
||
prompt_win.erase()
|
||
prompt_win.addstr(">> ")
|
||
prompt_win.refresh()
|
||
curses.echo()
|
||
try:
|
||
user_input = prompt_win.getstr().decode("utf-8").strip()
|
||
except:
|
||
continue
|
||
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()
|
||
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
|
||
|
||
# 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()
|