#!/usr/bin/env python3 """ Romarchive AI Chat - A GUI chat application for Romarchive AI Uses tkinter for the GUI and OpenAI library for API communication """ import tkinter as tk from tkinter import ttk, scrolledtext, messagebox import json import os from openai import OpenAI from datetime import datetime class ConfigManager: """Manages configuration storage and retrieval""" CONFIG_FILE = "romarchive_config.json" @staticmethod def load_config(): """Load configuration from file""" if os.path.exists(ConfigManager.CONFIG_FILE): try: with open(ConfigManager.CONFIG_FILE, 'r') as f: return json.load(f) except (json.JSONDecodeError, IOError): return {} return {} @staticmethod def save_config(api_key): """Save configuration to file""" try: with open(ConfigManager.CONFIG_FILE, 'w') as f: json.dump({"api_key": api_key}, f) return True except IOError: return False class RomarchiveAIChat: """Main application class for Romarchive AI Chat""" ENDPOINT = "https://cows.info.gf/ai/v1" def __init__(self, root): self.root = root self.root.title("Romarchive AI Chat") self.root.geometry("800x600") self.root.minsize(600, 400) # Load configuration self.config = ConfigManager.load_config() self.api_key = self.config.get("api_key", "") # Initialize OpenAI client self.client = None self.initialize_client() # Chat history self.chat_history = [] # Create GUI self.create_gui() # Load any existing chat history self.load_chat_history() def initialize_client(self): """Initialize OpenAI client with configured endpoint""" if self.api_key: try: self.client = OpenAI( api_key=self.api_key, base_url=self.ENDPOINT, timeout=30.0 # Set a reasonable timeout ) # Test the connection try: self.client.models.list() print(f"Successfully connected to {self.ENDPOINT}") except Exception as test_error: print(f"Connection test failed: {test_error}") # Don't set client to None, let the API call fail with proper error except Exception as e: print(f"Error initializing client: {e}") self.client = None def load_models(self): """Load available models from the API""" if not self.client: return try: # Disable refresh button while loading self.refresh_btn.config(state=tk.DISABLED) self.status_var.set("Loading models...") self.status_label.config(foreground='blue') # Fetch models from API models_response = self.client.models.list() # Extract model IDs model_ids = [model.id for model in models_response.data] model_ids.sort() # Sort alphabetically if model_ids: # Update combobox with models self.model_combo['values'] = model_ids # Set default selection if no current selection if not self.model_var.get() and model_ids: self.model_var.set(model_ids[0]) self.status_var.set(f"Loaded {len(model_ids)} models") self.status_label.config(foreground='green') print(f"Loaded {len(model_ids)} models: {model_ids}") else: self.status_var.set("No models found") self.status_label.config(foreground='red') except Exception as e: error_msg = str(e) print(f"Error loading models: {error_msg}") self.status_var.set("Failed to load models") self.status_label.config(foreground='red') # Show error in chat display self.display_message("error", f"Error loading models: {error_msg}") finally: # Re-enable refresh button self.refresh_btn.config(state=tk.NORMAL) def create_gui(self): """Create the main GUI interface""" # Configure styles style = ttk.Style() style.theme_use('clam') # Main frame main_frame = ttk.Frame(self.root, padding="10") main_frame.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S)) # Configure grid weights self.root.columnconfigure(0, weight=1) self.root.rowconfigure(0, weight=1) main_frame.columnconfigure(0, weight=1) main_frame.rowconfigure(1, weight=1) # Header section header_frame = ttk.Frame(main_frame) header_frame.grid(row=0, column=0, columnspan=2, sticky=(tk.W, tk.E), pady=(0, 10)) header_frame.columnconfigure(1, weight=1) ttk.Label(header_frame, text="Romarchive AI Chat", font=('Arial', 16, 'bold')).grid(row=0, column=0, sticky=tk.W) # API Key configuration api_frame = ttk.LabelFrame(header_frame, text="API Configuration", padding="5") api_frame.grid(row=1, column=0, columnspan=2, sticky=(tk.W, tk.E), pady=(5, 0)) api_frame.columnconfigure(1, weight=1) ttk.Label(api_frame, text="API Key:").grid(row=0, column=0, sticky=tk.W, padx=(0, 5)) self.api_key_var = tk.StringVar(value=self.api_key) self.api_key_entry = ttk.Entry(api_frame, textvariable=self.api_key_var, show="*", width=50) self.api_key_entry.grid(row=0, column=1, sticky=(tk.W, tk.E), padx=(0, 5)) self.save_btn = ttk.Button(api_frame, text="Save", command=self.save_api_key) self.save_btn.grid(row=0, column=2, padx=(5, 0)) # Model selection ttk.Label(api_frame, text="Model:").grid(row=1, column=0, sticky=tk.W, padx=(0, 5)) self.model_var = tk.StringVar(value="") model_frame = ttk.Frame(api_frame) model_frame.grid(row=1, column=1, columnspan=2, sticky=tk.W) # Model selection will be populated dynamically self.model_combo = ttk.Combobox(model_frame, textvariable=self.model_var, state="readonly", width=25) self.model_combo.pack(side=tk.LEFT, padx=(0, 10)) # Refresh models button self.refresh_btn = ttk.Button(model_frame, text="Refresh Models", command=self.load_models) self.refresh_btn.pack(side=tk.LEFT) # Load models on startup self.root.after(100, self.load_models) # Endpoint display ttk.Label(api_frame, text=f"Endpoint: {self.ENDPOINT}", font=('Arial', 9), foreground='gray').grid(row=2, column=0, columnspan=3, sticky=tk.W, pady=(3, 0)) # Status indicator self.status_var = tk.StringVar(value="Ready") self.status_label = ttk.Label(api_frame, textvariable=self.status_var, font=('Arial', 9), foreground='green') self.status_label.grid(row=3, column=0, columnspan=3, sticky=tk.W, pady=(3, 0)) # Chat display area chat_frame = ttk.LabelFrame(main_frame, text="Chat", padding="5") chat_frame.grid(row=1, column=0, columnspan=2, sticky=(tk.W, tk.E, tk.N, tk.S), pady=(10, 5)) chat_frame.columnconfigure(0, weight=1) chat_frame.rowconfigure(0, weight=1) self.chat_display = scrolledtext.ScrolledText( chat_frame, wrap=tk.WORD, state=tk.DISABLED, font=('Arial', 10), padx=10, pady=10 ) self.chat_display.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S)) # Input area input_frame = ttk.Frame(main_frame) input_frame.grid(row=2, column=0, columnspan=2, sticky=(tk.W, tk.E), pady=(5, 0)) input_frame.columnconfigure(0, weight=1) self.input_var = tk.StringVar() self.input_entry = ttk.Entry(input_frame, textvariable=self.input_var, font=('Arial', 10)) self.input_entry.grid(row=0, column=0, sticky=(tk.W, tk.E), padx=(0, 5)) self.input_entry.bind('', lambda e: self.send_message()) self.send_btn = ttk.Button(input_frame, text="Send", command=self.send_message) self.send_btn.grid(row=0, column=1) # Button frame for additional controls button_frame = ttk.Frame(main_frame) button_frame.grid(row=3, column=0, columnspan=2, sticky=(tk.W, tk.E), pady=(10, 0)) self.clear_btn = ttk.Button(button_frame, text="Clear Chat", command=self.clear_chat) self.clear_btn.pack(side=tk.LEFT, padx=(0, 5)) self.export_btn = ttk.Button(button_frame, text="Export Chat", command=self.export_chat) self.export_btn.pack(side=tk.LEFT) # Configure text tags for styling self.chat_display.tag_config("user", foreground="#0066cc", font=('Arial', 10, 'bold')) self.chat_display.tag_config("ai", foreground="#009933", font=('Arial', 10, 'bold')) self.chat_display.tag_config("timestamp", foreground='gray', font=('Arial', 8)) self.chat_display.tag_config("error", foreground='red', font=('Arial', 10)) self.chat_display.tag_config("info", foreground='gray', font=('Arial', 9)) # Focus on input field self.input_entry.focus() # Check if API key is configured if not self.api_key: self.status_var.set("Please configure API key") self.status_label.config(foreground='red') else: self.status_var.set("Ready to chat") self.status_label.config(foreground='green') def save_api_key(self): """Save the API key to configuration""" api_key = self.api_key_var.get().strip() if not api_key: messagebox.showerror("Error", "API key cannot be empty!") return if ConfigManager.save_config(api_key): self.api_key = api_key self.initialize_client() self.status_var.set("API key saved successfully") self.status_label.config(foreground='green') messagebox.showinfo("Success", "API key saved successfully!") else: messagebox.showerror("Error", "Failed to save API key!") self.status_var.set("Failed to save API key") self.status_label.config(foreground='red') def send_message(self): """Send message to Romarchive AI""" message = self.input_var.get().strip() if not message: return if not self.api_key: messagebox.showerror("Error", "Please configure API key first!") return if not self.client: messagebox.showerror("Error", "Failed to initialize API client. Please check your API key.") return # Check if a model is selected selected_model = self.model_var.get() if not selected_model: messagebox.showerror("Error", "Please select a model first!") return # Clear input self.input_var.set("") # Display user message self.display_message("user", f"You: {message}") # Add to chat history timestamp = datetime.now().isoformat() self.chat_history.append({"role": "user", "content": message, "timestamp": timestamp}) # Disable send button while processing self.send_btn.config(state=tk.DISABLED) self.status_var.set("Thinking...") self.status_label.config(foreground='blue') # Process in background (using after to avoid blocking) self.root.after(100, lambda: self.process_ai_response(message)) def process_ai_response(self, message): """Process AI response with streaming""" try: # Get selected model selected_model = self.model_var.get() print(f"Attempting to call API with model: {selected_model}") print(f"Endpoint: {self.ENDPOINT}") # Call OpenAI API with streaming stream = self.client.chat.completions.create( model=selected_model, messages=[ {"role": "system", "content": "You are a helpful assistant for Romarchive AI."}, {"role": "user", "content": message} ], max_tokens=2000, temperature=0.7, stream=True, # Enable streaming timeout=60.0 # Set a longer timeout for the request ) print(f"API call successful, starting stream...") # Display AI response with streaming self.chat_display.config(state=tk.NORMAL) # Add timestamp timestamp = datetime.now().strftime("%H:%M:%S") self.chat_display.insert(tk.END, f"[{timestamp}] ", "timestamp") self.chat_display.insert(tk.END, "AI: ", "ai") # Stream the response ai_response = "" chunk_count = 0 for chunk in stream: chunk_count += 1 if chunk.choices[0].delta.content: content = chunk.choices[0].delta.content ai_response += content self.chat_display.insert(tk.END, content, "ai") self.chat_display.see(tk.END) self.root.update_idletasks() # Update GUI print(f"Stream complete. Received {chunk_count} chunks. Response length: {len(ai_response)}") # Add newline after response self.chat_display.insert(tk.END, "\n\n") self.chat_display.config(state=tk.DISABLED) # Add to chat history timestamp = datetime.now().isoformat() self.chat_history.append({"role": "assistant", "content": ai_response, "timestamp": timestamp}) # Save chat history self.save_chat_history() # Update status self.status_var.set("Response received") self.status_label.config(foreground='green') except Exception as e: error_msg = str(e) print(f"Error in process_ai_response: {error_msg}") import traceback traceback.print_exc() # Check for specific timeout errors if "timeout" in error_msg.lower() or "handshake" in error_msg.lower(): error_msg = "Connection timeout. The Romarchive AI endpoint may be unreachable or experiencing issues. Please check your network connection and try again." self.display_message("error", f"Error: {error_msg}") self.status_var.set("Error occurred") self.status_label.config(foreground='red') messagebox.showerror("API Error", f"Failed to get response:\n{error_msg}") finally: # Re-enable send button self.send_btn.config(state=tk.NORMAL) self.input_entry.focus() def display_message(self, msg_type, message): """Display a message in the chat display""" self.chat_display.config(state=tk.NORMAL) # Add timestamp timestamp = datetime.now().strftime("%H:%M:%S") self.chat_display.insert(tk.END, f"[{timestamp}] ", "timestamp") # Add message with appropriate tag self.chat_display.insert(tk.END, message + "\n\n", msg_type) # Scroll to bottom self.chat_display.see(tk.END) self.chat_display.config(state=tk.DISABLED) def clear_chat(self): """Clear the chat history""" if messagebox.askyesno("Confirm", "Are you sure you want to clear the chat history?"): self.chat_history = [] self.chat_display.config(state=tk.NORMAL) self.chat_display.delete(1.0, tk.END) self.chat_display.config(state=tk.DISABLED) self.save_chat_history() self.status_var.set("Chat cleared") self.status_label.config(foreground='gray') def export_chat(self): """Export chat history to a file""" if not self.chat_history: messagebox.showinfo("Info", "No chat history to export") return try: filename = f"romarchive_chat_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" with open(filename, 'w') as f: json.dump(self.chat_history, f, indent=2) messagebox.showinfo("Success", f"Chat exported to {filename}") self.status_var.set(f"Exported to {filename}") self.status_label.config(foreground='green') except Exception as e: messagebox.showerror("Error", f"Failed to export chat: {e}") self.status_var.set("Export failed") self.status_label.config(foreground='red') def save_chat_history(self): """Save chat history to file""" try: with open("romarchive_chat_history.json", 'w') as f: json.dump(self.chat_history, f, indent=2) except Exception as e: print(f"Failed to save chat history: {e}") def load_chat_history(self): """Load chat history from file""" try: if os.path.exists("romarchive_chat_history.json"): with open("romarchive_chat_history.json", 'r') as f: self.chat_history = json.load(f) # Display loaded history for entry in self.chat_history: role = entry.get("role", "") content = entry.get("content", "") if role == "user": self.display_message("user", f"You: {content}") elif role == "assistant": self.display_message("ai", f"AI: {content}") except Exception as e: print(f"Failed to load chat history: {e}") def main(): """Main entry point for the application""" try: # Create root window root = tk.Tk() # Create and run application app = RomarchiveAIChat(root) # Start the main loop root.mainloop() except Exception as e: print(f"Error running application: {e}") import traceback traceback.print_exc() if __name__ == "__main__": main()