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

feat: per-key today usage panel under the accounts table

Browse Source 
- accounts page gains a #keys DataTable below the accounts table showing
  each API key's usage today: Key | Today | Tokens | Req, sorted by cost
- data comes from the Sub2API admin dashboard api-keys-trend (requests,
  tokens, key names) plus api-keys-usage (today_actual_cost), derived
  from the configured logs url and reusing the same admin API key
- key names use the same stable per-key colors as the logs page
- --once prints the same panel after the accounts snapshot; failures or
  a missing token surface in the status line without touching accounts

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
yunyaozhou
2026-07-21 12:32:10 +08:00
parent fd4e3256d4
commit 0e9b53dc79
5 changed files with 220 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sub2api_quota_tui.py
+131 -2
View File
@@ -18,7 +18,7 @@ from typing import Any
APP_NAME = "shusub2"
FALLBACK_VERSION = "0.2.2"
FALLBACK_VERSION = "0.2.3"
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"
@@ -397,6 +397,94 @@ def key_color(name: Any) -> str:
return KEY_COLOR_PALETTE[zlib.crc32(text.encode("utf-8")) % len(KEY_COLOR_PALETTE)]
def admin_api_base(logs_url: str) -> str:
parsed = urllib.parse.urlparse(str(logs_url or "").strip())
if not parsed.scheme or not parsed.netloc:
return ""
path = parsed.path.rstrip("/")
if path.endswith("/usage"):
path = path[: -len("/usage")]
if not path.endswith("/admin"):
return ""
return urllib.parse.urlunparse(parsed._replace(path=path, params="", query="", fragment=""))
def fetch_admin_json(url: str, token: str, timeout: int, body: dict[str, Any] | None = None) -> dict[str, Any]:
headers = {"Accept": "application/json"}
if str(token or "").strip():
headers["x-api-key"] = str(token).strip()
data = None
if body is not None:
headers["Content-Type"] = "application/json"
data = json.dumps(body).encode("utf-8")
req = urllib.request.Request(url, data=data, headers=headers)
with urllib.request.urlopen(req, timeout=timeout) as response:
payload = json.loads(response.read().decode("utf-8"))
if not isinstance(payload, dict):
raise RuntimeError("admin API did not return a JSON object")
return payload
def payload_data(payload: dict[str, Any]) -> dict[str, Any]:
data = payload.get("data")
return data if isinstance(data, dict) else payload
def fetch_key_usage_payload(logs_url: str, token: str, timeout: int, limit: int = 100) -> dict[str, Any]:
base = admin_api_base(logs_url)
if not base:
raise RuntimeError("cannot derive the admin API base from the logs url")
today = dt.date.today().isoformat()
query = urllib.parse.urlencode({"start_date": today, "end_date": today, "granularity": "day", "limit": limit})
trend_payload = fetch_admin_json(f"{base}/dashboard/api-keys-trend?{query}", token, timeout)
trend = payload_data(trend_payload).get("trend")
points = [point for point in trend if isinstance(point, dict)] if isinstance(trend, list) else []
ids = sorted({as_int(point.get("api_key_id")) for point in points if as_int(point.get("api_key_id"))})
stats: dict[str, Any] = {}
if ids:
costs_payload = fetch_admin_json(f"{base}/dashboard/api-keys-usage", token, timeout, body={"api_key_ids": ids})
raw_stats = payload_data(costs_payload).get("stats")
if isinstance(raw_stats, dict):
stats = raw_stats
return {"date": today, "trend": points, "stats": stats}
def normalize_key_rows(payload: dict[str, Any]) -> list[dict[str, Any]]:
stats = payload.get("stats") if isinstance(payload.get("stats"), dict) else {}
merged: dict[Any, dict[str, Any]] = {}
for point in payload.get("trend") or []:
if not isinstance(point, dict):
continue
key_id = as_int(point.get("api_key_id"))
name = str(point.get("key_name") or "").strip() or (f"#{key_id}" if key_id else "-")
row = merged.setdefault(key_id or name, {"id": key_id, "name": name, "requests": 0, "tokens": 0, "cost": 0.0})
row["requests"] += as_int(point.get("requests"))
row["tokens"] += as_int(point.get("tokens"))
for row in merged.values():
stat = stats.get(str(row["id"]))
if not isinstance(stat, dict):
stat = stats.get(row["id"])
if isinstance(stat, dict):
row["cost"] = as_float(stat.get("today_actual_cost"))
rows = list(merged.values())
rows.sort(key=lambda row: (-row["cost"], -row["tokens"], str(row["name"]).lower()))
return rows
def print_keys_once(payload: dict[str, Any]) -> None:
rows = normalize_key_rows(payload)
total_cost = sum(row["cost"] for row in rows)
print(f"keys today {payload.get('date') or '-'} | {len(rows)} keys | {format_cost(total_cost)}")
print("key today tokens req")
for row in rows:
print(
f"{str(row['name'])[:20]:<21} "
f"{format_cost(row['cost']):<10} "
f"{format_count(row['tokens']):<8} "
f"{format_count(row['requests'])}"
)
def nested_name(item: dict[str, Any], object_key: str, id_key: str, name_key: str = "") -> str:
nested = item.get(object_key)
if isinstance(nested, dict):
@@ -810,6 +898,8 @@ def run_textual(
self.payload: dict[str, Any] = {}
self.status_payload: dict[str, Any] = {}
self.status_error = ""
self.keys_payload: dict[str, Any] = {}
self.keys_error = ""
self.rows: list[dict[str, Any]] = []
self.row_by_key: dict[str, dict[str, Any]] = {}
@@ -818,6 +908,7 @@ def run_textual(
yield Static("", id="summary")
yield Input(placeholder="filter", id="filter")
yield DataTable(id="accounts")
yield DataTable(id="keys")
yield Static("", id="detail")
yield Static("", id="status")
yield Footer()
@@ -827,6 +918,10 @@ def run_textual(
table.cursor_type = "row"
table.zebra_stripes = True
table.add_columns("Name", "Provider", "Group", "Daily", "Today", "Tokens", "Req", "Kind", "5h", "7d", "Reset", "Status", "Availability")
keys_table = self.query_one("#keys", DataTable)
keys_table.cursor_type = "row"
keys_table.zebra_stripes = True
keys_table.add_columns("Key", "Today", "Tokens", "Req")
self.refresh_data(refresh=True)
self.set_interval(refresh_seconds, self.refresh_data)
@@ -853,12 +948,39 @@ def run_textual(
try:
self.payload = fetch_payload(api_url, timeout, refresh=refresh)
self.status_payload, self.status_error = fetch_optional_payload(status_url, timeout)
self.refresh_keys()
self.render_payload()
self.render_keys()
status_bits = [version_message, monitor_summary(self.status_payload, self.status_error), f"source {self.payload.get('source_name') or '-'}"]
if self.keys_error:
status_bits.append(f"keys: {self.keys_error}")
status.update(" | ".join(bit for bit in status_bits if bit) + f" | {api_url}")
except Exception as exc:
status.update(f"error: {exc}")
def refresh_keys(self) -> None:
self.keys_error = ""
self.keys_payload = {}
if not str(logs_token or "").strip():
self.keys_error = "logs token not configured"
return
try:
self.keys_payload = fetch_key_usage_payload(logs_url, logs_token, timeout)
except Exception as exc:
self.keys_error = str(exc)
def render_keys(self) -> None:
table = self.query_one("#keys", DataTable)
table.clear()
for row in normalize_key_rows(self.keys_payload):
color = key_color(row["name"])
table.add_row(
Text(str(row["name"]), style=color) if color else str(row["name"]),
format_cost(row["cost"]),
format_count(row["tokens"]),
format_count(row["requests"]),
)
def render_payload(self) -> None:
filter_text = self.query_one("#filter", Input).value
self.rows = normalize_account_rows(self.payload, filter_text)
@@ -1014,7 +1136,8 @@ def run_textual(
CSS = """
#summary { height: 1; padding: 0 1; color: $accent; }
#filter { height: 3; }
#accounts { height: 1fr; }
#accounts { height: 2fr; }
#keys { height: 1fr; border-top: solid $panel; }
#logs { height: 1fr; }
#detail { height: 3; padding: 0 1; border-top: solid $panel; }
#status { height: 1; padding: 0 1; color: $text-muted; }
@@ -1106,6 +1229,12 @@ def main(argv: list[str] | None = None) -> int:
return 0
status_payload, status_error = fetch_optional_payload(status_url, args.timeout)
print_once(fetch_payload(args.api_url, args.timeout, refresh=True), args.filter, status_payload, status_error)
if str(args.logs_token or "").strip():
print()
try:
print_keys_once(fetch_key_usage_payload(args.logs_url, args.logs_token, args.timeout))
except Exception as exc:
print(f"keys error: {exc}", file=sys.stderr)
return 0
return run_textual(
args.api_url,