Compare commits
19
Commits
3a042355b8
...
master
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
442c91ee66 | ||
|
|
858b469def | ||
|
|
2e4175f26b | ||
|
|
e72905fd78 | ||
|
|
aec5736394 | ||
|
|
e69a289361 | ||
|
|
b6e8864829 | ||
|
|
39b9288773 | ||
|
|
20233db282 | ||
|
|
c76beb764c | ||
|
|
6cb514ec13 | ||
|
|
5d9bd48013 | ||
|
|
453156148c | ||
|
|
e75a75e16c | ||
|
|
f1d1248d4b | ||
|
|
3197dedd4c | ||
|
|
918b1f3a5b | ||
|
|
c8e7489c26 | ||
|
|
82b70f88d1 |
No files matched your search
+293
@@ -0,0 +1,293 @@
|
||||
"""
|
||||
Terminal UI con scrollback usando Curses
|
||||
Scrollback vero con frecce e PgUp/PgDn
|
||||
"""
|
||||
import sys
|
||||
import signal
|
||||
import curses
|
||||
from typing import Optional, Callable
|
||||
|
||||
try:
|
||||
import curses
|
||||
CURSES_AVAILABLE = True
|
||||
except ImportError:
|
||||
CURSES_AVAILABLE = False
|
||||
|
||||
|
||||
class CursesUI:
|
||||
"""
|
||||
Terminal UI con scrollback usando curses.
|
||||
|
||||
Features:
|
||||
- Scrollback completo con PgUp/PgDn
|
||||
- Frecce su/giù per navigare
|
||||
- Resize automatico
|
||||
- Colors (se supportati)
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
scrollback: int = 10000,
|
||||
prompt: str = "❯ "
|
||||
):
|
||||
"""
|
||||
Inizializza il terminale UI con curses.
|
||||
|
||||
Args:
|
||||
scrollback: Numero di righe nello scrollback buffer
|
||||
prompt: Prompt da mostrare
|
||||
"""
|
||||
if not CURSES_AVAILABLE:
|
||||
raise ImportError("curses non è disponibile su questo sistema")
|
||||
|
||||
self.scrollback = scrollback
|
||||
self.prompt = prompt
|
||||
|
||||
# Buffer per scrollback
|
||||
self.lines: list = []
|
||||
|
||||
# Callback per l'agent
|
||||
self.agent_callback: Optional[Callable[[str], str]] = None
|
||||
|
||||
def _init_colors(self, stdscr):
|
||||
"""Inizializza i colori"""
|
||||
if curses.has_colors():
|
||||
curses.start_color()
|
||||
curses.use_default_colors()
|
||||
# Colori: 1=cyan, 2=green, 3=yellow, 4=red, 5=magenta, 6=blue
|
||||
curses.init_pair(1, curses.COLOR_CYAN, -1) # Header
|
||||
curses.init_pair(2, curses.COLOR_GREEN, -1) # Agent output
|
||||
curses.init_pair(3, curses.COLOR_YELLOW, -1) # Thinking
|
||||
curses.init_pair(4, curses.COLOR_RED, -1) # Error
|
||||
curses.init_pair(5, curses.COLOR_MAGENTA, -1) # Prompt
|
||||
curses.init_pair(6, curses.COLOR_BLUE, -1) # User input
|
||||
|
||||
def _add_line(self, text: str, color: int = 2):
|
||||
"""Aggiunge una linea al buffer"""
|
||||
self.lines.append((text, color))
|
||||
# Limita la dimensione del buffer
|
||||
if len(self.lines) > self.scrollback:
|
||||
self.lines.pop(0)
|
||||
|
||||
def _draw(self, stdscr):
|
||||
"""Disegna lo schermo"""
|
||||
stdscr.clear()
|
||||
|
||||
# Ottieni dimensioni
|
||||
max_y, max_x = stdscr.getmaxyx()
|
||||
|
||||
# Header
|
||||
header = "═" * (max_x - 4)
|
||||
stdscr.addstr(0, 0, f"╔{header}╗", curses.color_pair(1) | curses.A_BOLD)
|
||||
title = " LangCode Agent - Curses UI "
|
||||
stdscr.addstr(1, (max_x - len(title)) // 2, title, curses.color_pair(1) | curses.A_BOLD)
|
||||
stdscr.addstr(2, 0, f"╚{header}╝", curses.color_pair(1) | curses.A_BOLD)
|
||||
|
||||
# Calcola area visibile
|
||||
start_row = 4
|
||||
visible_rows = max_y - start_row - 2 # -2 per status e prompt
|
||||
|
||||
# Mostra linee visibili
|
||||
total_lines = len(self.lines)
|
||||
|
||||
# Scroll offset (0 = showing latest)
|
||||
# La scrolling position va da 0 a max(0, total_lines - visible_rows)
|
||||
self.scroll_offset = max(0, total_lines - visible_rows)
|
||||
|
||||
# Disegna le linee visibili
|
||||
for i in range(min(visible_rows, total_lines)):
|
||||
line_idx = self.scroll_offset + i
|
||||
if line_idx < total_lines:
|
||||
text, color = self.lines[line_idx]
|
||||
# Tronca se troppo lunga
|
||||
display_text = text[:max_x - 2] if len(text) > max_x - 2 else text
|
||||
try:
|
||||
stdscr.addstr(start_row + i, 0, display_text, curses.color_pair(color))
|
||||
except curses.error:
|
||||
pass
|
||||
|
||||
# Status bar
|
||||
status_y = max_y - 2
|
||||
scroll_info = f"Linee: {total_lines} | Scroll: {self.scroll_offset} | PgUp/PgDn=Frecce=Scroll | q=quit | Ctrl+C=exit"
|
||||
stdscr.addstr(status_y, 0, "─" * (max_x - len(scroll_info) - 1), curses.A_DIM)
|
||||
stdscr.addstr(status_y, max_x - len(scroll_info) - 1, scroll_info, curses.A_DIM)
|
||||
|
||||
# Prompt
|
||||
stdscr.addstr(max_y - 1, 0, f"\033[1;35m{self.prompt}\033[0m ", curses.color_pair(5) | curses.A_BOLD)
|
||||
|
||||
stdscr.refresh()
|
||||
|
||||
def _handle_input(self, stdscr):
|
||||
"""Gestisce l'input"""
|
||||
curses.curs_set(1) # Mostra cursore
|
||||
|
||||
# Abilita arrow keys e PgUp/PgDn
|
||||
stdscr.keypad(True)
|
||||
|
||||
# Configura timeout per read non-blocking
|
||||
stdscr.timeout(100) # 100ms timeout
|
||||
|
||||
input_buffer = ""
|
||||
while True:
|
||||
self._draw(stdscr)
|
||||
|
||||
try:
|
||||
key = stdscr.getch()
|
||||
except curses.error:
|
||||
continue
|
||||
|
||||
if key == -1:
|
||||
# Timeout - continua
|
||||
continue
|
||||
|
||||
elif key in [curses.KEY_ENTER, 10, 13]:
|
||||
# Enter - processa comando
|
||||
if input_buffer.strip():
|
||||
return input_buffer.strip()
|
||||
input_buffer = ""
|
||||
|
||||
elif key in [curses.KEY_UP]:
|
||||
# Scroll up nel buffer
|
||||
if self.scroll_offset > 0:
|
||||
self.scroll_offset -= 1
|
||||
|
||||
elif key in [curses.KEY_DOWN]:
|
||||
# Scroll down nel buffer
|
||||
max_y, max_x = stdscr.getmaxyx()
|
||||
visible_rows = max_y - 6
|
||||
if self.scroll_offset < max(0, len(self.lines) - visible_rows):
|
||||
self.scroll_offset += 1
|
||||
|
||||
elif key in [curses.KEY_PPAGE]:
|
||||
# PgUp - scroll su di una pagina
|
||||
max_y, _ = stdscr.getmaxyx()
|
||||
self.scroll_offset = max(0, self.scroll_offset - (max_y - 6))
|
||||
|
||||
elif key in [curses.KEY_NPAGE]:
|
||||
# PgDn - scroll giu di una pagina
|
||||
max_y, max_x = stdscr.getmaxyx()
|
||||
visible_rows = max_y - 6
|
||||
self.scroll_offset = min(max(0, len(self.lines) - visible_rows),
|
||||
self.scroll_offset + visible_rows)
|
||||
|
||||
elif key == 27: # ESC
|
||||
return "/exit"
|
||||
|
||||
elif key in [3, 26]: # Ctrl+C
|
||||
return "/exit"
|
||||
|
||||
elif key in [curses.KEY_BACKSPACE, 127, 8]:
|
||||
# Backspace
|
||||
input_buffer = input_buffer[:-1]
|
||||
|
||||
elif 32 <= key <= 126:
|
||||
# Carattere stampabile
|
||||
input_buffer += chr(key)
|
||||
|
||||
# Mostra input corrente da qualche parte se necessario
|
||||
|
||||
def start(self, agent_callback: Callable[[str], str]):
|
||||
"""
|
||||
Avvia il terminale interattivo con curses.
|
||||
"""
|
||||
self.agent_callback = agent_callback
|
||||
|
||||
def handle_sigint(signum, frame):
|
||||
sys.exit(0)
|
||||
|
||||
old_handler = signal.signal(signal.SIGINT, handle_sigint)
|
||||
|
||||
try:
|
||||
curses.wrapper(self._main_loop)
|
||||
finally:
|
||||
signal.signal(signal.SIGINT, old_handler)
|
||||
|
||||
def _main_loop(self, stdscr):
|
||||
"""Loop principale curses"""
|
||||
self._init_colors(stdscr)
|
||||
curses.cbreak()
|
||||
curses.noecho()
|
||||
|
||||
# Aggiungi header
|
||||
self._add_line("╔═══════════════════════════════════════════════════════╗", 1)
|
||||
self._add_line("║ LangCode Agent - Curses UI con Scrollback ║", 1)
|
||||
self._add_line("╚═══════════════════════════════════════════════════════╝", 1)
|
||||
self._add_line("", 0)
|
||||
self._add_line("Scrivi un messaggio e premi Enter. Usa PgUp/PgDn per scrollare.", 6)
|
||||
self._add_line("Comandi: /help, /clear, /exit", 6)
|
||||
self._add_line("", 0)
|
||||
|
||||
while True:
|
||||
# Ottieni input
|
||||
stdscr.timeout(-1) # Bloccante per l'input
|
||||
command = self._handle_input(stdscr)
|
||||
|
||||
if command.lower() in ['/exit', '/quit']:
|
||||
break
|
||||
|
||||
if not command:
|
||||
continue
|
||||
|
||||
if command == '/help':
|
||||
self._add_line("╔═══════════════════════════════════════════════════════╗", 1)
|
||||
self._add_line("║ COMANDI ║", 1)
|
||||
self._add_line("╠═══════════════════════════════════════════════════════╣", 1)
|
||||
self._add_line("║ /help - Mostra questo messaggio ║", 1)
|
||||
self._add_line("║ /clear - Pulisci lo schermo ║", 1)
|
||||
self._add_line("║ /exit - Esci dall'agente ║", 1)
|
||||
self._add_line("║ PgUp - Scroll su ║", 1)
|
||||
self._add_line("║ PgDn - Scroll giu ║", 1)
|
||||
self._add_line("║ Frecce - Naviga nel scrollback ║", 1)
|
||||
self._add_line("╚═══════════════════════════════════════════════════════╝", 1)
|
||||
continue
|
||||
|
||||
if command == '/clear':
|
||||
self.lines.clear()
|
||||
continue
|
||||
|
||||
# Mostra comando utente
|
||||
self._add_line("", 0)
|
||||
self._add_line(f"👤 {command}", 6)
|
||||
|
||||
# Chiamata all'agent
|
||||
self._add_line("", 0)
|
||||
self._add_line("🤔 Pensando...", 3)
|
||||
|
||||
try:
|
||||
response = agent_callback(command)
|
||||
|
||||
# Rimuovi "thinking" e mostra risposta
|
||||
if self.lines and "🤔 Pensando..." in self.lines[-1][0]:
|
||||
self.lines.pop()
|
||||
|
||||
# Aggiungi risposta
|
||||
for line in response.split('\n'):
|
||||
self._add_line(f"🤖 {line}", 2)
|
||||
except Exception as e:
|
||||
self._add_line(f"❌ Errore: {str(e)}", 4)
|
||||
|
||||
self._add_line("", 0)
|
||||
|
||||
|
||||
def create_curses_ui(
|
||||
scrollback: int = 10000,
|
||||
prompt: str = "❯ "
|
||||
) -> CursesUI:
|
||||
"""Factory function per creare CursesUI"""
|
||||
return CursesUI(
|
||||
scrollback=scrollback,
|
||||
prompt=prompt
|
||||
)
|
||||
|
||||
|
||||
# Test standalone
|
||||
if __name__ == "__main__":
|
||||
def test_callback(text):
|
||||
return f"Ricevuto: {text}\n\nRisposta multi-linea:\n1. Prima linea\n2. Seconda\n3. Terza\n4. Quarta\n5. Quinta"
|
||||
|
||||
print("Avvio CursesUI di test...")
|
||||
ui = create_curses_ui()
|
||||
try:
|
||||
ui.start(test_callback)
|
||||
except KeyboardInterrupt:
|
||||
print("\nUscito")
|
||||
+10
-1
@@ -428,7 +428,6 @@ class EnhancedCLI:
|
||||
key_bindings=self.key_bindings,
|
||||
auto_suggest=AutoSuggestFromHistory(),
|
||||
multiline=True,
|
||||
cursor=CursorShape.BEAM,
|
||||
mouse_support=self.enable_mouse,
|
||||
)
|
||||
|
||||
@@ -656,6 +655,16 @@ class EnhancedCLI:
|
||||
Args:
|
||||
agent_callback: Function that takes user input and returns agent response
|
||||
"""
|
||||
import signal
|
||||
import sys
|
||||
|
||||
# Register signal handler for Ctrl+C
|
||||
def handle_sigint(signum, frame):
|
||||
print("\n\n👋 Arrivederci!")
|
||||
sys.exit(0)
|
||||
|
||||
signal.signal(signal.SIGINT, handle_sigint)
|
||||
|
||||
# Print welcome
|
||||
print(self.welcome_message)
|
||||
|
||||
|
||||
+182
-8
@@ -29,11 +29,43 @@ sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
VERSION = "1.0.0"
|
||||
|
||||
|
||||
def strip_ansi(text: str) -> str:
|
||||
"""Rimuove codici ANSI escape sequences per terminali semplici"""
|
||||
import re
|
||||
ansi_pattern = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])')
|
||||
return ansi_pattern.sub('', text)
|
||||
|
||||
|
||||
def cmd_chat(args):
|
||||
"""Avvia l'agent in modalità chat interattiva con EnhancedCLI"""
|
||||
"""Avvia l'agent in modalità chat interattiva"""
|
||||
import signal
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
from main import create_langcode_agent, process_command
|
||||
from langchain_core.messages import HumanMessage
|
||||
from enhanced_cli import create_enhanced_cli
|
||||
|
||||
# Signal handler for Ctrl+C
|
||||
def handle_sigint(signum, frame):
|
||||
print("\n\n👋 Arrivederci!")
|
||||
sys.exit(0)
|
||||
|
||||
# Register signal handler
|
||||
signal.signal(signal.SIGINT, handle_sigint)
|
||||
|
||||
# Thread-safe flag for thinking indicator
|
||||
thinking_active = threading.Event()
|
||||
thinking_active.set()
|
||||
|
||||
def thinking_indicator():
|
||||
"""Mostra indicatore di pensiero mentre aspetta"""
|
||||
dots = [" ", ". ", ".. ", "..."]
|
||||
idx = 0
|
||||
while thinking_active.is_set():
|
||||
print(f"\r🤖 Pensando{dots[idx % 4]} ", end="", flush=True)
|
||||
idx += 1
|
||||
time.sleep(0.5)
|
||||
print("\r" + " " * 20 + "\r", end="") # Clear line
|
||||
|
||||
print("🚀 Inizializzo LangCode Agent...")
|
||||
|
||||
@@ -73,12 +105,152 @@ def cmd_chat(args):
|
||||
except Exception as e:
|
||||
return f"Errore: {str(e)}"
|
||||
|
||||
# Usa EnhancedCLI con history e autocomplete
|
||||
cli = create_enhanced_cli(
|
||||
history_file="./.langcode/.cli_history",
|
||||
theme="default"
|
||||
)
|
||||
cli.run(agent_callback)
|
||||
# Default: CLI semplice, EnhancedCLI/TerminalUI con flag
|
||||
use_enhanced = getattr(args, 'enhanced', False)
|
||||
use_terminal = getattr(args, 'terminal', False)
|
||||
|
||||
if use_terminal:
|
||||
try:
|
||||
from terminal_ui import create_terminal_ui
|
||||
term = create_terminal_ui(
|
||||
scrollback=10000,
|
||||
prompt="❯ "
|
||||
)
|
||||
term.start(agent_callback)
|
||||
return
|
||||
except ImportError:
|
||||
print("⚠️ pyte non installato. Esegui: pip install pyte")
|
||||
print("📝 Uso CLI semplice...")
|
||||
except Exception as e:
|
||||
print(f"⚠️ TerminalUI errore: {e}")
|
||||
print("📝 Uso CLI semplice...")
|
||||
|
||||
if use_enhanced:
|
||||
import errno
|
||||
import fcntl
|
||||
import select
|
||||
import threading
|
||||
|
||||
def timeout_handler():
|
||||
print("\n⚠️ EnhancedCLI timeout - uso CLI semplice")
|
||||
|
||||
try:
|
||||
from enhanced_cli import create_enhanced_cli
|
||||
cli = create_enhanced_cli(
|
||||
history_file="./.langcode/.cli_history",
|
||||
theme="default"
|
||||
)
|
||||
|
||||
# Run with timeout detection
|
||||
timeout_seconds = 30
|
||||
timer = threading.Timer(timeout_seconds, timeout_handler)
|
||||
timer.daemon = True
|
||||
timer.start()
|
||||
|
||||
cli.run(agent_callback)
|
||||
timer.cancel()
|
||||
return
|
||||
except (KeyboardInterrupt, EOFError, OSError) as e:
|
||||
# EnhancedCLI hang - fall back to simple
|
||||
print(f"⚠️ EnhancedCLI bloccato: {e}")
|
||||
print("📝 Uso CLI semplice...")
|
||||
except Exception as e:
|
||||
print(f"⚠️ EnhancedCLI errore: {e}")
|
||||
print("📝 Uso CLI semplice...")
|
||||
|
||||
# CLI semplice con input()
|
||||
# Reset terminal state dopo eventuale EnhancedCLI fallito
|
||||
import subprocess
|
||||
try:
|
||||
subprocess.run(["stty", "sane"], check=False, capture_output=True)
|
||||
except:
|
||||
pass
|
||||
|
||||
print("\n" + "=" * 60)
|
||||
print(" CLI SEMPLICE - digita /exit per uscire")
|
||||
print("=" * 60 + "\n")
|
||||
|
||||
while True:
|
||||
try:
|
||||
sys.stdout.flush()
|
||||
sys.stdout.write("❯ ")
|
||||
sys.stdout.flush()
|
||||
user_input = sys.stdin.readline().strip()
|
||||
|
||||
if not user_input:
|
||||
continue
|
||||
|
||||
if user_input.lower() in ["/exit", "/quit"]:
|
||||
print("👋 Arrivederci!")
|
||||
break
|
||||
|
||||
if user_input.startswith("/"):
|
||||
if user_input == "/help":
|
||||
print("""
|
||||
Comandi disponibili:
|
||||
/help - Mostra questo messaggio
|
||||
/status - Mostra stato del sistema
|
||||
/exit - Esci
|
||||
""")
|
||||
continue
|
||||
elif user_input == "/status":
|
||||
print("Agent: in esecuzione")
|
||||
continue
|
||||
else:
|
||||
print(f"Comando sconosciuto: {user_input}")
|
||||
continue
|
||||
|
||||
# Mostra input utente
|
||||
print(f"👤 {user_input}\n")
|
||||
|
||||
# Chiamata all'agent CON indicatore di pensiero
|
||||
thinking_active.set()
|
||||
thinker = threading.Thread(target=thinking_indicator, daemon=True)
|
||||
thinker.start()
|
||||
|
||||
try:
|
||||
response = agent_callback(user_input)
|
||||
finally:
|
||||
thinking_active.clear()
|
||||
thinker.join(timeout=1)
|
||||
|
||||
# Strip ANSI codes per terminali semplici
|
||||
clean_response = strip_ansi(response)
|
||||
|
||||
# Output con scrollback via less
|
||||
lines = clean_response.split('\n')
|
||||
if len(lines) > 10:
|
||||
import tempfile
|
||||
from datetime import datetime
|
||||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
output_file = f".langcode/output_{timestamp}.txt"
|
||||
|
||||
# Save to file
|
||||
os.makedirs(".langcode", exist_ok=True)
|
||||
with open(output_file, "w") as f:
|
||||
f.write(clean_response)
|
||||
|
||||
print(f"🤖 Output lungo salvato in: {output_file}")
|
||||
print("💡 Visualizza con: less {0} (usa frecce per scrollare, q per uscire)\n".format(output_file))
|
||||
|
||||
# Try to show with less
|
||||
import subprocess
|
||||
try:
|
||||
subprocess.run(["less", "-R", output_file], check=False)
|
||||
except FileNotFoundError:
|
||||
# less not available, just cat first part
|
||||
print("🤖 " + "\n🤖 ".join(lines[:20]))
|
||||
if len(lines) > 20:
|
||||
print(f"\n... (altre {len(lines)-20} righe nel file {output_file})")
|
||||
print()
|
||||
else:
|
||||
print(f"🤖 {clean_response}\n")
|
||||
|
||||
except KeyboardInterrupt:
|
||||
print("\n👋 Arrivederci!")
|
||||
break
|
||||
except EOFError:
|
||||
break
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Inizializzazione fallita: {str(e)}")
|
||||
@@ -455,6 +627,8 @@ Esempi:
|
||||
|
||||
# === CHAT (default) ===
|
||||
chat_parser = subparsers.add_parser("chat", help="Avvia l'agent in modalità chat")
|
||||
chat_parser.add_argument("--enhanced", "-e", action="store_true", help="Usa EnhancedCLI (sperimentale)")
|
||||
chat_parser.add_argument("--terminal", "-t", action="store_true", help="Usa TerminalUI con scrollback (pyte)")
|
||||
chat_parser.set_defaults(func=cmd_chat)
|
||||
|
||||
# === SERVER ===
|
||||
|
||||
+224
@@ -0,0 +1,224 @@
|
||||
"""
|
||||
Terminal UI con scrollback semplice
|
||||
Usa input() standard + visualizzazione scrollback
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
import signal
|
||||
from typing import Optional, Callable
|
||||
|
||||
|
||||
class TerminalUI:
|
||||
"""
|
||||
Terminal UI con scrollback.
|
||||
Uso semplice: input() per input, history buffer per scrollback.
|
||||
"""
|
||||
|
||||
def __init__(self, scrollback: int = 10000, prompt: str = "❯ "):
|
||||
self.scrollback = scrollback
|
||||
self.prompt = prompt
|
||||
|
||||
# History buffer
|
||||
self.history: list = []
|
||||
|
||||
# Callback
|
||||
self.agent_callback: Optional[Callable[[str], str]] = None
|
||||
|
||||
def _print(self, text: str):
|
||||
"""Stampa con colori base"""
|
||||
# Semplice - solo reset e bold
|
||||
if text.startswith("👤"):
|
||||
print(f"\033[1;36m{text}\033[0m")
|
||||
elif text.startswith("🤖"):
|
||||
print(f"\033[1;32m{text}\033[0m")
|
||||
elif text.startswith("🤔"):
|
||||
print(f"\033[1;33m{text}\033[0m")
|
||||
elif text.startswith("❌"):
|
||||
print(f"\033[1;31m{text}\033[0m")
|
||||
else:
|
||||
print(text)
|
||||
|
||||
def _wrap_line(self, text: str, width: int = 80) -> list:
|
||||
"""Wrap linea a width"""
|
||||
lines = []
|
||||
while len(text) > width:
|
||||
lines.append(text[:width])
|
||||
text = text[width:]
|
||||
if text:
|
||||
lines.append(text)
|
||||
return lines
|
||||
|
||||
def _add_history(self, text: str):
|
||||
"""Aggiunge a history"""
|
||||
self.history.append(text)
|
||||
while len(self.history) > self.scrollback:
|
||||
self.history.pop(0)
|
||||
|
||||
def _show_history(self, lines: int = None):
|
||||
"""Mostra ultime N linee di history"""
|
||||
if lines is None:
|
||||
# Determina righe visibili
|
||||
try:
|
||||
size = os.get_terminal_size()
|
||||
lines = size.lines - 5 # spazio per header, status, prompt
|
||||
except:
|
||||
lines = 20
|
||||
|
||||
# Mostra ultime righe
|
||||
start = max(0, len(self.history) - lines)
|
||||
for line in self.history[start:]:
|
||||
self._print(line)
|
||||
|
||||
def _process(self, cmd: str):
|
||||
"""Processa comando"""
|
||||
if not cmd.strip():
|
||||
return
|
||||
|
||||
self._add_history(f"👤 {cmd}")
|
||||
|
||||
# Thinking
|
||||
print("\033[1;33m🤔 Pensando...\033[0m")
|
||||
|
||||
try:
|
||||
resp = self.agent_callback(cmd)
|
||||
|
||||
# Rimuovi thinking (print sopra con risposta)
|
||||
print("\033[2A\033[K", end="") # move up 2, clear line
|
||||
|
||||
# Stampa risposta
|
||||
for line in resp.split('\n'):
|
||||
if line.strip():
|
||||
self._add_history(f"🤖 {line}")
|
||||
self._print(f"🤖 {line}")
|
||||
|
||||
except Exception as e:
|
||||
print("\033[2A\033[K", end="")
|
||||
err_msg = f"❌ {str(e)}"
|
||||
self._add_history(err_msg)
|
||||
self._print(err_msg)
|
||||
|
||||
def _scroll_help(self):
|
||||
"""Mostra aiuto scroll"""
|
||||
print("\n" + "=" * 60)
|
||||
print("SCROLLBACK - Premi q per uscire, frecce per navigare")
|
||||
print("=" * 60)
|
||||
print()
|
||||
|
||||
# Paginator semplice
|
||||
page_size = 30
|
||||
start = 0
|
||||
total = len(self.history)
|
||||
|
||||
while start < total:
|
||||
# Mostra pagina
|
||||
for i in range(start, min(start + page_size, total)):
|
||||
self._print(self.history[i])
|
||||
|
||||
start += page_size
|
||||
|
||||
if start < total:
|
||||
print()
|
||||
print(f"[{start}/{total} righe] - Premi ENTER per continuare, q per uscire")
|
||||
try:
|
||||
choice = input().strip()
|
||||
if choice.lower() == 'q':
|
||||
break
|
||||
except:
|
||||
break
|
||||
|
||||
print()
|
||||
print("=" * 60)
|
||||
print("FINE SCROLLBACK")
|
||||
print("=" * 60)
|
||||
|
||||
def start(self, agent_callback: Callable[[str], str]):
|
||||
"""Avvia UI"""
|
||||
self.agent_callback = agent_callback
|
||||
|
||||
# Signal handler
|
||||
def sigint_handler(signum, frame):
|
||||
print("\n\033[1;33m👋 Arrivederci!\033[0m\n")
|
||||
sys.exit(0)
|
||||
|
||||
signal.signal(signal.SIGINT, sigint_handler)
|
||||
|
||||
# Header
|
||||
print()
|
||||
print("\033[1;36m" + "=" * 50 + "\033[0m")
|
||||
print("\033[1;36m=\033[0m" + " " * 15 + "\033[1;33mLangCode Agent\033[0m" + " " * 15 + "\033[1;36m=\033[0m")
|
||||
print("\033[1;36m" + "=" * 50 + "\033[0m")
|
||||
print()
|
||||
print("digita /help per aiuto, /exit per uscire")
|
||||
print()
|
||||
|
||||
while True:
|
||||
try:
|
||||
# Input
|
||||
user_input = input(f"\033[1;35m{self.prompt}\033[0m ").strip()
|
||||
except EOFError:
|
||||
break
|
||||
except KeyboardInterrupt:
|
||||
print("\n\033[1;33m👋 Arrivederci!\033[0m")
|
||||
break
|
||||
|
||||
if not user_input:
|
||||
continue
|
||||
|
||||
# Comandi speciali
|
||||
if user_input.lower() == '/exit':
|
||||
print("\033[1;33m👋 Arrivederci!\033[0m")
|
||||
break
|
||||
|
||||
elif user_input.lower() == '/quit':
|
||||
print("\033[1;33m👋 Arrivederci!\033[0m")
|
||||
break
|
||||
|
||||
elif user_input.lower() == '/help':
|
||||
print()
|
||||
print("\033[1;36m" + "=" * 40 + "\033[0m")
|
||||
print("COMANDI:")
|
||||
print(" /help - Questo messaggio")
|
||||
print(" /clear - Pulisci schermo")
|
||||
print(" /history - Mostra cronologia")
|
||||
print(" /scroll - Entra in scrollback")
|
||||
print(" /exit, /quit - Esci")
|
||||
print()
|
||||
print("SCROLLBACK:")
|
||||
print(" /scroll - Mostra tutto lo scrollback")
|
||||
print(" Frecce su/giù - Naviga cronologia (in some terminals)")
|
||||
print()
|
||||
print("\033[1;36m" + "=" * 40 + "\033[0m")
|
||||
print()
|
||||
continue
|
||||
|
||||
elif user_input.lower() == '/clear':
|
||||
os.system('clear' if os.name != 'nt' else 'cls')
|
||||
continue
|
||||
|
||||
elif user_input.lower() == '/history':
|
||||
print()
|
||||
self._show_history(50)
|
||||
print()
|
||||
continue
|
||||
|
||||
elif user_input.lower() == '/scroll':
|
||||
self._scroll_help()
|
||||
continue
|
||||
|
||||
# Processa comando normale
|
||||
self._process(user_input)
|
||||
print()
|
||||
|
||||
|
||||
def create_terminal_ui(scrollback: int = 10000, prompt: str = "❯ ") -> TerminalUI:
|
||||
return TerminalUI(scrollback=scrollback, prompt=prompt)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
def test(text):
|
||||
lines = [f"Ricevuto: {text}"]
|
||||
lines += [f"Linea {i}: contenuto di esempio" for i in range(1, 31)]
|
||||
return "\n".join(lines)
|
||||
|
||||
ui = create_terminal_ui()
|
||||
ui.start(test)
|
||||
Reference in new issue
Block a user