Oprava, po které to ale blbne

Po příkazu se ptá stále dokola
This commit is contained in:
sinuhet 2025-03-22 23:24:23 +01:00
parent 49d93096d4
commit 2e2a18272a
5 changed files with 315 additions and 159 deletions

View File

@ -14,16 +14,17 @@ from configparser import ConfigParser
logger = logging.getLogger("robovojtik.api_interface") logger = logging.getLogger("robovojtik.api_interface")
config = ConfigParser() config = ConfigParser()
config.read("config.ini") config.read("config.ini")
try: try:
openai.api_key = config["OpenAI"]["api_key"] openai.api_key = config["OpenAI"]["api_key"]
except KeyError: 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.") raise ValueError("API key not found in config.ini. Zadejte 'api_key' do [OpenAI].")
ASSISTANT_ID = config["OpenAI"].get("assistant_id", None) ASSISTANT_ID = config["OpenAI"].get("assistant_id", None)
if not ASSISTANT_ID: if not ASSISTANT_ID:
raise ValueError("assistant_id not found in config.ini. Please set the 'assistant_id' under [OpenAI] section.") raise ValueError("assistant_id not found in config.ini. Zadejte 'assistant_id' do [OpenAI].")
# Globální proměnná pro uchování ID konverzačního vlákna # Globální vlákno konverzace
thread_id = None thread_id = None
def vytvor_nove_vlakno(): def vytvor_nove_vlakno():
@ -31,19 +32,26 @@ def vytvor_nove_vlakno():
Vytvoří nové vlákno konverzace s asistentem pomocí OpenAI API. Vytvoří nové vlákno konverzace s asistentem pomocí OpenAI API.
""" """
global thread_id global thread_id
thread = openai.beta.threads.create() try:
thread_id = thread.id thread = openai.beta.threads.create()
logger.debug(f"Vytvořeno nové vlákno: {thread_id}") thread_id = thread.id
return thread_id logger.debug(f"Vytvořeno nové vlákno: {thread_id}")
return thread_id
except Exception as e:
logger.exception("Chyba při vytváření nového vlákna konverzace.")
raise
def posli_dotaz_do_assistenta(prompt): def posli_dotaz_do_assistenta(prompt):
""" """
Odesílá dotaz v přirozeném jazyce do asistenta a vrací jeho odpověď. Odesílá dotaz v přirozeném jazyce do asistenta a vrací jeho odpověď.
Ošetřuje stav "queued", "running", "pending" i "in_progress" tak, aby cyklus čekal na dokončení.
Vrací poslední zprávu od asistenta.
""" """
global thread_id global thread_id
if thread_id is None: if thread_id is None:
vytvor_nove_vlakno() vytvor_nove_vlakno()
logger.debug(f"Odesílám dotaz: {prompt}") logger.debug(f"Odesílám dotaz: {prompt}")
openai.beta.threads.messages.create( openai.beta.threads.messages.create(
thread_id=thread_id, thread_id=thread_id,
role="user", role="user",
@ -55,20 +63,75 @@ def posli_dotaz_do_assistenta(prompt):
) )
while True: while True:
run_status = openai.beta.threads.runs.retrieve(thread_id=thread_id, run_id=run.id) run_status = openai.beta.threads.runs.retrieve(thread_id=thread_id, run_id=run.id)
if run_status.status == "completed": if run_status.status == "requires_action":
handle_required_action(run_status, run.id, thread_id)
elif run_status.status == "completed":
break break
time.sleep(1) elif run_status.status in ["queued", "running", "pending", "in_progress"]:
time.sleep(1)
else:
logger.warning(f"Neočekávaný stav run_status: {run_status.status}")
time.sleep(1)
break
messages = openai.beta.threads.messages.list(thread_id=thread_id) messages = openai.beta.threads.messages.list(thread_id=thread_id)
answer = messages.data[0].content[0].text.value answer = ""
for msg in reversed(messages.data):
if msg.role == "assistant" and msg.content and msg.content[0].text:
answer = msg.content[0].text.value
break
logger.debug(f"Asistent odpověděl: {answer}") logger.debug(f"Asistent odpověděl: {answer}")
return answer return answer
def handle_required_action(run_status, run_id, thread_id):
"""
Vyřídí volání funkcí, které asistent vyžaduje (execute_shell_command, create_script, read_file).
"""
from shell_functions import spust_prikaz, vytvor_skript, read_file_content
try:
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
logger.debug(f"Asistent volá funkci: {tool_name}")
try:
arguments = json.loads(tool_call.function.arguments)
except Exception as e:
logger.exception("Chyba při parsování argumentů pro funkci.")
tool_outputs.append({
"tool_call_id": tool_call.id,
"output": "Chyba: Nepodařilo se přečíst argumenty."
})
continue
if tool_name == "execute_shell_command":
vysledek = spust_prikaz(arguments["command"])
elif tool_name == "create_script":
vysledek = vytvor_skript(arguments["file_name"], arguments["content"])
elif tool_name == "read_file":
vysledek = read_file_content(arguments["path"])
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
)
except Exception as e:
logger.exception("Chyba při handle_required_action.")
def posli_prikaz_do_assistenta(command): def posli_prikaz_do_assistenta(command):
""" """
Odesílá schválený příkaz k vykonání do asistenta a vrací jeho odpověď. Odesílá schválený příkaz k vykonání do asistenta (např. cmd:).
Před odesláním se příkaz vyčistí. Ošetřuje stavy podobně jako posli_dotaz_do_assistenta.
""" """
from shell_functions import spust_prikaz, vytvor_skript # Importujeme zde, abychom předešli cyklickým závislostem.
global thread_id global thread_id
if thread_id is None: if thread_id is None:
vytvor_nove_vlakno() vytvor_nove_vlakno()
@ -81,38 +144,28 @@ def posli_prikaz_do_assistenta(command):
while True: while True:
run_status = openai.beta.threads.runs.retrieve(thread_id=thread_id, run_id=run.id) run_status = openai.beta.threads.runs.retrieve(thread_id=thread_id, run_id=run.id)
if run_status.status == "requires_action": if run_status.status == "requires_action":
tool_calls = run_status.required_action.submit_tool_outputs.tool_calls handle_required_action(run_status, run.id, thread_id)
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": elif run_status.status == "completed":
break break
time.sleep(1) elif run_status.status in ["queued", "running", "pending", "in_progress"]:
time.sleep(1)
else:
logger.warning(f"Neočekávaný stav run_status: {run_status.status}")
time.sleep(1)
break
messages = openai.beta.threads.messages.list(thread_id=thread_id) messages = openai.beta.threads.messages.list(thread_id=thread_id)
answer = messages.data[0].content[0].text.value answer = ""
for msg in reversed(messages.data):
if msg.role == "assistant" and msg.content and msg.content[0].text:
answer = msg.content[0].text.value
break
logger.debug(f"Výsledek spuštěného příkazu: {answer}") logger.debug(f"Výsledek spuštěného příkazu: {answer}")
return answer return answer
def volani_asistenta(prompt, spinner_func=None): def volani_asistenta(prompt, spinner_func=None):
""" """
Spustí dotaz do asistenta v samostatném vlákně a během čekání volá spinner_func. 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. Vrací odpověď asistenta nebo chybovou hlášku.
""" """
result_container = {} result_container = {}
@ -135,3 +188,10 @@ def volani_asistenta(prompt, spinner_func=None):
idx += 1 idx += 1
thread.join() thread.join()
return result_container.get('answer', "") return result_container.get('answer', "")
def reset_thread():
"""
Resetuje konverzační vlákno tím, že nastaví globální thread_id na None.
"""
global thread_id
thread_id = None

