zzstoatzz.io
search for standard sites
pub-search.waow.tech
search
zig
blog
atproto
1#!/usr/bin/env python3
2# /// script
3# requires-python = ">=3.11"
4# dependencies = ["httpx", "rich", "plotext", "typer"]
5# ///
6"""pub-search terminal dashboard
7
8usage:
9 uv run scripts/dashboard.py # default view
10 uv run scripts/dashboard.py --days 14 # longer timeline
11"""
12
13import httpx
14import plotext as plt
15import typer
16from rich.console import Console
17from rich.panel import Panel
18from rich.table import Table
19
20API_BASE = "https://leaflet-search-backend.fly.dev"
21console = Console()
22app = typer.Typer(add_completion=False)
23
24
25def fetch_stats() -> dict:
26 """fetch /stats endpoint"""
27 resp = httpx.get(f"{API_BASE}/stats", timeout=30)
28 resp.raise_for_status()
29 return resp.json()
30
31
32def fetch_dashboard() -> dict:
33 """fetch /api/dashboard endpoint"""
34 resp = httpx.get(f"{API_BASE}/api/dashboard", timeout=30)
35 resp.raise_for_status()
36 return resp.json()
37
38
39def display_overview(stats: dict) -> None:
40 """show document/publication counts"""
41 table = Table(show_header=False, box=None, padding=(0, 2), expand=True)
42 table.add_column(style="dim")
43 table.add_column(style="bold green", justify="right")
44
45 table.add_row("documents", f"{stats['documents']:,}")
46 table.add_row("publications", f"{stats['publications']:,}")
47 table.add_row("embeddings", f"{stats['embeddings']:,}")
48
49 embed_pct = (stats['embeddings'] / stats['documents'] * 100) if stats['documents'] > 0 else 0
50 table.add_row("embedded", f"{embed_pct:.0f}%")
51
52 console.print(Panel(table, title="[bold]index[/]", border_style="blue", expand=False))
53
54
55def display_usage(stats: dict) -> None:
56 """show usage and similarity cache stats"""
57 hits = stats.get('cache_hits', 0)
58 misses = stats.get('cache_misses', 0)
59 total = hits + misses
60 hit_rate = (hits / total * 100) if total > 0 else 0
61
62 table = Table(show_header=False, box=None, padding=(0, 2), expand=True)
63 table.add_column(style="dim")
64 table.add_column(style="bold cyan", justify="right")
65
66 table.add_row("searches", f"{stats.get('searches', 0):,}")
67 table.add_row("errors", f"{stats.get('errors', 0):,}")
68 table.add_row("similar cache hit", f"{hit_rate:.0f}% ({hits}/{total})")
69
70 console.print(Panel(table, title="[bold]usage[/]", border_style="cyan", expand=False))
71
72
73def display_latency(stats: dict) -> None:
74 """show latency percentiles"""
75 timing = stats.get('timing', {})
76 if not timing:
77 return
78
79 table = Table(box=None, padding=(0, 1), expand=True)
80 table.add_column("endpoint", style="dim")
81 table.add_column("p50", justify="right", style="green")
82 table.add_column("p95", justify="right", style="yellow")
83 table.add_column("p99", justify="right", style="red")
84 table.add_column("count", justify="right", style="dim")
85
86 for endpoint in ['search', 'similar', 'tags', 'popular']:
87 if endpoint in timing:
88 t = timing[endpoint]
89 table.add_row(
90 endpoint,
91 f"{t['p50_ms']:.0f}ms",
92 f"{t['p95_ms']:.0f}ms",
93 f"{t['p99_ms']:.0f}ms",
94 f"{t['count']:,}",
95 )
96
97 console.print(Panel(table, title="[bold]latency[/]", border_style="magenta", expand=False))
98
99
100def display_timeline(dashboard: dict, days: int) -> None:
101 """show indexing activity chart"""
102 timeline = dashboard.get('timeline', [])[:days]
103 if not timeline:
104 return
105
106 timeline = list(reversed(timeline)) # oldest first
107 dates = [d['date'][-5:] for d in timeline] # MM-DD
108 counts = [d['count'] for d in timeline]
109
110 plt.clear_figure()
111 plt.theme("dark")
112 plt.title("documents indexed per day")
113 plt.bar(dates, counts, color="cyan")
114 plt.plotsize(70, 12)
115 plt.show()
116 print()
117
118
119def display_latency_chart(stats: dict) -> None:
120 """bar chart of p50 latencies by endpoint"""
121 timing = stats.get('timing', {})
122 if not timing:
123 return
124
125 endpoints = []
126 p50s = []
127 for endpoint in ['search', 'similar', 'tags', 'popular']:
128 if endpoint in timing:
129 endpoints.append(endpoint)
130 p50s.append(timing[endpoint]['p50_ms'])
131
132 plt.clear_figure()
133 plt.theme("dark")
134 plt.title("p50 latency by endpoint (ms)")
135 plt.bar(endpoints, p50s, color="cyan")
136 plt.plotsize(50, 10)
137 plt.show()
138 print()
139
140
141@app.command()
142def main(
143 days: int = typer.Option(7, "-d", "--days", help="days of timeline to show"),
144) -> None:
145 """pub-search terminal dashboard"""
146 console.print("\n[bold cyan]pub-search[/] dashboard\n")
147
148 try:
149 stats = fetch_stats()
150 dashboard = fetch_dashboard()
151 except httpx.HTTPError as e:
152 console.print(f"[red]error fetching data:[/] {e}")
153 raise typer.Exit(1)
154
155 display_overview(stats)
156 display_usage(stats)
157 display_latency(stats)
158 print()
159 display_timeline(dashboard, days)
160 display_latency_chart(stats)
161
162
163if __name__ == "__main__":
164 app()