terminal_chatbot.py

#!/usr/bin/env python3
"""
Terminal Chatbot for VNC-Perplexity API Server
Interactive command-line interface for testing and using the API
"""

import asyncio
import json
import sys
import time
from datetime import datetime
from typing import Optional, Dict, Any, List
import readline
import os

try:
    import httpx
    import rich
    from rich.console import Console
    from rich.table import Table
    from rich.panel import Panel
    from rich.text import Text
    from rich.markdown import Markdown
    from rich.progress import Progress, SpinnerColumn, TextColumn
except ImportError as e:
    print(f"Missing dependencies: {e}")
    print("Please install with: pip install httpx rich")
    sys.exit(1)

class TerminalChatbot:
    """Interactive terminal chatbot for VNC-Perplexity API"""
    
    def __init__(self, base_url: str = "http://localhost:8000"):
        self.base_url = base_url.rstrip('/')
        self.console = Console()
        self.client = httpx.AsyncClient(timeout=300.0)  # 5 minute timeout
        self.conversation_history: List[Dict[str, str]] = []
        self.current_model = "perplexity-pro"
        self.api_mode = "openai"  # Can be "openai" or "anthropic"
        
        # Commands that don't require API calls
        self.system_commands = {
            'help': self._show_help,
            'status': self._show_status,
            'health': self._check_health,
            'models': self._list_models,
            'history': self._show_history,
            'clear': self._clear_history,
            'model': self._change_model,
            'mode': self._change_mode,
            'config': self._show_config,
            'exit': self._exit,
            'quit': self._exit
        }
        
    async def start(self):
        """Start the interactive chatbot"""
        await self._show_welcome()
        
        # Check server connectivity
        if not await self._check_connectivity():
            self.console.print("❌ Could not connect to API server", style="red")
            return
            
        # Main interaction loop
        while True:
            try:
                # Get user input
                prompt = self._get_user_input()
                
                if not prompt.strip():
                    continue
                    
                # Check for system commands
                if prompt.lower() in self.system_commands:
                    await self.system_commands[prompt.lower()]()
                    continue
                
                # Check for system commands with parameters
                parts = prompt.lower().split()
                if parts[0] in self.system_commands:
                    await self.system_commands[parts[0]](*parts[1:])
                    continue
                
                # Process as chat message
                await self._process_chat_message(prompt)
                
            except KeyboardInterrupt:
                self.console.print("\n\n👋 Goodbye!", style="blue")
                break
            except EOFError:
                self.console.print("\n\n👋 Goodbye!", style="blue")
                break
            except Exception as e:
                self.console.print(f"❌ Error: {e}", style="red")
    
    async def _show_welcome(self):
        """Show welcome message"""
        welcome_text = """
# 🤖 VNC-Perplexity API Terminal Chatbot

Welcome! This chatbot connects to your VNC-Perplexity API server.

**Quick Commands:**
- `help` - Show all commands
- `status` - Check server status  
- `health` - Check server health
- `models` - List available models
- `history` - Show conversation history
- `clear` - Clear conversation history
- `model <name>` - Change model
- `mode <openai|anthropic>` - Change API mode
- `exit` or `quit` - Exit chatbot

**Usage:**
Just type your question and I'll send it to Perplexity via the VNC automation!
        """
        self.console.print(Panel(Markdown(welcome_text), title="🚀 Welcome", border_style="blue"))
    
    def _get_user_input(self) -> str:
        """Get user input with readline support"""
        try:
            prompt_text = f"[{self.api_mode}:{self.current_model}] 💬 "
            return input(prompt_text)
        except (KeyboardInterrupt, EOFError):
            raise
    
    async def _check_connectivity(self) -> bool:
        """Check if the API server is reachable"""
        try:
            response = await self.client.get(f"{self.base_url}/health")
            return response.status_code == 200
        except:
            return False
    
    async def _process_chat_message(self, message: str):
        """Process a chat message through the API"""
        # Add to conversation history
        self.conversation_history.append({"role": "user", "content": message})
        
        try:
            with Progress(
                SpinnerColumn(),
                TextColumn("[progress.description]{task.description}"),
                console=self.console
            ) as progress:
                task = progress.add_task("🔍 Searching with Perplexity...", total=None)
                
                if self.api_mode == "openai":
                    response = await self._send_openai_request(message)
                else:
                    response = await self._send_anthropic_request(message)
                
                progress.remove_task(task)
            
            if response:
                # Display response
                self.console.print(Panel(
                    Markdown(response),
                    title="🤖 Response",
                    border_style="green"
                ))
                
                # Add to conversation history
                self.conversation_history.append({"role": "assistant", "content": response})
            
        except Exception as e:
            self.console.print(f"❌ Error processing message: {e}", style="red")
    
    async def _send_openai_request(self, message: str) -> Optional[str]:
        """Send request using OpenAI format"""
        try:
            payload = {
                "model": self.current_model,
                "messages": [{"role": "user", "content": message}],
                "temperature": 0.7,
                "max_tokens": 1000
            }
            
            response = await self.client.post(
                f"{self.base_url}/v1/chat/completions",
                json=payload,
                headers={"Content-Type": "application/json"}
            )
            
            if response.status_code == 200:
                data = response.json()
                return data["choices"][0]["message"]["content"]
            else:
                error_data = response.json() if response.headers.get("content-type") == "application/json" else {"error": response.text}
                self.console.print(f"❌ API Error ({response.status_code}): {error_data.get('error', 'Unknown error')}", style="red")
                return None
                
        except Exception as e:
            self.console.print(f"❌ Request failed: {e}", style="red")
            return None
    
    async def _send_anthropic_request(self, message: str) -> Optional[str]:
        """Send request using Anthropic format"""
        try:
            payload = {
                "model": self.current_model,
                "max_tokens": 1000,
                "messages": [{"role": "user", "content": message}]
            }
            
            response = await self.client.post(
                f"{self.base_url}/v1/messages",
                json=payload,
                headers={"Content-Type": "application/json"}
            )
            
            if response.status_code == 200:
                data = response.json()
                return data["content"][0]["text"]
            else:
                error_data = response.json() if response.headers.get("content-type") == "application/json" else {"error": response.text}
                self.console.print(f"❌ API Error ({response.status_code}): {error_data.get('error', 'Unknown error')}", style="red")
                return None
                
        except Exception as e:
            self.console.print(f"❌ Request failed: {e}", style="red")
            return None
    
    async def _show_help(self):
        """Show help information"""
        help_text = """
# 🤖 VNC-Perplexity API Chatbot Commands

## System Commands
- `help` - Show this help message
- `status` - Check server and VNC status
- `health` - Quick health check
- `models` - List available models
- `history` - Show conversation history
- `clear` - Clear conversation history
- `model <name>` - Change current model
- `mode <openai|anthropic>` - Change API mode
- `config` - Show current configuration
- `exit` or `quit` - Exit the chatbot

## Chat Usage
Just type your question naturally! Examples:
- "What is quantum computing?"
- "Explain machine learning algorithms"
- "Latest news about AI developments"

## Notes
- The chatbot uses VNC automation to interact with Perplexity.ai
- Responses may take 30-60 seconds due to browser automation
- Each query starts a fresh search (no conversation context on Perplexity side)
        """
        self.console.print(Panel(Markdown(help_text), title="📖 Help", border_style="cyan"))
    
    async def _show_status(self):
        """Show detailed server status"""
        try:
            response = await self.client.get(f"{self.base_url}/status")
            if response.status_code == 200:
                data = response.json()
                
                # Create status table
                table = Table(title="🖥️ Server Status")
                table.add_column("Component", style="cyan")
                table.add_column("Status", style="green")
                table.add_column("Details", style="yellow")
                
                # Server info
                uptime = data.get("uptime_seconds", 0)
                uptime_str = f"{uptime // 3600}h {(uptime % 3600) // 60}m {uptime % 60}s"
                table.add_row("Server", "Running", f"Uptime: {uptime_str}")
                table.add_row("Requests", "Active", f"Processed: {data.get('requests_processed', 0)}")
                
                # VNC status
                vnc_data = data.get("vnc", {})
                vnc_status = "✅ Running" if vnc_data.get("vnc_running", False) else "❌ Stopped"
                workers = vnc_data.get("active_workers", 0)
                table.add_row("VNC", vnc_status, f"Workers: {workers}/5")
                
                # Response processor
                processor_data = data.get("response_processor", {})
                responses_stored = processor_data.get("responses_stored", 0)
                table.add_row("Processor", "Ready", f"Responses: {responses_stored}")
                
                self.console.print(table)
            else:
                self.console.print("❌ Failed to get status", style="red")
                
        except Exception as e:
            self.console.print(f"❌ Error getting status: {e}", style="red")
    
    async def _check_health(self):
        """Quick health check"""
        try:
            start_time = time.time()
            response = await self.client.get(f"{self.base_url}/health")
            response_time = time.time() - start_time
            
            if response.status_code == 200:
                data = response.json()
                vnc_status = "✅" if data.get("vnc_running", False) else "❌"
                workers = data.get("workers_active", 0)
                
                self.console.print(f"✅ Server: Healthy ({response_time:.2f}s)", style="green")
                self.console.print(f"{vnc_status} VNC: {data.get('vnc_running', False)}", style="green" if data.get('vnc_running', False) else "red")
                self.console.print(f"👥 Workers: {workers}/5", style="green" if workers > 0 else "yellow")
            else:
                self.console.print("❌ Server: Unhealthy", style="red")
                
        except Exception as e:
            self.console.print(f"❌ Health check failed: {e}", style="red")
    
    async def _list_models(self):
        """List available models"""
        try:
            response = await self.client.get(f"{self.base_url}/v1/models")
            if response.status_code == 200:
                data = response.json()
                models = data.get("data", [])
                
                table = Table(title="🤖 Available Models")
                table.add_column("Model ID", style="cyan")
                table.add_column("Owner", style="green")
                table.add_column("Current", style="yellow")
                
                for model in models:
                    current = "✅" if model["id"] == self.current_model else ""
                    table.add_row(model["id"], model["owned_by"], current)
                
                self.console.print(table)
            else:
                self.console.print("❌ Failed to get models", style="red")
                
        except Exception as e:
            self.console.print(f"❌ Error getting models: {e}", style="red")
    
    async def _show_history(self):
        """Show conversation history"""
        if not self.conversation_history:
            self.console.print("📝 No conversation history", style="yellow")
            return
        
        self.console.print(f"📝 Conversation History ({len(self.conversation_history)} messages):")
        
        for i, msg in enumerate(self.conversation_history, 1):
            role = msg["role"]
            content = msg["content"]
            
            if role == "user":
                self.console.print(f"\n{i}. 👤 You:")
                self.console.print(f"   {content}", style="cyan")
            else:
                self.console.print(f"\n{i}. 🤖 Assistant:")
                self.console.print(Panel(Markdown(content[:200] + "..." if len(content) > 200 else content), border_style="green"))
    
    async def _clear_history(self):
        """Clear conversation history"""
        self.conversation_history.clear()
        self.console.print("🗑️ Conversation history cleared", style="green")
    
    async def _change_model(self, model_name: str = None):
        """Change the current model"""
        if not model_name:
            await self._list_models()
            self.console.print("\n💡 Usage: model <model_name>", style="yellow")
            return
        
        self.current_model = model_name
        self.console.print(f"🔄 Changed model to: {model_name}", style="green")
    
    async def _change_mode(self, mode: str = None):
        """Change API mode (openai/anthropic)"""
        if not mode or mode not in ["openai", "anthropic"]:
            self.console.print("💡 Usage: mode <openai|anthropic>", style="yellow")
            self.console.print(f"Current mode: {self.api_mode}", style="cyan")
            return
        
        self.api_mode = mode
        self.console.print(f"🔄 Changed API mode to: {mode}", style="green")
    
    async def _show_config(self):
        """Show current configuration"""
        table = Table(title="⚙️ Current Configuration")
        table.add_column("Setting", style="cyan")
        table.add_column("Value", style="green")
        
        table.add_row("API Server", self.base_url)
        table.add_row("API Mode", self.api_mode)
        table.add_row("Model", self.current_model)
        table.add_row("History Length", str(len(self.conversation_history)))
        
        self.console.print(table)
    
    async def _exit(self):
        """Exit the chatbot"""
        await self.client.aclose()
        self.console.print("👋 Goodbye!", style="blue")
        sys.exit(0)

async def main():
    """Main entry point"""
    import argparse
    
    parser = argparse.ArgumentParser(description="Terminal Chatbot for VNC-Perplexity API Server")
    parser.add_argument("--url", default="http://localhost:8000", help="API server URL")
    parser.add_argument("--model", default="perplexity-pro", help="Default model to use")
    parser.add_argument("--mode", default="openai", choices=["openai", "anthropic"], help="API mode")
    
    args = parser.parse_args()
    
    # Create and start chatbot
    chatbot = TerminalChatbot(base_url=args.url)
    chatbot.current_model = args.model
    chatbot.api_mode = args.mode
    
    try:
        await chatbot.start()
    except KeyboardInterrupt:
        print("\n👋 Goodbye!")
    finally:
        await chatbot.client.aclose()

if __name__ == "__main__":
    asyncio.run(main())