68
main.py
View File

@ -3,33 +3,97 @@ Module: main.py
--------------- ---------------
Hlavní spouštěcí skript pro Robovojtíka. Hlavní spouštěcí skript pro Robovojtíka.
Importuje moduly pro API komunikaci, shellové funkce a uživatelské rozhraní. Importuje moduly pro API komunikaci, shellové funkce a uživatelské rozhraní.
Zde také definujeme seznam OpenAI funkcí (FUNCTIONS) včetně read_file.
""" """
import argparse import argparse
import queue import queue
import logging import logging
import logging.handlers import logging.handlers
import ui import ui
# Definice OpenAI funkcí, které asistent může volat
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": "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
}
},
{
"name": "read_file",
"description": "Načte soubor o maximální velikosti 10kB a vrátí jeho obsah (případně zkrácený).",
"parameters": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Cesta k souboru, který se má přečíst."
}
},
"required": ["path"],
"additionalProperties": False
}
}
]
def main(): def main():
parser = argparse.ArgumentParser(description="Robovojtík interaktivní shell asistent") parser = argparse.ArgumentParser(description="Robovojtík interaktivní shell asistent")
parser.add_argument("--log", action="store_true", help="Zapne logování do souboru robovojtik.log") parser.add_argument("--log", action="store_true", help="Zapne logování do souboru robovojtik.log")
args = parser.parse_args() args = parser.parse_args()
listener = None listener = None
if args.log: if args.log:
import logging.handlers
log_queue = queue.Queue(-1) log_queue = queue.Queue(-1)
logger = logging.getLogger("robovojtik") logger = logging.getLogger("robovojtik")
logger.setLevel(logging.DEBUG) logger.setLevel(logging.DEBUG)
qh = logging.handlers.QueueHandler(log_queue) qh = logging.handlers.QueueHandler(log_queue)
logger.addHandler(qh) logger.addHandler(qh)
fh = logging.FileHandler("robovojtik.log") fh = logging.FileHandler("robovojtik.log", mode="a", encoding="utf-8")
fh.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")) fh.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(name)s - %(message)s"))
listener = logging.handlers.QueueListener(log_queue, fh) listener = logging.handlers.QueueListener(log_queue, fh)
listener.start() listener.start()
logger.debug("Logování zapnuto.") logger.debug("Logování zapnuto.")
# Předáme seznam FUNCTIONS do UI
ui.set_functions(FUNCTIONS)
ui.main_ui() ui.main_ui()
if listener: if listener:
listener.stop() listener.stop()
if __name__ == "__main__": if __name__ == "__main__":
main() main()

