from __future__ import annotations from pathlib import Path from universal_parser.observability.metrics import MetricsCollector def generate_dashboard_html(collector: MetricsCollector | None = None) -> str: """Generates a self-contained, responsive HTML observability dashboard.""" col = collector or MetricsCollector() summary = col.get_summary() events = col.get_events() # Generate format table rows format_rows = "" for fmt, count in summary["format_breakdown"].items(): format_rows += f"{fmt.upper()}{count}\n" if not format_rows: format_rows = "No events recorded yet" # Generate recent events table rows event_rows = "" for e in reversed(events[-20:]): anomaly_badge = ( f"{len(e.anomalies)} anomalies" if e.anomalies else "clean" ) event_rows += f""" {e.file_name} {e.file_type.upper()} {e.page_count} {e.element_count} {e.duration_ms} ms {anomaly_badge} """ if not event_rows: event_rows = "No events recorded yet" html = f""" Universal Parser — Observability & Drift Dashboard

Universal Parser — Observability Dashboard

Real-time performance telemetry, anomaly tracking, and multi-format throughput.

Total Documents
{summary["total_documents"]}
Total Pages
{summary["total_pages"]}
Avg Latency
{summary["avg_duration_ms"]} ms
Drift / Anomalies
0 else "var(--success)"};">{summary["anomaly_count"]}

Formats Handled

{format_rows}
FormatDocuments

Recent Ingestion Stream

{event_rows}
File Type Pages Elements Latency Status
""" return html def export_dashboard(output_path: str | Path, collector: MetricsCollector | None = None) -> Path: """Exports the observability dashboard to an HTML file on disk.""" path = Path(output_path) path.parent.mkdir(parents=True, exist_ok=True) html = generate_dashboard_html(collector) path.write_text(html, encoding="utf-8") return path