shujakuin/shusub2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add compact unified dashboard

Browse Source
This commit is contained in:
yunyaozhou
2026-07-27 03:26:32 +08:00
parent 545c1abbe6
commit 69f72c09b5
5 changed files with 518 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sub2api_quota_tui.py
+356 -17
View File
@@ -19,7 +19,7 @@ from typing import Any
APP_NAME = "shusub2"
FALLBACK_VERSION = "0.2.4"
FALLBACK_VERSION = "0.2.5"
DEFAULT_API_URL = "http://127.0.0.1:18318/api/tui/accounts"
DEFAULT_CONFIG_FILE = "~/.config/shusub2/api-url"
DEFAULT_STATUS_CONFIG_FILE = "~/.config/shusub2/status-url"
@@ -1172,7 +1172,7 @@ def run_textual(
errors_limit: int,
errors_time_range: str,
version_message: str = "",
start_page: str = "accounts",
start_page: str = "dashboard",
) -> int:
try:
from rich.text import Text
@@ -1183,6 +1183,336 @@ def run_textual(
print("Textual is required. Run with: uv run --with textual python sub2api_quota_tui.py", file=sys.stderr)
return 2
class DashboardScreen(Screen[None]):
AUTO_FOCUS = "#accounts"
BINDINGS = [
("r", "refresh", "Refresh"),
("/", "focus_filter", "Filter"),
("a", "focus_accounts", "Accounts"),
("l", "focus_logs", "Logs"),
("e", "focus_errors", "Errors"),
]
def __init__(self) -> None:
super().__init__()
self.accounts_payload: dict[str, Any] = {}
self.status_payload: dict[str, Any] = {}
self.monitor_error = ""
self.accounts_error = ""
self.logs_payload: dict[str, Any] = {}
self.logs_error = ""
self.errors_payload: dict[str, Any] = {}
self.errors_error = ""
self.account_rows: list[dict[str, Any]] = []
self.log_rows: list[dict[str, Any]] = []
self.error_rows: list[dict[str, Any]] = []
self.account_by_key: dict[str, dict[str, Any]] = {}
self.log_by_key: dict[str, dict[str, Any]] = {}
self.error_by_key: dict[str, dict[str, Any]] = {}
self.active_table = "accounts"
def compose(self) -> ComposeResult:
yield Header(show_clock=True)
yield Static("", id="summary")
yield Input(placeholder="filter", id="filter")
yield DataTable(id="accounts")
yield DataTable(id="logs")
yield DataTable(id="errors")
yield Static("", id="detail")
yield Static("", id="status")
yield Footer()
@staticmethod
def configure_table(table: DataTable, columns: tuple[tuple[str, int], ...]) -> None:
table.cursor_type = "row"
table.zebra_stripes = True
for label, width in columns:
table.add_column(label, width=width)
def on_mount(self) -> None:
self.configure_table(
self.query_one("#accounts", DataTable),
(
("ACCOUNT", 16),
("Group", 5),
("Today", 8),
("Daily", 10),
("5h", 9),
("7d", 9),
("Avail", 7),
),
)
self.configure_table(
self.query_one("#logs", DataTable),
(
("LOG KEY", 12),
("Account", 12),
("Model", 16),
("Cost", 8),
("Latency", 7),
("Time", 11),
),
)
self.configure_table(
self.query_one("#errors", DataTable),
(
("ERR", 4),
("Status", 6),
("Account", 12),
("Model", 13),
("Phase", 8),
("Type", 10),
("Time", 11),
),
)
self.refresh_all(refresh=True)
self.set_interval(refresh_seconds, self.refresh_accounts)
self.set_interval(logs_refresh_seconds, self.refresh_logs)
self.set_interval(errors_refresh_seconds, self.refresh_errors)
self.focus_table("accounts")
def action_refresh(self) -> None:
self.refresh_all(refresh=True)
def action_focus_filter(self) -> None:
self.query_one("#filter", Input).focus()
def action_focus_accounts(self) -> None:
self.focus_table("accounts")
def action_focus_logs(self) -> None:
self.focus_table("logs")
def action_focus_errors(self) -> None:
self.focus_table("errors")
def focus_table(self, table_id: str) -> None:
self.active_table = table_id
self.query_one(f"#{table_id}", DataTable).focus()
self.render_active_detail()
def on_input_changed(self, event: Input.Changed) -> None:
if event.input.id == "filter":
self.render_tables()
def on_input_submitted(self, event: Input.Submitted) -> None:
if event.input.id == "filter":
self.focus_table(self.active_table)
def refresh_all(self, refresh: bool = False) -> None:
self.query_one("#status", Static).update("refreshing accounts, logs, and errors...")
self.refresh_accounts(refresh=refresh)
self.refresh_logs()
self.refresh_errors()
def refresh_accounts(self, refresh: bool = False) -> None:
self.accounts_error = ""
try:
self.accounts_payload = fetch_payload(api_url, timeout, refresh=refresh)
self.status_payload, self.monitor_error = fetch_optional_payload(status_url, timeout)
except Exception as exc:
self.accounts_error = str(exc)
self.render_accounts()
self.render_meta()
def refresh_logs(self) -> None:
self.logs_error = ""
if not str(logs_token or "").strip():
self.logs_payload = {}
self.logs_error = "admin token not configured"
else:
try:
self.logs_payload = fetch_logs_payload(logs_url, logs_token, timeout, logs_limit)
except Exception as exc:
self.logs_error = str(exc)
self.render_logs()
self.render_meta()
def refresh_errors(self) -> None:
self.errors_error = ""
if not str(logs_token or "").strip():
self.errors_payload = {}
self.errors_error = "admin token not configured"
else:
try:
self.errors_payload = fetch_merged_errors(
default_error_sources(errors_cn_url, errors_us_url),
logs_token,
timeout,
errors_limit,
errors_time_range,
)
except Exception as exc:
self.errors_error = str(exc)
self.render_errors()
self.render_meta()
def render_tables(self) -> None:
self.render_accounts()
self.render_logs()
self.render_errors()
self.render_meta()
self.render_active_detail()
def render_accounts(self) -> None:
filter_text = self.query_one("#filter", Input).value
self.account_rows = normalize_account_rows(self.accounts_payload, filter_text)
table = self.query_one("#accounts", DataTable)
table.clear()
self.account_by_key = {}
for index, row in enumerate(self.account_rows):
key = f"account-{row['id']}-{index}"
self.account_by_key[key] = row
table.add_row(
row["name"],
row["routing_group"],
format_cost(row["today_cost_usd"]),
row["daily_quota_cell"],
row["five_hour"],
row["weekly"],
monitor_availability(row, self.status_payload),
key=key,
)
def render_logs(self) -> None:
filter_text = self.query_one("#filter", Input).value
self.log_rows = normalize_log_rows(self.logs_payload, filter_text)
table = self.query_one("#logs", DataTable)
table.clear()
self.log_by_key = {}
for index, row in enumerate(self.log_rows):
key = f"log-{row['id']}-{index}"
self.log_by_key[key] = row
color = key_color(row["key"])
table.add_row(
Text(row["key"], style=color) if color else row["key"],
row["account"],
row["model"],
format_cost(row["cost"]),
format_seconds(row["duration_ms"]),
row["time"],
key=key,
)
def render_errors(self) -> None:
filter_text = self.query_one("#filter", Input).value
self.error_rows = normalize_error_rows(self.errors_payload, filter_text)
table = self.query_one("#errors", DataTable)
table.clear()
self.error_by_key = {}
for index, row in enumerate(self.error_rows):
key = f"error-{row['node']}-{row['id']}-{index}"
self.error_by_key[key] = row
table.add_row(
row["node"],
str(row["status_code"]),
row["account"],
row["model"],
row["phase"],
row["type"],
row["time"],
key=key,
)
def errors_total(self) -> int:
sources = self.errors_payload.get("sources") if isinstance(self.errors_payload.get("sources"), dict) else {}
totals = [
as_int(info.get("total"))
for info in sources.values()
if isinstance(info, dict) and info.get("ok")
]
return sum(totals) if totals else len(self.errors_payload.get("items") or [])
def render_meta(self) -> None:
totals = self.accounts_payload.get("totals") if isinstance(self.accounts_payload.get("totals"), dict) else {}
account_total = as_int(totals.get("total_accounts")) or len(self.accounts_payload.get("accounts") or [])
usable = as_int(totals.get("usable_accounts"))
logs_total = as_int(logs_envelope(self.logs_payload).get("total"))
if not logs_total:
logs_total = len(log_items(self.logs_payload))
summary = (
f"accounts {len(self.account_rows)}/{account_total} ({usable} usable, "
f"{format_cost(totals.get('today_cost_usd'))}) | "
f"logs {len(self.log_rows)}/{logs_total} | "
f"errors {len(self.error_rows)}/{self.errors_total()}"
)
self.query_one("#summary", Static).update(summary)
status_bits = [version_message]
if self.accounts_error:
status_bits.append(f"accounts: {self.accounts_error}")
else:
status_bits.extend(
(
monitor_summary(self.status_payload, self.monitor_error),
f"source {self.accounts_payload.get('source_name') or '-'}",
)
)
shared_admin_error = self.logs_error and self.logs_error == self.errors_error
if shared_admin_error:
status_bits.append(f"logs/errors: {self.logs_error}")
else:
if self.logs_error:
status_bits.append(f"logs: {self.logs_error}")
if self.errors_error:
status_bits.append(f"errors: {self.errors_error}")
if not self.errors_error and self.errors_payload:
sources = self.errors_payload.get("sources") if isinstance(self.errors_payload.get("sources"), dict) else {}
source_bits = []
for node in ("cn", "us"):
info = sources.get(node) if isinstance(sources.get(node), dict) else {}
if info:
source_bits.append(f"{node} {'ok' if info.get('ok') else 'err'}")
if source_bits:
status_bits.append("errors " + "/".join(source_bits))
self.query_one("#status", Static).update(" | ".join(bit for bit in status_bits if bit) or "ready")
def render_active_detail(self) -> None:
rows_by_table = {
"accounts": self.account_rows,
"logs": self.log_rows,
"errors": self.error_rows,
}
rows = rows_by_table[self.active_table]
if not rows:
self.query_one("#detail", Static).update(f"no {self.active_table}")
return
self.render_detail(self.active_table, rows[0])
def render_detail(self, table_id: str, row: dict[str, Any]) -> None:
if table_id == "accounts":
detail = (
f"{row['name']} | {row['provider']} | group {row['routing_group']} | {row['kind_label']} | "
f"daily {row['daily_quota_cell']} | today {format_cost(row['today_cost_usd'])}, "
f"{format_count(row['today_tokens'])} tokens, {format_count(row['today_requests'])} req | "
f"{monitor_detail(row, self.status_payload)}"
)
error = str(row.get("error") or "").strip()
if error:
detail += f" | {error}"
elif table_id == "logs":
detail = log_detail_line(row)
else:
detail = error_detail_line(row)
self.query_one("#detail", Static).update(detail)
def on_data_table_row_highlighted(self, event: DataTable.RowHighlighted) -> None:
table_id = str(event.control.id or "")
if table_id not in {"accounts", "logs", "errors"}:
return
if event.control.has_focus:
self.active_table = table_id
if table_id != self.active_table:
return
row_maps = {
"accounts": self.account_by_key,
"logs": self.log_by_key,
"errors": self.error_by_key,
}
row = row_maps[table_id].get(str(event.row_key.value))
if row:
self.render_detail(table_id, row)
class AccountsScreen(Screen[None]):
AUTO_FOCUS = "#accounts"
BINDINGS = [
@@ -1274,7 +1604,9 @@ def run_textual(
def render_keys(self) -> None:
table = self.query_one("#keys", DataTable)
table.clear()
for row in normalize_key_rows(self.keys_payload):
rows = normalize_key_rows(self.keys_payload)
table.display = bool(rows)
for row in rows:
color = key_color(row["name"])
table.add_row(
Text(str(row["name"]), style=color) if color else str(row["name"]),
@@ -1556,12 +1888,13 @@ def run_textual(
class Sub2APIQuotaApp(App[None]):
CSS = """
#summary { height: 1; padding: 0 1; color: $accent; }
#filter { height: 3; }
#accounts { height: 2fr; }
#keys { height: 1fr; border-top: solid $panel; }
#logs { height: 1fr; }
#errors { height: 1fr; }
#detail { height: 3; padding: 0 1; border-top: solid $panel; }
#filter { height: 1; border: none; padding: 0 1; }
#filter:focus { border: none; }
#accounts { height: 1fr; min-height: 3; }
#keys { height: auto; max-height: 6; }
#logs { height: 1fr; min-height: 3; }
#errors { height: 1fr; min-height: 3; }
#detail { height: 2; padding: 0 1; background: $surface-lighten-1; }
#status { height: 1; padding: 0 1; color: $text-muted; }
"""
BINDINGS = [
@@ -1569,12 +1902,14 @@ def run_textual(
]
def on_mount(self) -> None:
if start_page == "logs":
if start_page == "accounts":
self.push_screen(AccountsScreen())
elif start_page == "logs":
self.push_screen(LogsScreen())
elif start_page == "errors":
self.push_screen(ErrorsScreen())
else:
self.push_screen(AccountsScreen())
self.push_screen(DashboardScreen())
Sub2APIQuotaApp().run()
return 0
@@ -1640,8 +1975,9 @@ def build_parser() -> argparse.ArgumentParser:
)
parser.add_argument("--timeout", type=int, default=env_int("SUB2API_QUOTA_TUI_TIMEOUT", DEFAULT_TIMEOUT_SECONDS))
parser.add_argument("--once", action="store_true", help="print one snapshot and exit")
parser.add_argument("--logs", action="store_true", help="start on the request logs page (with --once: print a logs snapshot)")
parser.add_argument("--errors", action="store_true", help="start on the merged cn+us errors page (with --once: print an errors snapshot)")
parser.add_argument("--accounts", action="store_true", help="start on the dedicated accounts page (with --once: print the accounts snapshot)")
parser.add_argument("--logs", action="store_true", help="start on the dedicated request logs page (with --once: print a logs snapshot)")
parser.add_argument("--errors", action="store_true", help="start on the dedicated merged cn+us errors page (with --once: print an errors snapshot)")
parser.add_argument("--filter", default="", help="initial filter for --once output")
return parser
@@ -1652,8 +1988,9 @@ def main(argv: list[str] | None = None) -> int:
logs_limit = max(1, min(args.logs_limit, 1000))
errors_limit = max(1, min(args.errors_limit, 500))
errors_time_range = str(args.errors_time_range or DEFAULT_ERRORS_TIME_RANGE).strip() or DEFAULT_ERRORS_TIME_RANGE
if args.logs and args.errors:
print("choose only one of --logs or --errors", file=sys.stderr)
selected_pages = sum(bool(value) for value in (args.accounts, args.logs, args.errors))
if selected_pages > 1:
print("choose only one of --accounts, --logs, or --errors", file=sys.stderr)
return 2
if args.save_config or args.install:
config_path = write_api_url_config(args.api_url)
@@ -1711,8 +2048,10 @@ def main(argv: list[str] | None = None) -> int:
except Exception as exc:
print(f"keys error: {exc}", file=sys.stderr)
return 0
start_page = "accounts"
if args.logs:
start_page = "dashboard"
if args.accounts:
start_page = "accounts"
elif args.logs:
start_page = "logs"
elif args.errors:
start_page = "errors"