app.py

# VVPS Observatory app -- docs/StoxoMDs/Conv_2026-06-07_HF_CPU_Spaces_Exploration.md#space-a-vvps-observatory
"""VVPS Observatory — Gradio dashboard + background polling scheduler.

Private HF Space that externally monitors the Stoxopia VPS.
Polls REST endpoints (zero new attack surface), alerts via Telegram,
logs history to HF Dataset.
"""

import threading
import time
import logging
from datetime import datetime, timezone

import gradio as gr

import config
from poller import Poller
from alerts import AlertEngine
from history import HistoryLogger

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
logger = logging.getLogger(__name__)

# --- Global instances ---
poller = Poller()
alert_engine = AlertEngine()
history_logger = HistoryLogger()


# --- Background polling thread ---
def _polling_loop():
    """Runs in a daemon thread, polls every POLL_INTERVAL_SECONDS."""
    logger.info(
        "Polling loop started (interval=%ds, target=%s)",
        config.POLL_INTERVAL_SECONDS,
        config.VPS_BASE_URL,
    )
    while True:
        try:
            snapshot = poller.poll_once()

            # Evaluate alerts
            alert_engine.evaluate(snapshot)
            alert_engine.evaluate_freshness(
                snapshot.freshness, poller.consecutive_failures
            )

            # Buffer for history
            history_logger.append(snapshot)
            history_logger.maybe_flush()

        except Exception as e:
            logger.error("Polling loop error: %s", e)

        time.sleep(config.POLL_INTERVAL_SECONDS)


# Start polling in background
_poll_thread = threading.Thread(target=_polling_loop, daemon=True)
_poll_thread.start()


# --- Gradio Dashboard ---
def _status_tab():
    """Current health status display."""
    snapshot = poller.latest
    if snapshot is None:
        return "⏳ Waiting for first poll...", "", "", ""

    ts = datetime.fromtimestamp(snapshot.timestamp, tz=timezone.utc)
    time_str = ts.strftime("%Y-%m-%d %H:%M:%S UTC")

    # Status indicator
    status_map = {
        "healthy": "🟢 HEALTHY",
        "degraded": "🟡 DEGRADED",
        "unreachable": "🔴 UNREACHABLE",
    }
    status_display = status_map.get(snapshot.health_status, "⚪ UNKNOWN")

    # Components table
    comp_lines = []
    for comp, st in snapshot.components.items():
        icon = "🟢" if st == "healthy" else "🔴"
        comp_lines.append(f"  {icon} {comp}: {st}")
    components_str = "\n".join(comp_lines) if comp_lines else "  No data yet"

    # Latency
    latency_str = f"  Response time: {snapshot.latency_ms:.0f}ms"
    if snapshot.error:
        latency_str += f"\n  Error: {snapshot.error}"

    return (
        f"## {status_display}\n\nLast check: {time_str}",
        f"### Components\n```\n{components_str}\n```",
        f"### Latency\n```\n{latency_str}\n```",
        f"### Active Alerts: {len(alert_engine.active_alerts)}",
    )


def _freshness_tab():
    """Per-section freshness display."""
    snapshot = poller.latest
    if snapshot is None or not snapshot.freshness:
        return "⏳ Waiting for freshness data..."

    freshness = snapshot.freshness
    sources = freshness.get("sources", [])
    if not sources:
        return "No freshness sources returned."

    lines = [
        "| Source | Status | Last Sync | Records |",
        "|--------|--------|-----------|---------|",
    ]
    for src in sources:
        name = src.get("source_name", "unknown")
        status = src.get("status", "unknown")
        icon = "🟢" if status == "healthy" else "🟡" if status == "stale" else "🔴"
        last_sync = src.get("last_sync", "never")
        records = src.get("records_updated", "?")
        lines.append(f"| {icon} {name} | {status} | {last_sync} | {records} |")

    return "\n".join(lines)


