perplexica-search-tui/perplexica-search-tui.py

#!/usr/bin/env python3
"""
🔮 Perplexica CLI - Beautiful terminal interface for Perplexica API
"""

import json
import sys
from typing import Optional, List, Tuple
import httpx
import click
from rich.console import Console
from rich.panel import Panel
from rich.markdown import Markdown
from rich.table import Table
from rich.live import Live
from rich.spinner import Spinner
from rich.prompt import Prompt, Confirm
from rich.syntax import Syntax
from rich.text import Text
from rich.columns import Columns
import asyncio

console = Console()

# Configuration
# NOTE: Set API_URL below to your own Perplexica API endpoint.
API_URL = "http://localhost:3000/api/search"  # <-- Set this to your own API endpoint

console.print("This script is configured to use a private API endpoint. Make sure you've set API_URL to your own endpoint and are connected.")

# Focus modes with emojis
FOCUS_MODES = {
    "webSearch": {"emoji": "🌐", "name": "Web Search", "desc": "Search the entire web"},
    "academicSearch": {"emoji": "🎓", "name": "Academic Search", "desc": "Search academic papers and research"},
    "writingAssistant": {"emoji": "✍️", "name": "Writing Assistant", "desc": "Help with writing and editing"},
    "wolframAlphaSearch": {"emoji": "🧮", "name": "Wolfram Alpha", "desc": "Computational and mathematical queries"},
    "youtubeSearch": {"emoji": "📺", "name": "YouTube Search", "desc": "Search YouTube videos"},
    "redditSearch": {"emoji": "🤖", "name": "Reddit Search", "desc": "Search Reddit discussions (not currently working in Perplexica)"}
}

# Optimization modes
OPTIMIZATION_MODES = {
    "speed": {"emoji": "⚡", "name": "Speed", "desc": "Fast responses"},
    "balanced": {"emoji": "⚖️", "name": "Balanced", "desc": "Balanced speed and quality"}
}

class PerplexicaCLI:
    def __init__(self):
        self.history: List[Tuple[str, str]] = []
        self.focus_mode = "webSearch"
        self.optimization_mode = "balanced"
        self.chat_model = {"provider": "openai", "name": "gpt-4o-mini"}
        self.embedding_model = {"provider": "openai", "name": "text-embedding-3-large"}
        
    def display_welcome(self):
        """Display welcome message with ASCII art"""
        welcome_text = """
[bold cyan]╔═══════════════════════════════════════╗
║      🔮 Perplexica Search CLI 🔮      ║
║ AI-Powered Search in your Terminal!   ║
╚═══════════════════════════════════════╝[/bold cyan]
"""
        console.print(welcome_text)
        
    def display_focus_modes(self):
        """Display available focus modes in a beautiful table"""
        table = Table(title="🎯 Focus Modes", show_header=True, header_style="bold magenta")
        table.add_column("Mode", style="cyan", no_wrap=True)
        table.add_column("Icon", justify="center")
        table.add_column("Description", style="dim")
        
        for key, mode in FOCUS_MODES.items():
            table.add_row(key, mode["emoji"], f"{mode['name']} - {mode['desc']}")
            
        console.print(table)
        
    def get_mode_emoji(self) -> str:
        """Get emoji for current focus mode"""
        return FOCUS_MODES.get(self.focus_mode, {}).get("emoji", "🔍")
        
    def format_sources(self, sources: List[dict]) -> None:
        """Format and display sources"""
        if not sources:
            return
            
        console.print("\n📚 [bold yellow]Sources:[/bold yellow]")
        for i, source in enumerate(sources, 1):
            metadata = source.get("metadata", {})
            title = metadata.get("title", "Untitled")
            url = metadata.get("url", "")
            
            source_panel = Panel(
                f"[bold]{title}[/bold]\n[link={url}]{url}[/link]\n\n[dim]{source.get('pageContent', '')[:200]}...[/dim]",
                title=f"[{i}]",
                border_style="blue"
            )
            console.print(source_panel)
            
    async def stream_response(self, query: str, system_instructions: Optional[str] = None):
        """Stream response from API"""
        payload = {
            "chatModel": self.chat_model,
            "embeddingModel": self.embedding_model,
            "optimizationMode": self.optimization_mode,
            "focusMode": self.focus_mode,
            "query": query,
            "history": self.history,
            "stream": True
        }
        
        if system_instructions:
            payload["systemInstructions"] = system_instructions
            
        async with httpx.AsyncClient() as client:
            response_text = ""
            sources = []
            
            with Live(Spinner("dots", text="[cyan]Thinking...[/cyan]"), refresh_per_second=10) as live:
                async with client.stream('POST', API_URL, json=payload, timeout=60.0) as response:
                    response.raise_for_status()
                    
                    async for line in response.aiter_lines():
                        if line.strip():
                            try:
                                data = json.loads(line)
                                
                                if data["type"] == "sources":
                                    sources = data["data"]
                                    live.update("[green]✓ Sources found[/green]")
                                    
                                elif data["type"] == "response":
                                    response_text += data["data"]
                                    # Update live display with markdown
                                    md = Markdown(response_text)
                                    panel = Panel(
                                        md,
                                        title=f"{self.get_mode_emoji()} Response",
                                        border_style="green",
                                        expand=False
                                    )
                                    live.update(panel)
                                    
                                elif data["type"] == "done":
                                    break
                                    
                            except json.JSONDecodeError:
                                console.print(f"[red]Error parsing response: {line}[/red]")
                                
            # Add to history
            self.history.append(["human", query])
            self.history.append(["assistant", response_text])
            
            # Display sources
            self.format_sources(sources)
            
    async def search(self, query: str, system_instructions: Optional[str] = None):
        """Perform a search query"""
        mode_info = FOCUS_MODES[self.focus_mode]
        console.print(f"\n{mode_info['emoji']} [bold cyan]Searching with {mode_info['name']}...[/bold cyan]")
        
        try:
            await self.stream_response(query, system_instructions)
        except httpx.HTTPError as e:
            console.print(f"[red]❌ HTTP Error: {e}[/red]")
        except Exception as e:
            console.print(f"[red]❌ Error: {e}[/red]")
            
    def interactive_mode(self):
        """Run in interactive mode"""
        self.display_welcome()
        
        while True:
            # Display current mode
            mode_info = FOCUS_MODES[self.focus_mode]
            opt_info = OPTIMIZATION_MODES[self.optimization_mode]
            
            status = Text()
            status.append("Mode: ", style="bold")
            status.append(f"{mode_info['emoji']} {mode_info['name']}", style="cyan")
            status.append(" | ", style="dim")
            status.append("Speed: ", style="bold")
            status.append(f"{opt_info['emoji']} {opt_info['name']}", style="yellow")
            
            console.print(Panel(status, expand=False))
            
            # Get user input
            console.print("\n[bold]Commands:[/bold] /mode, /speed, /clear, /help, /exit")
            query = Prompt.ask(f"\n{self.get_mode_emoji()} [bold]Ask anything[/bold]")
            
            if query.startswith("/"):
                self.handle_command(query)
            elif query.strip():
                # Check for system instructions
                instructions = None
                if "|" in query:
                    query_parts = query.split("|", 1)
                    query = query_parts[0].strip()
                    instructions = query_parts[1].strip()
                    console.print(f"[dim]Using instructions: {instructions}[/dim]")
                    
                asyncio.run(self.search(query, instructions))
            
    def handle_command(self, command: str):
        """Handle special commands"""
        cmd = command.lower().strip()
        
        if cmd == "/mode":
            self.display_focus_modes()
            new_mode = Prompt.ask("Select focus mode", default=self.focus_mode)
            if new_mode in FOCUS_MODES:
                self.focus_mode = new_mode
                console.print(f"[green]✓ Switched to {FOCUS_MODES[new_mode]['name']}[/green]")
            else:
                console.print("[red]Invalid mode[/red]")
                
        elif cmd == "/speed":
            table = Table(title="⚡ Optimization Modes")
            table.add_column("Mode", style="cyan")
            table.add_column("Icon", justify="center")
            table.add_column("Description")
            
            for key, mode in OPTIMIZATION_MODES.items():
                table.add_row(key, mode["emoji"], mode["desc"])
                
            console.print(table)
            new_mode = Prompt.ask("Select optimization mode", default=self.optimization_mode)
            if new_mode in OPTIMIZATION_MODES:
                self.optimization_mode = new_mode
                console.print(f"[green]✓ Switched to {OPTIMIZATION_MODES[new_mode]['name']}[/green]")
                
        elif cmd == "/clear":
            self.history = []
            console.clear()
            self.display_welcome()
            console.print("[green]✓ History cleared[/green]")
            
        elif cmd == "/help":
            help_text = """
[bold cyan]🔮 Perplexica CLI Help[/bold cyan]

[bold]Commands:[/bold]
  /mode     - Change search focus mode
  /speed    - Change optimization mode
  /clear    - Clear chat history and screen
  /help     - Show this help message
  /exit     - Exit the application

[bold]Tips:[/bold]
  • Use | to add system instructions: "query | write in Spanish"
  • Press Ctrl+C to interrupt a search
  • History is maintained for context-aware responses
"""
            console.print(Panel(help_text, title="Help", border_style="blue"))
            
        elif cmd == "/exit":
            if Confirm.ask("Are you sure you want to exit?"):
                console.print("[yellow]👋 Goodbye![/yellow]")
                sys.exit(0)
        else:
            console.print(f"[red]Unknown command: {command}[/red]")

@click.command()
@click.option('--mode', '-m', type=click.Choice(list(FOCUS_MODES.keys())),
              help='Focus mode for search')
@click.option('--query', '-q', help='Direct query (non-interactive mode)')
@click.option('--speed', '-s', type=click.Choice(['speed', 'balanced']),
              default='balanced', help='Optimization mode')
@click.option('--version', is_flag=True, help='Show version and exit')
def main(mode: Optional[str], query: Optional[str], speed: str, version: bool):
    """
    🔮 Perplexica CLI - Beautiful AI-powered search in your terminal
    """
    VERSION = "1.0.0"
    if version:
        console.print(f"[bold cyan]Perplexica CLI version {VERSION}[/bold cyan]")
        sys.exit(0)

    cli = PerplexicaCLI()
    cli.optimization_mode = speed

    if mode:
        cli.focus_mode = mode

    # Support reading query from stdin if --query is "-"
    if query == "-":
        query = sys.stdin.read().strip()

    if query:
        # Non-interactive mode
        asyncio.run(cli.search(query))
    else:
        # Interactive mode
        try:
            cli.interactive_mode()
        except KeyboardInterrupt:
            console.print("\n[yellow]👋 Goodbye![/yellow]")
            sys.exit(0)

if __name__ == "__main__":
    # Check minimum Python version
    import platform
    if sys.version_info < (3, 8):
        sys.stderr.write("Perplexica CLI requires Python 3.8 or newer.\n")
        sys.exit(1)
    main()