diff --git a/README.md b/README.md index be6ab0e..3312f5d 100644 --- a/README.md +++ b/README.md @@ -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`： diff --git a/pyproject.toml b/pyproject.toml index d4463ee..c613c62 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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" diff --git a/sub2api_quota_tui.py b/sub2api_quota_tui.py index 83d4a44..addfa55 100644 --- a/sub2api_quota_tui.py +++ b/sub2api_quota_tui.py @@ -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") diff --git a/tests/test_payload.py b/tests/test_payload.py index 9661ee8..8d7fb8a 100644 --- a/tests/test_payload.py +++ b/tests/test_payload.py @@ -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) diff --git a/uv.lock b/uv.lock index f0d1adf..b826193 100644 --- a/uv.lock +++ b/uv.lock @@ -85,7 +85,7 @@ wheels = [ [[package]] name = "shusub2" -version = "0.3.0" +version = "0.3.1" source = { editable = "." } dependencies = [ { name = "textual" },