from __future__ import annotations from collections.abc import Mapping from typing import cast import htpy as h from htpy import Node, Renderable from repub.components import ( action_button, app_shell, inline_link, muted_action_link, stat_card, status_badge, table_section, ) from repub.pages.runs import live_work_section, relative_time_formatter_script def dashboard_header( *, path_prefix: str = "/admin", reader_app_url: str | None = None ) -> 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.div(class_="flex flex-wrap gap-2")[ ( muted_action_link( href=reader_app_url, label="Open AnyNews", target="_blank", rel="noopener noreferrer", ) if reader_app_url is not None else None ), muted_action_link( href=f"{path_prefix}/publisher", label="Publisher View", ), ], ] ] 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.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 _source_feed_time( source_feed: Mapping[str, object], *, iso_key: str, label_key: str, class_name: str, data_attr: str | None = None, inline: bool = False, ) -> Node: iso_value = source_feed.get(iso_key) label = str(source_feed[label_key]) if iso_value is not None: attrs = { "datetime": str(iso_value), "title": str(iso_value), "class": class_name, } if data_attr is not None: attrs[data_attr] = str(iso_value) return h.time(attrs)[label] if inline: return h.span(class_=class_name)[label] return h.p(class_=class_name)[label] def _source_feed_row( source_feed: Mapping[str, object], *, show_feed_url: bool, compact_mobile: bool ) -> tuple[Node, ...]: last_updated = _source_feed_time( source_feed, iso_key="last_updated_iso", label_key="last_updated", class_name="font-medium text-slate-900", ) next_run = _source_feed_time( source_feed, iso_key="next_run_at", label_key="next_run", class_name="font-medium text-slate-900", data_attr="data-next-run-at", ) mobile_meta = ( h.div(class_="mt-2 grid gap-1 text-xs text-slate-500 md:hidden")[ h.p(class_="flex flex-wrap gap-x-1.5")[ h.span(class_="font-medium text-slate-600")["Updated"], _source_feed_time( source_feed, iso_key="last_updated_iso", label_key="last_updated", class_name="font-medium text-slate-700", inline=True, ), ], h.p(class_="flex flex-wrap gap-x-1.5")[ h.span(class_="font-medium text-slate-600")["Next"], _source_feed_time( source_feed, iso_key="next_run_at", label_key="next_run", class_name="font-medium text-slate-700", data_attr="data-next-run-at", inline=True, ), ], ] if compact_mobile else None ) feed_url_cells = ( ( h.div(class_="min-w-64")[ inline_link( href=str(source_feed["feed_href"]), label=str(source_feed["feed_href"]), tone="amber", ) ], ) if show_feed_url else () ) 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"]) ], mobile_meta, ], *feed_url_cells, status_badge( label=str(source_feed["feed_status_label"]), tone=str(source_feed["feed_status_tone"]), ), last_updated, next_run, action_button( label="Run now", disabled=bool(source_feed["run_disabled"]), post_path=cast(str | None, source_feed.get("run_post_path")), ), ) def published_feeds_table( *, source_feeds: tuple[Mapping[str, object], ...] | None = None, manage_sources_href: str | None = "/admin/sources", show_heading: bool = True, show_feed_url: bool = True, compact_mobile: bool = False, ) -> Renderable: rows = tuple( _source_feed_row( source_feed, show_feed_url=show_feed_url, compact_mobile=compact_mobile, ) for source_feed in (source_feeds or ()) ) feed_url_headers = ("Feed URL",) if show_feed_url else () use_compact_columns = compact_mobile and not show_feed_url header_classes = ( ( "w-[48%] px-3 py-2.5 text-left text-xs font-semibold uppercase tracking-[0.18em] whitespace-nowrap text-slate-500 md:w-[32%] sm:pl-4", "w-[26%] px-2.5 py-2.5 text-left text-xs font-semibold uppercase tracking-[0.18em] whitespace-nowrap text-slate-500 md:w-[16%]", "hidden px-2.5 py-2.5 text-left text-xs font-semibold uppercase tracking-[0.18em] whitespace-nowrap text-slate-500 md:table-cell md:w-[22%]", "hidden px-2.5 py-2.5 text-left text-xs font-semibold uppercase tracking-[0.18em] whitespace-nowrap text-slate-500 md:table-cell md:w-[18%]", "w-[26%] px-2.5 py-2.5 text-right text-xs font-semibold uppercase tracking-[0.18em] whitespace-nowrap text-slate-500 md:w-[12%]", ) if use_compact_columns else None ) cell_classes = ( ( "w-[48%] py-3 pr-3 pl-3 text-sm font-medium text-slate-950 md:w-[32%] sm:pl-4", "w-[26%] px-2.5 py-3 align-top text-sm whitespace-nowrap text-slate-600 md:w-[16%]", "hidden px-2.5 py-3 align-top text-sm whitespace-nowrap text-slate-600 md:table-cell md:w-[22%]", "hidden px-2.5 py-3 align-top text-sm whitespace-nowrap text-slate-600 md:table-cell md:w-[18%]", "w-[26%] px-2.5 py-3 text-right align-top text-sm whitespace-nowrap text-slate-600 md:w-[12%]", ) if use_compact_columns else None ) return table_section( eyebrow="Published feeds" if show_heading else None, title="Published feeds" if show_heading else None, empty_message="No feeds have been published yet.", headers=( "Source", *feed_url_headers, "Status", "Last updated", "Next run", "Actions", ), rows=rows, header_classes=header_classes, cell_classes=cell_classes, table_class=( "relative w-full min-w-0 divide-y divide-slate-200 table-fixed" if use_compact_columns else "relative w-full min-w-[64rem] divide-y divide-slate-200 table-auto" ), actions=( muted_action_link(href=manage_sources_href, label="Manage sources") if manage_sources_href is not None else None ), ) def dashboard_page() -> Renderable: return dashboard_page_with_data() def dashboard_page_with_data( *, current_path: str = "/admin", path_prefix: str = "/admin", snapshot: Mapping[str, str] | None = None, running_executions: tuple[Mapping[str, object], ...] | None = None, queued_executions: tuple[Mapping[str, object], ...] | None = None, source_feeds: tuple[Mapping[str, object], ...] | None = None, reader_app_url: str | None = None, ) -> Renderable: running_items = running_executions or () queued_items = queued_executions or () source_items = source_feeds or () return app_shell( current_path=current_path, source_count=len(source_items), running_count=len(running_items), content=( dashboard_header(path_prefix=path_prefix, reader_app_url=reader_app_url), operational_snapshot(snapshot=snapshot), live_work_section( running_executions=running_items, queued_executions=queued_items, ), published_feeds_table( source_feeds=source_items, manage_sources_href=f"{path_prefix}/sources", ), relative_time_formatter_script(), ), )