From 311236d28f6cb2456df4b7bd2af92077f6694cd0 Mon Sep 17 00:00:00 2001 From: yunyaozhou Date: Tue, 4 Aug 2026 00:16:34 +0800 Subject: [PATCH] feat: add cached pricing workspace --- README.md | 396 ++++++++++-------------- pyproject.toml | 4 +- sub2api_quota_tui.py | 700 ++++++++++++++++++++++++++++++++++-------- tests/test_payload.py | 430 +++++++++++++++++++++++++- uv.lock | 2 +- 5 files changed, 1156 insertions(+), 376 deletions(-) diff --git a/README.md b/README.md index 04f809b..be6ab0e 100644 --- a/README.md +++ b/README.md @@ -8,190 +8,175 @@ - 对应服务：无 - 对应 stack：无 - Secret 边界：本地用户配置 -- 分发方式：`uvx` / `uv tool install`。 +- 目标 API：Pricing Monitor `GET /api/ui-data?view=workspace` +- 分发方式：`uvx` / `uv tool install` - 包名：`shusub2` - 命令名：`shusub2` - 本地状态路径：`~/.config/shusub2/` -- 目标 API：`cliproxy-codex-quota` 的 `/api/tui/accounts`、可选 `sub2api-status` `/api/status`、可选 Pricing Monitor `/api/ui-data?view=accounts`，以及可选 Sub2API `/api/v1/admin/usage` 和 `/api/v1/admin/ops/errors`。 -- 备注：账号数据无 secret，不需要 SSH、数据库访问、API key、OAuth credentials 或 plaintext env；请求与错误日志可选持有 Sub2API admin API key，仅存本机 `~/.config/shusub2/logs-token`（0600）或 `SHUSUB2_LOGS_TOKEN`。 +- 默认数据源：Pricing Monitor `GET /api/ui-data?view=workspace` -Single-page terminal dashboard for the token-safe Sub2API account feed, -request logs, and merged cn/us error logs. Dedicated full-table views remain -available for each data set. +`shusub2` 是 Sub2API 聚合运维 TUI。默认模式只读取 server2 Pricing Monitor +后台生成的内存 snapshot，不直接访问 Accounts helper、`sub2api-status`、Sub2API +admin API、PostgreSQL、SSH 或 pricing upstream。每次数据刷新只请求一次 workspace +projection；Dashboard、Accounts、Sources、Requests 和 Errors 共用同一份进程内缓存。后台 snapshot +读取之外，默认启动仍会进行一次短超时的 Gitea 版本检查；使用 +`--no-version-check` 或 `SHUSUB2_NO_VERSION_CHECK=1` 可关闭它。 ## Quick Start -Run once from a public Git repo with `uvx`: +已安装时直接运行： ```bash -uvx --from git+https://gitea.shujk.top/shujakuin/shusub2.git shusub2 \ - --api-url https://codex.server2.shujk.top/1232131231313123/api/tui/accounts -``` - -Bootstrap a new machine from `uvx`: save the API URL, install `shusub2` as a -user command, then run it later as `shusub2`: - -```bash -uvx --from git+https://gitea.shujk.top/shujakuin/shusub2.git shusub2 \ - --api-url https://codex.server2.shujk.top/1232131231313123/api/tui/accounts \ - --install shusub2 ``` -Install as a user command: +默认 workspace URL 是 Tailscale HTTPS 入口： + +```text +https://price.tailbeb9ad.ts.net/api/ui-data?view=workspace +``` + +覆盖并保存 workspace URL： + +```bash +shusub2 \ + --workspace-url https://price.tailbeb9ad.ts.net/api/ui-data?view=workspace \ + --save-config +``` + +从 Gitea 安装为用户命令： ```bash uv tool install git+https://gitea.shujk.top/shujakuin/shusub2.git -shusub2 --api-url https://codex.server2.shujk.top/1232131231313123/api/tui/accounts --save-config +shusub2 ``` -Configure the default public API URL for `shusub2`: +也可以一次性运行： + +```bash +uvx --from git+https://gitea.shujk.top/shujakuin/shusub2.git shusub2 +``` + +## Data Contract + +Pricing Monitor 在后台按固定节奏采集并裁剪数据。普通 workspace HTTP 读取只复制 +最近一次内存 snapshot，不触发 SSH、生产 SQL、旧 helper refresh 或 pricing source +refresh。组件失败时服务端保留对应 last-good 数据，并在 `components` / `state` 中标记 +`stale`、`partial` 和更新时间。 + +Workspace 当前提供以下有界 projection： + +- `accounts`：账号摘要、quota window、当日用量和 provider/group/status 字段。 +- `status`：channel monitor 摘要。 +- `sources`：Pricing Monitor 已有的脱敏 source 余额与健康状态。 +- `traffic.requests`：最近的有界请求样本。 +- `traffic.errors`：有界错误聚合，保留聚合次数和状态链路。 +- `traffic.keys`：24 小时 key 使用聚合。 + +投影允许展示 account/key 名称、稳定 ID、model、instance 和运维状态，但不包含 API +key 原文、access/refresh token、cookie、密码、数据库凭据或请求/响应正文。ID 始终按 +字符串处理，避免 JavaScript 或 Python 客户端误损失大整数精度。 + +## Interface + +默认首页是紧凑 Dashboard。页面信息架构为： + +```text +Dashboard | Accounts | Sources | Requests | Errors +``` + +快捷键： + +- `a`：聚焦 Accounts，或从独立页面打开 Accounts。 +- `p`：打开 Sources。 +- `k`：在 Dashboard 聚焦 Key usage。 +- `l`：聚焦 Requests，或从独立页面打开 Requests。 +- `e`：聚焦 Errors，或从独立页面打开 Errors。 +- `d`：从独立页面返回 Dashboard。 +- `/`：聚焦当前页面 filter；Dashboard filter 同时作用于全部表。 +- `r`：重新读取 workspace snapshot。该操作不会要求服务端立即采集上游。 + +窄终端中的长 key、account 和 model 使用横向滚动，不截短原字段。独立页面可通过 +以下参数启动： + +```bash +shusub2 --accounts +shusub2 --sources # --pricing 保留为兼容别名 +shusub2 --requests # --logs 保留为兼容别名 +shusub2 --errors +``` + +加上 `--once` 可输出对应 snapshot 并退出： + +```bash +shusub2 --once --requests +``` + +Accounts 只在 canonical 名称 `{family}-quota-{source}` 或 +`{family}-quotaonly-{source}` 与唯一健康 source 精确匹配时显示 CNY 余额。不会根据 +provider、URL、display name 或模糊文本推断；source 为 error/stale 时保留名称和状态， +但不伪造 CNY 数值。 + +Requests 展示 key、account、model、token bucket、actual cost、first-token latency、 +duration 和 decode throughput。Errors 展示 instance、聚合次数、status path、key、 +account、model、phase/type/owner 和时间。Key usage 使用 workspace 的 24 小时窗口， +不再宣称是客户端本地“今日”查询。 + +## Legacy Direct + +`--legacy-direct` 仅用于迁移对账和故障诊断。它恢复旧拓扑： + +- Accounts helper `/api/tui/accounts` +- `sub2api-status` `/api/status` +- Pricing Monitor `view=accounts` +- Sub2API admin usage/key/error API + +```bash +shusub2 --legacy-direct --accounts +shusub2 --legacy-direct --requests +``` + +显式传入空 `--workspace-url ''` 也会进入 legacy direct 模式。legacy Requests、Errors +和 Key usage 需要 Sub2API admin API key；默认 workspace 模式不会读取该凭据文件。 + +legacy token 仅允许放在 `SHUSUB2_LOGS_TOKEN` 或权限为 `0600` 的 +`~/.config/shusub2/logs-token`： ```bash mkdir -p ~/.config/shusub2 chmod 700 ~/.config/shusub2 -printf '%s\n' 'https://codex.server2.shujk.top/1232131231313123/api/tui/accounts' > ~/.config/shusub2/api-url -chmod 600 ~/.config/shusub2/api-url -shusub2 -``` - -Environment variables override the config file: - -- `SHUSUB2_API_URL` -- `SUB2API_QUOTA_TUI_API_URL` -- `SHUSUB2_API_URL_FILE` -- `SHUSUB2_STATUS_URL` -- `SHUSUB2_STATUS_URL_FILE` -- `SHUSUB2_PRICING_URL` -- `SHUSUB2_PRICING_URL_FILE` -- `SHUSUB2_LOGS_URL` -- `SHUSUB2_LOGS_URL_FILE` -- `SHUSUB2_LOGS_TOKEN` -- `SHUSUB2_LOGS_TOKEN_FILE` -- `SHUSUB2_LOGS_REFRESH_SECONDS` -- `SHUSUB2_LOGS_LIMIT` -- `SHUSUB2_ERRORS_CN_URL` -- `SHUSUB2_ERRORS_CN_URL_FILE` -- `SHUSUB2_ERRORS_US_URL` -- `SHUSUB2_ERRORS_US_URL_FILE` -- `SHUSUB2_ERRORS_REFRESH_SECONDS` -- `SHUSUB2_ERRORS_LIMIT` -- `SHUSUB2_ERRORS_TIME_RANGE` -- `SHUSUB2_VERSION_CHECK_URL` -- `SHUSUB2_VERSION_CHECK_TIMEOUT` -- `SHUSUB2_NO_VERSION_CHECK` - -## Unified Dashboard - -`shusub2` opens a compact dashboard for Accounts, Keys, Logs, and Errors. Its -summary also reports Pricing Monitor upstream health and CNY total. The -Accounts table attaches a source balance only when an account has a canonical -quota-source name and the sanitized Pricing Monitor projection has one -unambiguous matching source. Press `a`, `k`, `l`, or `e` to focus a dashboard -table; press `p` to open the full Upstreams source-balance page. `/` filters -the active table and `r` refreshes all data immediately. Automatic Accounts -and Upstreams, Keys/Logs, and Errors refresh defaults to every five minutes. The client requests -gzip-compressed JSON and transparently decodes it when the upstream supports it. -The two-line detail area follows the selected row. Accounts, Keys, Logs, and -Errors use green, yellow, magenta, and red section styling respectively. - -The dashboard sizes every column from the fetched content so wide terminals show -complete keys, accounts, and models. On a narrow terminal, a table scrolls -horizontally instead of truncating a field: - -```text -ACCOUNT | Group | Source | Src CNY | Src state | Today | Daily | 5h | 7d | Avail -UPSTREAMS (p) | Source | Kind | Balance | CNY | Unit | Status | Updated | Age -KEY | Today | Tokens | Req -LOG KEY | Account | Model | First | Duration | Tok/s | Input | Output | Cache | Tokens | Cost | Time | Age -ERR | Status | Key | Account | Model | Time | Age -``` - -The Keys table shows today's usage sorted by actual cost, including key name, -tokens, and request count. It uses the same admin API key as Logs and Errors; -without that token the table is hidden and the summary remains compact. - -Use `--accounts`, `--pricing`, `--logs`, or `--errors` to start a dedicated -full-table view. The corresponding `--once` form still prints only that data set. - -## Request Logs Page - -The middle dashboard table shows request logs, and `l` focuses it. -`shusub2 --logs` starts the dedicated logs page, while -`shusub2 --once --logs` prints one logs snapshot to stdout. - -The logs page reads the latest requests (default 100, `--logs-limit`) from the -Sub2API admin usage API and refreshes every five minutes by default -(`--logs-refresh-seconds` / `SHUSUB2_LOGS_REFRESH_SECONDS`). Columns: - -```text -Key | Account | Model | Effort | Type | Input | Output | Cache | Tokens | Cost | First | Duration | Tok/s | Time | Age -``` - -`Type` is the Sub2API `request_type` (`sync` / `stream` / `ws_v2` / `cyber`). -`Tokens` is input + output + cache write + cache read; the detail line below -the table shows the per-bucket breakdown, actual cost, first-token latency, -decode speed, upstream model mapping, user, and request id. - -`Effort` is the request's `reasoning_effort` (`-` when absent). `First` is the -first-token latency and `Duration` the total request time, both shown in -seconds. `Tok/s` is the decode throughput computed as -`output_tokens / (duration - first_token)`; it shows `-` when there is no -output or no positive decode window. `Input`, `Output`, and `Cache` are token -buckets; `Cache` combines cache write and cache read. `Tokens` is their total. -`Age` is relative to local current time (`now`, `5m ago`, `2h ago`, etc.). In the TUI each -API key name is rendered in a stable per-key color so rows from the same -key are easy to group visually (`--once --logs` output stays plain text). - -The logs URL defaults to `https://sub2apicn.shujk.top/api/v1/admin/usage` and -can be overridden with `--logs-url` / `SHUSUB2_LOGS_URL` / -`~/.config/shusub2/logs-url`. - -The page needs a Sub2API **admin API key** (generated in the Sub2API web UI -under Settings). Configure it once: - -```bash -mkdir -p ~/.config/shusub2 && chmod 700 ~/.config/shusub2 printf '%s\n' '' > ~/.config/shusub2/logs-token chmod 600 ~/.config/shusub2/logs-token ``` -or via `SHUSUB2_LOGS_TOKEN`. The key is sent as the `x-api-key` header and is -never written anywhere else by the client. Without a token the dashboard Logs -and Errors tables stay empty and show a configuration hint; the Accounts table -keeps working without any secret. +## Configuration -## Errors Page +Workspace-first 配置： -The bottom dashboard table shows merged errors, and `e` focuses it. -`shusub2 --errors` starts the dedicated errors page, while -`shusub2 --once --errors` prints one merged snapshot to stdout. +- `--workspace-url` / `SHUSUB2_WORKSPACE_URL` / + `SHUSUB2_WORKSPACE_URL_FILE` +- `--refresh-seconds` / `SUB2API_QUOTA_TUI_REFRESH_SECONDS` +- `--timeout` / `SUB2API_QUOTA_TUI_TIMEOUT` +- `--accounts` / `--sources` / `--requests` / `--errors` +- `--once` +- `--save-config` +- `--install` +- `--version-check-url` / `SHUSUB2_VERSION_CHECK_URL` +- `--no-version-check` / `SHUSUB2_NO_VERSION_CHECK` -The page pulls the latest admin ops error logs from **both** cn and us in -parallel, then merges them by `created_at`: +Legacy-only 配置： -- cn default: `https://sub2apicn.shujk.top/api/v1/admin/ops/errors` -- us default: `https://us.sub2.shujk.top/api/v1/admin/ops/errors` +- `--legacy-direct` +- `--api-url` / `SUB2API_QUOTA_TUI_API_URL` +- `--status-url` / `SHUSUB2_STATUS_URL` +- `--pricing-url` / `SHUSUB2_PRICING_URL` +- `--logs-url` / `SHUSUB2_LOGS_URL` +- `--logs-token` / `SHUSUB2_LOGS_TOKEN` +- `--errors-cn-url` / `SHUSUB2_ERRORS_CN_URL` +- `--errors-us-url` / `SHUSUB2_ERRORS_US_URL` +- `--logs-limit`、`--errors-limit`、`--errors-time-range` -`us.sub2.shujk.top` is the DNS-only standard HTTPS direct entry for the same -server4 backend. It avoids the Cloudflare 1010 behavior sometimes returned by -`sub2apius.shujk.top` to non-browser clients while avoiding the fixed `:19857` -port in the default client configuration. Use `--errors-us-url` to select a -different direct origin when required. - -The dashboard and dedicated errors page refresh every five minutes by default -(`--errors-refresh-seconds` / `SHUSUB2_ERRORS_REFRESH_SECONDS`). Query window -defaults to `24h` (`--errors-time-range` / `SHUSUB2_ERRORS_TIME_RANGE`); each source uses -`page=1` and `page_size` from `--errors-limit` (default 100, max 500). - -Columns: - -```text -Node | Status | Key | Account | Model | Phase | Type | Owner | Time | Age -``` - -`Age` is relative to local current time (`now`, `5m ago`, `2h ago`, etc.). -Auth reuses the same admin API key as the logs page. +URL config 文件默认位于 `~/.config/shusub2/`。环境变量优先于 config 文件。 ## Local Development @@ -200,96 +185,33 @@ cd apps/sub2api-quota-tui uv run shusub2 ``` -Run inside zellij: +运行完整测试和静态检查： ```bash -zellij action new-pane --name sub2api-quota -- \ - bash -lc 'cd /home/shujakuin/infra/apps/sub2api-quota-tui && uv run shusub2' +uv run python -m unittest discover -s tests -p 'test_*.py' -v +uv run python -m py_compile sub2api_quota_tui.py +uv lock --check ``` -Configuration: +## Maintenance Notes -- `--api-url` / `SUB2API_QUOTA_TUI_API_URL` -- `--status-url` / `SHUSUB2_STATUS_URL` -- `--pricing-url` / `SHUSUB2_PRICING_URL` -- `--logs-url` / `SHUSUB2_LOGS_URL` -- `--logs-token` / `SHUSUB2_LOGS_TOKEN` -- `--logs-refresh-seconds` / `SHUSUB2_LOGS_REFRESH_SECONDS` -- `--logs-limit` / `SHUSUB2_LOGS_LIMIT` -- `--errors-refresh-seconds` / `SHUSUB2_ERRORS_REFRESH_SECONDS` -- `--errors-limit` / `SHUSUB2_ERRORS_LIMIT` -- `--errors-time-range` / `SHUSUB2_ERRORS_TIME_RANGE` -- `--accounts` / `--pricing` / `--logs` / `--errors` -- `--save-config` -- `--install` -- `--refresh-seconds` / `SUB2API_QUOTA_TUI_REFRESH_SECONDS` -- `--timeout` / `SUB2API_QUOTA_TUI_TIMEOUT` +Workspace adapter 只负责把 server projection 映射为现有渲染模型，不得在普通刷新中 +重新引入 legacy HTTP 请求。新增 projection 字段时应同时验证：字段 allowlist、字符串 ID、 +finite number、payload 上限、单请求 cache、错误聚合次数和 legacy fallback。 -The dashboard reads `/api/tui/accounts`, can optionally read `sub2api-status` -`/api/status` for channel monitor health, and reads Pricing Monitor's -`/api/ui-data?view=accounts` for upstream source balances. The default pricing URL -is the Tailscale HTTPS service; `--pricing-url`, `SHUSUB2_PRICING_URL`, or -`~/.config/shusub2/pricing-url` can override it. This is a cached, read-only -projection collected by Pricing Monitor on its own cadence: `shusub2` neither -contacts upstream providers directly nor asks Pricing Monitor to refresh. The Accounts table does not need SSH, -database access, API keys, OAuth credentials, or plaintext env files. The Keys, -Logs, and Errors tables authenticate to the Sub2API admin APIs with -an admin API key stored only in `~/.config/shusub2/logs-token` (0600) or -`SHUSUB2_LOGS_TOKEN`. +`cliproxy-codex-quota` 与 `sub2api-status` 当前仍可作为 Aggregation Hub 的过渡采集输入和 +双读对照，但不再是默认 TUI 客户端直连依赖。移除这些过渡输入前，必须先完成 server2 +聚合字段对账和 AstrBot `view=alerts` 连续性验证。 -When `--status-url` is omitted, `shusub2` infers a sibling `/api/status` URL -from account URLs ending in `/api/tui/accounts`, so the public Codex endpoint -automatically enables monitor availability. On startup it also checks the -public Gitea repo for a newer package version and prints a short upgrade hint -when one is available. +## Tests And Release -On the dedicated `--accounts` page, a per-key panel shows today's usage for -each API key (`Key | Today | Tokens | Req`, sorted by cost, key names in the -same per-key colors as the logs page). It combines the Sub2API admin -`dashboard/api-keys-trend` and `dashboard/api-keys-usage` endpoints and -needs the same admin API key as the logs page; without a token the panel -stays empty and the status line shows a hint. `--once` prints the same -panel after the accounts table when a token is configured. +发布遵循本仓库 `client-tui-gitea-uv` 流程：完成 scoped diff 审阅和测试后，将 +`apps/sub2api-quota-tui/` 同步到独立 Gitea 仓库，再通过 git source 执行 +`uv tool install --force`。安装后检查 `uv-receipt.toml`，确保来源仍是 Gitea git URL， +而不是本地目录。 -The dedicated Accounts page keeps the full columns for scanning inside zellij: +## Secret Boundary -```text -Name | Provider | Group | Source | Src CNY | Src state | Daily | Today | Tokens | Req | Kind | 5h | 7d | Reset | Status | Availability -``` - -`Provider` distinguishes `openai` and `anthropic` accounts from the public -`platform` field returned by the API. - -`Daily` is shown as `used/limit` when Sub2API has `quota_daily_*` fields in -`accounts.extra`; otherwise it is `-`. - -`5h` / `7d` are only the account feed's used/remaining quota percentages. They -are not upstream provider balances and therefore never claim to be a cash or -source-account balance. - -`Source`, `Src CNY`, and `Src state` use Pricing Monitor's cached, sanitized -`view=accounts` projection. A balance is attached only for exact canonical -account names of the form `{family}-quota-{source}` or -`{family}-quotaonly-{source}`, where `{source}` is one or more lowercase -alphanumeric segments joined by single hyphens, and a single normalized source -name. Provider, URL, display-name, and fuzzy matching are deliberately not used. Accounts -without a unique mapping show `-`; an `error` or `stale` source retains its -source/state label but does not show a CNY amount. - -The dedicated `--pricing` page shows Pricing Monitor source rows such as -`code-plan`, `codexapis`, `testvideo`, `kedaya`, `ycy`, and `mdkj`: raw available -balance with its source unit, derived CNY when supplied, health state, and last -successful collection time. It reads only the existing sanitized source projection. - -`Group` is the derived Sub2API tier alias group. Higher tiers win when multiple -aliases exist: `id < slow < fast < sfast`. The table shows `sfast` first, then -`fast`, `slow`, `id`, and ungrouped accounts. - -When `--status-url` is configured, the status line shows channel monitor -health, and the selected account detail shows the matching monitor status when -one exists. Monitor binding first uses the shared `base_url_hash` emitted by -the account API and `sub2api-status`; name-token matching remains only as a -fallback for older status payloads. - -`Availability` is derived from the bound channel monitor. Accounts without a -matching monitor show `-`. +默认 workspace 模式不需要任何客户端 secret，也不读取 legacy token 文件。服务端 +workspace 不聚合原始凭据或请求/响应正文。legacy direct 模式的 admin API key 仅保存在 +本机用户配置或进程环境中，不得写入仓库、文档、日志或测试 fixture。 diff --git a/pyproject.toml b/pyproject.toml index ae4a10c..d4463ee 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,7 +1,7 @@ [project] name = "shusub2" -version = "0.2.15" -description = "Terminal UI for Sub2API account quota and daily usage" +version = "0.3.0" +description = "Aggregated operations TUI for Sub2API" readme = "README.md" requires-python = ">=3.11" license = { text = "MIT" } diff --git a/sub2api_quota_tui.py b/sub2api_quota_tui.py index e9fb393..83d4a44 100644 --- a/sub2api_quota_tui.py +++ b/sub2api_quota_tui.py @@ -3,15 +3,19 @@ from __future__ import annotations import argparse +import copy import datetime as dt import gzip import importlib.metadata +import io import json import math import os import re import subprocess import sys +import threading +import time import urllib.parse import urllib.request import zlib @@ -21,7 +25,9 @@ from typing import Any APP_NAME = "shusub2" -FALLBACK_VERSION = "0.2.15" +FALLBACK_VERSION = "0.3.0" +DEFAULT_WORKSPACE_URL = "https://price.tailbeb9ad.ts.net/api/ui-data?view=workspace" +DEFAULT_WORKSPACE_URL_CONFIG_FILE = "~/.config/shusub2/workspace-url" 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" @@ -44,6 +50,15 @@ DEFAULT_ERRORS_LIMIT = 100 DEFAULT_ERRORS_TIME_RANGE = "24h" DEFAULT_TIMEOUT_SECONDS = 10 DEFAULT_VERSION_CHECK_TIMEOUT_SECONDS = 2 +MAX_JSON_RESPONSE_BYTES = 4 * 1024 * 1024 +MAX_WORKSPACE_JSON_BYTES = 768 * 1024 +MAX_WORKSPACE_ACCOUNTS = 512 +MAX_WORKSPACE_MONITORS = 256 +MAX_WORKSPACE_SOURCES = 64 +MAX_WORKSPACE_INSTANCES = 16 +MAX_WORKSPACE_REQUESTS = 100 +MAX_WORKSPACE_ERRORS = 256 +MAX_WORKSPACE_KEYS = 256 MONITOR_OK_STATUSES = {"operational", "ok", "success"} MONITOR_FAILED_STATUSES = {"error", "failed", "failure"} MONITOR_STOPWORDS = {"response", "responses", "monitor"} @@ -78,6 +93,14 @@ def configured_url(env_names: tuple[str, ...], config_path: str, default: str = return default +def default_workspace_url() -> str: + return configured_url( + ("SHUSUB2_WORKSPACE_URL",), + os.environ.get("SHUSUB2_WORKSPACE_URL_FILE", DEFAULT_WORKSPACE_URL_CONFIG_FILE), + DEFAULT_WORKSPACE_URL, + ) + + def default_api_url() -> str: return configured_url( ("SHUSUB2_API_URL", "SUB2API_QUOTA_TUI_API_URL"), @@ -116,6 +139,12 @@ def default_logs_token() -> str: ) +def workspace_url_config_file_path() -> Path: + return Path( + os.environ.get("SHUSUB2_WORKSPACE_URL_FILE", DEFAULT_WORKSPACE_URL_CONFIG_FILE) + ).expanduser() + + def config_file_path() -> Path: return Path(os.environ.get("SHUSUB2_API_URL_FILE", DEFAULT_CONFIG_FILE)).expanduser() @@ -266,10 +295,22 @@ def as_float(value: Any) -> float: def as_int(value: Any) -> int: try: - if value is None or str(value).strip() == "": + if value is None: return 0 - return int(float(value)) - except Exception: + if isinstance(value, bool): + return int(value) + if isinstance(value, int): + return value + if isinstance(value, float): + return int(value) if math.isfinite(value) else 0 + text = str(value).strip() + if not text: + return 0 + if re.fullmatch(r"[+-]?[0-9]+", text): + return int(text) + number = float(text) + return int(number) if math.isfinite(number) else 0 + except (TypeError, ValueError, OverflowError): return 0 @@ -377,24 +418,54 @@ def add_refresh_param(api_url: str, refresh: bool) -> str: return urllib.parse.urlunparse(parsed._replace(query=urllib.parse.urlencode(query))) -def decode_json_response(response: Any, error_message: str) -> dict[str, Any]: - raw = response.read() +def decode_json_response( + response: Any, + error_message: str, + *, + maximum_bytes: int = MAX_JSON_RESPONSE_BYTES, +) -> dict[str, Any]: + limit = max(1, int(maximum_bytes)) + raw = response.read(limit + 1) + if len(raw) > limit: + raise RuntimeError(error_message) content_encoding = str(response.headers.get("Content-Encoding", "")).lower() if "gzip" in {value.strip() for value in content_encoding.split(",")}: - raw = gzip.decompress(raw) + try: + with gzip.GzipFile(fileobj=io.BytesIO(raw)) as compressed: + raw = compressed.read(limit + 1) + except (OSError, EOFError, zlib.error) as exc: + raise RuntimeError(error_message) from exc + if len(raw) > limit: + raise RuntimeError(error_message) data = json.loads(raw.decode("utf-8")) if not isinstance(data, dict): raise RuntimeError(error_message) return data -def fetch_payload(api_url: str, timeout: int, *, refresh: bool = False) -> dict[str, Any]: +def fetch_payload( + api_url: str, + timeout: int, + *, + refresh: bool = False, + maximum_bytes: int = MAX_JSON_RESPONSE_BYTES, + use_proxy: bool = True, +) -> dict[str, Any]: req = urllib.request.Request( add_refresh_param(api_url, refresh), headers={"Accept": "application/json", "Accept-Encoding": "gzip"}, ) - with urllib.request.urlopen(req, timeout=timeout) as response: - return decode_json_response(response, "API did not return a JSON object") + open_request = ( + urllib.request.urlopen + if use_proxy + else urllib.request.build_opener(urllib.request.ProxyHandler({})).open + ) + with open_request(req, timeout=timeout) as response: + return decode_json_response( + response, + "API did not return a JSON object", + maximum_bytes=maximum_bytes, + ) def fetch_optional_payload(url: str, timeout: int) -> tuple[dict[str, Any], str]: @@ -434,6 +505,212 @@ def fetch_optional_pricing_payload(pricing_url: str, timeout: int) -> tuple[dict return {}, "unavailable" +def workspace_values_are_finite(value: Any) -> bool: + pending = [value] + visited = 0 + while pending: + current = pending.pop() + visited += 1 + if visited > 100_000: + return False + if isinstance(current, float) and not math.isfinite(current): + return False + if isinstance(current, dict): + pending.extend(current.values()) + elif isinstance(current, list): + pending.extend(current) + return True + + +def fetch_workspace_payload(workspace_url: str, timeout: int) -> dict[str, Any]: + payload = fetch_payload( + workspace_url, + timeout, + maximum_bytes=MAX_WORKSPACE_JSON_BYTES, + use_proxy=False, + ) + traffic = payload.get("traffic") + if ( + payload.get("view") != "workspace" + or not isinstance(payload.get("state"), dict) + or not isinstance(payload.get("components"), dict) + or not isinstance(payload.get("accounts"), dict) + or not isinstance(payload.get("status"), dict) + or not isinstance(payload.get("sources"), list) + or not isinstance(traffic, dict) + or not workspace_values_are_finite(payload) + ): + raise RuntimeError("Pricing Monitor returned an invalid workspace projection") + channel_monitors = payload["status"].get("channel_monitors") + bounded_fields = ( + ("accounts", payload["accounts"].get("accounts"), MAX_WORKSPACE_ACCOUNTS), + ( + "monitors", + channel_monitors.get("items") if isinstance(channel_monitors, dict) else None, + MAX_WORKSPACE_MONITORS, + ), + ("instances", traffic.get("instances"), MAX_WORKSPACE_INSTANCES), + ("requests", traffic.get("requests"), MAX_WORKSPACE_REQUESTS), + ("errors", traffic.get("errors"), MAX_WORKSPACE_ERRORS), + ("keys", traffic.get("keys"), MAX_WORKSPACE_KEYS), + ) + for label, rows, maximum in bounded_fields: + if not isinstance(rows, list) or len(rows) > maximum or any( + not isinstance(row, dict) for row in rows + ): + raise RuntimeError(f"Pricing Monitor returned invalid workspace {label}") + if len(payload["sources"]) > MAX_WORKSPACE_SOURCES or not all( + isinstance(source, dict) for source in payload["sources"] + ): + raise RuntimeError("Pricing Monitor returned invalid workspace sources") + return payload + + +def workspace_accounts_payload(payload: dict[str, Any]) -> dict[str, Any]: + value = payload.get("accounts") + return value if isinstance(value, dict) else {} + + +def workspace_status_payload(payload: dict[str, Any]) -> dict[str, Any]: + value = payload.get("status") + return value if isinstance(value, dict) else {} + + +def workspace_pricing_payload(payload: dict[str, Any]) -> dict[str, Any]: + return { + "service": payload.get("service"), + "view": "accounts", + "generated_at": payload.get("generated_at"), + "sources": payload.get("sources") if isinstance(payload.get("sources"), list) else [], + } + + +def workspace_logs_payload(payload: dict[str, Any]) -> dict[str, Any]: + traffic = payload.get("traffic") if isinstance(payload.get("traffic"), dict) else {} + raw_items = traffic.get("requests") if isinstance(traffic.get("requests"), list) else [] + items = [] + for raw in raw_items: + if not isinstance(raw, dict): + continue + item = dict(raw) + item["_node"] = str(raw.get("instance") or raw.get("node") or "-") + if "total_cost" not in item: + item["total_cost"] = ( + raw.get("cost") + if raw.get("cost") is not None + else raw.get("actual_cost") + ) + items.append(item) + return { + "data": {"items": items, "total": len(items)}, + "generated_at": traffic.get("generated_at") or payload.get("generated_at"), + "workspace": True, + } + + +def workspace_keys_payload(payload: dict[str, Any]) -> dict[str, Any]: + traffic = payload.get("traffic") if isinstance(payload.get("traffic"), dict) else {} + return { + "items": traffic.get("keys") if isinstance(traffic.get("keys"), list) else [], + "generated_at": traffic.get("generated_at") or payload.get("generated_at"), + "window_hours": as_int(traffic.get("window_hours")), + "workspace": True, + } + + +def workspace_errors_payload(payload: dict[str, Any]) -> dict[str, Any]: + traffic = payload.get("traffic") if isinstance(payload.get("traffic"), dict) else {} + raw_items = traffic.get("errors") if isinstance(traffic.get("errors"), list) else [] + items = [] + for index, raw in enumerate(raw_items): + if not isinstance(raw, dict): + continue + item = dict(raw) + item.setdefault("id", index + 1) + item["_node"] = str(raw.get("instance") or raw.get("node") or "-") + item.setdefault("created_at", raw.get("latest_at")) + item.setdefault("requested_model", raw.get("model")) + item.setdefault("phase", raw.get("error_source")) + item.setdefault("type", raw.get("error_type")) + item.setdefault("message", raw.get("error_detail") or "") + items.append(item) + instances = traffic.get("instances") if isinstance(traffic.get("instances"), list) else [] + sources = {} + for raw in instances: + if not isinstance(raw, dict): + continue + node = str(raw.get("instance") or raw.get("name") or "-") + sources[node] = { + "ok": raw.get("ok") is not False, + "total": as_int(raw.get("error_total")), + "fetched": sum(1 for item in items if item.get("_node") == node), + "error": str(raw.get("error") or ""), + } + return { + "items": items, + "sources": sources, + "time_range": f"{as_int(traffic.get('window_hours')) or 24}h", + "limit": as_int(traffic.get("limit")), + "generated_at": traffic.get("generated_at") or payload.get("generated_at"), + "workspace": True, + } + + +def workspace_state_summary(payload: dict[str, Any], client_error: str = "") -> str: + if client_error: + return "workspace endpoint unavailable (using last good)" + state = payload.get("state") if isinstance(payload.get("state"), dict) else {} + components = payload.get("components") if isinstance(payload.get("components"), dict) else {} + stale = sorted( + str(name) + for name, value in components.items() + if isinstance(value, dict) and value.get("stale") is True + ) + if stale: + return f"workspace partial ({', '.join(stale)} stale)" + if state.get("ok") is True: + return "workspace ok" + return "workspace partial" if state.get("partial") is True else "workspace unknown" + + +class WorkspaceCache: + def __init__(self, url: str, timeout: int, ttl_seconds: int) -> None: + self.url = str(url or "").strip() + self.timeout = max(1, int(timeout)) + self.ttl_seconds = max(1, int(ttl_seconds)) + self.payload: dict[str, Any] = {} + self.fetched_at = 0.0 + self.last_attempt_at = 0.0 + self.retry_seconds = min(30, self.ttl_seconds) + self.error = "" + self.network_fetches = 0 + self.lock = threading.RLock() + + def get(self, *, force: bool = False) -> dict[str, Any]: + with self.lock: + now = time.monotonic() + if not force: + if self.error and self.last_attempt_at and now - self.last_attempt_at < self.retry_seconds: + if self.payload: + return copy.deepcopy(self.payload) + raise RuntimeError(self.error) + if not self.error and self.payload and now - self.fetched_at < self.ttl_seconds: + return copy.deepcopy(self.payload) + self.last_attempt_at = now + self.network_fetches += 1 + try: + payload = fetch_workspace_payload(self.url, self.timeout) + except Exception as exc: + self.error = "workspace unavailable" + if self.payload: + return copy.deepcopy(self.payload) + raise RuntimeError(self.error) from exc + self.payload = copy.deepcopy(payload) + self.fetched_at = time.monotonic() + self.error = "" + return copy.deepcopy(self.payload) + + def logs_request_url(logs_url: str, limit: int) -> str: parsed = urllib.parse.urlparse(logs_url) query = urllib.parse.parse_qsl(parsed.query, keep_blank_values=True) @@ -584,6 +861,29 @@ def fetch_key_usage_payload(logs_url: str, token: str, timeout: int, limit: int def normalize_key_rows(payload: dict[str, Any], filter_text: str = "") -> list[dict[str, Any]]: needle = filter_text.strip().lower() + workspace_items = payload.get("items") if isinstance(payload.get("items"), list) else [] + if workspace_items: + rows = [] + for item in workspace_items: + if not isinstance(item, dict): + continue + key_id = str(item.get("api_key_id") or "").strip() + node = str(item.get("instance") or item.get("node") or "-").strip() or "-" + name = str(item.get("api_key_name") or "").strip() or (f"#{key_id}" if key_id else "-") + if needle and needle not in f"{node} {name}".lower(): + continue + rows.append( + { + "id": key_id, + "node": node, + "name": name, + "requests": as_int(item.get("request_count")), + "tokens": as_int(item.get("token_count")), + "cost": as_float(item.get("actual_cost") if item.get("actual_cost") is not None else item.get("cost")), + } + ) + rows.sort(key=lambda row: (-row["cost"], -row["tokens"], str(row["node"]), str(row["name"]).lower())) + return rows stats = payload.get("stats") if isinstance(payload.get("stats"), dict) else {} merged: dict[Any, dict[str, Any]] = {} for point in payload.get("trend") or []: @@ -593,7 +893,7 @@ def normalize_key_rows(payload: dict[str, Any], filter_text: str = "") -> list[d name = str(point.get("key_name") or "").strip() or (f"#{key_id}" if key_id else "-") if needle and needle not in name.lower(): continue - row = merged.setdefault(key_id or name, {"id": key_id, "name": name, "requests": 0, "tokens": 0, "cost": 0.0}) + row = merged.setdefault(key_id or name, {"id": key_id, "node": "-", "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(): @@ -610,10 +910,12 @@ def normalize_key_rows(payload: dict[str, Any], filter_text: str = "") -> list[d 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") + period = f"{as_int(payload.get('window_hours')) or 24}h" if payload.get("workspace") else f"today {payload.get('date') or '-'}" + print(f"keys {period} | {len(rows)} keys | {format_cost(total_cost)}") + print("node key cost tokens req") for row in rows: print( + f"{str(row['node'])[:8]:<9} " f"{str(row['name'])[:20]:<21} " f"{format_cost(row['cost']):<10} " f"{format_count(row['tokens']):<8} " @@ -639,6 +941,7 @@ def normalize_log_rows(payload: dict[str, Any], filter_text: str = "") -> list[d needle = filter_text.strip().lower() rows = [] for item in log_items(payload): + node = str(item.get("_node") or item.get("instance") or item.get("node") or "-").strip() or "-" key_name = nested_name(item, "api_key", "api_key_id", "api_key_name") account_name = nested_name(item, "account", "account_id", "account_name") user_name = nested_name(item, "user", "user_id", "user_name") @@ -648,6 +951,7 @@ def normalize_log_rows(payload: dict[str, Any], filter_text: str = "") -> list[d type_label = log_type_label(item) row = { "id": as_int(item.get("id")), + "node": node, "key": key_name, "account": account_name, "user": user_name, @@ -674,12 +978,12 @@ def normalize_log_rows(payload: dict[str, Any], filter_text: str = "") -> list[d } if needle: haystack = " ".join( - (key_name, account_name, user_name, model, upstream_model, effort, type_label, row["request_id"]) + (node, key_name, account_name, user_name, model, upstream_model, effort, type_label, row["request_id"]) ).lower() if needle not in haystack: continue rows.append(row) - rows.sort(key=lambda row: (row["created_at"], row["id"]), reverse=True) + rows.sort(key=lambda row: (row["created_at"], row["id"], row["node"]), reverse=True) return rows @@ -691,7 +995,7 @@ def logs_summary_line(payload: dict[str, Any], shown: int) -> str: cost = sum(as_float(item.get("total_cost")) for item in log_items(payload)) tokens = sum(log_total_tokens(item) for item in log_items(payload)) return ( - f"{fetched} logs | showing {shown}/{len(log_items(payload))} of latest {page_size or '-'} | " + f"{fetched} requests | showing {shown}/{len(log_items(payload))} of latest {page_size or '-'} | " f"total {format_count(total)} records | page cost {format_cost(cost)} | {format_count(tokens)} tokens" ) @@ -703,7 +1007,7 @@ def log_detail_line(row: dict[str, Any]) -> str: if row["effort"] != "-": model = f"{model} ({row['effort']})" detail = ( - f"{row['time']} | key {row['key']} | account {row['account']} | user {row['user']} | {model} | {row['type']} | " + f"{row['time']} | {row['node']} | key {row['key']} | account {row['account']} | user {row['user']} | {model} | {row['type']} | " f"tokens in {format_count(row['input_tokens'])} out {format_count(row['output_tokens'])} " f"cache_w {format_count(row['cache_creation_tokens'])} cache_r {format_count(row['cache_read_tokens'])} | " f"cost {format_cost(row['cost'])} (actual {format_cost(row['actual_cost'])}) | " @@ -718,9 +1022,10 @@ def log_detail_line(row: dict[str, Any]) -> str: def print_logs_once(payload: dict[str, Any], filter_text: str = "") -> None: rows = normalize_log_rows(payload, filter_text) print(logs_summary_line(payload, len(rows))) - print("key account model effort type input output cache tokens cost first duration tok/s time age") + print("node key account model effort type input output cache tokens cost first duration tok/s time age") for row in rows: print( + f"{row['node'][:8]:<9} " f"{row['key'][:20]:<21} " f"{row['account'][:20]:<21} " f"{row['model'][:24]:<25} " @@ -873,13 +1178,13 @@ def normalize_error_rows(payload: dict[str, Any], filter_text: str = "") -> list user_email = f"#{as_int(item.get('user_id'))}" model = str(item.get("requested_model") or item.get("model") or "-").strip() or "-" upstream_model = str(item.get("upstream_model") or "").strip() - phase = str(item.get("phase") or "-").strip() or "-" - error_type = str(item.get("type") or "-").strip() or "-" + phase = str(item.get("phase") or item.get("error_source") or "-").strip() or "-" + error_type = str(item.get("type") or item.get("error_type") or "-").strip() or "-" owner = str(item.get("error_owner") or "-").strip() or "-" source = str(item.get("error_source") or "-").strip() or "-" platform = str(item.get("platform") or "-").strip() or "-" status_code = as_int(item.get("status_code")) - message = str(item.get("message") or "").strip() + message = str(item.get("message") or item.get("error_detail") or "").strip() request_id = str(item.get("request_id") or "").strip() client_request_id = str(item.get("client_request_id") or "").strip() group_name = str(item.get("group_name") or "").strip() or "-" @@ -887,6 +1192,9 @@ def normalize_error_rows(payload: dict[str, Any], filter_text: str = "") -> list "id": as_int(item.get("id")), "node": node, "status_code": status_code, + "count": max(1, as_int(item.get("error_count")) or 1), + "inbound_status_code": as_int(item.get("inbound_status_code")), + "upstream_status_code": as_int(item.get("upstream_status_code")), "key": key_name, "account": account_name, "user": user_email or "-", @@ -957,8 +1265,14 @@ def error_detail_line(row: dict[str, Any]) -> str: message = row["message"] or "-" if len(message) > 180: message = message[:177] + "..." + status_path = [ + f"final {row['status_code']}" if row.get("status_code") else "", + f"in {row['inbound_status_code']}" if row.get("inbound_status_code") else "", + f"up {row['upstream_status_code']}" if row.get("upstream_status_code") else "", + ] + status_path_text = "/".join(value for value in status_path if value) or "-" detail = ( - f"{row['time']} | {row['node']} | status {row['status_code']} | key {row['key']} | account {row['account']} | " + f"{row['time']} | {row['node']} | count {row.get('count', 1)} | {status_path_text} | key {row['key']} | account {row['account']} | " f"user {row['user']} | {model} | {row['platform']} | phase {row['phase']} | type {row['type']} | " f"owner {row['owner']} | source {row['source']} | group {row['group']} | {message}" ) @@ -981,10 +1295,11 @@ def print_errors_once(payload: dict[str, Any], filter_text: str = "") -> None: print(f"{node}: ok fetched {as_int(info.get('fetched'))} total {as_int(info.get('total'))} | {info.get('url') or '-'}") else: print(f"{node}: error {info.get('error') or 'unknown'} | {info.get('url') or '-'}") - print("node status key account model phase type owner time age") + print("node count status key account model phase type owner time age") for row in rows: print( f"{row['node']:<5} " + f"{format_count(row.get('count', 1)):<6} " f"{row['status_code']:<7} " f"{row['key'][:20]:<21} " f"{row['account'][:20]:<21} " @@ -1316,7 +1631,7 @@ def pricing_summary(payload: dict[str, Any], rows: list[dict[str, Any]] | None = ] total_cny = sum(cny_values) if cny_values else None total_label = f" | CNY {format_cny(total_cny)}" if total_cny is not None else "" - return f"upstreams {healthy}/{len(source_rows)} healthy{total_label}" + return f"sources {healthy}/{len(source_rows)} healthy{total_label}" def pricing_detail_line(row: dict[str, Any]) -> str: @@ -1333,13 +1648,13 @@ def pricing_detail_line(row: dict[str, Any]) -> str: def account_pricing_detail(row: dict[str, Any]) -> str: source = str(row.get("pricing_source") or "-") if source == "-": - return "upstream -" + return "source -" cny = format_cny(row.get("pricing_cny")) raw_balance = format_amount(row.get("pricing_balance")) unit = str(row.get("pricing_unit") or "-") state = str(row.get("pricing_state") or "-") updated = str(row.get("pricing_updated") or "-") - return f"upstream {source} | CNY {cny} | raw {raw_balance} {unit} | {state} | updated {updated}" + return f"source {source} | CNY {cny} | raw {raw_balance} {unit} | {state} | updated {updated}" def print_pricing_once(payload: dict[str, Any], filter_text: str = "") -> None: @@ -1384,7 +1699,7 @@ def print_once( if status_payload or status_error: print(monitor_summary(status_payload or {}, status_error)) if pricing_error: - print(f"upstreams: {pricing_error}") + print(f"sources: {pricing_error}") elif pricing_payload: print(pricing_summary(pricing_payload)) print("name provider group source src cny src state daily today tokens req kind 5h 7d reset status availability") @@ -1426,6 +1741,8 @@ def run_textual( errors_time_range: str, version_message: str = "", start_page: str = "dashboard", + workspace_url: str = "", + legacy_direct: bool = False, ) -> int: try: from rich.text import Text @@ -1436,15 +1753,59 @@ def run_textual( print("Textual is required. Run with: uv run --with textual python sub2api_quota_tui.py", file=sys.stderr) return 2 + workspace_cache = WorkspaceCache( + workspace_url or DEFAULT_WORKSPACE_URL, + timeout, + min(refresh_seconds, logs_refresh_seconds, errors_refresh_seconds), + ) + + def load_accounts(*, force: bool = False) -> dict[str, Any]: + if legacy_direct: + return fetch_payload(api_url, timeout, refresh=force) + return workspace_accounts_payload(workspace_cache.get(force=force)) + + def load_status() -> tuple[dict[str, Any], str]: + if legacy_direct: + return fetch_optional_payload(status_url, timeout) + payload = workspace_cache.get() + return workspace_status_payload(payload), workspace_cache.error + + def load_pricing() -> tuple[dict[str, Any], str]: + if legacy_direct: + return fetch_optional_pricing_payload(pricing_url, timeout) + payload = workspace_cache.get() + return workspace_pricing_payload(payload), workspace_cache.error + + def load_logs(*, force: bool = False) -> dict[str, Any]: + if legacy_direct: + return fetch_logs_payload(logs_url, logs_token, timeout, logs_limit) + return workspace_logs_payload(workspace_cache.get(force=force)) + + def load_keys() -> dict[str, Any]: + if legacy_direct: + return fetch_key_usage_payload(logs_url, logs_token, timeout) + return workspace_keys_payload(workspace_cache.get()) + + def load_errors(*, force: bool = False) -> dict[str, Any]: + if legacy_direct: + return fetch_merged_errors( + default_error_sources(errors_cn_url, errors_us_url), + logs_token, + timeout, + errors_limit, + errors_time_range, + ) + return workspace_errors_payload(workspace_cache.get(force=force)) + class DashboardScreen(Screen[None]): AUTO_FOCUS = "#accounts" BINDINGS = [ ("r", "refresh", "Refresh"), - ("/", "focus_filter", "Filter"), + ("/", "focus_filter", "Global filter"), ("a", "focus_accounts", "Accounts"), - ("p", "show_pricing", "Upstreams"), - ("k", "focus_keys", "Keys"), - ("l", "focus_logs", "Logs"), + ("p", "show_pricing", "Sources"), + ("k", "focus_keys", "Key usage"), + ("l", "focus_logs", "Requests"), ("e", "focus_errors", "Errors"), ] @@ -1475,7 +1836,7 @@ def run_textual( def compose(self) -> ComposeResult: yield Header(show_clock=True) yield Static("", id="summary") - yield Input(placeholder="filter", id="filter") + yield Input(placeholder="filter all tables", id="filter") yield DataTable(id="accounts") yield DataTable(id="keys") yield DataTable(id="logs") @@ -1494,18 +1855,20 @@ def run_textual( table.add_column(label) def on_mount(self) -> None: + self.app.sub_title = "Dashboard" self.configure_table( self.query_one("#accounts", DataTable), ("ACCOUNT", "Group", "Source", "Src CNY", "Src state", "Today", "Daily", "5h", "7d", "Avail"), ) self.configure_table( self.query_one("#keys", DataTable), - ("KEY", "Today", "Tokens", "Req"), + ("Node", "KEY", "Today" if legacy_direct else "24h", "Tokens", "Req"), ) self.configure_table( self.query_one("#logs", DataTable), ( - "LOG KEY", + "Node", + "REQUEST KEY", "Account", "Model", "First", @@ -1522,9 +1885,9 @@ def run_textual( ) self.configure_table( self.query_one("#errors", DataTable), - ("ERR", "Status", "Key", "Account", "Model", "Time", "Age"), + ("Node", "Count", "Status", "Key", "Account", "Model", "Time", "Age"), ) - self.refresh_all(refresh=True) + self.refresh_all(refresh=legacy_direct) self.set_interval(refresh_seconds, self.refresh_accounts) self.set_interval(refresh_seconds, self.refresh_pricing) self.set_interval(logs_refresh_seconds, self.refresh_logs) @@ -1568,7 +1931,8 @@ def run_textual( self.focus_table(self.active_table) def refresh_all(self, refresh: bool = False) -> None: - self.query_one("#status", Static).update("refreshing accounts, upstream balances, keys, logs, and errors...") + mode = "direct sources" if legacy_direct else "aggregated workspace" + self.query_one("#status", Static).update(f"refreshing {mode}...") self.refresh_accounts(refresh=refresh) self.refresh_pricing() self.refresh_logs() @@ -1577,28 +1941,31 @@ def run_textual( 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) + self.accounts_payload = load_accounts(force=refresh) + self.status_payload, self.monitor_error = load_status() except Exception as exc: self.accounts_error = str(exc) self.render_accounts() self.render_meta() def refresh_pricing(self) -> None: - self.pricing_payload, self.pricing_error = fetch_optional_pricing_payload( - pricing_url, timeout - ) + self.pricing_error = "" + try: + self.pricing_payload, self.pricing_error = load_pricing() + except Exception as exc: + self.pricing_payload = {} + self.pricing_error = str(exc) self.render_accounts() self.render_meta() def refresh_logs(self) -> None: self.logs_error = "" - if not str(logs_token or "").strip(): + if legacy_direct and 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) + self.logs_payload = load_logs() except Exception as exc: self.logs_error = str(exc) self.render_logs() @@ -1607,12 +1974,12 @@ def run_textual( def refresh_keys(self) -> None: self.keys_error = "" - if not str(logs_token or "").strip(): + if legacy_direct and not str(logs_token or "").strip(): self.keys_payload = {} self.keys_error = "admin token not configured" else: try: - self.keys_payload = fetch_key_usage_payload(logs_url, logs_token, timeout) + self.keys_payload = load_keys() except Exception as exc: self.keys_payload = {} self.keys_error = str(exc) @@ -1620,18 +1987,12 @@ def run_textual( def refresh_errors(self) -> None: self.errors_error = "" - if not str(logs_token or "").strip(): + if legacy_direct and 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, - ) + self.errors_payload = load_errors() except Exception as exc: self.errors_error = str(exc) self.render_errors() @@ -1681,6 +2042,7 @@ def run_textual( self.key_by_key[key] = row color = key_color(row["name"]) table.add_row( + row["node"], Text(str(row["name"]), style=color) if color else str(row["name"]), format_cost(row["cost"]), format_count(row["tokens"]), @@ -1700,6 +2062,7 @@ def run_textual( self.log_by_key[key] = row color = key_color(row["key"]) table.add_row( + row["node"], Text(row["key"], style=color) if color else row["key"], row["account"], row["model"], @@ -1727,6 +2090,7 @@ def run_textual( self.error_by_key[key] = row table.add_row( row["node"], + format_count(row.get("count", 1)), str(row["status_code"]), row["key"], row["account"], @@ -1763,18 +2127,23 @@ def run_textual( ) summary.append(" | ", style="dim") if self.pricing_error: - summary.append("upstreams unavailable", style="bright_blue") + summary.append("sources unavailable", style="bright_blue") else: summary.append(pricing_summary(self.pricing_payload), style="bright_blue") summary.append(" | ", style="dim") summary.append(f"keys {len(self.key_rows)} ({key_summary.removeprefix('keys ')})", style="yellow") summary.append(" | ", style="dim") - summary.append(f"logs {len(self.log_rows)}/{logs_total}", style="magenta") + summary.append(f"requests {len(self.log_rows)}/{logs_total}", style="magenta") summary.append(" | ", style="dim") summary.append(f"errors {len(self.error_rows)}/{self.errors_total()}", style="red") self.query_one("#summary", Static).update(summary) status_bits = [version_message] + if not legacy_direct: + try: + status_bits.append(workspace_state_summary(workspace_cache.get(), workspace_cache.error)) + except Exception: + status_bits.append("workspace unavailable") if self.accounts_error: status_bits.append(f"accounts: {self.accounts_error}") else: @@ -1785,15 +2154,16 @@ def run_textual( ) ) if self.pricing_error: - status_bits.append(f"upstreams: {self.pricing_error}") + status_bits.append(f"sources: {self.pricing_error}") elif self.pricing_payload: - status_bits.append(pricing_summary(self.pricing_payload) + f" | {pricing_url}") + endpoint = pricing_url if legacy_direct else workspace_cache.url + status_bits.append(pricing_summary(self.pricing_payload) + f" | {endpoint}") 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}") + status_bits.append(f"requests/errors: {self.logs_error}") else: if self.logs_error: - status_bits.append(f"logs: {self.logs_error}") + status_bits.append(f"requests: {self.logs_error}") if self.errors_error: status_bits.append(f"errors: {self.errors_error}") if self.keys_error and self.keys_error != self.logs_error: @@ -1801,7 +2171,7 @@ def run_textual( 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"): + for node in sorted(sources): 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'}") @@ -1839,7 +2209,7 @@ def run_textual( detail += f" | {error}" elif table_id == "keys": detail = ( - f"{row['name']} | today {format_cost(row['cost'])} | " + f"{row['node']} | {row['name']} | {'today' if legacy_direct else '24h'} {format_cost(row['cost'])} | " f"{format_count(row['tokens'])} tokens | {format_count(row['requests'])} req" ) elif table_id == "logs": @@ -1871,8 +2241,9 @@ def run_textual( BINDINGS = [ ("r", "refresh", "Refresh"), ("/", "focus_filter", "Filter"), - ("p", "show_pricing", "Upstreams"), - ("l", "show_logs", "Logs"), + ("d", "show_dashboard", "Dashboard"), + ("p", "show_pricing", "Sources"), + ("l", "show_logs", "Requests"), ("e", "show_errors", "Errors"), ] @@ -1899,6 +2270,7 @@ def run_textual( yield Footer() def on_mount(self) -> None: + self.app.sub_title = "Accounts" table = self.query_one("#accounts", DataTable) table.cursor_type = "row" table.zebra_stripes = True @@ -1906,8 +2278,8 @@ def run_textual( 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) + keys_table.add_columns("Node", "Key", "Today" if legacy_direct else "24h", "Tokens", "Req") + self.refresh_data(refresh=legacy_direct) self.set_interval(refresh_seconds, self.refresh_data) def action_refresh(self) -> None: @@ -1916,6 +2288,9 @@ def run_textual( def action_focus_filter(self) -> None: self.query_one("#filter", Input).focus() + def action_show_dashboard(self) -> None: + self.app.switch_screen(DashboardScreen()) + def action_show_pricing(self) -> None: self.app.switch_screen(PricingScreen()) @@ -1937,33 +2312,34 @@ def run_textual( status = self.query_one("#status", Static) status.update("refreshing...") try: - self.payload = fetch_payload(api_url, timeout, refresh=refresh) - self.status_payload, self.status_error = fetch_optional_payload(status_url, timeout) - self.pricing_payload, self.pricing_error = fetch_optional_pricing_payload( - pricing_url, timeout - ) + self.payload = load_accounts(force=refresh) + self.status_payload, self.status_error = load_status() + self.pricing_payload, self.pricing_error = load_pricing() 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 not legacy_direct: + status_bits.insert(1, workspace_state_summary(workspace_cache.get(), workspace_cache.error)) if self.pricing_error: - status_bits.append(f"upstreams: {self.pricing_error}") + status_bits.append(f"sources: {self.pricing_error}") else: status_bits.append(pricing_summary(self.pricing_payload)) 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}") + endpoint = api_url if legacy_direct else workspace_cache.url + status.update(" | ".join(bit for bit in status_bits if bit) + f" | {endpoint}") 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(): + if legacy_direct and 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) + self.keys_payload = load_keys() except Exception as exc: self.keys_error = str(exc) @@ -1975,6 +2351,7 @@ def run_textual( for row in rows: color = key_color(row["name"]) table.add_row( + row["node"], Text(str(row["name"]), style=color) if color else str(row["name"]), format_cost(row["cost"]), format_count(row["tokens"]), @@ -2044,8 +2421,9 @@ def run_textual( BINDINGS = [ ("r", "refresh", "Refresh"), ("/", "focus_filter", "Filter"), + ("d", "show_dashboard", "Dashboard"), ("a", "show_accounts", "Accounts"), - ("l", "show_logs", "Logs"), + ("l", "show_logs", "Requests"), ("e", "show_errors", "Errors"), ] @@ -2058,13 +2436,14 @@ def run_textual( def compose(self) -> ComposeResult: yield Header(show_clock=True) yield Static("", id="summary") - yield Input(placeholder="filter upstream sources", id="filter") + yield Input(placeholder="filter sources", id="filter") yield DataTable(id="pricing") yield Static("", id="detail") yield Static("", id="status") yield Footer() def on_mount(self) -> None: + self.app.sub_title = "Sources" table = self.query_one("#pricing", DataTable) table.cursor_type = "row" table.zebra_stripes = True @@ -2073,11 +2452,14 @@ def run_textual( self.set_interval(refresh_seconds, self.refresh_data) def action_refresh(self) -> None: - self.refresh_data() + self.refresh_data(force=True) def action_focus_filter(self) -> None: self.query_one("#filter", Input).focus() + def action_show_dashboard(self) -> None: + self.app.switch_screen(DashboardScreen()) + def action_show_accounts(self) -> None: self.app.switch_screen(AccountsScreen()) @@ -2095,22 +2477,28 @@ def run_textual( if event.input.id == "filter": self.query_one("#pricing", DataTable).focus() - def refresh_data(self) -> None: + def refresh_data(self, force: bool = False) -> None: status = self.query_one("#status", Static) - status.update("refreshing upstream balances from Pricing Monitor...") + status.update("refreshing source balances from workspace...") try: - self.payload = fetch_pricing_payload(pricing_url, timeout) + if legacy_direct: + self.payload = fetch_pricing_payload(pricing_url, timeout) + else: + self.payload = workspace_pricing_payload(workspace_cache.get(force=force)) self.render_payload() - status_bits = [version_message, pricing_summary(self.payload), pricing_url] + endpoint = pricing_url if legacy_direct else workspace_cache.url + status_bits = [version_message, pricing_summary(self.payload), endpoint] + if not legacy_direct: + status_bits.insert(1, workspace_state_summary(workspace_cache.get(), workspace_cache.error)) status.update(" | ".join(bit for bit in status_bits if bit)) except Exception: self.payload = {} self.rows = [] self.row_by_key = {} self.query_one("#pricing", DataTable).clear() - self.query_one("#summary", Static).update("upstreams unavailable") - self.query_one("#detail", Static).update("no upstream balances") - status.update("upstreams unavailable") + self.query_one("#summary", Static).update("sources unavailable") + self.query_one("#detail", Static).update("no source balances") + status.update("sources unavailable") def render_payload(self) -> None: filter_text = self.query_one("#filter", Input).value @@ -2148,8 +2536,9 @@ def run_textual( BINDINGS = [ ("r", "refresh", "Refresh"), ("/", "focus_filter", "Filter"), + ("d", "show_dashboard", "Dashboard"), ("a", "show_accounts", "Accounts"), - ("p", "show_pricing", "Upstreams"), + ("p", "show_pricing", "Sources"), ("e", "show_errors", "Errors"), ] @@ -2169,10 +2558,12 @@ def run_textual( yield Footer() def on_mount(self) -> None: + self.app.sub_title = "Requests" table = self.query_one("#logs", DataTable) table.cursor_type = "row" table.zebra_stripes = True table.add_columns( + "Node", "Key", "Account", "Model", @@ -2193,11 +2584,14 @@ def run_textual( self.set_interval(logs_refresh_seconds, self.refresh_data) def action_refresh(self) -> None: - self.refresh_data() + self.refresh_data(force=True) def action_focus_filter(self) -> None: self.query_one("#filter", Input).focus() + def action_show_dashboard(self) -> None: + self.app.switch_screen(DashboardScreen()) + def action_show_accounts(self) -> None: self.app.switch_screen(AccountsScreen()) @@ -2215,19 +2609,22 @@ def run_textual( if event.input.id == "filter": self.query_one("#logs", DataTable).focus() - def refresh_data(self) -> None: + def refresh_data(self, force: bool = False) -> None: status = self.query_one("#status", Static) - if not str(logs_token or "").strip(): + if legacy_direct and not str(logs_token or "").strip(): status.update(logs_token_hint()) return - status.update("refreshing logs...") + status.update("refreshing requests...") try: - self.payload = fetch_logs_payload(logs_url, logs_token, timeout, logs_limit) + self.payload = load_logs(force=force) self.render_payload() - status_bits = [version_message, f"latest {logs_limit} requests | {logs_url}"] + endpoint = logs_url if legacy_direct else workspace_cache.url + status_bits = [version_message, f"latest {logs_limit} requests | {endpoint}"] + if not legacy_direct: + status_bits.insert(1, workspace_state_summary(workspace_cache.get(), workspace_cache.error)) status.update(" | ".join(bit for bit in status_bits if bit)) except Exception as exc: - status.update(f"logs error: {exc}") + status.update(f"requests error: {exc}") def render_payload(self) -> None: filter_text = self.query_one("#filter", Input).value @@ -2240,6 +2637,7 @@ def run_textual( self.row_by_key[key] = row color = key_color(row["key"]) table.add_row( + row["node"], Text(row["key"], style=color) if color else row["key"], row["account"], row["model"], @@ -2274,9 +2672,10 @@ def run_textual( BINDINGS = [ ("r", "refresh", "Refresh"), ("/", "focus_filter", "Filter"), + ("d", "show_dashboard", "Dashboard"), ("a", "show_accounts", "Accounts"), - ("p", "show_pricing", "Upstreams"), - ("l", "show_logs", "Logs"), + ("p", "show_pricing", "Sources"), + ("l", "show_logs", "Requests"), ] def __init__(self) -> None: @@ -2295,19 +2694,23 @@ def run_textual( yield Footer() def on_mount(self) -> None: + self.app.sub_title = "Errors" table = self.query_one("#errors", DataTable) table.cursor_type = "row" table.zebra_stripes = True - table.add_columns("Node", "Status", "Key", "Account", "Model", "Phase", "Type", "Owner", "Time", "Age") + table.add_columns("Node", "Count", "Status", "Key", "Account", "Model", "Phase", "Type", "Owner", "Time", "Age") self.refresh_data() self.set_interval(errors_refresh_seconds, self.refresh_data) def action_refresh(self) -> None: - self.refresh_data() + self.refresh_data(force=True) def action_focus_filter(self) -> None: self.query_one("#filter", Input).focus() + def action_show_dashboard(self) -> None: + self.app.switch_screen(DashboardScreen()) + def action_show_accounts(self) -> None: self.app.switch_screen(AccountsScreen()) @@ -2325,24 +2728,18 @@ def run_textual( if event.input.id == "filter": self.query_one("#errors", DataTable).focus() - def refresh_data(self) -> None: + def refresh_data(self, force: bool = False) -> None: status = self.query_one("#status", Static) - if not str(logs_token or "").strip(): + if legacy_direct and not str(logs_token or "").strip(): status.update(logs_token_hint()) return - status.update("refreshing errors from cn + us...") + status.update("refreshing errors...") try: - self.payload = fetch_merged_errors( - default_error_sources(errors_cn_url, errors_us_url), - logs_token, - timeout, - errors_limit, - errors_time_range, - ) + self.payload = load_errors(force=force) self.render_payload() sources = self.payload.get("sources") if isinstance(self.payload.get("sources"), dict) else {} bits = [] - for node in ("cn", "us"): + for node in sorted(sources): info = sources.get(node) if isinstance(sources.get(node), dict) else {} if not info: continue @@ -2351,6 +2748,8 @@ def run_textual( else: bits.append(f"{node} {info.get('error') or 'error'}") status_bits = [version_message, f"every {errors_refresh_seconds}s", " | ".join(bits)] + if not legacy_direct: + status_bits.insert(1, workspace_state_summary(workspace_cache.get(), workspace_cache.error)) status.update(" | ".join(bit for bit in status_bits if bit)) except Exception as exc: status.update(f"errors error: {exc}") @@ -2366,6 +2765,7 @@ def run_textual( self.row_by_key[key] = row table.add_row( row["node"], + format_count(row.get("count", 1)), str(row["status_code"]), row["key"], row["account"], @@ -2390,6 +2790,8 @@ def run_textual( self.query_one("#detail", Static).update(error_detail_line(row)) class Sub2APIQuotaApp(App[None]): + TITLE = "shusub2" + SUB_TITLE = "Dashboard" CSS = """ #summary { height: 1; padding: 0 1; color: $foreground; } #filter { height: 1; border: none; padding: 0 1; } @@ -2438,19 +2840,21 @@ def logs_token_hint() -> str: def build_parser() -> argparse.ArgumentParser: - parser = argparse.ArgumentParser(description="Sub2API quota and daily usage TUI") + parser = argparse.ArgumentParser(description="Sub2API aggregated operations TUI") + parser.add_argument("--workspace-url", default=default_workspace_url(), help="Pricing Monitor view=workspace aggregate URL") + parser.add_argument("--legacy-direct", action="store_true", help="use the former multi-endpoint Accounts/Status/Admin topology for diagnostics") parser.add_argument("--api-url", default=default_api_url()) parser.add_argument("--status-url", default=default_status_url(), help="optional sub2api-status /api/status URL for channel monitor health") parser.add_argument("--pricing-url", default=default_pricing_url(), help="Pricing Monitor /api/ui-data?view=accounts URL for upstream source balances") parser.add_argument("--logs-url", default=default_logs_url(), help="Sub2API admin usage logs URL (default: sub2apicn /api/v1/admin/usage)") - parser.add_argument("--logs-token", default=default_logs_token(), help="Sub2API admin API key for logs/errors pages (prefer SHUSUB2_LOGS_TOKEN or ~/.config/shusub2/logs-token)") + parser.add_argument("--logs-token", default=None, help="Sub2API admin API key used only with --legacy-direct") parser.add_argument("--errors-cn-url", default=default_errors_cn_url(), help="cn Sub2API admin ops errors URL (default: sub2apicn /api/v1/admin/ops/errors)") parser.add_argument( "--errors-us-url", default=default_errors_us_url(), help="us Sub2API admin ops errors URL (default: DNS-only us.sub2 /api/v1/admin/ops/errors; CF sub2apius often blocks non-browser clients)", ) - parser.add_argument("--save-config", action="store_true", help="persist --api-url/--pricing-url/--logs-url/--logs-token/--errors-*-url to ~/.config/shusub2/ before running") + parser.add_argument("--save-config", action="store_true", help="persist workspace and legacy endpoint options to ~/.config/shusub2/ before running") parser.add_argument("--install", action="store_true", help="persist config, install shusub2 as a uv tool, then exit") parser.add_argument("--version-check-url", default=os.environ.get("SHUSUB2_VERSION_CHECK_URL", DEFAULT_VERSION_CHECK_URL)) parser.add_argument( @@ -2491,9 +2895,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("--accounts", action="store_true", help="start on the dedicated accounts page (with --once: print the accounts snapshot)") - parser.add_argument("--pricing", action="store_true", help="start on the Pricing Monitor upstream balances page (with --once: print source balances)") - parser.add_argument("--logs", action="store_true", help="start on the dedicated request logs page (with --once: print a logs snapshot)") + parser.add_argument("--accounts", action="store_true", help="start on the Accounts page (with --once: print Accounts and Key usage)") + parser.add_argument("--pricing", "--sources", dest="pricing", action="store_true", help="start on the Sources page (legacy alias: --pricing)") + parser.add_argument("--logs", "--requests", dest="logs", action="store_true", help="start on the Requests page (legacy alias: --logs)") 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 @@ -2501,15 +2905,25 @@ def build_parser() -> argparse.ArgumentParser: def main(argv: list[str] | None = None) -> int: args = build_parser().parse_args(argv) - status_url = str(args.status_url or "").strip() or inferred_status_url(args.api_url) + workspace_url = str(args.workspace_url or "").strip() + legacy_direct = bool(args.legacy_direct or not workspace_url) + logs_token = str(args.logs_token or "").strip() + if legacy_direct and args.logs_token is None: + logs_token = default_logs_token() + status_url = ( + str(args.status_url or "").strip() or inferred_status_url(args.api_url) + ) if legacy_direct else "" 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 selected_pages = sum(bool(value) for value in (args.accounts, args.pricing, args.logs, args.errors)) if selected_pages > 1: - print("choose only one of --accounts, --pricing, --logs, or --errors", file=sys.stderr) + print("choose only one of --accounts, --sources/--pricing, --requests/--logs, or --errors", file=sys.stderr) return 2 if args.save_config or args.install: + if workspace_url: + workspace_path = write_config_value(workspace_url_config_file_path(), workspace_url) + print(f"saved workspace url to {workspace_path}") config_path = write_api_url_config(args.api_url) print(f"saved api url to {config_path}") if str(args.pricing_url or "").strip(): @@ -2518,8 +2932,8 @@ def main(argv: list[str] | None = None) -> int: if str(args.logs_url or "").strip(): logs_url_path = write_config_value(logs_url_config_file_path(), args.logs_url) print(f"saved logs url to {logs_url_path}") - if str(args.logs_token or "").strip(): - logs_token_path = write_config_value(logs_token_config_file_path(), args.logs_token) + if logs_token: + logs_token_path = write_config_value(logs_token_config_file_path(), logs_token) print(f"saved logs token to {logs_token_path}") if str(args.errors_cn_url or "").strip(): errors_cn_path = write_config_value(errors_cn_url_config_file_path(), args.errors_cn_url) @@ -2538,29 +2952,61 @@ def main(argv: list[str] | None = None) -> int: if args.once: if version_message: print(version_message) + if not legacy_direct: + try: + workspace = fetch_workspace_payload(workspace_url, args.timeout) + except Exception: + print("workspace unavailable", file=sys.stderr) + return 1 + accounts_payload = workspace_accounts_payload(workspace) + status_payload = workspace_status_payload(workspace) + pricing_payload = workspace_pricing_payload(workspace) + logs_payload = workspace_logs_payload(workspace) + keys_payload = workspace_keys_payload(workspace) + errors_payload = workspace_errors_payload(workspace) + if args.pricing: + print_pricing_once(pricing_payload, args.filter) + return 0 + if args.logs: + print_logs_once(logs_payload, args.filter) + return 0 + if args.errors: + print_errors_once(errors_payload, args.filter) + return 0 + print_once( + accounts_payload, + args.filter, + status_payload, + "", + pricing_payload, + "", + ) + print() + print_keys_once(keys_payload) + return 0 if args.pricing: try: print_pricing_once( fetch_pricing_payload(args.pricing_url, args.timeout), args.filter ) except Exception: - print("upstreams unavailable", file=sys.stderr) + print("sources unavailable", file=sys.stderr) return 1 return 0 if args.logs: - if not str(args.logs_token or "").strip(): + if not logs_token: print(logs_token_hint(), file=sys.stderr) return 2 - print_logs_once(fetch_logs_payload(args.logs_url, args.logs_token, args.timeout, logs_limit), args.filter) + print_logs_once(fetch_logs_payload(args.logs_url, logs_token, args.timeout, logs_limit), args.filter) return 0 if args.errors: - if not str(args.logs_token or "").strip(): + if not logs_token: print(logs_token_hint(), file=sys.stderr) return 2 print_errors_once( fetch_merged_errors( default_error_sources(args.errors_cn_url, args.errors_us_url), - args.logs_token, + logs_token, args.timeout, errors_limit, errors_time_range, @@ -2578,10 +3024,10 @@ def main(argv: list[str] | None = None) -> int: pricing_payload, pricing_error, ) - if str(args.logs_token or "").strip(): + if logs_token: print() try: - print_keys_once(fetch_key_usage_payload(args.logs_url, args.logs_token, args.timeout)) + print_keys_once(fetch_key_usage_payload(args.logs_url, logs_token, args.timeout)) except Exception as exc: print(f"keys error: {exc}", file=sys.stderr) return 0 @@ -2599,7 +3045,7 @@ def main(argv: list[str] | None = None) -> int: status_url, args.pricing_url, args.logs_url, - args.logs_token, + logs_token, args.errors_cn_url, args.errors_us_url, max(1, args.refresh_seconds), @@ -2611,6 +3057,8 @@ def main(argv: list[str] | None = None) -> int: errors_time_range, version_message, start_page=start_page, + workspace_url=workspace_url, + legacy_direct=legacy_direct, ) diff --git a/tests/test_payload.py b/tests/test_payload.py index 7136a29..9661ee8 100644 --- a/tests/test_payload.py +++ b/tests/test_payload.py @@ -1,13 +1,17 @@ from __future__ import annotations import contextlib +import copy import io import os from pathlib import Path import importlib.util import sys import tempfile +import threading +import time import unittest +from unittest import mock def load_module(): @@ -20,6 +24,117 @@ def load_module(): return module +def workspace_payload_fixture() -> dict[str, object]: + return { + "service": "sub2api-pricing-monitor", + "view": "workspace", + "generated_at": "2026-08-03T12:00:00Z", + "state": {"ok": True, "partial": False}, + "components": { + "accounts": {"ok": True, "stale": False}, + "status": {"ok": True, "stale": False}, + "traffic": {"ok": True, "stale": False}, + }, + "sources": [ + { + "name": "code-plan", + "source_kind": "newapi", + "health_state": "healthy", + "balance_available": True, + "balance": {"available": 5000000, "available_cny": 10, "unit": "quota"}, + } + ], + "accounts": { + "source_name": "workspace", + "totals": { + "total_accounts": 1, + "usable_accounts": 1, + "today_cost_usd": 1.25, + "today_tokens": 300, + "today_requests": 2, + }, + "accounts": [ + { + "id": "9007199254740993", + "name": "oai-quota-code-plan", + "routing_group": "fast", + "provider": "openai", + "kind": "quota_limited", + "usable": True, + } + ], + }, + "status": { + "channel_monitors": { + "items": [ + { + "id": "9007199254740995", + "name": "oai-quota-code-plan", + "provider": "openai", + "latest_status": "success", + } + ] + } + }, + "traffic": { + "window_hours": 24, + "sample_limit": 1000, + "sample_limited": False, + "partial": False, + "instances": [ + {"instance": "server6", "ok": True, "error_total": 4}, + {"instance": "server4", "ok": True, "error_total": 0}, + ], + "requests": [ + { + "instance": "server6", + "id": "9007199254740997", + "created_at": "2026-08-03T11:59:00Z", + "api_key_id": "7", + "api_key_name": "wmy", + "account_id": "9", + "account_name": "oai-quota-code-plan", + "model": "gpt-5.5", + "input_tokens": 100, + "output_tokens": 50, + "cache_read_tokens": 25, + "actual_cost": 0.25, + "duration_ms": 1000, + } + ], + "errors": [ + { + "instance": "server6", + "latest_at": "2026-08-03T11:58:00Z", + "error_count": 4, + "status_code": 502, + "inbound_status_code": 500, + "upstream_status_code": 502, + "api_key_id": "7", + "api_key_name": "wmy", + "account_id": "9", + "account_name": "oai-quota-code-plan", + "model": "gpt-5.5", + "error_type": "api_error", + "error_source": "upstream_http", + } + ], + "keys": [ + { + "instance": "server6", + "api_key_id": "7", + "api_key_name": "wmy", + "status": "active", + "request_count": 2, + "token_count": 175, + "actual_cost": 0.25, + "latest_at": "2026-08-03T11:59:00Z", + } + ], + }, + } + + class Sub2APIQuotaTUITests(unittest.TestCase): def test_normalize_rows_sorts_by_usage_and_formats_windows(self) -> None: mod = load_module() @@ -190,7 +305,7 @@ class Sub2APIQuotaTUITests(unittest.TestCase): self.assertIsNone(unavailable_rows[0]["balance"]) self.assertIsNone(unavailable_rows[0]["balance_cny"]) self.assertEqual([row["name"] for row in mod.normalize_pricing_rows(payload, "keday")], ["kedaya"]) - self.assertEqual(mod.pricing_summary(payload, rows), "upstreams 1/2 healthy | CNY ¥10") + self.assertEqual(mod.pricing_summary(payload, rows), "sources 1/2 healthy | CNY ¥10") out = io.StringIO() with contextlib.redirect_stdout(out): mod.print_pricing_once(payload) @@ -314,8 +429,8 @@ class Sub2APIQuotaTUITests(unittest.TestCase): def __exit__(self, exc_type, exc, traceback): return False - def read(self): - return compressed + def read(self, size=-1): + return compressed if size < 0 else compressed[:size] old_urlopen = mod.urllib.request.urlopen mod.urllib.request.urlopen = lambda request, timeout: (requests.append(request) or FakeResponse()) @@ -330,13 +445,35 @@ class Sub2APIQuotaTUITests(unittest.TestCase): class PlainResponse: headers = {} - def read(self): - return mod.json.dumps(payload).encode("utf-8") + def read(self, size=-1): + raw = mod.json.dumps(payload).encode("utf-8") + return raw if size < 0 else raw[:size] self.assertEqual(mod.decode_json_response(PlainResponse(), "plain JSON failed"), payload) self.assertEqual(len(requests), 4) self.assertTrue(all(request.get_header("Accept-encoding") == "gzip" for request in requests)) + def test_json_decoder_bounds_plain_and_gzip_payloads(self) -> None: + mod = load_module() + + class Response: + def __init__(self, raw: bytes, encoding: str = "") -> None: + self.raw = raw + self.headers = {"Content-Encoding": encoding} if encoding else {} + + def read(self, size=-1): + return self.raw if size < 0 else self.raw[:size] + + oversized = mod.json.dumps({"value": "x" * 256}).encode("utf-8") + with self.assertRaisesRegex(RuntimeError, "bounded"): + mod.decode_json_response(Response(oversized), "bounded response", maximum_bytes=64) + with self.assertRaisesRegex(RuntimeError, "bounded"): + mod.decode_json_response( + Response(mod.gzip.compress(oversized), "gzip"), + "bounded response", + maximum_bytes=64, + ) + def test_default_refresh_intervals_are_five_minutes(self) -> None: mod = load_module() @@ -613,6 +750,181 @@ def sample_logs_payload() -> dict: } +class WorkspaceTests(unittest.TestCase): + def test_workspace_adapters_preserve_ids_and_map_compact_traffic(self) -> None: + mod = load_module() + payload = workspace_payload_fixture() + + accounts = mod.workspace_accounts_payload(payload) + status = mod.workspace_status_payload(payload) + pricing = mod.workspace_pricing_payload(payload) + logs = mod.workspace_logs_payload(payload) + keys = mod.workspace_keys_payload(payload) + errors = mod.workspace_errors_payload(payload) + + self.assertEqual(accounts["accounts"][0]["id"], "9007199254740993") + self.assertEqual(status["channel_monitors"]["items"][0]["id"], "9007199254740995") + self.assertEqual(pricing["sources"][0]["name"], "code-plan") + self.assertEqual(logs["data"]["items"][0]["id"], "9007199254740997") + self.assertEqual(logs["data"]["items"][0]["instance"], "server6") + normalized_accounts = mod.normalize_account_rows(accounts, pricing_payload=pricing) + normalized_logs = mod.normalize_log_rows(logs) + self.assertEqual(normalized_accounts[0]["id"], 9007199254740993) + self.assertEqual(normalized_logs[0]["id"], 9007199254740997) + self.assertEqual(normalized_logs[0]["cost"], 0.25) + self.assertEqual(mod.as_int("9007199254740999"), 9007199254740999) + self.assertEqual(mod.normalize_key_rows(keys)[0]["name"], "wmy") + self.assertEqual(mod.normalize_key_rows(keys)[0]["cost"], 0.25) + self.assertEqual(errors["items"][0]["error_count"], 4) + self.assertEqual(errors["items"][0]["phase"], "upstream_http") + self.assertEqual(errors["sources"]["server6"]["total"], 4) + self.assertEqual(errors["sources"]["server4"]["total"], 0) + self.assertEqual( + mod.workspace_state_summary(payload, "workspace unavailable"), + "workspace endpoint unavailable (using last good)", + ) + fallback_error = copy.deepcopy(errors) + fallback_error["items"][0].pop("api_key_name", None) + fallback_error["items"][0].pop("account_name", None) + fallback_error["items"][0]["api_key_id"] = "9007199254740999" + fallback_error["items"][0]["account_id"] = "9007199254740998" + fallback_row = mod.normalize_error_rows(fallback_error)[0] + self.assertEqual(fallback_row["key"], "#9007199254740999") + self.assertEqual(fallback_row["account"], "#9007199254740998") + normalized_errors = mod.normalize_error_rows(errors) + self.assertEqual(normalized_errors[0]["count"], 4) + self.assertIn("count 4", mod.error_detail_line(normalized_errors[0])) + + def test_workspace_payload_validation_rejects_nonfinite_and_oversized_rows(self) -> None: + mod = load_module() + nonfinite = workspace_payload_fixture() + nonfinite["traffic"]["requests"][0]["actual_cost"] = float("nan") + with mock.patch.object(mod, "fetch_payload", return_value=nonfinite): + with self.assertRaisesRegex(RuntimeError, "invalid workspace projection"): + mod.fetch_workspace_payload("https://workspace.example.test/data", 3) + + oversized = workspace_payload_fixture() + oversized["traffic"]["requests"] = [ + {"id": str(index)} for index in range(mod.MAX_WORKSPACE_REQUESTS + 1) + ] + with mock.patch.object(mod, "fetch_payload", return_value=oversized): + with self.assertRaisesRegex(RuntimeError, "invalid workspace requests"): + mod.fetch_workspace_payload("https://workspace.example.test/data", 3) + + def test_workspace_cache_coalesces_reads_and_returns_copies(self) -> None: + mod = load_module() + calls: list[str] = [] + call_lock = threading.Lock() + + def fake_request(url: str, timeout: int): + with call_lock: + calls.append(url) + time.sleep(0.03) + return workspace_payload_fixture() + + cache = mod.WorkspaceCache("https://workspace.example.test/api/ui-data?view=workspace", 3, 300) + results: list[dict[str, object]] = [] + threads = [ + threading.Thread(target=lambda: results.append(cache.get())) + for _ in range(12) + ] + with mock.patch.object(mod, "fetch_workspace_payload", side_effect=fake_request): + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=2) + self.assertEqual(len(calls), 1) + self.assertEqual(len(results), 12) + results[0]["view"] = "mutated" + self.assertEqual(cache.get()["view"], "workspace") + cache.get(force=True) + self.assertEqual(len(calls), 2) + + def test_workspace_cache_backs_off_failures_with_and_without_last_good(self) -> None: + mod = load_module() + cache = mod.WorkspaceCache("https://workspace.example.test/data", 3, 300) + with mock.patch.object( + mod, + "fetch_workspace_payload", + side_effect=[ + workspace_payload_fixture(), + RuntimeError("private failure"), + workspace_payload_fixture(), + ], + ) as fetch: + first = cache.get() + stale = cache.get(force=True) + repeated = cache.get() + self.assertEqual(fetch.call_count, 2) + cache.last_attempt_at -= cache.retry_seconds + 1 + recovered = cache.get() + self.assertEqual(first["view"], "workspace") + self.assertEqual(stale["view"], "workspace") + self.assertEqual(repeated["view"], "workspace") + self.assertEqual(recovered["view"], "workspace") + self.assertEqual(fetch.call_count, 3) + self.assertEqual(cache.network_fetches, 3) + self.assertEqual(cache.error, "") + + empty = mod.WorkspaceCache("https://workspace.example.test/data", 3, 300) + with mock.patch.object(mod, "fetch_workspace_payload", side_effect=RuntimeError("private failure")) as fetch: + with self.assertRaisesRegex(RuntimeError, "workspace unavailable"): + empty.get() + with self.assertRaisesRegex(RuntimeError, "workspace unavailable"): + empty.get() + fetch.assert_called_once() + + def test_default_main_uses_one_workspace_request_without_loading_admin_token(self) -> None: + mod = load_module() + out = io.StringIO() + with ( + mock.patch.object(mod, "fetch_workspace_payload", return_value=workspace_payload_fixture()) as workspace_fetch, + mock.patch.object(mod, "default_logs_token", side_effect=AssertionError("admin token must not be read")) as token_read, + mock.patch.object(mod, "fetch_payload", side_effect=AssertionError("legacy accounts must not be read")), + mock.patch.object(mod, "fetch_optional_payload", side_effect=AssertionError("legacy status must not be read")), + mock.patch.object(mod, "fetch_optional_pricing_payload", side_effect=AssertionError("legacy pricing must not be read")), + contextlib.redirect_stdout(out), + ): + rc = mod.main( + [ + "--once", + "--workspace-url", + "https://workspace.example.test/api/ui-data?view=workspace", + "--no-version-check", + ] + ) + + self.assertEqual(rc, 0) + workspace_fetch.assert_called_once_with( + "https://workspace.example.test/api/ui-data?view=workspace", 10 + ) + token_read.assert_not_called() + self.assertIn("oai-quota-code-plan", out.getvalue()) + self.assertIn("wmy", out.getvalue()) + + def test_empty_workspace_url_falls_back_to_legacy_direct_mode(self) -> None: + mod = load_module() + accounts = workspace_payload_fixture()["accounts"] + with ( + mock.patch.object(mod, "default_logs_token", return_value="") as token_read, + mock.patch.object(mod, "fetch_payload", return_value=accounts) as account_fetch, + mock.patch.object(mod, "fetch_optional_payload", return_value=({}, "")), + mock.patch.object(mod, "fetch_optional_pricing_payload", return_value=({"sources": []}, "")), + mock.patch.object(mod, "fetch_workspace_payload", side_effect=AssertionError("workspace must not be read")), + contextlib.redirect_stdout(io.StringIO()), + ): + rc = mod.main(["--once", "--workspace-url", "", "--no-version-check"]) + + self.assertEqual(rc, 0) + token_read.assert_called_once_with() + account_fetch.assert_called_once() + + def test_sources_and_requests_flags_are_aliases(self) -> None: + mod = load_module() + self.assertTrue(mod.build_parser().parse_args(["--sources"]).pricing) + self.assertTrue(mod.build_parser().parse_args(["--requests"]).logs) + + class Sub2APILogsTests(unittest.TestCase): def test_normalize_log_rows_maps_columns_and_sorts_newest_first(self) -> None: mod = load_module() @@ -737,7 +1049,7 @@ class Sub2APILogsTests(unittest.TestCase): err = io.StringIO() try: with contextlib.redirect_stderr(err): - rc = mod.main(["--once", "--logs", "--no-version-check"]) + rc = mod.main(["--once", "--logs", "--legacy-direct", "--no-version-check"]) finally: if old_token is not None: os.environ["SHUSUB2_LOGS_TOKEN"] = old_token @@ -915,7 +1227,7 @@ class Sub2APILogsTests(unittest.TestCase): err = io.StringIO() try: with contextlib.redirect_stderr(err): - rc = mod.main(["--once", "--errors", "--no-version-check"]) + rc = mod.main(["--once", "--errors", "--legacy-direct", "--no-version-check"]) finally: if old_token is not None: os.environ["SHUSUB2_LOGS_TOKEN"] = old_token @@ -1022,6 +1334,7 @@ class DashboardLayoutTests(unittest.IsolatedAsyncioTestCase): 100, 100, "24h", + legacy_direct=True, ) finally: App.run = original_run @@ -1061,6 +1374,102 @@ class DashboardLayoutTests(unittest.IsolatedAsyncioTestCase): self.assertEqual(app.screen.rows[0]["name"], "code-plan") + async def test_workspace_dashboard_handles_initial_endpoint_failure(self) -> None: + from textual.app import App + + mod = load_module() + captured: dict[str, object] = {} + original_run = App.run + App.run = lambda self, *args, **kwargs: captured.setdefault("app", self) + try: + rc = mod.run_textual( + "", + "", + "", + "", + "", + "", + "", + 999, + 999, + 999, + 1, + 100, + 100, + "24h", + workspace_url="https://workspace.example.test/api/ui-data?view=workspace", + legacy_direct=False, + ) + finally: + App.run = original_run + + self.assertEqual(rc, 0) + app = captured["app"] + with mock.patch.object( + mod, + "fetch_workspace_payload", + side_effect=RuntimeError("private endpoint failure"), + ) as fetch: + async with app.run_test(size=(100, 30)) as pilot: + await pilot.pause() + await pilot.pause() + self.assertEqual(type(app.screen).__name__, "DashboardScreen") + self.assertIn("workspace unavailable", str(app.screen.query_one("#status").render())) + self.assertEqual(fetch.call_count, 1) + + async def test_workspace_dashboard_uses_one_snapshot_and_has_return_navigation(self) -> None: + from textual.app import App + + mod = load_module() + captured: dict[str, object] = {} + original_run = App.run + App.run = lambda self, *args, **kwargs: captured.setdefault("app", self) + try: + rc = mod.run_textual( + "", + "", + "", + "", + "", + "", + "", + 999, + 999, + 999, + 1, + 100, + 100, + "24h", + workspace_url="https://workspace.example.test/api/ui-data?view=workspace", + legacy_direct=False, + ) + finally: + App.run = original_run + + self.assertEqual(rc, 0) + app = captured["app"] + with mock.patch.object(mod, "fetch_workspace_payload", return_value=workspace_payload_fixture()) as fetch: + async with app.run_test(size=(100, 30)) as pilot: + await pilot.pause() + await pilot.pause() + screen = app.screen + self.assertEqual(type(screen).__name__, "DashboardScreen") + self.assertEqual(app.sub_title, "Dashboard") + self.assertEqual(screen.account_rows[0]["name"], "oai-quota-code-plan") + self.assertEqual(screen.error_rows[0]["count"], 4) + self.assertEqual(len(fetch.call_args_list), 1) + + await pilot.press("p") + await pilot.pause() + self.assertEqual(type(app.screen).__name__, "PricingScreen") + self.assertEqual(app.sub_title, "Sources") + await pilot.press("d") + await pilot.pause() + self.assertEqual(type(app.screen).__name__, "DashboardScreen") + self.assertEqual(app.sub_title, "Dashboard") + self.assertEqual(len(fetch.call_args_list), 1) + + class PageSelectionTests(unittest.TestCase): def test_once_pricing_prints_pricing_monitor_sources(self) -> None: mod = load_module() @@ -1080,7 +1489,7 @@ class PageSelectionTests(unittest.TestCase): out = io.StringIO() try: with contextlib.redirect_stdout(out): - rc = mod.main(["--once", "--pricing", "--no-version-check"]) + rc = mod.main(["--once", "--pricing", "--legacy-direct", "--no-version-check"]) finally: mod.fetch_pricing_payload = original_fetch @@ -1097,12 +1506,12 @@ class PageSelectionTests(unittest.TestCase): err = io.StringIO() try: with contextlib.redirect_stderr(err): - rc = mod.main(["--once", "--pricing", "--no-version-check"]) + rc = mod.main(["--once", "--pricing", "--legacy-direct", "--no-version-check"]) finally: mod.fetch_pricing_payload = original_fetch self.assertEqual(rc, 1) - self.assertEqual(err.getvalue().strip(), "upstreams unavailable") + self.assertEqual(err.getvalue().strip(), "sources unavailable") def test_once_accounts_includes_only_mapped_pricing_source_balance(self) -> None: mod = load_module() @@ -1136,6 +1545,7 @@ class PageSelectionTests(unittest.TestCase): with contextlib.redirect_stdout(out): rc = mod.main( [ + "--legacy-direct", "--once", "--api-url", "https://accounts.example.test/api/tui/accounts", diff --git a/uv.lock b/uv.lock index af63f3f..f0d1adf 100644 --- a/uv.lock +++ b/uv.lock @@ -85,7 +85,7 @@ wheels = [ [[package]] name = "shusub2" -version = "0.2.15" +version = "0.3.0" source = { editable = "." } dependencies = [ { name = "textual" },