用纯 Python 构建可交互 Dashboard：Prefab 响应式 UI 与静态 HTML 导出教程

import base64
import subprocess
import sys
from pathlib import Path

from IPython.display import FileLink, HTML, display

PREFAB_VERSION = "0.20.2"
APP_PATH = Path("/content/prefab_advanced_tutorial_app.py")
HTML_PATH = Path("/content/prefab_advanced_dashboard.html")

subprocess.check_call([
    sys.executable,
    "-m",
    "pip",
    "install",
    "-q",
    f"prefab-ui=={PREFAB_VERSION}",
])

python3 -m venv .venv
source .venv/bin/activate
python -m pip install 'prefab-ui==0.20.2'

from __future__ import annotations

import random
from collections import Counter, defaultdict
from datetime import date, timedelta
from typing import Any

REGIONS = ["APAC", "EMEA", "NA", "LATAM"]
PIPELINES = ["LLM Triage", "Risk Scoring", "Warehouse Load"]
STATES = ["Completed", "Completed", "Completed", "Late", "Failed"]
PRIORITIES = ["P0", "P1", "P2", "P3"]


def build_runs(days: int = 10) -> list[dict[str, Any]]:
    random.seed(42)
    today = date.today()
    runs: list[dict[str, Any]] = []

    for day_offset in range(days):
        run_date = today - timedelta(days=days - day_offset - 1)
        for region in REGIONS:
            for index in range(random.randint(8, 16)):
                state = random.choice(STATES)
                duration = random.randint(40, 220)
                runs.append({
                    "run_id": f"{run_date:%m%d}-{region[:2]}-{index:03d}",
                    "date": run_date.isoformat(),
                    "region": region,
                    "pipeline": random.choice(PIPELINES),
                    "state": state,
                    "priority": random.choice(PRIORITIES),
                    "duration_s": duration,
                    "sla_gap": round(max(0, duration - 120) / 60, 1),
                })

    return runs


def summarize_by_region(runs: list[dict[str, Any]]) -> list[dict[str, Any]]:
    buckets: dict[str, list[dict[str, Any]]] = defaultdict(list)
    for run in runs:
        buckets[run["region"]].append(run)

    rows: list[dict[str, Any]] = []
    for region, items in buckets.items():
        failures = sum(1 for item in items if item["state"] == "Failed")
        late = sum(1 for item in items if item["state"] == "Late")
        success_rate = round(100 * (len(items) - failures - late * 0.35) / len(items), 1)
        rows.append({
            "region": region,
            "runs": len(items),
            "failures": failures,
            "late": late,
            "success_rate": success_rate,
            "avg_latency": round(sum(item["duration_s"] for item in items) / len(items), 1),
        })
    return sorted(rows, key=lambda row: row["region"])


def make_watchlist(runs: list[dict[str, Any]]) -> list[dict[str, Any]]:
    risky = [run for run in runs if run["state"] in {"Failed", "Late"}]
    return sorted(risky, key=lambda run: (run["priority"], -run["sla_gap"]))[:8]


if __name__ == "__main__":
    runs = build_runs()
    status_counts = Counter(run["state"] for run in runs)
    print("status_counts", dict(status_counts))
    print("regions", summarize_by_region(runs))
    print("watchlist_size", len(make_watchlist(runs)))

status_counts {'Completed': ..., 'Late': ..., 'Failed': ...}
regions [{'region': 'APAC', 'runs': ..., 'success_rate': ...}, ...]
watchlist_size 8

initial_state = {
    "selected_region": "All",
    "line_rows": DAILY_BY_REGION["All"],
    "table_rows": RUNS_BY_REGION["All"],
    "status_rows": STATUS_BY_REGION["All"],
    "pipeline_rows": PIPELINE_BY_REGION["All"],
    "region_kpis": KPI_BY_REGION["All"],
    "selected_run": None,
    "slo_target": 94,
    "operator_name": "Colab Builder",
    "notes": [{"note": "Click a run row to inspect it, then add triage notes here."}],
    "watchlist": WATCHLIST,
    "compact": False,
    "dark_mode": False,
}

from prefab_ui.app import PrefabApp
from prefab_ui.components import Badge, Button, Card, CardContent, CardHeader, CardTitle, Column, Grid, H2, Muted, Row, Slider, Tabs, Tab
from prefab_ui.rx import STATE

with PrefabApp(
    title="Prefab Advanced Colab Tutorial",
    state=initial_state,
    css_class="max-w-7xl mx-auto p-6",
) as app:
    with Column(gap=6):
        with Row(justify="between", align="center", gap=4):
            with Column(gap=1):
                H2("Prefab Advanced Operations Dashboard")
                Muted("Everything below is composed in Python.")
            Badge(f"Region: {STATE.selected_region}", variant="info")

        with Tabs(value="overview"):
            with Tab("Overview", value="overview"):
                with Card():
                    with CardHeader():
                        CardTitle("Interactive controls")
                    with CardContent():
                        Slider(
                            value=STATE.slo_target,
                            min=80,
                            max=99,
                            step=0.5,
                        )

from prefab_ui.actions import SetState, ShowToast

REGION_ACTIONS = {
    region: [
        SetState("selected_region", region),
        SetState("line_rows", DAILY_BY_REGION[region]),
        SetState("table_rows", RUNS_BY_REGION[region]),
        SetState("status_rows", STATUS_BY_REGION[region]),
        SetState("pipeline_rows", PIPELINE_BY_REGION[region]),
        SetState("region_kpis", KPI_BY_REGION[region]),
        SetState("selected_run", None),
        ShowToast(f"Region set to {region}", variant="info", duration=1800),
    ]
    for region in REGIONS
}

from prefab_ui.actions import SetState
from prefab_ui.components import DataTable, DataTableColumn
from prefab_ui.components.control_flow import If
from prefab_ui.rx import EVENT, STATE

DataTable(
    columns=[
        DataTableColumn(key="run_id", header="Run ID", sortable=True),
        DataTableColumn(key="date", header="Date", sortable=True),
        DataTableColumn(key="pipeline", header="Pipeline", sortable=True),
        DataTableColumn(key="region", header="Region", sortable=True),
        DataTableColumn(key="state", header="State", sortable=True),
        DataTableColumn(key="priority", header="Priority", sortable=True),
    ],
    rows=STATE.table_rows,
    search=True,
    paginated=True,
    pageSize=12,
    onRowClick=SetState("selected_run", EVENT),
)

with If(STATE.selected_run):
    # 这里渲染 selected_run 的详情卡片
    pass

from prefab_ui.actions import AppendState, PopState, ShowToast
from prefab_ui.components import Button, Form, Input
from prefab_ui.components.control_flow import ForEach

with Form(
    on_submit=[
        AppendState("notes", EVENT),
        ShowToast("Note added", variant="success"),
    ]
):
    Input(
        name="note",
        placeholder="Example: Escalate APAC LLM Triage failures",
        required=True,
    )
    Button("Add note", buttonType="submit")

Button(
    "Remove latest note",
    onClick=PopState("notes", -1),
)

with ForEach("notes") as note:
    # 这里渲染 note.note
    pass

prefab export /content/prefab_advanced_tutorial_app.py -o /content/prefab_advanced_dashboard.html

prefab serve /content/prefab_advanced_tutorial_app.py --reload

html_bytes = HTML_PATH.read_bytes()
encoded = base64.b64encode(html_bytes).decode("utf-8")

display(HTML(f"""
<iframe
  src="data:text/html;base64,{encoded}"
  width="100%"
  height="950"
  style="border: 1px solid #ddd; border-radius: 12px;"
></iframe>
"""))