Cómo diseñar paneles interactivos basados ​​en Python con componentes de interfaz de usuario reactivos prefabricados y exportación de HTML estático

importar aleatorio desde colecciones importar contador, defaultdict desde fecha y hora fecha de importación, timedelta desde prefab_ui.actions importar AppendState, OpenLink, PopState, SetState, ShowToast, ToggleState desde prefab_ui.app importar PrefabApp desde prefab_ui.components importar (alerta, descripción de alerta, título de alerta, insignia, botón, tarjeta, contenido de tarjeta, descripción de tarjeta, pie de página de tarjeta, encabezado de tarjeta, CardTitle, Código, Columna, DataTable, DataTableColumn, Formulario, Cuadrícula, H2, Entrada, Markdown, Sirena, Métrica, Silenciado, Progreso, Anillo, Fila, Control deslizante, Pequeño, Cambiar, Tabulador, Tabuladores, Texto) de prefab_ui.components.charts importar (BarChart, ChartSeries, LineChart, PieChart, RadarChart, ScatterChart, Sparkline) de prefab_ui.components.control_flow importar de lo contrario, para cada uno, si desde prefab_ui.rx importar EVENTO, ESTADO random.seed(42) HOY = date.today() FECHAS = [TODAY – timedelta(days=29 – i) for i in range(30)]
REGIONES = [“All”, “APAC”, “EMEA”, “NA”, “LATAM”]
TUBERÍAS = [
“Customer 360 ETL”,
“Invoice OCR”,
“LLM Triage”,
“Risk Scoring”,
“Forecast Sync”,
“Warehouse Load”,
]
DUEÑOS = [“Data Platform”, “AI Apps”, “Revenue Ops”, “Risk Engineering”]
ESTADOS = [“Completed”, “Completed”, “Completed”, “Completed”, “Late”, “Failed”]
PRIORIDADES = [“P0”, “P1”, “P2”, “P3”]
corre = []
daily_region_rows = []
para d en FECHAS: para región en REGIONES[1:]: region_bias = { “APAC”: 0,96, “EMEA”: 0,94, “NA”: 0,97, “LATAM”: 0,91, }[region]
volumen = random.randint(32, 78) fallas = 0 tarde = 0 costo_total = 0.0 latencia_total = 0.0 ingresos_total = 0.0 para i en rango(volumen): tubería = random.choice(PIPELINES) propietario = random.choice(PROPIETARIOS) estado = random.choices( ESTADOS, pesos =[
region_bias * 10,
6,
4,
3,
1.2,
max(0.2, (1 – region_bias) * 16),
],k=1, )[0]
duración = max( 12, int( random.gauss(95, 35) + (20 if state == “Tarde” else 0) + (45 if state == “Falló” else 0) ), ) costo = round(max(0.09, random.lognormvariate(-1.15, 0.55) + duración / 1800), 2) ingresos = round(random.uniform(1.2, 8.5) * (1.3 if state == “Completed” else 0.6), 2) prioridad = random.choices(PRIORIDADES, pesos=[1, 3, 7, 10],k=1)[0]
if estado == “Falló”: fallas += 1 if estado == “Tarde”: tarde += 1 costo_total += costo latencia_total += duración ingresos_total += ingresos if d >= HOY – timedelta(días=10) y (estado en {“Falló”, “Tarde”} o random.random() < 0.05): ejecuta.append({ "run_id": f"{d.strftime('%m%d')}-{región[:2]}-{len(runs)+1:04d}", "date": d.strftime("%Y-%m-%d"), "pipeline": pipeline, "owner": propietario, "region": región, "state": estado, "priority": prioridad, "duration_s": duración, "cost_usd": costo, "revenue_k": ingresos, "sla_gap": round(max(0, duración - 120) / 60, 1), }) daily_region_rows.append({ "date": d.strftime("%b %d"), "region": región, "runs": volumen, "failures": fallas, "late": late, "success_rate": round(100 * (volumen - fallas - late * 0.35) / volumen, 1), "avg_latency": round(total_latency / volumen, 1), "cost_usd": round(total_cost, 2), "revenue_k": round(total_revenue, 1), }) ejecuciones = ordenadas (ejecuciones, clave = lambda r: (r["priority"]r.["state"] != "Error", -r["duration_s"]) )[:80] def agregado_daily(filas): by_date = defaultdict(lambda: { "date": "", "runs": 0, "failures": 0, "late": 0, "cost_usd": 0.0, "revenue_k": 0.0, "latency_weighted": 0.0, }) para r en filas: depósito = by_date[r["date"]]balde["date"] =r["date"] balde["runs"] +=r["runs"] balde["failures"] +=r["failures"] balde["late"] +=r["late"] balde["cost_usd"] +=r["cost_usd"] balde["revenue_k"] +=r["revenue_k"] balde["latency_weighted"] +=r["avg_latency"] *r["runs"] fuera = [] para d en [x.strftime("%b %d") for x in DATES]: b = por_fecha[d] si b["runs"]: b["success_rate"] = ronda(100 * (b["runs"] - b["failures"] - b["late"] * 0,35) / segundo["runs"]1) segundo["avg_latency"] = ronda(b["latency_weighted"] / b["runs"]1) segundo["cost_usd"] = ronda(b["cost_usd"]2) segundo["revenue_k"] = ronda(b["revenue_k"]1) del b["latency_weighted"] out.append(dict(b)) devuelve def agregado_regiones(filas): by_region = defaultdict(lambda: { "region": "", "runs": 0, "failures": 0, "late": 0, "cost_usd": 0.0, "revenue_k": 0.0, "latency_weighted": 0.0, }) para r en filas: 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"] fuera = [] para región en REGIONES[1:]: b = por_región[region] b["success_rate"] = ronda(100 * (b["runs"] - b["failures"] - b["late"] * 0,35) / segundo["runs"]1) segundo["avg_latency"] = ronda(b["latency_weighted"] / b["runs"]1) segundo["cost_usd"] = ronda(b["cost_usd"]2) segundo["revenue_k"] = ronda(b["revenue_k"]1) segundo["roi"] = ronda(b["revenue_k"] / máx(1, b["cost_usd"]), 1) del b["latency_weighted"] out.append(dict(b)) devuelve def make_status_rows(table_rows): recuentos = Contador(r["state"] para r en table_rows) retorno [{"state": k, "count": v} for k, v in counts.items()] def make_pipeline_rows(table_rows): recuentos = Contador(r["pipeline"] para r en table_rows) retorno [{"pipeline": k, "count": v} for k, v in counts.most_common()] def make_kpis(región, filas_diarias, filas_tabla): recuento_ejecuciones = suma(r["runs"] para r en daily_rows) fallas = suma (r["failures"] para r en daily_rows) tarde = suma(r["late"] para r en filas_diarias) costo = suma(r["cost_usd"] para r en filas_diarias) ingresos = suma(r["revenue_k"] para r en filas_diarias) return { "región": región, "ejecuciones": recuento_ejecuciones, "tasa_éxito": ronda(100 * (recuento_ejecuciones - fallas - tarde * 0.35) / max(1, recuento_ejecuciones), 1), "latencia_avg": ronda(suma(r["avg_latency"] *r["runs"] para r en filas_diarias) / max(1, run_count), 1), "cost_usd": round(costo, 2), "revenue_k": round(ingresos, 1), "roi": round(ingresos / max(1, costo), 1), "open_issues": len(table_rows), "p0p1": suma(1 para r en table_rows si r["priority"] en {"P0", "P1"}), "tasa_de_fallos": round(100 * fallos/max(1, recuento_ejecuciones), 2), "chispa": [r["success_rate"] para r en filas_diarias[-14:]], } DAILY_BY_REGION = {"Todos": agregado_daily(daily_region_rows)} REGION_ROWS = agregado_regions(daily_region_rows) para la región en REGIONS[1:]: DAILY_BY_REGION[region] = [r for r in daily_region_rows if r["region"] == región]RUNS_BY_REGION = {región: [r for r in runs if region == "All" or r["region"] == región]para la región en REGIONS } STATUS_BY_REGION = { región: make_status_rows(RUNS_BY_REGION[region]) para la región en REGIONS } PIPELINE_BY_REGION = { región: make_pipeline_rows(RUNS_BY_REGION[region]) para la región en REGIONS } KPI_BY_REGION = { región: make_kpis(región, DAILY_BY_REGION[region]CORRE_POR_REGION[region]) para la región en REGIONES } WATCHLIST = ordenado (ejecuciones, clave = lambda r: (r["priority"]r.["state"] != "Error", -r["sla_gap"]) )[:8] DISPERSIÓN_ROWS = [ { "region": r["region"]"tasa_éxito": r["success_rate"]"costo_usd": r["cost_usd"]"latencia_promedio": r["avg_latency"]} para r en REGION_ROWS ]RADAR_ROWS = [ {"metric": "Success", **{r["region"]:r["success_rate"] para r en REGION_ROWS}}, {"metric": "ROI", **{r["region"]: mín(100, r["roi"] * 8) para r en REGION_ROWS}}, {"metric": "Latencia", **{r["region"]: máx(0, 100 - r["avg_latency"] / 2) para r en REGION_ROWS}}, {"métrica": "Costo", **{r["region"]: máx(0, 100 - r["cost_usd"] / 20) para r en REGION_ROWS}}, ]REGION_ACTIONS = {región: [ SetState("selected_region", region), SetState("line_rows", DAILY_BY_REGION[region]), SetState("table_rows", RUNS_BY_REGION[region]), SetState("estado_filas", STATUS_BY_REGION[region]), SetState("pipeline_rows", PIPELINE_BY_REGION[region]), SetState("region_kpis", KPI_BY_REGION[region]), SetState("selected_run", Ninguno), ShowToast(f"Región establecida en {región}", variante="info", duración=1800), ]para la región en REGIONES } '''