"""Query analytics HTML dashboard (v1.98).
Visual complement to v1.89 (query_stats) + v1.90 (promote candidates)
+ v1.96 (query × doc affinity). Same design as v1.86 content health UI:
inline CSS, zero JS framework, zero CDN, ?refresh=N for wall screens.
Sections:
* Banner: n_unique_queries, n_events, avg_sources_per_query.
* Top queries (by count).
* Promote candidates (count ≥ threshold, not already a preset).
* Query × doc affinity — top N queries × top M docs as a heat-table.
* Isolated queries — those with 0 cited docs (retrieval gap).
Style mirrors the v1.86 dashboard so ops have one visual language.
"""
from __future__ import annotations
import html as _h
from typing import Any, Dict, List, Optional
def _esc(s: object) -> str:
return _h.escape(str(s), quote=True)
def _table(headers: List[str], rows: List[List[str]],
empty_msg: str = "— no data yet —") -> str:
if not rows:
return (f'
{_esc(empty_msg)}
')
th = "".join(f'{_esc(h)} | '
for h in headers)
body = ""
for row in rows:
tds = "".join(
f'{c} | '
for c in row
)
body += f'{tds}
'
return (f'')
def _card(title: str, body_html: str, hint: str = "") -> str:
hint_html = (f''
f'{_esc(hint)}
') if hint else ""
return (
f''
f'
{_esc(title)}
'
f'{hint_html}'
f'{body_html}'
)
def _cell_color(count: int, max_count: int) -> str:
"""Heatmap cell background: empty (0) → neutral, max → deep blue."""
if count <= 0 or max_count <= 0:
return "#f9fafb"
ratio = min(1.0, count / max_count)
# Blend from #f9fafb (light) to #1e40af (deep blue)
# Simple mix: higher count = darker blue
r = int(0xf9 + (0x1e - 0xf9) * ratio)
g = int(0xfa + (0x40 - 0xfa) * ratio)
b = int(0xfb + (0xaf - 0xfb) * ratio)
return f"#{r:02x}{g:02x}{b:02x}"
def _heatmap(queries: List[Dict], docs: List[str],
pair_lookup: Dict, ) -> str:
"""Render a query × doc heat table. pair_lookup maps (fp, doc_id) → count."""
if not queries or not docs:
return ('— not enough data for matrix —
')
max_count = 0
for q in queries:
for did in docs:
c = pair_lookup.get((q["fingerprint"], did), 0)
if c > max_count:
max_count = c
header_cells = "".join(
f'{_esc(did)} | '
for did in docs
)
body_rows = ""
for q in queries:
row_label = f'{_esc(q.get("sample") or q["fingerprint"])[:40]}'
row_cells = ""
for did in docs:
c = pair_lookup.get((q["fingerprint"], did), 0)
bg = _cell_color(c, max_count)
text = str(c) if c > 0 else ""
color = "#fff" if c > max_count * 0.6 else "#1f2937"
row_cells += (
f'{text} | '
)
body_rows += (
f'| '
f'{row_label} | {row_cells}
'
)
return (
f''
f'| query ↓ / doc → | '
f'{header_cells}
'
f'{body_rows}
'
)
def render_query_analytics_ui(
summary: Dict[str, Any],
top_queries: List[Dict[str, Any]],
promote_candidates: List[Dict[str, Any]],
matrix_queries: List[Dict[str, Any]],
matrix_docs: List[str],
matrix_pairs: Dict,
refresh_sec: int = 0,
) -> str:
"""Render the HTML. All inputs are plain dicts/lists — pure render.
* ``matrix_pairs``: dict keyed by ``(fingerprint, doc_id)`` → count.
Caller pre-computes this so the renderer is O(N·M) lookup only.
"""
# Banner
banner = (
f''
f'
Query analytics overview
'
f'
'
f'
unique queries'
f'
{summary.get("n_unique_queries", 0)}
'
f'
total events'
f'
{summary.get("n_events", 0)}
'
f'
avg sources'
f'
'
f'{summary.get("avg_sources_per_query", 0.0):.2f}
'
f'
'
)
# Top queries table
top_rows = [
[
_esc(row.get("sample") or row["fingerprint"])[:60],
str(row["count"]),
f'{row.get("avg_sources", 0.0):.2f}',
f'{row.get("avg_latency_ms", 0.0):.0f}',
]
for row in top_queries
]
top_block = _card(
"Top queries (by count)",
_table(["query", "count", "avg_sources", "avg_latency_ms"],
top_rows, empty_msg="no queries recorded yet"),
hint="user questions ordered by frequency — "
"top-of-list queries deserve saved presets",
)
# Promote candidates
promote_rows = [
[
_esc(row["suggested_preset_name"])[:40],
_esc(row.get("sample") or row["fingerprint"])[:50],
str(row["count"]),
f'{row.get("avg_sources", 0.0):.2f}',
f'{row.get("avg_latency_ms", 0.0):.0f}',
]
for row in promote_candidates
]
promote_block = _card(
f'Promote candidates ({len(promote_candidates)})',
_table(
["suggested_name", "sample", "count",
"avg_sources", "avg_latency_ms"],
promote_rows,
empty_msg="no candidates — everyone's already a preset, "
"or nothing crosses the threshold",
),
hint="queries asked ≥ 3× that aren't already saved as presets — "
"click POST /v1/admin/queries/promote to ship them all",
)
# Heatmap
heat_block = _card(
f'Query × doc affinity matrix ({len(matrix_queries)} × '
f'{len(matrix_docs)})',
_heatmap(matrix_queries, matrix_docs, matrix_pairs),
hint="darker = more often cited together. cells reveal "
"which docs carry which questions",
)
meta_html = ""
if refresh_sec > 0:
meta_html = f''
return f"""
tau-rag · query analytics
{meta_html}
🔎 tau-rag · query analytics
v1.89 fingerprints + v1.90 promote candidates +
v1.96 query × doc affinity, combined into one view.
{'Auto-refresh every ' + str(int(refresh_sec)) + 's.'
if refresh_sec > 0 else ''}
{banner}
{top_block}
{promote_block}
{heat_block}
"""
__all__ = ["render_query_analytics_ui"]