sysadmin
22a1c4c2df
When Gitea credentials are missing or the queue fetch fails closed, show "Not loaded" instead of "No open items." and keep pagination marked unavailable. Successful empty inventories still show the empty state with complete pagination proof. Closes #458
119 lines
3.8 KiB
Python
119 lines
3.8 KiB
Python
"""HTML views for the live Gitea queue dashboard (#429)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import html
|
|
|
|
from webui.queue_loader import PaginationMeta, QueueItem, QueueSnapshot
|
|
|
|
|
|
def _badge_html(badges: tuple[str, ...]) -> str:
|
|
if not badges:
|
|
return ""
|
|
chips = "".join(
|
|
f'<span class="badge badge-{html.escape(b)}">{html.escape(b)}</span>'
|
|
for b in badges
|
|
)
|
|
return f'<span class="badges">{chips}</span>'
|
|
|
|
|
|
def _pagination_html(label: str, meta: PaginationMeta | None) -> str:
|
|
if meta is None:
|
|
return (
|
|
f"<p class='muted'><strong>{html.escape(label)} pagination:</strong> "
|
|
"unavailable</p>"
|
|
)
|
|
status = "complete" if meta.inventory_complete else "partial"
|
|
more = "yes" if meta.has_more else "no"
|
|
return (
|
|
f"<p class='meta'><strong>{html.escape(label)} pagination ({status}):</strong> "
|
|
f"returned {meta.returned_count} · per_page {meta.per_page} · "
|
|
f"pages_fetched {meta.pages_fetched} · has_more {more} · "
|
|
f"final_page {'yes' if meta.is_final_page else 'no'}</p>"
|
|
)
|
|
|
|
|
|
def _queue_table(
|
|
*,
|
|
title: str,
|
|
items: tuple[QueueItem, ...],
|
|
columns: tuple[tuple[str, str], ...],
|
|
fetch_failed: bool = False,
|
|
) -> str:
|
|
if fetch_failed:
|
|
return (
|
|
f"<h3>{html.escape(title)}</h3>"
|
|
"<p class='muted'>Not loaded — queue fetch did not complete.</p>"
|
|
)
|
|
if not items:
|
|
return f"<h3>{html.escape(title)}</h3><p class='muted'>No open items.</p>"
|
|
|
|
headers = "".join(f"<th>{html.escape(label)}</th>" for _, label in columns)
|
|
rows = []
|
|
for item in items:
|
|
cells = []
|
|
for key, _ in columns:
|
|
if key == "number":
|
|
cells.append(f"<td>#{item.number}</td>")
|
|
elif key == "title":
|
|
cells.append(
|
|
f"<td>{html.escape(item.title)}{_badge_html(item.badges)}</td>"
|
|
)
|
|
else:
|
|
cells.append(f"<td>{html.escape(item.extra.get(key, ''))}</td>")
|
|
rows.append("<tr>" + "".join(cells) + "</tr>")
|
|
|
|
body = (
|
|
f"<h3>{html.escape(title)}</h3>"
|
|
f"<table class='registry'><thead><tr>{headers}</tr></thead>"
|
|
f"<tbody>{''.join(rows)}</tbody></table>"
|
|
)
|
|
return body
|
|
|
|
|
|
def render_queue_page(snapshot: QueueSnapshot) -> str:
|
|
error_block = ""
|
|
if snapshot.fetch_error:
|
|
error_block = (
|
|
f'<div class="stub"><p><strong>Queue unavailable:</strong> '
|
|
f"{html.escape(snapshot.fetch_error)}</p></div>"
|
|
)
|
|
|
|
fetch_failed = bool(snapshot.fetch_error)
|
|
pr_section = _queue_table(
|
|
title="Open pull requests",
|
|
items=snapshot.prs,
|
|
fetch_failed=fetch_failed,
|
|
columns=(
|
|
("number", "#"),
|
|
("title", "Title"),
|
|
("branch", "Branch"),
|
|
("head_sha", "Head"),
|
|
("mergeable", "Mergeable"),
|
|
("linked_issue", "Linked issue"),
|
|
),
|
|
)
|
|
issue_section = _queue_table(
|
|
title="Open issues",
|
|
items=snapshot.issues,
|
|
fetch_failed=fetch_failed,
|
|
columns=(
|
|
("number", "#"),
|
|
("title", "Title"),
|
|
("labels", "Labels"),
|
|
("assignee", "Assignee"),
|
|
("state", "State"),
|
|
),
|
|
)
|
|
|
|
return (
|
|
"<h2>Live queue</h2>"
|
|
f"<p class='meta'>Repository: <code>{html.escape(snapshot.repo_label)}</code> "
|
|
f"· project <code>{html.escape(snapshot.project_id)}</code></p>"
|
|
f"{error_block}"
|
|
f"{_pagination_html('PR', snapshot.pr_pagination)}"
|
|
f"{pr_section}"
|
|
f"{_pagination_html('Issue', snapshot.issue_pagination)}"
|
|
f"{issue_section}"
|
|
"<p class='muted'>Read-only MVP — claims, reviews, and merges stay in Gitea/MCP tools.</p>"
|
|
) |