def _history_tab():
    """7-day uptime and latency summary."""
    history = poller.history
    if not history:
        return "⏳ No history yet (accumulates over time)..."

    total = len(history)
    healthy = sum(1 for s in history if s.health_status == "healthy")
    uptime_pct = (healthy / total * 100) if total > 0 else 0

    latencies = [s.latency_ms for s in history if s.latency_ms > 0]
    avg_latency = sum(latencies) / len(latencies) if latencies else 0
    max_latency = max(latencies) if latencies else 0

    unreachable_count = sum(1 for s in history if s.health_status == "unreachable")

    hours_covered = (total * config.POLL_INTERVAL_SECONDS) / 3600

    summary = f"""### Uptime Summary ({hours_covered:.1f}h window)

| Metric | Value |
|--------|-------|
| Uptime | {uptime_pct:.1f}% ({healthy}/{total} checks) |
| Avg Latency | {avg_latency:.0f}ms |
| Max Latency | {max_latency:.0f}ms |
| Unreachable Events | {unreachable_count} |
| Total Checks | {total} |
"""
    return summary


def _alerts_tab():
    """Recent alert log."""
    log = alert_engine.alert_log
    if not log:
        return "### No alerts fired yet ✅"

    lines = [
        "### Recent Alerts\n",
        "| Time | Condition | Resolved |",
        "|------|-----------|----------|",
    ]
    for alert in reversed(log[-20:]):
        ts = datetime.fromtimestamp(alert.fired_at, tz=timezone.utc)
        time_str = ts.strftime("%m-%d %H:%M UTC")
        resolved = "✅" if alert.resolved else "🔴 Active"
        lines.append(f"| {time_str} | {alert.condition} | {resolved} |")

    return "\n".join(lines)


# --- Build Gradio app ---
with gr.Blocks(
    title="VVPS Observatory",
    theme=gr.themes.Soft(),
) as app:
    gr.Markdown("# 🔭 VVPS Observatory\nExternal health monitor for Stoxopia VPS")

    with gr.Tabs():
        with gr.TabItem("Status"):
            status_out = gr.Markdown()
            components_out = gr.Markdown()
            latency_out = gr.Markdown()
            alerts_summary = gr.Markdown()

            def refresh_status():
                return _status_tab()

            app.load(
                refresh_status,
                outputs=[status_out, components_out, latency_out, alerts_summary],
            )
            # Auto-refresh every 60s
            refresh_btn = gr.Button("🔄 Refresh", size="sm")
            refresh_btn.click(
                refresh_status,
                outputs=[status_out, components_out, latency_out, alerts_summary],
            )

        with gr.TabItem("Freshness"):
            freshness_out = gr.Markdown()

            def refresh_freshness():
                return _freshness_tab()

            app.load(refresh_freshness, outputs=[freshness_out])
            refresh_btn2 = gr.Button("🔄 Refresh", size="sm")
            refresh_btn2.click(refresh_freshness, outputs=[freshness_out])

        with gr.TabItem("History"):
            history_out = gr.Markdown()

            def refresh_history():
                return _history_tab()

            app.load(refresh_history, outputs=[history_out])
            refresh_btn3 = gr.Button("🔄 Refresh", size="sm")
            refresh_btn3.click(refresh_history, outputs=[history_out])

        with gr.TabItem("Alerts"):
            alerts_out = gr.Markdown()

            def refresh_alerts():
                return _alerts_tab()

            app.load(refresh_alerts, outputs=[alerts_out])
            refresh_btn4 = gr.Button("🔄 Refresh", size="sm")
            refresh_btn4.click(refresh_alerts, outputs=[alerts_out])

    gr.Markdown(
        f"---\n*Polling every {config.POLL_INTERVAL_SECONDS}s | "
        f"Target: {config.VPS_BASE_URL} | "
        f"History flush: every {config.HISTORY_WRITE_INTERVAL_SECONDS}s*"
    )

if __name__ == "__main__":
    auth = None
    if config.DASHBOARD_PASSWORD:
        auth = (config.DASHBOARD_USERNAME, config.DASHBOARD_PASSWORD)
    # ssr_mode=False required: Gradio 5 SSR health check can't authenticate
    app.launch(server_name="0.0.0.0", server_port=7860, auth=auth, ssr_mode=False)