使用 Prefab 反應式 UI 元件與靜態 HTML 匯出，設計 Python 優先的互動儀錶板

In this tutorial, we build a Prefab application that demonstrates how to create interactive dashboards entirely in Python. We use Prefab’s component-based Python interface to design a polished operations dashboard with reactive state, charts, tables, filters, forms, tabs, alerts, metrics, and client-side actions. We generate realistic pipeline monitoring data, connect it to live UI controls, and export the final app as a static HTML dashboard that we can preview directly inside Google Colab. Through this workflow, we learn how Prefab lets us move from Python data logic to a modern React-powered user interface without having to write frontend code manually. Installing Prefab in Colab Copy CodeCopiedUse a different Browserimport os import sys import base64 import subprocess from pathlib import Path from IPython.display import HTML, display, FileLink 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}", ]) APP_CODE = "" We set up the Colab environment by importing the required Python utilities and defining the Prefab version, app path, and HTML export path. We install the pinned prefab-ui package so that the tutorial runs consistently without version-related issues. We also initialize an empty APP_CODE string, which we use to build the complete Prefab application step by step. Copy CodeCopiedUse a different BrowserAPP_CODE += r''' Generating Synthetic Operations Data Copy CodeCopiedUse a different Browserimport random from collections import Counter, defaultdict from datetime import date, timedelta from prefab_ui.actions import AppendState, OpenLink, PopState, SetState, ShowToast, ToggleState from prefab_ui.app import PrefabApp from prefab_ui.components import ( Alert, AlertDescription, AlertTitle, Badge, Button, Card, CardContent, CardDescription, CardFooter, CardHeader, CardTitle, Code, Column, DataTable, DataTableColumn, Form, Grid, H2, Input, Markdown, Mermaid, Metric, Muted, Progress, Ring, Row, Slider, Small, Switch, Tab, Tabs, Text ) from prefab_ui.components.charts import ( BarChart, ChartSeries, LineChart, PieChart, RadarChart, ScatterChart, Sparkline ) from prefab_ui.components.control_flow import Else, ForEach, If from prefab_ui.rx import EVENT, STATE random.seed(42) TODAY = date.today() DATES = [TODAY - timedelta(days=29 - i) for i in range(30)] REGIONS = ["All", "APAC", "EMEA", "NA", "LATAM"] PIPELINES = [ "Customer 360 ETL", "Invoice OCR", "LLM Triage", "Risk Scoring", "Forecast Sync", "Warehouse Load", ] OWNERS = ["Data Platform", "AI Apps", "Revenue Ops", "Risk Engineering"] STATES = ["Completed", "Completed", "Completed", "Completed", "Late", "Failed"] PRIORITIES = ["P0", "P1", "P2", "P3"] runs = [] daily_region_rows = [] for d in DATES: for region in REGIONS[1:]: region_bias = { "APAC": 0.96, "EMEA": 0.94, "NA": 0.97, "LATAM": 0.91, }[region] volume = random.randint(32, 78) failures = 0 late = 0 total_cost = 0.0 total_latency = 0.0 total_revenue = 0.0 for i in range(volume): pipeline = random.choice(PIPELINES) owner = random.choice(OWNERS) state = random.choices( STATES, weights=[ region_bias * 10, 6, 4, 3, 1.2, max(0.2, (1 - region_bias) * 16), ], k=1, )[0] duration = max( 12, int( random.gauss(95, 35) + (20 if state == "Late" else 0) + (45 if state == "Failed" else 0) ), ) cost = round(max(0.09, random.lognormvariate(-1.15, 0.55) + duration / 1800), 2) revenue = round(random.uniform(1.2, 8.5) * (1.3 if state == "Completed" else 0.6), 2) priority = random.choices(PRIORITIES, weights=[1, 3, 7, 10], k=1)[0] if state == "Failed": failures += 1 if state == "Late": late += 1 total_cost += cost total_latency += duration total_revenue += revenue if d >= TODAY - timedelta(days=10) and (state in {"Failed", "Late"} or random.random() < 0.05): runs.append({ "run_id": f"{d.strftime('%m%d')}-{region[:2]}-{len(runs)+1:04d}", "date": d.strftime("%Y-%m-%d"), "pipeline": pipeline, "owner": owner, "region": region, "state": state, "priority": priority, "duration_s": duration, "cost_usd": cost, "revenue_k": revenue, "sla_gap": round(max(0, duration - 120) / 60, 1), }) daily_region_rows.append({ "date": d.strftime("%b %d"), "region": region, "runs": volume, "failures": failures, "late": late, "success_rate": round(100 * (volume - failures - late * 0.35) / volume, 1), "avg_latency": round(total_latency / volume, 1), "cost_usd": round(total_cost, 2), "revenue_k": round(total_revenue, 1), }) runs = sorted( runs, key=lambda r: (r["priority"], r["state"] != "Failed", -r["duration_s"]) )[:80] def aggregate_daily(rows): by_date = defaultdict(lambda: { "date": "", "runs": 0, "failures": 0, "late": 0, "cost_usd": 0.0, "revenue_k": 0.0, "latency_weighted": 0.0, }) for r in rows: bucket = by_date[r["date"]] bucket["date"] = r["date"] bucket["runs"] += r["runs"] bucket["failures"] += r["failures"] bucket["late"] += r["late"] bucket["cost_usd"] += r["cost_usd"] bucket["revenue_k"] += r["revenue_k"] bucket["latency_weighted"] += r["avg_latency"] * r["runs"] out = [] for d in [x.strftime("%b %d") for x in DATES]: b = by_date[d] if b["runs"]: b["success_rate"] = round(100 * (b["runs"] - b["failures"] - b["late"] * 0.35) / b["runs"], 1) b["avg_latency"] = round(b["latency_weighted"] / b["runs"], 1) b["cost_usd"] = round(b["cost_usd"], 2) b["revenue_k"] = round(b["revenue_k"], 1) del b["latency_weighted"] out.append(dict(b)) return out def aggregate_regions(rows): by_region = defaultdict(lambda: { "region": "", "runs": 0, "failures": 0, "late": 0, "cost_usd": 0.0, "revenue_k": 0.0, "latency_weighted": 0.0, }) for r in rows: b = by_region[r["region"]] b["region"] = r["region"] b["runs"] += r["runs"] b["failures"] += r["failures"] b["late"] += r["late"] b["cost_usd"] += r["cost_usd"] b["revenue_k"] += r["revenue_k"] b["latency_weighted"] += r["avg_latency"] * r["runs"] out = [] for region in REGIONS[1:]: b = by_region[region] b["success_rate"] = round(100 * (b["runs"] - b["failures"] - b["late"] * 0.35) / b["runs"], 1) b["avg_latency"] = round(b["latency_weighted"] / b["runs"], 1) b["cost_usd"] = round(b["cost_usd"], 2) b["revenue_k"] = round(b["revenue_k"], 1) b["roi"] = round(b["revenue_k"] / max(1, b["cost_usd"]), 1) del b["latency_weighted"] out.append(dict(b)) return out def make_status_rows(table_rows): counts = Counter(r["state"] for r in table_rows) return [{"state": k, "count": v} for k, v in counts.items()] def make_pipeline_rows(table_rows): counts = Counter(r["pipeline"] for r in table_rows) return [{"pipeline": k, "count": v} for k, v in counts.most_common()] def make_kpis(region, daily_rows, table_rows): runs_count = sum(r["runs"] for r in daily_rows) failures = sum(r["failures"] for r in daily_rows) late = sum(r["late"] for r in daily_rows) cost = sum(r["cost_usd"] for r in daily_rows) revenue = sum(r["revenue_k"] for r in daily_rows) return { "region": region, "runs": runs_count, "success_rate": round(100 * (runs_count - failures - late * 0.35) / max(1, runs_count), 1), "avg_latency": round(sum(r["avg_latency"] * r["runs"] for r in daily_rows) / max(1, runs_count), 1), "cost_usd": round(cost, 2), "revenue_k": round(revenue, 1), "roi": round(revenue / max(1, cost), 1), "open_issues": len(table_rows), "p0p1": sum(1 for r in table_rows if r["priority"] in {"P0", "P1"}), "failure_rate": round(100 * failures / max(1, runs_count), 2), "spark": [r["success_rate"] for r in daily_rows[-14:]], } DAILY_BY_REGION = {"All": aggregate_daily(daily_region_rows)} REGION_ROWS = aggregate_regions(daily_region_rows) for region in REGIONS[1:]: DAILY_BY_REGION[region] = [r for r in daily_region_rows if r["region"] == region] RUNS_BY_REGION = { region: [r for r in runs if region == "All" or r["region"] == region] for region in REGIONS } STATUS_BY_REGION = { region: make_status_rows(RUNS_BY_REGION[region]) for region in REGIONS } PIPELINE_BY_REGION = { region: make_pipeline_rows(RUNS_BY_RE