feat: use calendar-day workspace traffic

This commit is contained in:
2026-08-04 01:36:40 +08:00
parent 311236d28f
commit d9a9fbaaad
5 changed files with 168 additions and 19 deletions
+7 -4
View File
@@ -71,7 +71,7 @@ Workspace 当前提供以下有界 projection
- `sources`Pricing Monitor 已有的脱敏 source 余额与健康状态。
- `traffic.requests`:最近的有界请求样本。
- `traffic.errors`:有界错误聚合,保留聚合次数和状态链路。
- `traffic.keys`24 小时 key 使用聚合
- `traffic.keys``Asia/Shanghai` 自然日聚合的 key 使用量,payload 同时携带统计日期和时区
投影允许展示 account/key 名称、稳定 ID、model、instance 和运维状态,但不包含 API
key 原文、access/refresh token、cookie、密码、数据库凭据或请求/响应正文。ID 始终按
@@ -119,8 +119,9 @@ provider、URL、display name 或模糊文本推断;source 为 error/stale 时
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 小时窗口,
不再宣称是客户端本地“今日”查询。
account、model、phase/type/owner 和时间。Key usage、Requests、Errors 和对应 account
当天用量都使用 workspace 服务器声明的 `Asia/Shanghai` 自然日,不依赖客户端本地日期或滚动
24 小时窗口;每个 Traffic projection 显式携带日期、时区和 `[00:00, 次日 00:00)` 边界。
## Legacy Direct
@@ -137,7 +138,9 @@ shusub2 --legacy-direct --requests
```
显式传入空 `--workspace-url ''` 也会进入 legacy direct 模式。legacy Requests、Errors
和 Key usage 需要 Sub2API admin API key;默认 workspace 模式不会读取该凭据文件。
和 Key usage 需要 Sub2API admin API key;默认 workspace 模式不会读取该凭据文件。legacy
Key usage 也固定按 `Asia/Shanghai` 当日查询;legacy Errors 的 `--errors-time-range` 保留为与
上游诊断 API 对账的独立兼容参数,不能代表 workspace 的当天 Traffic 口径。
legacy token 仅允许放在 `SHUSUB2_LOGS_TOKEN` 或权限为 `0600`
`~/.config/shusub2/logs-token`
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "shusub2"
version = "0.3.0"
version = "0.3.1"
description = "Aggregated operations TUI for Sub2API"
readme = "README.md"
requires-python = ">=3.11"
+99 -12
View File
@@ -22,10 +22,11 @@ import zlib
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
from typing import Any
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
APP_NAME = "shusub2"
FALLBACK_VERSION = "0.3.0"
FALLBACK_VERSION = "0.3.1"
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"
@@ -59,6 +60,7 @@ MAX_WORKSPACE_INSTANCES = 16
MAX_WORKSPACE_REQUESTS = 100
MAX_WORKSPACE_ERRORS = 256
MAX_WORKSPACE_KEYS = 256
USAGE_TIMEZONE = "Asia/Shanghai"
MONITOR_OK_STATUSES = {"operational", "ok", "success"}
MONITOR_FAILED_STATUSES = {"error", "failed", "failure"}
MONITOR_STOPWORDS = {"response", "responses", "monitor"}
@@ -522,6 +524,61 @@ def workspace_values_are_finite(value: Any) -> bool:
return True
def workspace_traffic_period(traffic: Any) -> dict[str, str] | None:
"""Validate the server-owned local calendar-day Traffic contract."""
if not isinstance(traffic, dict) or traffic.get("period_kind") != "calendar_day":
return None
date_text = str(traffic.get("date") or "").strip()
timezone_text = str(traffic.get("timezone") or "").strip()
if timezone_text != USAGE_TIMEZONE:
return None
started_at = parse_time(traffic.get("started_at"))
ends_at = parse_time(traffic.get("ends_at"))
if not date_text or not timezone_text or started_at is None or ends_at is None:
return None
if started_at.tzinfo is None or ends_at.tzinfo is None:
return None
try:
date = dt.date.fromisoformat(date_text)
zone = ZoneInfo(timezone_text)
except (ValueError, ZoneInfoNotFoundError):
return None
local_start = dt.datetime.combine(date, dt.time.min, tzinfo=zone)
local_end = local_start + dt.timedelta(days=1)
if (
started_at.astimezone(dt.timezone.utc) != local_start.astimezone(dt.timezone.utc)
or ends_at.astimezone(dt.timezone.utc) != local_end.astimezone(dt.timezone.utc)
):
return None
return {
"period_kind": "calendar_day",
"date": date.isoformat(),
"timezone": zone.key,
"started_at": started_at.astimezone(dt.timezone.utc).isoformat(),
"ends_at": ends_at.astimezone(dt.timezone.utc).isoformat(),
}
def workspace_traffic_is_current(payload: dict[str, Any], *, now: dt.datetime | None = None) -> bool:
traffic = payload.get("traffic") if isinstance(payload.get("traffic"), dict) else {}
period = workspace_traffic_period(traffic)
if period is None:
return False
current = now or dt.datetime.now(dt.timezone.utc)
if current.tzinfo is None:
current = current.replace(tzinfo=dt.timezone.utc)
started_at = parse_time(period["started_at"])
ends_at = parse_time(period["ends_at"])
return bool(started_at and ends_at and started_at <= current < ends_at)
def workspace_period_label(traffic: Any) -> str:
period = workspace_traffic_period(traffic)
if period is None:
return "today"
return f"today {period['date']} ({period['timezone']})"
def fetch_workspace_payload(workspace_url: str, timeout: int) -> dict[str, Any]:
payload = fetch_payload(
workspace_url,
@@ -541,6 +598,8 @@ def fetch_workspace_payload(workspace_url: str, timeout: int) -> dict[str, Any]:
or not workspace_values_are_finite(payload)
):
raise RuntimeError("Pricing Monitor returned an invalid workspace projection")
if workspace_traffic_period(traffic) is None:
raise RuntimeError("Pricing Monitor returned an invalid workspace traffic period")
channel_monitors = payload["status"].get("channel_monitors")
bounded_fields = (
("accounts", payload["accounts"].get("accounts"), MAX_WORKSPACE_ACCOUNTS),
@@ -610,10 +669,13 @@ def workspace_logs_payload(payload: dict[str, Any]) -> dict[str, Any]:
def workspace_keys_payload(payload: dict[str, Any]) -> dict[str, Any]:
traffic = payload.get("traffic") if isinstance(payload.get("traffic"), dict) else {}
period = workspace_traffic_period(traffic) or {}
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")),
"period_kind": period.get("period_kind", ""),
"date": period.get("date", ""),
"timezone": period.get("timezone", ""),
"workspace": True,
}
@@ -649,7 +711,7 @@ def workspace_errors_payload(payload: dict[str, Any]) -> dict[str, Any]:
return {
"items": items,
"sources": sources,
"time_range": f"{as_int(traffic.get('window_hours')) or 24}h",
"time_range": workspace_period_label(traffic),
"limit": as_int(traffic.get("limit")),
"generated_at": traffic.get("generated_at") or payload.get("generated_at"),
"workspace": True,
@@ -689,12 +751,17 @@ class WorkspaceCache:
def get(self, *, force: bool = False) -> dict[str, Any]:
with self.lock:
now = time.monotonic()
current_payload = bool(self.payload) and workspace_traffic_is_current(self.payload)
if not force:
if self.error and self.last_attempt_at and now - self.last_attempt_at < self.retry_seconds:
if self.payload:
if current_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:
if (
not self.error
and current_payload
and now - self.fetched_at < self.ttl_seconds
):
return copy.deepcopy(self.payload)
self.last_attempt_at = now
self.network_fetches += 1
@@ -702,9 +769,14 @@ class WorkspaceCache:
payload = fetch_workspace_payload(self.url, self.timeout)
except Exception as exc:
self.error = "workspace unavailable"
if self.payload:
if current_payload:
return copy.deepcopy(self.payload)
raise RuntimeError(self.error) from exc
if not workspace_traffic_is_current(payload):
self.error = "workspace current-day traffic unavailable"
if current_payload:
return copy.deepcopy(self.payload)
raise RuntimeError(self.error)
self.payload = copy.deepcopy(payload)
self.fetched_at = time.monotonic()
self.error = ""
@@ -840,11 +912,18 @@ def payload_data(payload: dict[str, Any]) -> dict[str, Any]:
return data if isinstance(data, dict) else payload
def usage_day_date(now: dt.datetime | None = None) -> str:
current = now or dt.datetime.now(dt.timezone.utc)
if current.tzinfo is None:
current = current.replace(tzinfo=dt.timezone.utc)
return current.astimezone(ZoneInfo(USAGE_TIMEZONE)).date().isoformat()
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()
today = usage_day_date()
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")
@@ -907,10 +986,18 @@ def normalize_key_rows(payload: dict[str, Any], filter_text: str = "") -> list[d
return rows
def usage_period_label(payload: dict[str, Any]) -> str:
if not payload.get("workspace"):
return f"today {payload.get('date') or '-'}"
date = str(payload.get("date") or "").strip()
timezone = str(payload.get("timezone") or "").strip()
return f"today {date}{f' ({timezone})' if timezone else ''}" if date else "today"
def print_keys_once(payload: dict[str, Any]) -> None:
rows = normalize_key_rows(payload)
total_cost = sum(row["cost"] for row in rows)
period = f"{as_int(payload.get('window_hours')) or 24}h" if payload.get("workspace") else f"today {payload.get('date') or '-'}"
period = usage_period_label(payload)
print(f"keys {period} | {len(rows)} keys | {format_cost(total_cost)}")
print("node key cost tokens req")
for row in rows:
@@ -1862,7 +1949,7 @@ def run_textual(
)
self.configure_table(
self.query_one("#keys", DataTable),
("Node", "KEY", "Today" if legacy_direct else "24h", "Tokens", "Req"),
("Node", "KEY", "Today", "Tokens", "Req"),
)
self.configure_table(
self.query_one("#logs", DataTable),
@@ -2209,7 +2296,7 @@ def run_textual(
detail += f" | {error}"
elif table_id == "keys":
detail = (
f"{row['node']} | {row['name']} | {'today' if legacy_direct else '24h'} {format_cost(row['cost'])} | "
f"{row['node']} | {row['name']} | {usage_period_label(self.keys_payload)} {format_cost(row['cost'])} | "
f"{format_count(row['tokens'])} tokens | {format_count(row['requests'])} req"
)
elif table_id == "logs":
@@ -2278,7 +2365,7 @@ def run_textual(
keys_table = self.query_one("#keys", DataTable)
keys_table.cursor_type = "row"
keys_table.zebra_stripes = True
keys_table.add_columns("Node", "Key", "Today" if legacy_direct else "24h", "Tokens", "Req")
keys_table.add_columns("Node", "Key", "Today", "Tokens", "Req")
self.refresh_data(refresh=legacy_direct)
self.set_interval(refresh_seconds, self.refresh_data)
@@ -2891,7 +2978,7 @@ def build_parser() -> argparse.ArgumentParser:
parser.add_argument(
"--errors-time-range",
default=os.environ.get("SHUSUB2_ERRORS_TIME_RANGE", DEFAULT_ERRORS_TIME_RANGE),
help="ops errors time_range query (5m/30m/1h/6h/24h/7d/30d, default 24h)",
help="legacy-direct ops errors time_range query (5m/30m/1h/6h/24h/7d/30d, default 24h)",
)
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")
+60 -1
View File
@@ -2,6 +2,7 @@ from __future__ import annotations
import contextlib
import copy
import datetime as dt
import io
import os
from pathlib import Path
@@ -25,6 +26,9 @@ def load_module():
def workspace_payload_fixture() -> dict[str, object]:
zone = dt.timezone(dt.timedelta(hours=8), name="Asia/Shanghai")
local_start = dt.datetime.combine(dt.datetime.now(zone).date(), dt.time.min, tzinfo=zone)
local_end = local_start + dt.timedelta(days=1)
return {
"service": "sub2api-pricing-monitor",
"view": "workspace",
@@ -77,7 +81,11 @@ def workspace_payload_fixture() -> dict[str, object]:
}
},
"traffic": {
"window_hours": 24,
"period_kind": "calendar_day",
"date": local_start.date().isoformat(),
"timezone": "Asia/Shanghai",
"started_at": local_start.isoformat(),
"ends_at": local_end.isoformat(),
"sample_limit": 1000,
"sample_limited": False,
"partial": False,
@@ -775,6 +783,11 @@ class WorkspaceTests(unittest.TestCase):
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(keys["period_kind"], "calendar_day")
self.assertEqual(keys["date"], payload["traffic"]["date"])
self.assertEqual(keys["timezone"], "Asia/Shanghai")
self.assertIn(f"today {keys['date']} (Asia/Shanghai)", mod.usage_period_label(keys))
self.assertEqual(errors["time_range"], f"today {keys['date']} (Asia/Shanghai)")
self.assertEqual(errors["items"][0]["error_count"], 4)
self.assertEqual(errors["items"][0]["phase"], "upstream_http")
self.assertEqual(errors["sources"]["server6"]["total"], 4)
@@ -811,6 +824,28 @@ class WorkspaceTests(unittest.TestCase):
with self.assertRaisesRegex(RuntimeError, "invalid workspace requests"):
mod.fetch_workspace_payload("https://workspace.example.test/data", 3)
invalid_period = workspace_payload_fixture()
invalid_period["traffic"].pop("ends_at")
with mock.patch.object(mod, "fetch_payload", return_value=invalid_period):
with self.assertRaisesRegex(RuntimeError, "invalid workspace traffic period"):
mod.fetch_workspace_payload("https://workspace.example.test/data", 3)
utc_period = workspace_payload_fixture()
utc_start = dt.datetime.now(dt.timezone.utc).replace(
hour=0, minute=0, second=0, microsecond=0
)
utc_period["traffic"].update(
{
"date": utc_start.date().isoformat(),
"timezone": "UTC",
"started_at": utc_start.isoformat(),
"ends_at": (utc_start + dt.timedelta(days=1)).isoformat(),
}
)
with mock.patch.object(mod, "fetch_payload", return_value=utc_period):
with self.assertRaisesRegex(RuntimeError, "invalid workspace traffic period"):
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] = []
@@ -840,6 +875,30 @@ class WorkspaceTests(unittest.TestCase):
cache.get(force=True)
self.assertEqual(len(calls), 2)
def test_workspace_cache_rejects_previous_calendar_day(self) -> None:
mod = load_module()
self.assertEqual(
mod.usage_day_date(dt.datetime(2026, 8, 3, 16, tzinfo=dt.timezone.utc)),
"2026-08-04",
)
payload = workspace_payload_fixture()
yesterday = dt.datetime.now(dt.timezone.utc) - dt.timedelta(days=1)
local_start = yesterday.astimezone(dt.timezone(dt.timedelta(hours=8))).replace(
hour=0, minute=0, second=0, microsecond=0
)
local_end = local_start + dt.timedelta(days=1)
payload["traffic"].update(
{
"date": local_start.date().isoformat(),
"started_at": local_start.isoformat(),
"ends_at": local_end.isoformat(),
}
)
cache = mod.WorkspaceCache("https://workspace.example.test/data", 3, 300)
with mock.patch.object(mod, "fetch_workspace_payload", return_value=payload):
with self.assertRaisesRegex(RuntimeError, "current-day traffic unavailable"):
cache.get()
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)
Generated
+1 -1
View File
@@ -85,7 +85,7 @@ wheels = [
[[package]]
name = "shusub2"
version = "0.3.0"
version = "0.3.1"
source = { editable = "." }
dependencies = [
{ name = "textual" },