from __future__ import annotations from collections.abc import Mapping import htpy as h from htpy import Node, Renderable from repub.components import ( admin_sidebar, header_action_link, inline_button, inline_link, muted_action_link, stat_card, status_badge, table_section, ) def _text(values: Mapping[str, object], key: str) -> str: return str(values[key]) def _running_execution_row(execution: Mapping[str, object]) -> tuple[Node, ...]: status_tone = "running" if _text(execution, "status") != "Succeeded" else "done" return ( h.div[ h.div(class_="font-semibold text-slate-950")[_text(execution, "source")], h.p(class_="mt-0.5 font-mono text-[11px] text-slate-500")[ _text(execution, "slug") ], ], h.div[ h.p(class_="font-medium text-slate-900")[ f"#{_text(execution, 'execution_id')}" ], h.p(class_="mt-0.5 text-[11px] text-slate-500")[ f"job {_text(execution, 'job_id')}" ], ], h.div[ h.p(class_="font-medium text-slate-900")[_text(execution, "started_at")], h.p(class_="mt-0.5 text-[11px] text-slate-500")[ _text(execution, "runtime") ], ], status_badge(label=_text(execution, "status"), tone=status_tone), h.div(class_="min-w-56 whitespace-normal")[ h.p(class_="font-medium text-slate-900")[_text(execution, "stats")], h.p(class_="mt-0.5 text-[11px] text-slate-500")[_text(execution, "worker")], ], h.div(class_="flex flex-nowrap items-center gap-3")[ inline_link( href=_text(execution, "log_href"), label="View log", tone="amber", ), inline_button(label="Stop", tone="danger"), ], ) def dashboard_header() -> Renderable: return h.section[ h.div( class_="flex flex-col gap-4 sm:flex-row sm:items-start sm:justify-between" )[ h.div[ h.h1(class_="text-3xl font-semibold tracking-tight text-slate-950")[ "Republisher" ], h.p(class_="mt-1 text-sm text-slate-600")[ "Operational status and live executions." ], ], h.div(class_="flex flex-wrap gap-2")[ header_action_link(href="/sources/create", label="Create source"), muted_action_link(href="/sources", label="View sources"), ], ] ] def operational_snapshot(*, snapshot: Mapping[str, str] | None = None) -> Renderable: values = snapshot or { "running_now": "0", "upcoming_today": "0", "failures_24h": "0", "artifact_footprint": "0 B", } return h.section[ h.div(class_="mb-3 flex items-end justify-between gap-4")[ h.div[ h.p( class_="text-xs font-semibold uppercase tracking-[0.22em] text-slate-500" )["Overview"], h.h2(class_="mt-1 text-xl font-semibold tracking-tight text-slate-950")[ "Operational snapshot" ], ], h.p(class_="text-xs text-slate-500")["Live values from the database"], ], h.dl(class_="grid gap-3 md:grid-cols-2 xl:grid-cols-4")[ stat_card( label="Running now", value=values["running_now"], detail="Currently active job executions.", ), stat_card( label="Upcoming today", value=values["upcoming_today"], detail="Enabled jobs that are ready for their next run.", ), stat_card( label="Failures in 24h", value=values["failures_24h"], detail="Recent failed executions recorded by the scheduler.", ), stat_card( label="Artifact footprint", value=values["artifact_footprint"], detail="Current artifact size under the output path.", ), ], ] def running_executions_table( *, running_executions: tuple[Mapping[str, object], ...] | None = None ) -> Renderable: rows = tuple( _running_execution_row(execution) for execution in (running_executions or ()) ) headers = ("Source", "Execution", "Started", "Status", "Stats", "Actions") def render_row(row: tuple[Node, ...]) -> Renderable: first_cell, *other_cells = row return h.tr(class_="align-top")[ h.td(class_="py-3 pr-6 pl-4 text-sm font-medium text-slate-950 sm:pl-4")[ first_cell ], ( h.td( class_="px-3 py-3 align-top text-sm whitespace-nowrap text-slate-600" )[cell] for cell in other_cells ), ] return h.section[ h.div(class_="mb-3 flex items-end justify-between gap-4")[ h.div[ h.p( class_="text-xs font-semibold uppercase tracking-[0.22em] text-amber-600" )["Live work"], h.h2(class_="mt-1 text-xl font-semibold text-slate-950")[ "Running executions" ], h.p(class_="mt-1 text-sm text-slate-600")[ "Dashboard keeps only the in-flight executions visible here. The full run history lives on the Runs page." ], ], muted_action_link(href="/runs", label="Open runs"), ], h.div( class_="overflow-hidden rounded-2xl bg-white shadow-sm ring-1 ring-slate-200" )[ h.div(class_="overflow-x-auto")[ h.table( class_="w-full min-w-[70rem] divide-y divide-slate-200 table-auto" )[ h.thead(class_="bg-stone-50")[ h.tr[ ( h.th( scope="col", class_="px-3 py-2.5 text-left text-[11px] font-semibold uppercase tracking-[0.18em] whitespace-nowrap text-slate-500 first:pl-4", )[header] for header in headers ) ] ], h.tbody(class_="divide-y divide-slate-200 bg-white")[ (render_row(row) for row in rows) ], ] ] ], ] def _source_feed_row(source_feed: Mapping[str, object]) -> tuple[Node, ...]: last_updated_iso = source_feed.get("last_updated_iso") last_updated = ( h.time( datetime=str(last_updated_iso), title=str(last_updated_iso), class_="font-medium text-slate-900", )[str(source_feed["last_updated"])] if last_updated_iso is not None else h.p(class_="font-medium text-slate-900")[str(source_feed["last_updated"])] ) return ( h.div[ h.div(class_="font-semibold text-slate-950")[str(source_feed["source"])], h.p(class_="mt-0.5 font-mono text-[11px] text-slate-500")[ str(source_feed["slug"]) ], ], h.div(class_="min-w-64")[ inline_link( href=str(source_feed["feed_href"]), label=str(source_feed["feed_href"]), tone="amber", ) ], status_badge( label=str(source_feed["feed_status_label"]), tone=str(source_feed["feed_status_tone"]), ), last_updated, h.p(class_="font-medium text-slate-900")[ str(source_feed["artifact_footprint"]) ], ) def published_feeds_table( *, source_feeds: tuple[Mapping[str, object], ...] | None = None ) -> Renderable: rows = tuple(_source_feed_row(source_feed) for source_feed in (source_feeds or ())) return table_section( eyebrow="Published feeds", title="Published feeds", subtitle="Per-source public feed paths under /feeds, with current availability and disk usage.", headers=("Source", "Feed URL", "Status", "Last updated", "Disk usage"), rows=rows, actions=muted_action_link(href="/sources", label="Manage sources"), ) def dashboard_page() -> Renderable: return dashboard_page_with_data() def dashboard_page_with_data( *, snapshot: Mapping[str, str] | None = None, running_executions: tuple[Mapping[str, object], ...] | None = None, source_feeds: tuple[Mapping[str, object], ...] | None = None, ) -> Renderable: return h.main( id="morph", class_="min-h-screen lg:grid lg:grid-cols-[18rem_minmax(0,1fr)]", )[ admin_sidebar(current_path="/"), h.div(class_="px-4 py-4 sm:px-5 lg:px-6 lg:py-5")[ h.div(class_="mx-auto max-w-7xl space-y-5")[ dashboard_header(), operational_snapshot(snapshot=snapshot), running_executions_table(running_executions=running_executions), published_feeds_table(source_feeds=source_feeds), ] ], ]