768 lines
26 KiB
Python
768 lines
26 KiB
Python
"""
|
|
Text User Interface for MaxClaw Agent configuration management
|
|
Enhanced with search/filter, export/import, and connection testing
|
|
"""
|
|
import curses
|
|
import sys
|
|
import json
|
|
import yaml
|
|
from pathlib import Path
|
|
from typing import Optional, List, Dict, Any
|
|
from config_manager import config_manager
|
|
|
|
|
|
def draw_menu(stdscr, selected_row_idx, options, title="MaxClaw Agent Configuration TUI"):
|
|
"""Draw a menu with options"""
|
|
stdscr.clear()
|
|
h, w = stdscr.getmaxyx()
|
|
|
|
# Title
|
|
stdscr.attron(curses.A_BOLD)
|
|
stdscr.addstr(0, (w - len(title)) // 2, title)
|
|
stdscr.attroff(curses.A_BOLD)
|
|
|
|
# Instructions
|
|
instructions = "Use arrow keys to navigate, Enter to select, 'q' to quit"
|
|
stdscr.addstr(2, (w - len(instructions)) // 2, instructions)
|
|
|
|
# Menu options
|
|
for idx, option in enumerate(options):
|
|
x = w // 2 - len(option) // 2
|
|
y = h // 2 - len(options) // 2 + idx
|
|
if idx == selected_row_idx:
|
|
stdscr.attron(curses.A_REVERSE)
|
|
stdscr.addstr(y, x, option)
|
|
stdscr.attroff(curses.A_REVERSE)
|
|
else:
|
|
stdscr.addstr(y, x, option)
|
|
|
|
stdscr.refresh()
|
|
|
|
|
|
def get_string_input(stdscr, prompt, y, x, max_length=50):
|
|
"""Get string input from user"""
|
|
curses.echo()
|
|
stdscr.addstr(y, x, prompt)
|
|
stdscr.refresh()
|
|
input_win = curses.newwin(1, max_length, y, x + len(prompt))
|
|
input_win.box()
|
|
input_win.refresh()
|
|
curses.curs_set(1)
|
|
result = input_win.getstr(0, 0, max_length).decode('utf-8')
|
|
curses.curs_set(0)
|
|
curses.noecho()
|
|
return result
|
|
|
|
|
|
def get_password_input(stdscr, prompt, y, x, max_length=50):
|
|
"""Get password input from user (hidden)"""
|
|
curses.noecho()
|
|
stdscr.addstr(y, x, prompt)
|
|
stdscr.refresh()
|
|
input_win = curses.newwin(1, max_length, y, x + len(prompt))
|
|
input_win.box()
|
|
input_win.refresh()
|
|
curses.curs_set(1)
|
|
result = ""
|
|
while True:
|
|
ch = input_win.getch()
|
|
if ch in [curses.KEY_ENTER, 10, 13]: # Enter key
|
|
break
|
|
elif ch in [curses.KEY_BACKSPACE, 127, 8]: # Backspace
|
|
if len(result) > 0:
|
|
result = result[:-1]
|
|
input_win.addstr(0, len(result), " ")
|
|
input_win.move(0, len(result))
|
|
elif ch >= 32 and ch <= 126 and len(result) < max_length: # Printable characters
|
|
result += chr(ch)
|
|
input_win.addstr(0, len(result)-1, "*")
|
|
input_win.refresh()
|
|
curses.curs_set(0)
|
|
return result
|
|
|
|
|
|
def show_message(stdscr, message, wait_for_key=True):
|
|
"""Show a message to the user"""
|
|
h, w = stdscr.getmaxyx()
|
|
stdscr.addstr(h//2, (w - len(message)) // 2, message)
|
|
stdscr.refresh()
|
|
if wait_for_key:
|
|
stdscr.getch()
|
|
|
|
|
|
def filter_configurations(configs: Dict[str, Any], search_term: str) -> Dict[str, Any]:
|
|
"""Filter configurations based on search term"""
|
|
if not search_term:
|
|
return configs
|
|
|
|
search_term = search_term.lower()
|
|
filtered = {}
|
|
|
|
for name, config in configs.items():
|
|
# Search in name
|
|
if search_term in name.lower():
|
|
filtered[name] = config
|
|
continue
|
|
|
|
# Search in provider
|
|
provider = config.get("provider", "").lower()
|
|
if search_term in provider:
|
|
filtered[name] = config
|
|
continue
|
|
|
|
# Search in model name
|
|
model_name = config.get("model_name", "").lower()
|
|
if search_term in model_name:
|
|
filtered[name] = config
|
|
continue
|
|
|
|
# Search in base URL
|
|
base_url = config.get("base_url", "").lower()
|
|
if search_term in base_url:
|
|
filtered[name] = config
|
|
continue
|
|
|
|
return filtered
|
|
|
|
|
|
def main_tui(stdscr):
|
|
"""Main TUI function"""
|
|
# Initialize colors if available
|
|
if curses.has_colors():
|
|
curses.start_color()
|
|
curses.init_pair(1, curses.COLOR_BLACK, curses.COLOR_WHITE)
|
|
curses.init_pair(2, curses.COLOR_GREEN, curses.COLOR_BLACK)
|
|
curses.init_pair(3, curses.COLOR_RED, curses.COLOR_BLACK)
|
|
|
|
# Main menu options
|
|
main_options = [
|
|
"View Configurations",
|
|
"Search Configurations",
|
|
"Add Configuration",
|
|
"Remove Configuration",
|
|
"Set Active Configuration",
|
|
"View Active Configuration",
|
|
"Test Configuration",
|
|
"Export Configurations",
|
|
"Import Configurations",
|
|
"Exit"
|
|
]
|
|
|
|
current_selection = 0
|
|
|
|
while True:
|
|
draw_menu(stdscr, current_selection, main_options)
|
|
|
|
key = stdscr.getch()
|
|
|
|
if key == curses.KEY_UP and current_selection > 0:
|
|
current_selection -= 1
|
|
elif key == curses.KEY_DOWN and current_selection < len(main_options) - 1:
|
|
current_selection += 1
|
|
elif key == curses.KEY_ENTER or key in [10, 13]:
|
|
# Enter key pressed
|
|
if current_selection == 0: # View Configurations
|
|
view_configurations_tui(stdscr)
|
|
elif current_selection == 1: # Search Configurations
|
|
search_configurations_tui(stdscr)
|
|
elif current_selection == 2: # Add Configuration
|
|
add_configuration_tui(stdscr)
|
|
elif current_selection == 3: # Remove Configuration
|
|
remove_configuration_tui(stdscr)
|
|
elif current_selection == 4: # Set Active Configuration
|
|
set_active_configuration_tui(stdscr)
|
|
elif current_selection == 5: # View Active Configuration
|
|
view_active_configuration_tui(stdscr)
|
|
elif current_selection == 6: # Test Configuration
|
|
test_configuration_tui(stdscr)
|
|
elif current_selection == 7: # Export Configurations
|
|
export_configurations_tui(stdscr)
|
|
elif current_selection == 8: # Import Configurations
|
|
import_configurations_tui(stdscr)
|
|
elif current_selection == 9: # Exit
|
|
break
|
|
elif key == ord('q') or key == ord('Q'):
|
|
break
|
|
|
|
|
|
def view_configurations_tui(stdscr, configs=None, title="Configurations"):
|
|
"""View all configurations in TUI"""
|
|
if configs is None:
|
|
configs = config_manager.get_configurations()
|
|
active = config_manager.get_active_configuration()
|
|
|
|
stdscr.clear()
|
|
h, w = stdscr.getmaxyx()
|
|
|
|
# Title
|
|
stdscr.attron(curses.A_BOLD)
|
|
stdscr.addstr(0, (w - len(title)) // 2, title)
|
|
stdscr.attroff(curses.A_BOLD)
|
|
|
|
if not configs:
|
|
stdscr.addstr(2, 0, "No configurations found.")
|
|
else:
|
|
# Header
|
|
header = f"{'Name':<20} {'Provider':<12} {'Model':<20} {'Active'}"
|
|
stdscr.addstr(2, 0, header)
|
|
stdscr.addstr(3, 0, "-" * len(header))
|
|
|
|
# Configurations
|
|
for idx, (name, config) in enumerate(configs.items()):
|
|
y = 4 + idx
|
|
if y >= h - 2: # Prevent overflow
|
|
break
|
|
|
|
marker = "*" if name == active else " "
|
|
provider = config.get("provider", "unknown")[:11]
|
|
model = config.get("model_name", "unknown")[:19]
|
|
line = f"{marker} {name:<19} {provider:<12} {model:<20}"
|
|
stdscr.addstr(y, 0, line)
|
|
|
|
# Instructions
|
|
instructions = "Press any key to return to main menu"
|
|
stdscr.addstr(h-2, (w - len(instructions)) // 2, instructions)
|
|
stdscr.refresh()
|
|
stdscr.getch()
|
|
|
|
|
|
def search_configurations_tui(stdscr):
|
|
"""Search/filter configurations"""
|
|
stdscr.clear()
|
|
h, w = stdscr.getmaxyx()
|
|
|
|
# Title
|
|
title = "Search Configurations"
|
|
stdscr.attron(curses.A_BOLD)
|
|
stdscr.addstr(0, (w - len(title)) // 2, title)
|
|
stdscr.attroff(curses.A_BOLD)
|
|
|
|
# Get search term
|
|
search_term = get_string_input(stdscr, "Search term (name, provider, model, URL): ", 2, 0, 50)
|
|
|
|
if not search_term:
|
|
show_message(stdscr, "Search cancelled.")
|
|
return
|
|
|
|
# Get all configurations and filter
|
|
all_configs = config_manager.get_configurations()
|
|
filtered_configs = filter_configurations(all_configs, search_term)
|
|
|
|
# Show filtered results
|
|
if not filtered_configs:
|
|
show_message(stdscr, f"No configurations found matching '{search_term}'.")
|
|
else:
|
|
view_configurations_tui(stdscr, filtered_configs, f"Search Results for '{search_term}'")
|
|
|
|
|
|
def add_configuration_tui(stdscr):
|
|
"""Add a new configuration via TUI"""
|
|
stdscr.clear()
|
|
h, w = stdscr.getmaxyx()
|
|
|
|
# Title
|
|
title = "Add New Configuration"
|
|
stdscr.attron(curses.A_BOLD)
|
|
stdscr.addstr(0, (w - len(title)) // 2, title)
|
|
stdscr.attroff(curses.A_BOLD)
|
|
|
|
# Get configuration name
|
|
name = get_string_input(stdscr, "Configuration name: ", 2, 0)
|
|
if not name:
|
|
show_message(stdscr, "Error: Configuration name cannot be empty")
|
|
return
|
|
|
|
if config_manager.configuration_exists(name):
|
|
overwrite_prompt = f"Configuration '{name}' already exists. Overwrite? (y/n): "
|
|
overwrite = get_string_input(stdscr, overwrite_prompt, 3, 0, 1)
|
|
if overwrite.lower() != 'y':
|
|
show_message(stdscr, "Cancelled.")
|
|
return
|
|
|
|
# Select provider
|
|
providers = ["OpenAI", "Ollama", "OpenRouter"]
|
|
provider_selection = 0
|
|
|
|
while True:
|
|
stdscr.clear()
|
|
stdscr.attron(curses.A_BOLD)
|
|
stdscr.addstr(0, (w - len(title)) // 2, title)
|
|
stdscr.attroff(curses.A_BOLD)
|
|
|
|
stdscr.addstr(2, 0, f"Name: {name}")
|
|
stdscr.addstr(3, 0, "Select provider:")
|
|
|
|
for idx, provider in enumerate(providers):
|
|
y = 4 + idx
|
|
if idx == provider_selection:
|
|
stdscr.attron(curses.A_REVERSE)
|
|
stdscr.addstr(y, 2, f"> {provider}")
|
|
stdscr.attroff(curses.A_REVERSE)
|
|
else:
|
|
stdscr.addstr(y, 2, f" {provider}")
|
|
|
|
stdscr.addstr(h-2, 0, "Use arrow keys to select, Enter to confirm")
|
|
stdscr.refresh()
|
|
|
|
key = stdscr.getch()
|
|
if key == curses.KEY_UP and provider_selection > 0:
|
|
provider_selection -= 1
|
|
elif key == curses.KEY_DOWN and provider_selection < len(providers) - 1:
|
|
provider_selection += 1
|
|
elif key == curses.KEY_ENTER or key in [10, 13]:
|
|
break
|
|
|
|
provider_map = {"OpenAI": "openai", "Ollama": "ollama", "OpenRouter": "openrouter"}
|
|
provider = provider_map[providers[provider_selection]]
|
|
config = {"provider": provider}
|
|
|
|
# Get model name
|
|
model_name = get_string_input(stdscr, f"Model name for {provider} (e.g., gpt-4o, gemma2:9b): ", 5, 0)
|
|
if not model_name:
|
|
show_message(stdscr, "Error: Model name cannot be empty")
|
|
return
|
|
config["model_name"] = model_name
|
|
|
|
# Get provider-specific settings
|
|
y_pos = 6
|
|
if provider == "openai":
|
|
api_key = get_password_input(stdscr, "OpenAI API key: ", y_pos, 0)
|
|
config["api_key"] = api_key
|
|
y_pos += 1
|
|
elif provider == "ollama":
|
|
base_url = get_string_input(stdscr, f"Ollama base URL (default: http://localhost:11434): ", y_pos, 0)
|
|
if not base_url:
|
|
base_url = "http://localhost:11434"
|
|
config["base_url"] = base_url
|
|
y_pos += 1
|
|
elif provider == "openrouter":
|
|
api_key = get_password_input(stdscr, "OpenRouter API key: ", y_pos, 0)
|
|
config["api_key"] = api_key
|
|
y_pos += 1
|
|
|
|
base_url = get_string_input(stdscr, f"OpenRouter base URL (default: https://openrouter.ai/api/v1): ", y_pos, 0)
|
|
if not base_url:
|
|
base_url = "https://openrouter.ai/api/v1"
|
|
config["base_url"] = base_url
|
|
y_pos += 1
|
|
|
|
# Ask if this should be the active configuration
|
|
set_active_prompt = "Set as active configuration? (y/n): "
|
|
set_active = get_string_input(stdscr, set_active_prompt, y_pos, 0, 1)
|
|
|
|
# Save the configuration
|
|
try:
|
|
config_manager.add_configuration(name, config)
|
|
message = f"Configuration '{name}' saved successfully."
|
|
if set_active.lower() == 'y':
|
|
config_manager.set_active_configuration(name)
|
|
message += f" Configuration '{name}' set as active."
|
|
show_message(stdscr, message)
|
|
except Exception as e:
|
|
show_message(stdscr, f"Error saving configuration: {e}")
|
|
|
|
|
|
def remove_configuration_tui(stdscr):
|
|
"""Remove a configuration via TUI"""
|
|
configs = config_manager.get_configurations()
|
|
if not configs:
|
|
show_message(stdscr, "No configurations to remove.")
|
|
return
|
|
|
|
stdscr.clear()
|
|
h, w = stdscr.getmaxyx()
|
|
|
|
# Title
|
|
title = "Remove Configuration"
|
|
stdscr.attron(curses.A_BOLD)
|
|
stdscr.addstr(0, (w - len(title)) // 2, title)
|
|
stdscr.attroff(curses.A_BOLD)
|
|
|
|
# List configurations
|
|
config_names = list(configs.keys())
|
|
selection = 0
|
|
|
|
while True:
|
|
stdscr.clear()
|
|
stdscr.attron(curses.A_BOLD)
|
|
stdscr.addstr(0, (w - len(title)) // 2, title)
|
|
stdscr.attroff(curses.A_BOLD)
|
|
|
|
stdscr.addstr(2, 0, "Select configuration to remove:")
|
|
|
|
for idx, name in enumerate(config_names):
|
|
y = 3 + idx
|
|
if y >= h - 4: # Prevent overflow
|
|
break
|
|
|
|
active_marker = " *"
|
|
if name == config_manager.get_active_configuration():
|
|
active_marker = " (active)"
|
|
|
|
if idx == selection:
|
|
stdscr.attron(curses.A_REVERSE)
|
|
stdscr.addstr(y, 2, f"> {name}{active_marker}")
|
|
stdscr.attroff(curses.A_REVERSE)
|
|
else:
|
|
stdscr.addstr(y, 2, f" {name}{active_marker}")
|
|
|
|
stdscr.addstr(h-2, 0, "Use arrow keys to select, Enter to confirm, 'c' to cancel")
|
|
stdscr.refresh()
|
|
|
|
key = stdscr.getch()
|
|
if key == curses.KEY_UP and selection > 0:
|
|
selection -= 1
|
|
elif key == curses.KEY_DOWN and selection < len(config_names) - 1:
|
|
selection += 1
|
|
elif key == curses.KEY_ENTER or key in [10, 13]:
|
|
# Confirm removal
|
|
name_to_remove = config_names[selection]
|
|
active = config_manager.get_active_configuration()
|
|
|
|
if name_to_remove == active:
|
|
confirm_prompt = f"'{name_to_remove}' is the active configuration. Remove it anyway? (y/n): "
|
|
confirm = get_string_input(stdscr, confirm_prompt, h-3, 0, 1)
|
|
if confirm.lower() != 'y':
|
|
show_message(stdscr, "Cancelled.")
|
|
return
|
|
|
|
confirm_prompt = f"Are you sure you want to remove configuration '{name_to_remove}'? (y/n): "
|
|
confirm = get_string_input(stdscr, confirm_prompt, h-3, 0, 1)
|
|
if confirm.lower() == 'y':
|
|
if config_manager.remove_configuration(name_to_remove):
|
|
show_message(stdscr, f"Configuration '{name_to_remove}' removed successfully.")
|
|
if name_to_remove == active:
|
|
show_message(stdscr, "Active configuration has been unset.")
|
|
else:
|
|
show_message(stdscr, f"Error: Failed to remove configuration '{name_to_remove}'.")
|
|
else:
|
|
show_message(stdscr, "Cancelled.")
|
|
return
|
|
elif key == ord('c') or key == ord('C'):
|
|
show_message(stdscr, "Cancelled.")
|
|
return
|
|
|
|
|
|
def set_active_configuration_tui(stdscr):
|
|
"""Set active configuration via TUI"""
|
|
configs = config_manager.get_configurations()
|
|
if not configs:
|
|
show_message(stdscr, "No configurations available.")
|
|
return
|
|
|
|
stdscr.clear()
|
|
h, w = stdscr.getmaxyx()
|
|
|
|
# Title
|
|
title = "Set Active Configuration"
|
|
stdscr.attron(curses.A_BOLD)
|
|
stdscr.addstr(0, (w - len(title)) // 2, title)
|
|
stdscr.attroff(curses.A_BOLD)
|
|
|
|
# List configurations
|
|
config_names = list(configs.keys())
|
|
selection = 0
|
|
|
|
while True:
|
|
stdscr.clear()
|
|
stdscr.attron(curses.A_BOLD)
|
|
stdscr.addstr(0, (w - len(title)) // 2, title)
|
|
stdscr.attroff(curses.A_BOLD)
|
|
|
|
stdscr.addstr(2, 0, "Select configuration to set as active:")
|
|
|
|
for idx, name in enumerate(config_names):
|
|
y = 3 + idx
|
|
if y >= h - 4: # Prevent overflow
|
|
break
|
|
|
|
current_active = config_manager.get_active_configuration()
|
|
active_marker = " (current)" if name == current_active else ""
|
|
|
|
if idx == selection:
|
|
stdscr.attron(curses.A_REVERSE)
|
|
stdscr.addstr(y, 2, f"> {name}{active_marker}")
|
|
stdscr.attroff(curses.A_REVERSE)
|
|
else:
|
|
stdscr.addstr(y, 2, f" {name}{active_marker}")
|
|
|
|
stdscr.addstr(h-2, 0, "Use arrow keys to select, Enter to confirm, 'c' to cancel")
|
|
stdscr.refresh()
|
|
|
|
key = stdscr.getch()
|
|
if key == curses.KEY_UP and selection > 0:
|
|
selection -= 1
|
|
elif key == curses.KEY_DOWN and selection < len(config_names) - 1:
|
|
selection += 1
|
|
elif key == curses.KEY_ENTER or key in [10, 13]:
|
|
# Set as active
|
|
name_to_set = config_names[selection]
|
|
if config_manager.set_active_configuration(name_to_set):
|
|
show_message(stdscr, f"Configuration '{name_to_set}' set as active.")
|
|
else:
|
|
show_message(stdscr, f"Error: Failed to set configuration '{name_to_set}' as active.")
|
|
return
|
|
elif key == ord('c') or key == ord('C'):
|
|
show_message(stdscr, "Cancelled.")
|
|
return
|
|
|
|
|
|
def view_active_configuration_tui(stdscr):
|
|
"""View active configuration via TUI"""
|
|
active_name = config_manager.get_active_configuration()
|
|
if not active_name:
|
|
show_message(stdscr, "No active configuration set.")
|
|
return
|
|
|
|
config = config_manager.get_configuration(active_name)
|
|
if not config:
|
|
show_message(stdscr, f"Error: Active configuration '{active_name}' not found.")
|
|
return
|
|
|
|
stdscr.clear()
|
|
h, w = stdscr.getmaxyx()
|
|
|
|
# Title
|
|
title = "Active Configuration"
|
|
stdscr.attron(curses.A_BOLD)
|
|
stdscr.addstr(0, (w - len(title)) // 2, title)
|
|
stdscr.attroff(curses.A_BOLD)
|
|
|
|
# Display configuration details
|
|
details = [
|
|
f"Name: {active_name}",
|
|
f"Provider: {config.get('provider', 'unknown')}",
|
|
f"Model: {config.get('model_name', 'unknown')}"
|
|
]
|
|
|
|
# Add provider-specific details
|
|
provider = config.get('provider')
|
|
if provider == "openai":
|
|
details.append(f"API Key: {'Set' if config.get('api_key') else 'Not set'}")
|
|
elif provider == "ollama":
|
|
base_url = config.get('base_url', 'http://localhost:11434')
|
|
details.append(f"Base URL: {base_url}")
|
|
elif provider == "openrouter":
|
|
details.append(f"API Key: {'Set' if config.get('api_key') else 'Not set'}")
|
|
base_url = config.get('base_url', 'https://openrouter.ai/api/v1')
|
|
details.append(f"Base URL: {base_url}")
|
|
|
|
# Display details
|
|
for idx, detail in enumerate(details):
|
|
y = 2 + idx
|
|
if y >= h - 2: # Prevent overflow
|
|
break
|
|
stdscr.addstr(y, 0, detail)
|
|
|
|
# Instructions
|
|
instructions = "Press any key to return to main menu"
|
|
stdscr.addstr(h-2, (w - len(instructions)) // 2, instructions)
|
|
stdscr.refresh()
|
|
stdscr.getch()
|
|
|
|
|
|
def test_configuration_tui(stdscr):
|
|
"""Test connection to a configuration"""
|
|
configs = config_manager.get_configurations()
|
|
if not configs:
|
|
show_message(stdscr, "No configurations to test.")
|
|
return
|
|
|
|
stdscr.clear()
|
|
h, w = stdscr.getmaxyx()
|
|
|
|
# Title
|
|
title = "Test Configuration Connection"
|
|
stdscr.attron(curses.A_BOLD)
|
|
stdscr.addstr(0, (w - len(title)) // 2, title)
|
|
stdscr.attroff(curses.A_BOLD)
|
|
|
|
# List configurations
|
|
config_names = list(configs.keys())
|
|
selection = 0
|
|
|
|
while True:
|
|
stdscr.clear()
|
|
stdscr.attron(curses.A_BOLD)
|
|
stdscr.addstr(0, (w - len(title)) // 2, title)
|
|
stdscr.attroff(curses.A_BOLD)
|
|
|
|
stdscr.addstr(2, 0, "Select configuration to test:")
|
|
|
|
for idx, name in enumerate(config_names):
|
|
y = 3 + idx
|
|
if y >= h - 4: # Prevent overflow
|
|
break
|
|
|
|
if idx == selection:
|
|
stdscr.attron(curses.A_REVERSE)
|
|
stdscr.addstr(y, 2, f"> {name}")
|
|
stdscr.attroff(curses.A_REVERSE)
|
|
else:
|
|
stdscr.addstr(y, 2, f" {name}")
|
|
|
|
stdscr.addstr(h-2, 0, "Use arrow keys to select, Enter to test, 'c' to cancel")
|
|
stdscr.refresh()
|
|
|
|
key = stdscr.getch()
|
|
if key == curses.KEY_UP and selection > 0:
|
|
selection -= 1
|
|
elif key == curses.KEY_DOWN and selection < len(config_names) - 1:
|
|
selection += 1
|
|
elif key == curses.KEY_ENTER or key in [10, 13]:
|
|
# Test configuration
|
|
name_to_test = config_names[selection]
|
|
show_message(stdscr, f"Testing connection to '{name_to_test}'...", False)
|
|
|
|
success, message = config_manager.test_configuration_connection(name_to_test)
|
|
|
|
if success:
|
|
show_message(stdscr, f"✅ {message}")
|
|
else:
|
|
show_message(stdscr, f"❌ {message}")
|
|
return
|
|
elif key == ord('c') or key == ord('C'):
|
|
show_message(stdscr, "Cancelled.")
|
|
return
|
|
|
|
|
|
def export_configurations_tui(stdscr):
|
|
"""Export configurations to JSON or YAML"""
|
|
configs = config_manager.get_configurations()
|
|
if not configs:
|
|
show_message(stdscr, "No configurations to export.")
|
|
return
|
|
|
|
stdscr.clear()
|
|
h, w = stdscr.getmaxyx()
|
|
|
|
# Title
|
|
title = "Export Configurations"
|
|
stdscr.attron(curses.A_BOLD)
|
|
stdscr.addstr(0, (w - len(title)) // 2, title)
|
|
stdscr.attroff(curses.A_BOLD)
|
|
|
|
# Select format
|
|
formats = ["JSON", "YAML"]
|
|
format_selection = 0
|
|
|
|
while True:
|
|
stdscr.clear()
|
|
stdscr.attron(curses.A_BOLD)
|
|
stdscr.addstr(0, (w - len(title)) // 2, title)
|
|
stdscr.attroff(curses.A_BOLD)
|
|
|
|
stdscr.addstr(2, 0, "Select export format:")
|
|
|
|
for idx, fmt in enumerate(formats):
|
|
y = 3 + idx
|
|
if idx == format_selection:
|
|
stdscr.attron(curses.A_REVERSE)
|
|
stdscr.addstr(y, 2, f"> {fmt}")
|
|
stdscr.attroff(curses.A_REVERSE)
|
|
else:
|
|
stdscr.addstr(y, 2, f" {fmt}")
|
|
|
|
stdscr.addstr(h-2, 0, "Use arrow keys to select, Enter to confirm, 'c' to cancel")
|
|
stdscr.refresh()
|
|
|
|
key = stdscr.getch()
|
|
if key == curses.KEY_UP and format_selection > 0:
|
|
format_selection -= 1
|
|
elif key == curses.KEY_DOWN and format_selection < len(formats) - 1:
|
|
format_selection += 1
|
|
elif key == curses.KEY_ENTER or key in [10, 13]:
|
|
break
|
|
elif key == ord('c') or key == ord('C'):
|
|
show_message(stdscr, "Cancelled.")
|
|
return
|
|
|
|
# Get filename
|
|
filename = get_string_input(stdscr, "Filename (without extension): ", 5, 0, 30)
|
|
if not filename:
|
|
show_message(stdscr, "Cancelled.")
|
|
return
|
|
|
|
# Export configurations
|
|
try:
|
|
if format_selection == 0: # JSON
|
|
export_file = Path(f"{filename}.json")
|
|
with open(export_file, 'w') as f:
|
|
json.dump(configs, f, indent=2)
|
|
else: # YAML
|
|
export_file = Path(f"{filename}.yaml")
|
|
with open(export_file, 'w') as f:
|
|
yaml.dump(configs, f, default_flow_style=False)
|
|
|
|
show_message(stdscr, f"Configurations exported to {export_file}")
|
|
except Exception as e:
|
|
show_message(stdscr, f"Error exporting configurations: {e}")
|
|
|
|
|
|
def import_configurations_tui(stdscr):
|
|
"""Import configurations from JSON or YAML"""
|
|
stdscr.clear()
|
|
h, w = stdscr.getmaxyx()
|
|
|
|
# Title
|
|
title = "Import Configurations"
|
|
stdscr.attron(curses.A_BOLD)
|
|
stdscr.addstr(0, (w - len(title)) // 2, title)
|
|
stdscr.attroff(curses.A_BOLD)
|
|
|
|
# Get filename
|
|
filename = get_string_input(stdscr, "Filename (with extension .json or .yaml): ", 2, 0, 50)
|
|
if not filename:
|
|
show_message(stdscr, "Cancelled.")
|
|
return
|
|
|
|
# Check if file exists
|
|
file_path = Path(filename)
|
|
if not file_path.exists():
|
|
show_message(stdscr, f"Error: File '{filename}' not found.")
|
|
return
|
|
|
|
# Import configurations
|
|
try:
|
|
with open(file_path, 'r') as f:
|
|
if filename.endswith('.json'):
|
|
imported_configs = json.load(f)
|
|
elif filename.endswith(('.yaml', '.yml')):
|
|
imported_configs = yaml.safe_load(f)
|
|
else:
|
|
show_message(stdscr, "Error: Unsupported file format. Use .json or .yaml")
|
|
return
|
|
|
|
# Validate imported configurations
|
|
if not isinstance(imported_configs, dict):
|
|
show_message(stdscr, "Error: Invalid configuration format.")
|
|
return
|
|
|
|
# Ask for confirmation
|
|
confirm_prompt = f"Import {len(imported_configs)} configurations? (y/n): "
|
|
confirm = get_string_input(stdscr, confirm_prompt, 4, 0, 1)
|
|
if confirm.lower() != 'y':
|
|
show_message(stdscr, "Cancelled.")
|
|
return
|
|
|
|
# Import configurations
|
|
imported_count = 0
|
|
for name, config in imported_configs.items():
|
|
try:
|
|
config_manager.add_configuration(name, config)
|
|
imported_count += 1
|
|
except Exception as e:
|
|
show_message(stdscr, f"Warning: Could not import '{name}': {e}")
|
|
|
|
show_message(stdscr, f"Successfully imported {imported_count} configurations.")
|
|
except Exception as e:
|
|
show_message(stdscr, f"Error importing configurations: {e}")
|
|
|
|
|
|
def main():
|
|
"""Main entry point for the TUI"""
|
|
curses.wrapper(main_tui)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |