|
|
@@ -0,0 +1,305 @@
|
|
|
+#!/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()
|