začátek nové verze
This commit is contained in:
parent
9003d676ac
commit
d619fc0f3d
137
api_interface.py
Normal file
137
api_interface.py
Normal file
@ -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', "")
|
||||
@ -1,3 +1,3 @@
|
||||
[OpenAI]
|
||||
api_key =
|
||||
assistant_id =
|
||||
assistant_id =
|
||||
|
||||
35
main.py
Normal file
35
main.py
Normal file
@ -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()
|
||||
55
shell_functions.py
Normal file
55
shell_functions.py
Normal file
@ -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
|
||||
190
ui.py
Normal file
190
ui.py
Normal file
@ -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)
|
||||
Loading…
x
Reference in New Issue
Block a user