View File

@ -2,74 +2,45 @@
Module: markdown_parser.py Module: markdown_parser.py
---------------------------- ----------------------------
Rozšířený Markdown parser pro curses. Rozšířený Markdown parser pro curses.
...
Podporované funkce:
1. Nadpisy (řádky začínající #) -> zobrazeny tučně a podtrženě
2. Odrážky (řádky začínající - a *) -> zobrazeny s tučným textem a symbolickou tečkou
3. Víceřádkové bloky kódu (```):
vše mezi trojitými backticky je zobrazeno s curses.A_REVERSE
4. Inline formátování:
**text** -> curses.A_BOLD
*text* -> curses.A_UNDERLINE
`text` -> curses.A_REVERSE
Poznámka: nepodporujeme vnořené značky, a parser je pouze zjednodušený.
""" """
import re import re
import curses import curses
def parse_markdown(text): def parse_markdown(text):
"""
Hlavní funkce pro rozparsování zadaného textu na seznam (segment_text, attribute).
Respektuje víceřádkové bloky kódu (```), nadpisy, odrážky a volá parse_inline pro
zpracování běžných řádků.
"""
segments = [] segments = []
lines = text.splitlines() lines = text.splitlines()
in_code_block = False # Indikátor, zda se nacházíme uvnitř bloku ```. in_code_block = False
for line in lines: for line in lines:
stripped = line.strip() stripped = line.strip()
# Kontrola začátku/konce bloku kódu (```).
if stripped.startswith("```"): if stripped.startswith("```"):
# Přepínáme stav
in_code_block = not in_code_block in_code_block = not in_code_block
if in_code_block:
# Řádek obsahující samotné ```
# Pokud tam je něco navíc, ořežeme
code_block_marker = stripped[3:].strip()
# Můžeme sem zařadit logiku pro detekci jazyka, pokud by to bylo potřeba
else:
# Konec bloku kódu
pass
continue continue
if in_code_block: if in_code_block:
# Ve víceřádkovém kódu každou linku zobrazíme s curses.A_REVERSE # celé řádky ve vícerádkovém kódu
segments.append((line + "\n", curses.A_REVERSE)) segments.append((line + "\n", curses.A_REVERSE))
continue continue
# Pokud nejsme v bloku kódu, zpracujeme nadpisy, odrážky a inline formát. # Nadpisy
if stripped.startswith("#"): if stripped.startswith("#"):
# Nadpis: zjistíme, kolik # tam je
hash_count = 0 hash_count = 0
for ch in line: for ch in line:
if ch == '#': if ch == '#':
hash_count += 1 hash_count += 1
else: else:
break break
# Obsah nadpisu za #...
content = line[hash_count:].strip() content = line[hash_count:].strip()
# Atribut: tučný a podtržený
segments.append((content + "\n", curses.A_BOLD | curses.A_UNDERLINE)) segments.append((content + "\n", curses.A_BOLD | curses.A_UNDERLINE))
# Odrážky
elif stripped.startswith("-") or stripped.startswith("*"): elif stripped.startswith("-") or stripped.startswith("*"):
# Odrážka
content = line[1:].strip() content = line[1:].strip()
segments.append(("" + content + "\n", curses.A_BOLD)) segments.append(("" + content + "\n", curses.A_BOLD))
else: else:
# Běžný řádek -> inline formát # Běžný řádek
inline_segments = parse_inline(line) inline_segments = parse_inline(line)
for seg_text, seg_attr in inline_segments: for seg_text, seg_attr in inline_segments:
segments.append((seg_text, seg_attr)) segments.append((seg_text, seg_attr))
@ -79,14 +50,8 @@ def parse_markdown(text):
def parse_inline(text): def parse_inline(text):
"""
Rozparsuje inline Markdown značky (tučný, kurzíva, inline kód) v jednom řádku,
vrací seznam (text, attr).
"""
segments = [] segments = []
pos = 0 pos = 0
# Regulární výraz pro zachycení **text**, *text*, `text`
pattern = re.compile(r'(\*\*.*?\*\*|\*.*?\*|`.*?`)') pattern = re.compile(r'(\*\*.*?\*\*|\*.*?\*|`.*?`)')
for match in pattern.finditer(text): for match in pattern.finditer(text):
@ -110,19 +75,3 @@ def parse_inline(text):
if pos < len(text): if pos < len(text):
segments.append((text[pos:], 0)) segments.append((text[pos:], 0))
return segments return segments
if __name__ == "__main__":
# Krátký test
sample = """# Nadpis 1
Toto je normální text s **tučným** a *kurzívou*.
- Odrážka 1
* Odrážka 2
A teď blok kódu:
ls -la echo "Hello"
Další text s `inline kódem` a ## menším nadpisem?
## Nadpis 2
"""
segs = parse_markdown(sample)
for seg, attr in segs:
# Ukázkové vypsání v terminálu (bez curses) pro debug
print(f"{repr(seg)} attr={attr}")

View File

@ -2,13 +2,17 @@
Module: shell_functions.py Module: shell_functions.py
---------------------------- ----------------------------
Obsahuje funkce pro vykonávání lokálních shellových příkazů a tvorbu skriptů. Obsahuje funkce pro vykonávání lokálních shellových příkazů a tvorbu skriptů.
Přidána funkce read_file_content pro načtení souboru.
""" """
import subprocess import subprocess
import logging import logging
import os
logger = logging.getLogger("robovojtik.shell_functions") logger = logging.getLogger("robovojtik.shell_functions")
MAX_FILE_SIZE = 10 * 1024 # 10 kB
def clean_command(command): def clean_command(command):
""" """
Odstraní backticky a nepotřebné mezery z příkazu. Odstraní backticky a nepotřebné mezery z příkazu.
@ -25,31 +29,50 @@ def spust_prikaz(command):
output = subprocess.check_output(command, shell=True, stderr=subprocess.STDOUT, universal_newlines=True) output = subprocess.check_output(command, shell=True, stderr=subprocess.STDOUT, universal_newlines=True)
return output.strip() return output.strip()
except subprocess.CalledProcessError as e: except subprocess.CalledProcessError as e:
logger.exception("Chyba při vykonávání příkazu.")
return f"Chyba při vykonávání příkazu:\n{e.output}" return f"Chyba při vykonávání příkazu:\n{e.output}"
except Exception as e:
logger.exception("Obecná chyba při vykonávání příkazu.")
return f"Chyba při vykonávání příkazu: {str(e)}"
def vytvor_skript(nazev, obsah): def vytvor_skript(nazev, obsah):
""" """
Zapíše obsah do souboru s daným názvem a nastaví jej jako spustitelný. Zapíše obsah do souboru s daným názvem a nastaví jej jako spustitelný.
""" """
try: try:
with open(nazev, "w") as f: with open(nazev, "w", encoding="utf-8") as f:
f.write(obsah) f.write(obsah)
subprocess.call(f"chmod +x {nazev}", shell=True) subprocess.call(f"chmod +x {nazev}", shell=True)
logger.debug(f"Skript {nazev} vytvořen a nastaven jako spustitelný.") logger.debug(f"Skript {nazev} vytvořen a nastaven jako spustitelný.")
return f"Skript {nazev} byl úspěšně vytvořen." return f"Skript {nazev} byl úspěšně vytvořen."
except Exception as e: except Exception as e:
logger.exception("Chyba při tvorbě skriptu.")
return f"Nastala chyba při tvorbě skriptu: {str(e)}" return f"Nastala chyba při tvorbě skriptu: {str(e)}"
def run_command_locally_and_report(command, api_interface): def read_file_content(path: str) -> str:
""" """
Spustí příkaz lokálně, odešle jeho výstup jako report (s prefixem) Načte soubor na adrese 'path' do max. 10 kB.
do asistenta a vrátí tuple (output, assistant_response). Pokud je soubor větší, vrátí jen 10 kB a poznámku.
"""
try:
if not os.path.isfile(path):
return f"Chyba: Soubor '{path}' neexistuje nebo není souborem."
size = os.path.getsize(path)
partial = (size > MAX_FILE_SIZE)
with open(path, "rb") as f:
data = f.read(MAX_FILE_SIZE)
text = data.decode("utf-8", errors="replace")
if partial:
text += "\n\n[POZOR: Soubor byl delší než 10 kB, zobrazujeme pouze prvních 10 kB!]"
return text
except Exception as e:
logger.exception("Chyba při čtení souboru")
return f"Chyba při čtení souboru: {str(e)}"
def run_command_locally(command):
"""
Spustí příkaz lokálně a vrátí jeho výstup.
""" """
command = clean_command(command) command = clean_command(command)
output = spust_prikaz(command) output = spust_prikaz(command)
if not output.startswith("Chyba při vykonávání příkazu:"): return output
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

166
ui.py
View File

@ -2,14 +2,15 @@
Module: ui.py Module: ui.py
------------- -------------
Obsahuje interaktivní rozhraní Robovojtíka založené na knihovně curses. Obsahuje interaktivní rozhraní Robovojtíka založené na knihovně curses.
Implementuje barevné oddělení, trvalou hlavičku, historii chatu s formátováním Markdown, Implementuje barevné oddělení, trvalou hlavičku, historii chatu,
spinner pro indikaci čekání a vstupní oblast. spinner pro indikaci čekání a vstupní oblast. Také prefix 'readfile:'.
""" """
import curses import curses
import time import time
import threading import threading
import logging import logging
import re
import api_interface import api_interface
import shell_functions import shell_functions
@ -17,11 +18,19 @@ import markdown_parser
logger = logging.getLogger("robovojtik.ui") logger = logging.getLogger("robovojtik.ui")
# Definujeme globální proměnnou automode # Globální funkce pro integraci s main.py
FUNCTIONS = []
def set_functions(funcs):
global FUNCTIONS
FUNCTIONS = funcs
automode = False automode = False
def main_curses(stdscr): def main_curses(stdscr):
# Inicializace barev
curses.start_color() curses.start_color()
curses.use_default_colors() curses.use_default_colors()
curses.init_pair(1, curses.COLOR_WHITE, curses.COLOR_BLUE) # Hlavička curses.init_pair(1, curses.COLOR_WHITE, curses.COLOR_BLUE) # Hlavička
@ -39,7 +48,7 @@ def main_curses(stdscr):
header_height = 3 header_height = 3
prompt_height = 2 prompt_height = 2
left_width = width // 2 left_width = width // 2
right_width = width - left_width - 1 # oddělovač right_width = width - left_width - 1
chat_height = height - header_height - prompt_height chat_height = height - header_height - prompt_height
header_win = curses.newwin(header_height, left_width, 0, 0) header_win = curses.newwin(header_height, left_width, 0, 0)
@ -64,8 +73,7 @@ def main_curses(stdscr):
header_text = [ header_text = [
"Vítejte v Robovojtikovi!", "Vítejte v Robovojtikovi!",
"Zadejte dotaz, příkaz (prefix 'cmd:'), 'automat',", "Zadejte dotaz, příkaz (prefix 'cmd:'), 'automat', 'readfile:' apod.",
"nebo napiš 'napiš mi skript, ...' pro generování skriptu.",
"Pro ukončení zadejte 'vypni' nebo 'exit'." "Pro ukončení zadejte 'vypni' nebo 'exit'."
] ]
max_chars = max(0, left_width - 2) max_chars = max(0, left_width - 2)
@ -82,31 +90,34 @@ def main_curses(stdscr):
output_win.box() output_win.box()
output_win.refresh() output_win.refresh()
spinner_chars = ["|", "/", "-", "\\"]
def spinner_func(ch): def spinner_func(ch):
mid_x = left_width // 2 - 5 mid_x = left_width // 2 - 5
# Vypíšeme spinner do hlavičky
try: try:
header_win.addnstr(1, mid_x, f"Čekám... {ch}", left_width - mid_x - 1, curses.color_pair(5) | curses.A_BOLD) 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() header_win.refresh()
except curses.error: except curses.error:
pass pass
chat_history = []
def add_to_chat(text): def add_to_chat(text):
# Nové volání parseru parse_markdown z markdown_parser
segments = markdown_parser.parse_markdown(text) segments = markdown_parser.parse_markdown(text)
for seg_text, seg_attr in segments: for seg_text, seg_attr in segments:
chat_win.addstr(seg_text, seg_attr) try:
chat_win.addstr(seg_text, seg_attr)
except curses.error:
pass
chat_win.refresh() chat_win.refresh()
add_to_chat("Historie chatu:") add_to_chat("Historie chatu:")
global automode
while True: while True:
prompt_win.erase() prompt_win.erase()
prompt_win.addstr(">> ", curses.A_BOLD) try:
prompt_win.addstr(">> ", curses.A_BOLD)
except curses.error:
pass
prompt_win.refresh() prompt_win.refresh()
curses.echo() curses.echo()
try: try:
@ -120,83 +131,132 @@ def main_curses(stdscr):
add_to_chat("Ty: " + user_input) add_to_chat("Ty: " + user_input)
# Ukončení
if user_input.lower() in ["vypni", "exit"]: if user_input.lower() in ["vypni", "exit"]:
add_to_chat("Ukončuji Robovojtíka...") add_to_chat("Ukončuji Robovojtíka...")
time.sleep(1) time.sleep(1)
break break
# Přepnutí automódu
if user_input.lower() == "automat": if user_input.lower() == "automat":
global automode
automode = not automode automode = not automode
stav = "zapnut" if automode else "vypnut" stav = "zapnut" if automode else "vypnut"
add_to_chat(f"Automód byl nyní {stav}.") add_to_chat(f"Automód byl nyní {stav}.")
continue continue
if user_input.lower().startswith("skript:"): # readfile:
try: if user_input.lower().startswith("readfile:"):
_, rest = user_input.split("skript:", 1) path = user_input[9:].strip()
parts = rest.split(";", 1) add_to_chat(f"Načítám soubor: {path}")
if len(parts) < 2: from shell_functions import read_file_content
add_to_chat("Pro vytvoření skriptu uveďte název a popis oddělené středníkem.") content = read_file_content(path)
continue add_to_chat(f"Obsah souboru:\n{content}")
file_name = parts[0].strip() assistant_response = api_interface.volani_asistenta(
description = parts[1].strip() f"Analyzuj obsah tohoto souboru:\n{content}",
add_to_chat(f"Vytvářím skript '{file_name}' na základě popisu: {description}") spinner_func=spinner_func
generated_content = api_interface.posli_dotaz_do_assistenta("Vygeneruj skript podle popisu: " + description) )
add_to_chat("Generovaný obsah skriptu:\n" + generated_content) add_to_chat("Robovojtík odpovídá:\n" + assistant_response)
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 continue
# Přímý příkaz cmd:
if user_input.startswith("cmd:"): if user_input.startswith("cmd:"):
command = user_input[4:].strip() command = user_input[4:].strip()
add_to_chat(f"Rozpoznán přímý příkaz: {command}") add_to_chat(f"Rozpoznán příkaz: {command}")
if not automode: if not automode:
prompt_win.erase() prompt_win.erase()
prompt_win.addstr("Spustit tento příkaz? (y/n): ", curses.A_BOLD) try:
prompt_win.addstr("Spustit tento příkaz? (y/n): ", curses.A_BOLD)
except curses.error:
pass
prompt_win.refresh() prompt_win.refresh()
potvrzeni = prompt_win.getstr().decode("utf-8").strip().lower() try:
potvrzeni = prompt_win.getstr().decode("utf-8").strip().lower()
except Exception:
potvrzeni = ""
curses.flushinp() # vyprázdní vstupní buffer
add_to_chat("Ty: " + potvrzeni) add_to_chat("Ty: " + potvrzeni)
if potvrzeni != "y": if potvrzeni not in ("y", "yes", "ano"):
add_to_chat("Příkaz nebyl spuštěn.") add_to_chat("Příkaz nebyl spuštěn.")
continue continue
output, response = shell_functions.run_command_locally_and_report(command, api_interface) output = shell_functions.run_command_locally(command)
output_win.erase() output_win.erase()
output_win.addstr(1, 1, output) try:
output_win.box() lines = output.splitlines()
output_win.refresh() y = 1
add_to_chat("Robovojtík odpovídá:\n" + response) for line in lines:
output_win.addstr(y, 1, line)
y += 1
output_win.box()
output_win.refresh()
except curses.error:
pass
summary_prompt = (f"Analyzuj následující výstup příkazu. Neposkytuj žádné návrhy příkazů, "
f"jen shrň výsledky a uveď komentář:\n{output}")
assistant_summary = api_interface.volani_asistenta(summary_prompt, spinner_func=spinner_func)
add_to_chat("Robovojtík shrnuje výstup:\n" + assistant_summary)
continue continue
# Ostatní dotazy # Ostatní dotazy -> Asistent
assistant_response = api_interface.volani_asistenta(user_input, spinner_func=spinner_func) assistant_response = api_interface.volani_asistenta(user_input, spinner_func=spinner_func)
add_to_chat("Robovojtík odpovídá:\n" + assistant_response) add_to_chat("Robovojtík odpovídá:\n" + assistant_response)
if assistant_response.strip().lower().startswith("navrhovaný příkaz:"): # Pokud asistent navrhne příkaz, extrahujeme jej
match = re.search(r"`([^`]+)`", assistant_response)
if match:
navrhovany_prikaz = match.group(1)
elif assistant_response.strip().lower().startswith("navrhovaný příkaz:"):
lines = assistant_response.splitlines() lines = assistant_response.splitlines()
proposal_line = lines[0] proposal_line = lines[0]
navrhovany_prikaz = proposal_line[len("navrhovaný příkaz:"):].strip() navrhovany_prikaz = proposal_line[len("navrhovaný příkaz:"):].strip()
else:
navrhovany_prikaz = None
if navrhovany_prikaz:
add_to_chat(f"Navrhovaný příkaz: {navrhovany_prikaz}") add_to_chat(f"Navrhovaný příkaz: {navrhovany_prikaz}")
if not automode: if not automode:
prompt_win.erase() prompt_win.erase()
prompt_win.addstr("Spustit tento příkaz? (y/n): ", curses.A_BOLD) try:
prompt_win.addstr("Spustit tento příkaz? (y/n): ", curses.A_BOLD)
except curses.error:
pass
prompt_win.refresh() prompt_win.refresh()
potvrzeni = prompt_win.getstr().decode("utf-8").strip().lower() try:
potvrzeni = prompt_win.getstr().decode("utf-8").strip().lower()
except Exception:
potvrzeni = ""
curses.flushinp() # vyprázdní vstupní buffer
add_to_chat("Ty: " + potvrzeni) add_to_chat("Ty: " + potvrzeni)
if potvrzeni != "y": if potvrzeni not in ("y", "yes", "ano"):
add_to_chat("Příkaz nebyl spuštěn.") add_to_chat("Příkaz nebyl spuštěn.")
continue continue
output, response = shell_functions.run_command_locally_and_report(navrhovany_prikaz, api_interface) output = shell_functions.run_command_locally(navrhovany_prikaz)
output_win.erase() output_win.erase()
output_win.addstr(1, 1, output) try:
output_win.box() lines = output.splitlines()
output_win.refresh() y = 1
add_to_chat("Robovojtík odpovídá:\n" + response) for line in lines:
output_win.addstr(y, 1, line)
y += 1
output_win.box()
output_win.refresh()
except curses.error:
pass
summary_prompt = (
f"Výsledek příkazu:\n{output}\n\n"
"Prosím, stručně shrň výše uvedený výstup příkazu a "
"uveď pouze komentář bez návrhu dalších příkazů."
)
assistant_summary = api_interface.volani_asistenta(summary_prompt, spinner_func=spinner_func)
add_to_chat("Robovojtík shrnuje výstup:\n" + assistant_summary)
continue
prompt_win.erase() prompt_win.erase()
prompt_win.refresh() prompt_win.refresh()
def main_ui(): def main_ui():
curses.wrapper(main_curses) try:
curses.wrapper(main_curses)
except Exception as e:
logger.exception("Neočekávaná chyba v curses wrapperu.")
print(f"Chyba: {str(e)}")