#!/usr/bin/env python3 """Textual TUI for Sub2API quota and daily account usage.""" from __future__ import annotations import argparse import datetime as dt import importlib.metadata import json import os import re import subprocess import sys import urllib.parse import urllib.request import zlib from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path from typing import Any APP_NAME = "shusub2" FALLBACK_VERSION = "0.2.10" 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" DEFAULT_LOGS_API_URL = "https://sub2apicn.shujk.top/api/v1/admin/usage" DEFAULT_LOGS_URL_CONFIG_FILE = "~/.config/shusub2/logs-url" DEFAULT_LOGS_TOKEN_CONFIG_FILE = "~/.config/shusub2/logs-token" # sub2apius.shujk.top is Cloudflare-proxied and often returns 1010 for non-browser clients. # Use the fixed server4 origin for machine clients while still labeling the source as us. DEFAULT_ERRORS_CN_URL = "https://sub2apicn.shujk.top/api/v1/admin/ops/errors" DEFAULT_ERRORS_US_URL = "https://sub2api.server4.shujk.top:19857/api/v1/admin/ops/errors" DEFAULT_ERRORS_CN_URL_CONFIG_FILE = "~/.config/shusub2/errors-cn-url" DEFAULT_ERRORS_US_URL_CONFIG_FILE = "~/.config/shusub2/errors-us-url" DEFAULT_VERSION_CHECK_URL = "https://gitea.shujk.top/shujakuin/shusub2/raw/branch/main/pyproject.toml" DEFAULT_REFRESH_SECONDS = 60 DEFAULT_LOGS_REFRESH_SECONDS = 60 DEFAULT_ERRORS_REFRESH_SECONDS = 60 DEFAULT_LOGS_LIMIT = 100 DEFAULT_ERRORS_LIMIT = 100 DEFAULT_ERRORS_TIME_RANGE = "24h" DEFAULT_TIMEOUT_SECONDS = 10 DEFAULT_VERSION_CHECK_TIMEOUT_SECONDS = 2 MONITOR_OK_STATUSES = {"operational", "ok", "success"} MONITOR_FAILED_STATUSES = {"error", "failed", "failure"} MONITOR_STOPWORDS = {"response", "responses", "monitor"} INSTALL_COMMAND = "uv tool install --force git+https://gitea.shujk.top/shujakuin/shusub2.git" INSTALL_COMMAND_ARGS = ["uv", "tool", "install", "--force", "git+https://gitea.shujk.top/shujakuin/shusub2.git"] def env_int(name: str, default: int, *, minimum: int = 1) -> int: try: return max(minimum, int(os.environ.get(name, default))) except Exception: return default def configured_url(env_names: tuple[str, ...], config_path: str, default: str = "") -> str: for name in env_names: value = os.environ.get(name, "").strip() if value: return value config_file = Path(config_path).expanduser() try: for line in config_file.read_text(encoding="utf-8").splitlines(): value = line.strip() if value and not value.startswith("#"): return value except OSError: pass return default def default_api_url() -> str: return configured_url( ("SHUSUB2_API_URL", "SUB2API_QUOTA_TUI_API_URL"), os.environ.get("SHUSUB2_API_URL_FILE", DEFAULT_CONFIG_FILE), DEFAULT_API_URL, ) def default_status_url() -> str: return configured_url( ("SHUSUB2_STATUS_URL",), os.environ.get("SHUSUB2_STATUS_URL_FILE", DEFAULT_STATUS_CONFIG_FILE), ) def default_logs_url() -> str: return configured_url( ("SHUSUB2_LOGS_URL",), os.environ.get("SHUSUB2_LOGS_URL_FILE", DEFAULT_LOGS_URL_CONFIG_FILE), DEFAULT_LOGS_API_URL, ) def default_logs_token() -> str: return configured_url( ("SHUSUB2_LOGS_TOKEN",), os.environ.get("SHUSUB2_LOGS_TOKEN_FILE", DEFAULT_LOGS_TOKEN_CONFIG_FILE), ) def config_file_path() -> Path: return Path(os.environ.get("SHUSUB2_API_URL_FILE", DEFAULT_CONFIG_FILE)).expanduser() def logs_url_config_file_path() -> Path: return Path(os.environ.get("SHUSUB2_LOGS_URL_FILE", DEFAULT_LOGS_URL_CONFIG_FILE)).expanduser() def logs_token_config_file_path() -> Path: return Path(os.environ.get("SHUSUB2_LOGS_TOKEN_FILE", DEFAULT_LOGS_TOKEN_CONFIG_FILE)).expanduser() def default_errors_cn_url() -> str: return configured_url( ("SHUSUB2_ERRORS_CN_URL",), os.environ.get("SHUSUB2_ERRORS_CN_URL_FILE", DEFAULT_ERRORS_CN_URL_CONFIG_FILE), DEFAULT_ERRORS_CN_URL, ) def default_errors_us_url() -> str: return configured_url( ("SHUSUB2_ERRORS_US_URL",), os.environ.get("SHUSUB2_ERRORS_US_URL_FILE", DEFAULT_ERRORS_US_URL_CONFIG_FILE), DEFAULT_ERRORS_US_URL, ) def errors_cn_url_config_file_path() -> Path: return Path(os.environ.get("SHUSUB2_ERRORS_CN_URL_FILE", DEFAULT_ERRORS_CN_URL_CONFIG_FILE)).expanduser() def errors_us_url_config_file_path() -> Path: return Path(os.environ.get("SHUSUB2_ERRORS_US_URL_FILE", DEFAULT_ERRORS_US_URL_CONFIG_FILE)).expanduser() def write_config_value(path: Path, value: str) -> Path: text = str(value or "").strip() if not text: raise ValueError("config value is empty") path.parent.mkdir(parents=True, exist_ok=True) try: path.parent.chmod(0o700) except OSError: pass path.write_text(text + "\n", encoding="utf-8") try: path.chmod(0o600) except OSError: pass return path def write_api_url_config(api_url: str) -> Path: value = str(api_url or "").strip() if not value: raise ValueError("api url is empty") return write_config_value(config_file_path(), value) def run_install_command() -> int: try: return subprocess.run(INSTALL_COMMAND_ARGS, check=False).returncode except FileNotFoundError: print("uv command not found; install uv first, then run:", INSTALL_COMMAND, file=sys.stderr) return 127 def inferred_status_url(api_url: str) -> str: parsed = urllib.parse.urlparse(str(api_url or "").strip()) if not parsed.scheme or not parsed.netloc: return "" path = parsed.path.rstrip("/") if path.endswith("/api/tui/accounts"): status_path = path[: -len("/api/tui/accounts")] + "/api/status" elif path.endswith("/api/accounts"): status_path = path[: -len("/api/accounts")] + "/api/status" else: return "" return urllib.parse.urlunparse(parsed._replace(path=status_path, params="", query="", fragment="")) def current_version() -> str: try: return importlib.metadata.version(APP_NAME) except importlib.metadata.PackageNotFoundError: return FALLBACK_VERSION except Exception: return FALLBACK_VERSION def parse_version(value: Any) -> tuple[int, ...]: parts = [] for part in re.split(r"[^0-9]+", str(value or "")): if part: parts.append(int(part)) return tuple(parts) def version_is_newer(latest: str, current: str) -> bool: latest_parts = parse_version(latest) current_parts = parse_version(current) width = max(len(latest_parts), len(current_parts), 1) return latest_parts + (0,) * (width - len(latest_parts)) > current_parts + (0,) * (width - len(current_parts)) def latest_version_from_text(text: str) -> str: match = re.search(r'(?m)^version\s*=\s*"([^"]+)"', text) return "" if not match else match.group(1).strip() def fetch_latest_version(url: str, timeout: int) -> str: req = urllib.request.Request(url, headers={"Accept": "text/plain"}) with urllib.request.urlopen(req, timeout=timeout) as response: return latest_version_from_text(response.read(65536).decode("utf-8", errors="replace")) def version_update_message(latest: str, current: str | None = None) -> str: current = current or current_version() if not latest or not version_is_newer(latest, current): return "" return f"update available: {APP_NAME} {current} -> {latest}; run `{INSTALL_COMMAND}`" def check_version_update(url: str, timeout: int, *, disabled: bool = False) -> str: if disabled or os.environ.get("SHUSUB2_NO_VERSION_CHECK", "").strip().lower() in {"1", "true", "yes", "on"}: return "" try: return version_update_message(fetch_latest_version(url, timeout)) except Exception: return "" def as_float(value: Any) -> float: try: if value is None or str(value).strip() == "": return 0.0 return float(value) except Exception: return 0.0 def as_int(value: Any) -> int: try: if value is None or str(value).strip() == "": return 0 return int(float(value)) except Exception: return 0 def format_cost(value: Any) -> str: amount = as_float(value) if amount == 0: return "$0" if abs(amount) < 0.01: return f"${amount:.6f}".rstrip("0").rstrip(".") if abs(amount) < 10: return f"${amount:.3f}".rstrip("0").rstrip(".") return f"${amount:.2f}".rstrip("0").rstrip(".") def format_count(value: Any) -> str: number = as_int(value) if abs(number) >= 1_000_000: return f"{number / 1_000_000:.1f}M".rstrip("0").rstrip(".") if abs(number) >= 1_000: return f"{number / 1_000:.1f}K".rstrip("0").rstrip(".") return str(number) def format_percent(value: Any) -> str: if value is None or str(value).strip() == "": return "-" percent = as_float(value) rounded = round(percent, 1) if rounded.is_integer(): return f"{int(rounded)}%" return f"{rounded}%" def parse_time(value: Any) -> dt.datetime | None: text = str(value or "").strip() if not text: return None try: normalized = text[:-1] + "+00:00" if text.endswith("Z") else text return dt.datetime.fromisoformat(normalized) except Exception: return None def short_time(value: Any) -> str: parsed = parse_time(value) if not parsed: return "-" return parsed.astimezone().strftime("%m-%d %H:%M") def relative_age(value: Any, now: dt.datetime | None = None) -> str: parsed = parse_time(value) if not parsed: return "-" current = now or dt.datetime.now(dt.timezone.utc) if current.tzinfo is None: current = current.replace(tzinfo=dt.timezone.utc) elapsed = (current.astimezone(dt.timezone.utc) - parsed.astimezone(dt.timezone.utc)).total_seconds() if elapsed < -30: return f"in {max(1, int((-elapsed) / 60))}m" seconds = max(0, int(elapsed)) if seconds < 60: return "now" minutes = seconds // 60 if minutes < 60: return f"{minutes}m ago" hours = minutes // 60 if hours < 24: return f"{hours}h ago" return f"{hours // 24}d ago" def add_refresh_param(api_url: str, refresh: bool) -> str: if not refresh: return api_url parsed = urllib.parse.urlparse(api_url) query = urllib.parse.parse_qsl(parsed.query, keep_blank_values=True) query = [(key, value) for key, value in query if key != "refresh"] query.append(("refresh", "1")) return urllib.parse.urlunparse(parsed._replace(query=urllib.parse.urlencode(query))) def fetch_payload(api_url: str, timeout: int, *, refresh: bool = False) -> dict[str, Any]: req = urllib.request.Request(add_refresh_param(api_url, refresh), headers={"Accept": "application/json"}) with urllib.request.urlopen(req, timeout=timeout) as response: data = json.loads(response.read().decode("utf-8")) if not isinstance(data, dict): raise RuntimeError("API did not return a JSON object") return data def fetch_optional_payload(url: str, timeout: int) -> tuple[dict[str, Any], str]: if not str(url or "").strip(): return {}, "" try: return fetch_payload(url, timeout), "" except Exception as exc: return {}, str(exc) 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) query = [(key, value) for key, value in query if key not in {"page", "page_size", "limit"}] query.extend((("page", "1"), ("page_size", str(max(1, min(int(limit), 1000)))))) return urllib.parse.urlunparse(parsed._replace(query=urllib.parse.urlencode(query))) def fetch_logs_payload(logs_url: str, token: str, timeout: int, limit: int = DEFAULT_LOGS_LIMIT) -> dict[str, Any]: headers = {"Accept": "application/json"} if str(token or "").strip(): headers["x-api-key"] = str(token).strip() req = urllib.request.Request(logs_request_url(logs_url, limit), headers=headers) with urllib.request.urlopen(req, timeout=timeout) as response: data = json.loads(response.read().decode("utf-8")) if not isinstance(data, dict): raise RuntimeError("logs API did not return a JSON object") return data def logs_envelope(payload: dict[str, Any]) -> dict[str, Any]: data = payload.get("data") if isinstance(data, dict) and isinstance(data.get("items"), list): return data return payload def log_items(payload: dict[str, Any]) -> list[dict[str, Any]]: items = logs_envelope(payload).get("items") if not isinstance(items, list): return [] return [item for item in items if isinstance(item, dict)] def log_type_label(item: dict[str, Any]) -> str: text = str(item.get("request_type") or "").strip().lower() if text and text != "unknown": return text if item.get("openai_ws_mode"): return "ws_v2" if item.get("stream") is True: return "stream" if item.get("stream") is False: return "sync" return text or "-" def log_cache_tokens(item: dict[str, Any]) -> int: return as_int(item.get("cache_creation_tokens")) + as_int(item.get("cache_read_tokens")) def log_total_tokens(item: dict[str, Any]) -> int: return as_int(item.get("input_tokens")) + as_int(item.get("output_tokens")) + log_cache_tokens(item) def log_tokens_per_second(item: dict[str, Any]) -> float: output = as_int(item.get("output_tokens")) duration = as_int(item.get("duration_ms")) first = max(0, as_int(item.get("first_token_ms"))) decode_ms = duration - first if output <= 0 or decode_ms <= 0: return 0.0 return output * 1000.0 / decode_ms def format_rate(value: Any) -> str: rate = as_float(value) if rate <= 0: return "-" if rate >= 100: return f"{rate:.0f}/s" return f"{rate:.1f}/s" def format_seconds(value: Any) -> str: number = as_int(value) if number <= 0: return "-" return f"{number / 1000:.1f}s" KEY_COLOR_PALETTE = ( "cyan", "magenta", "green", "yellow", "bright_blue", "bright_cyan", "bright_magenta", "bright_green", "bright_yellow", "bright_red", ) def key_color(name: Any) -> str: text = str(name or "").strip() if not text or text == "-": return "" return KEY_COLOR_PALETTE[zlib.crc32(text.encode("utf-8")) % len(KEY_COLOR_PALETTE)] def admin_api_base(logs_url: str) -> str: parsed = urllib.parse.urlparse(str(logs_url or "").strip()) if not parsed.scheme or not parsed.netloc: return "" path = parsed.path.rstrip("/") if path.endswith("/usage"): path = path[: -len("/usage")] if not path.endswith("/admin"): return "" return urllib.parse.urlunparse(parsed._replace(path=path, params="", query="", fragment="")) def fetch_admin_json(url: str, token: str, timeout: int, body: dict[str, Any] | None = None) -> dict[str, Any]: headers = {"Accept": "application/json"} if str(token or "").strip(): headers["x-api-key"] = str(token).strip() data = None if body is not None: headers["Content-Type"] = "application/json" data = json.dumps(body).encode("utf-8") req = urllib.request.Request(url, data=data, headers=headers) with urllib.request.urlopen(req, timeout=timeout) as response: payload = json.loads(response.read().decode("utf-8")) if not isinstance(payload, dict): raise RuntimeError("admin API did not return a JSON object") return payload def payload_data(payload: dict[str, Any]) -> dict[str, Any]: data = payload.get("data") return data if isinstance(data, dict) else payload def fetch_key_usage_payload(logs_url: str, token: str, timeout: int, limit: int = 100) -> dict[str, Any]: base = admin_api_base(logs_url) if not base: raise RuntimeError("cannot derive the admin API base from the logs url") today = dt.date.today().isoformat() query = urllib.parse.urlencode({"start_date": today, "end_date": today, "granularity": "day", "limit": limit}) trend_payload = fetch_admin_json(f"{base}/dashboard/api-keys-trend?{query}", token, timeout) trend = payload_data(trend_payload).get("trend") points = [point for point in trend if isinstance(point, dict)] if isinstance(trend, list) else [] ids = sorted({as_int(point.get("api_key_id")) for point in points if as_int(point.get("api_key_id"))}) stats: dict[str, Any] = {} if ids: costs_payload = fetch_admin_json(f"{base}/dashboard/api-keys-usage", token, timeout, body={"api_key_ids": ids}) raw_stats = payload_data(costs_payload).get("stats") if isinstance(raw_stats, dict): stats = raw_stats return {"date": today, "trend": points, "stats": stats} def normalize_key_rows(payload: dict[str, Any], filter_text: str = "") -> list[dict[str, Any]]: needle = filter_text.strip().lower() stats = payload.get("stats") if isinstance(payload.get("stats"), dict) else {} merged: dict[Any, dict[str, Any]] = {} for point in payload.get("trend") or []: if not isinstance(point, dict): continue key_id = as_int(point.get("api_key_id")) name = str(point.get("key_name") or "").strip() or (f"#{key_id}" if key_id else "-") 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["requests"] += as_int(point.get("requests")) row["tokens"] += as_int(point.get("tokens")) for row in merged.values(): stat = stats.get(str(row["id"])) if not isinstance(stat, dict): stat = stats.get(row["id"]) if isinstance(stat, dict): row["cost"] = as_float(stat.get("today_actual_cost")) rows = list(merged.values()) rows.sort(key=lambda row: (-row["cost"], -row["tokens"], str(row["name"]).lower())) return rows def print_keys_once(payload: dict[str, Any]) -> None: rows = normalize_key_rows(payload) total_cost = sum(row["cost"] for row in rows) print(f"keys today {payload.get('date') or '-'} | {len(rows)} keys | {format_cost(total_cost)}") print("key today tokens req") for row in rows: print( f"{str(row['name'])[:20]:<21} " f"{format_cost(row['cost']):<10} " f"{format_count(row['tokens']):<8} " f"{format_count(row['requests'])}" ) def nested_name(item: dict[str, Any], object_key: str, id_key: str, name_key: str = "") -> str: nested = item.get(object_key) if isinstance(nested, dict): name = str(nested.get("name") or "").strip() if name: return name if name_key: name = str(item.get(name_key) or "").strip() if name: return name identifier = as_int(item.get(id_key)) return f"#{identifier}" if identifier else "-" def normalize_log_rows(payload: dict[str, Any], filter_text: str = "") -> list[dict[str, Any]]: needle = filter_text.strip().lower() rows = [] for item in log_items(payload): 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") model = str(item.get("model") or "-") upstream_model = str(item.get("upstream_model") or "").strip() effort = str(item.get("reasoning_effort") or "").strip() or "-" type_label = log_type_label(item) row = { "id": as_int(item.get("id")), "key": key_name, "account": account_name, "user": user_name, "model": model, "upstream_model": upstream_model, "effort": effort, "type": type_label, "tokens": log_total_tokens(item), "input_tokens": as_int(item.get("input_tokens")), "output_tokens": as_int(item.get("output_tokens")), "cache_creation_tokens": as_int(item.get("cache_creation_tokens")), "cache_read_tokens": as_int(item.get("cache_read_tokens")), "cache_tokens": log_cache_tokens(item), "cost": as_float(item.get("total_cost")), "actual_cost": as_float(item.get("actual_cost")), "duration_ms": as_int(item.get("duration_ms")), "first_token_ms": as_int(item.get("first_token_ms")), "tokens_per_second": log_tokens_per_second(item), "created_at": str(item.get("created_at") or ""), "time": short_time(item.get("created_at")), "age": relative_age(item.get("created_at")), "request_id": str(item.get("request_id") or ""), "raw": item, } if needle: haystack = " ".join( (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) return rows def logs_summary_line(payload: dict[str, Any], shown: int) -> str: envelope = logs_envelope(payload) total = as_int(envelope.get("total")) page_size = as_int(envelope.get("page_size")) fetched = short_time(dt.datetime.now(dt.timezone.utc).isoformat()) 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"total {format_count(total)} records | page cost {format_cost(cost)} | {format_count(tokens)} tokens" ) def log_detail_line(row: dict[str, Any]) -> str: model = row["model"] if row["upstream_model"] and row["upstream_model"] != row["model"]: model = f"{row['model']} -> {row['upstream_model']}" 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"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'])}) | " f"first {format_seconds(row['first_token_ms'])} duration {format_seconds(row['duration_ms'])} " f"{format_rate(row['tokens_per_second'])}" ) if row["request_id"]: detail += f" | {row['request_id']}" return detail 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") for row in rows: print( f"{row['key'][:20]:<21} " f"{row['account'][:20]:<21} " f"{row['model'][:24]:<25} " f"{row['effort']:<7} " f"{row['type']:<7} " f"{format_count(row['input_tokens']):<8} " f"{format_count(row['output_tokens']):<8} " f"{format_count(row['cache_tokens']):<8} " f"{format_count(row['tokens']):<8} " f"{format_cost(row['cost']):<10} " f"{format_seconds(row['first_token_ms']):<8} " f"{format_seconds(row['duration_ms']):<10} " f"{format_rate(row['tokens_per_second']):<8} " f"{row['time']:<11} " f"{row['age']}" ) def errors_request_url(errors_url: str, limit: int, time_range: str = DEFAULT_ERRORS_TIME_RANGE) -> str: parsed = urllib.parse.urlparse(errors_url) query = urllib.parse.parse_qsl(parsed.query, keep_blank_values=True) query = [(key, value) for key, value in query if key not in {"page", "page_size", "limit", "time_range"}] page_size = max(1, min(int(limit), 500)) query.extend( ( ("page", "1"), ("page_size", str(page_size)), ("time_range", str(time_range or DEFAULT_ERRORS_TIME_RANGE).strip() or DEFAULT_ERRORS_TIME_RANGE), ) ) return urllib.parse.urlunparse(parsed._replace(query=urllib.parse.urlencode(query))) def fetch_errors_payload( errors_url: str, token: str, timeout: int, limit: int = DEFAULT_ERRORS_LIMIT, time_range: str = DEFAULT_ERRORS_TIME_RANGE, ) -> dict[str, Any]: headers = {"Accept": "application/json"} if str(token or "").strip(): headers["x-api-key"] = str(token).strip() req = urllib.request.Request(errors_request_url(errors_url, limit, time_range), headers=headers) with urllib.request.urlopen(req, timeout=timeout) as response: data = json.loads(response.read().decode("utf-8")) if not isinstance(data, dict): raise RuntimeError("errors API did not return a JSON object") return data def error_items(payload: dict[str, Any]) -> list[dict[str, Any]]: data = payload.get("data") if isinstance(data, dict) and isinstance(data.get("items"), list): items = data.get("items") elif isinstance(payload.get("items"), list): items = payload.get("items") else: items = [] return [item for item in items if isinstance(item, dict)] def error_total(payload: dict[str, Any]) -> int: data = payload.get("data") if isinstance(data, dict) and data.get("total") is not None: return as_int(data.get("total")) if payload.get("total") is not None: return as_int(payload.get("total")) return len(error_items(payload)) def default_error_sources(cn_url: str, us_url: str) -> list[tuple[str, str]]: sources: list[tuple[str, str]] = [] if str(cn_url or "").strip(): sources.append(("cn", str(cn_url).strip())) if str(us_url or "").strip(): sources.append(("us", str(us_url).strip())) return sources def fetch_merged_errors( sources: list[tuple[str, str]], token: str, timeout: int, limit: int = DEFAULT_ERRORS_LIMIT, time_range: str = DEFAULT_ERRORS_TIME_RANGE, ) -> dict[str, Any]: source_status: dict[str, dict[str, Any]] = {} merged_items: list[dict[str, Any]] = [] def _one(node: str, url: str) -> tuple[str, str, dict[str, Any] | None, str]: try: return node, url, fetch_errors_payload(url, token, timeout, limit, time_range), "" except Exception as exc: return node, url, None, str(exc) if not sources: return {"items": [], "sources": {}, "time_range": time_range, "limit": limit} with ThreadPoolExecutor(max_workers=max(1, len(sources))) as pool: futures = [pool.submit(_one, node, url) for node, url in sources] for future in as_completed(futures): node, url, payload, error = future.result() if payload is None: source_status[node] = {"ok": False, "url": url, "total": 0, "fetched": 0, "error": error} continue items = error_items(payload) for item in items: row = dict(item) row["_node"] = node row["_source_url"] = url merged_items.append(row) source_status[node] = { "ok": True, "url": url, "total": error_total(payload), "fetched": len(items), "error": "", } merged_items.sort( key=lambda item: (str(item.get("created_at") or ""), as_int(item.get("id")), str(item.get("_node") or "")), reverse=True, ) return { "items": merged_items, "sources": source_status, "time_range": time_range, "limit": limit, "generated_at": dt.datetime.now(dt.timezone.utc).isoformat(), } def normalize_error_rows(payload: dict[str, Any], filter_text: str = "") -> list[dict[str, Any]]: needle = filter_text.strip().lower() rows = [] for item in payload.get("items") or []: if not isinstance(item, dict): continue node = str(item.get("_node") or item.get("node") or "-").strip() or "-" key_name = str(item.get("api_key_name") or "").strip() if not key_name and item.get("api_key_id") is not None: key_name = f"#{as_int(item.get('api_key_id'))}" if not key_name: key_name = "-" account_name = str(item.get("account_name") or "").strip() if not account_name and item.get("account_id") is not None: account_name = f"#{as_int(item.get('account_id'))}" if not account_name: account_name = "-" user_email = str(item.get("user_email") or "").strip() if not user_email and item.get("user_id") is not None: 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 "-" 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() 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 "-" row = { "id": as_int(item.get("id")), "node": node, "status_code": status_code, "key": key_name, "account": account_name, "user": user_email or "-", "model": model, "upstream_model": upstream_model, "phase": phase, "type": error_type, "owner": owner, "source": source, "platform": platform, "group": group_name, "message": message, "request_id": request_id, "client_request_id": client_request_id, "created_at": str(item.get("created_at") or ""), "time": short_time(item.get("created_at")), "age": relative_age(item.get("created_at")), "resolved": bool(item.get("resolved")), "raw": item, } if needle: haystack = " ".join( ( node, key_name, account_name, user_email, model, upstream_model, phase, error_type, owner, source, platform, group_name, message, request_id, client_request_id, str(status_code), ) ).lower() if needle not in haystack: continue rows.append(row) rows.sort(key=lambda row: (row["created_at"], row["id"], row["node"]), reverse=True) return rows def errors_summary_line(payload: dict[str, Any], shown: int) -> str: sources = payload.get("sources") if isinstance(payload.get("sources"), dict) else {} source_bits = [] for node in sorted(sources.keys()): info = sources.get(node) if isinstance(sources.get(node), dict) else {} if info.get("ok"): source_bits.append(f"{node} {as_int(info.get('fetched'))}/{as_int(info.get('total'))}") else: source_bits.append(f"{node} err") fetched = short_time(payload.get("generated_at") or dt.datetime.now(dt.timezone.utc).isoformat()) time_range = str(payload.get("time_range") or DEFAULT_ERRORS_TIME_RANGE) source_text = ", ".join(source_bits) if source_bits else "no sources" return f"{fetched} errors | showing {shown}/{len(payload.get('items') or [])} | {time_range} | {source_text}" def error_detail_line(row: dict[str, Any]) -> str: model = row["model"] if row["upstream_model"] and row["upstream_model"] != row["model"]: model = f"{row['model']} -> {row['upstream_model']}" message = row["message"] or "-" if len(message) > 180: message = message[:177] + "..." detail = ( f"{row['time']} | {row['node']} | status {row['status_code']} | 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}" ) if row["request_id"]: detail += f" | {row['request_id']}" if row["client_request_id"] and row["client_request_id"] != row["request_id"]: detail += f" | client {row['client_request_id']}" if row["resolved"]: detail += " | resolved" return detail def print_errors_once(payload: dict[str, Any], filter_text: str = "") -> None: rows = normalize_error_rows(payload, filter_text) print(errors_summary_line(payload, len(rows))) sources = payload.get("sources") if isinstance(payload.get("sources"), dict) else {} for node in sorted(sources.keys()): info = sources.get(node) if isinstance(sources.get(node), dict) else {} if info.get("ok"): 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") for row in rows: print( f"{row['node']:<5} " f"{row['status_code']:<7} " f"{row['key'][:20]:<21} " f"{row['account'][:20]:<21} " f"{row['model'][:24]:<25} " f"{row['phase'][:9]:<10} " f"{row['type'][:13]:<14} " f"{row['owner'][:8]:<9} " f"{row['time']:<11} " f"{row['age']}" ) def kind_label(kind: Any) -> str: return { "quota_limited": "quota", "pending_quota": "pending", "daily_limited": "daily", "pay_as_you_go": "usage", "disabled": "disabled", }.get(str(kind or ""), str(kind or "unknown")) def provider_label(account: dict[str, Any]) -> str: text = str(account.get("provider") or account.get("platform") or "").strip().lower() if not text: return "-" if "anthropic" in text or "claude" in text: return "anthropic" if "openai" in text or "chatgpt" in text: return "openai" return text def routing_group_sort_rank(value: Any) -> int: return {"sfast": 0, "fast": 1, "slow": 2, "id": 3, "": 4, "-": 4}.get(str(value or "").strip().lower(), 4) def status_kind(value: Any) -> str: text = str(value or "").strip().lower() if text in MONITOR_OK_STATUSES: return "ok" if text in MONITOR_FAILED_STATUSES: return "failed" return "unknown" def format_latency(value: Any) -> str: number = as_int(value) return "-" if number <= 0 else f"{number}ms" def monitor_items(status_payload: dict[str, Any]) -> list[dict[str, Any]]: monitors = status_payload.get("channel_monitors") if isinstance(status_payload.get("channel_monitors"), dict) else {} items = [] for item in monitors.get("items") or []: if isinstance(item, dict): items.append(item) return items def monitor_summary(status_payload: dict[str, Any], error: str = "") -> str: if error: return f"monitors error: {error}" monitors = status_payload.get("channel_monitors") if isinstance(status_payload.get("channel_monitors"), dict) else {} if not monitors: return "monitors -" enabled = as_int(monitors.get("enabled")) ok = as_int(monitors.get("latest_ok")) failed = as_int(monitors.get("latest_failed")) unknown = as_int(monitors.get("latest_unknown")) checked = short_time(monitors.get("latest_checked_max")) return f"monitors {ok} ok / {failed} failed / {unknown} unknown ({enabled} enabled, {checked})" def text_tokens(value: Any) -> set[str]: tokens = set() for token in "".join(ch.lower() if ch.isalnum() else " " for ch in str(value or "")).split(): if token and token not in MONITOR_STOPWORDS and not token.isdigit(): tokens.add(token) return tokens def matching_monitor(row: dict[str, Any], status_payload: dict[str, Any]) -> dict[str, Any] | None: base_url_hash = str(row.get("base_url_hash") or "").strip() if base_url_hash: for item in monitor_items(status_payload): if str(item.get("base_url_hash") or "").strip() == base_url_hash: return item account_tokens = text_tokens(row.get("name")) if not account_tokens: return None best: tuple[int, dict[str, Any]] | None = None for item in monitor_items(status_payload): monitor_tokens = text_tokens(item.get("name")) if not monitor_tokens: continue overlap = account_tokens & monitor_tokens if not overlap: continue subset_bonus = 10 if account_tokens <= monitor_tokens or monitor_tokens <= account_tokens else 0 score = subset_bonus + len(overlap) if best is None or score > best[0]: best = (score, item) return None if best is None else best[1] def monitor_detail(row: dict[str, Any], status_payload: dict[str, Any]) -> str: item = matching_monitor(row, status_payload) if not item: return "monitor -" status = str(item.get("latest_status") or "unknown") latency = format_latency(item.get("latency_ms")) checked = short_time(item.get("checked_at")) message = str(item.get("message") or "").strip() detail = f"monitor {status} {latency} checked {checked}" if message: detail += f" {message}" return detail def monitor_availability(row: dict[str, Any], status_payload: dict[str, Any]) -> str: item = matching_monitor(row, status_payload) if not item: return "-" if item.get("enabled") is False: return "disabled" raw_status = str(item.get("latest_status") or "").strip().lower() if not raw_status: return "unknown" kind = status_kind(raw_status) if kind == "ok": return "ok" if kind == "failed": return "failed" return raw_status def window_for(account: dict[str, Any], window_id: str) -> dict[str, Any]: for window in account.get("windows") or []: if isinstance(window, dict) and window.get("id") == window_id: return window return {} def window_cell(account: dict[str, Any], window_id: str) -> str: window = window_for(account, window_id) if not window or window.get("used_percent") is None: return "-" return f"{format_percent(window.get('used_percent'))}/{format_percent(window.get('remaining_percent'))}" def daily_quota_cell(account: dict[str, Any]) -> str: limit = account.get("daily_quota_limit") used = account.get("daily_quota_used") if limit is None: return "-" return f"{used or 0}/{limit}" def reset_cell(account: dict[str, Any]) -> str: daily_reset = short_time(account.get("daily_quota_reset_at")) if daily_reset != "-": return daily_reset five_hour = window_for(account, "five-hour") weekly = window_for(account, "weekly") for window in (five_hour, weekly): reset = short_time(window.get("reset")) if reset != "-": return reset return "-" def normalize_account_rows(payload: dict[str, Any], filter_text: str = "") -> list[dict[str, Any]]: needle = filter_text.strip().lower() rows = [] for account in payload.get("accounts") or []: if not isinstance(account, dict): continue haystack = " ".join( str(account.get(key) or "") for key in ( "id", "name", "routing_group", "provider", "kind", "status", "account_type", "platform", "plan", "daily_quota_timezone", ) ).lower() if needle and needle not in haystack: continue routing_group = str(account.get("routing_group") or "-") rows.append( { "id": as_int(account.get("id")), "name": str(account.get("name") or ""), "routing_group": routing_group, "base_url_hash": str(account.get("base_url_hash") or ""), "provider": provider_label(account), "kind": str(account.get("kind") or "unknown"), "kind_label": kind_label(account.get("kind")), "status": str(account.get("status") or ""), "account_type": str(account.get("account_type") or ""), "plan": str(account.get("plan") or ""), "priority": account.get("priority"), "today_cost_usd": as_float(account.get("today_cost_usd")), "today_actual_cost_usd": as_float(account.get("today_actual_cost_usd")), "today_tokens": as_int(account.get("today_tokens")), "today_requests": as_int(account.get("today_requests")), "daily_quota": account.get("daily_quota") if isinstance(account.get("daily_quota"), dict) else {}, "daily_quota_limit": account.get("daily_quota_limit"), "daily_quota_used": account.get("daily_quota_used"), "daily_quota_remaining": account.get("daily_quota_remaining"), "daily_quota_used_percent": account.get("daily_quota_used_percent"), "daily_quota_reset_at": account.get("daily_quota_reset_at"), "daily_quota_cell": daily_quota_cell(account), "quota_used_percent_max": account.get("quota_used_percent_max"), "five_hour": window_cell(account, "five-hour"), "weekly": window_cell(account, "weekly"), "reset": reset_cell(account), "latest_usage_at": short_time(account.get("latest_usage_at") or account.get("last_used_at")), "error": str(account.get("error") or ""), "raw": account, } ) rows.sort( key=lambda item: ( routing_group_sort_rank(item["routing_group"]), 1 if item["kind"] == "disabled" else 0, -as_float(item["today_cost_usd"]), -as_int(item["today_tokens"]), -as_float(item["daily_quota_used_percent"]), -as_float(item["quota_used_percent_max"]), str(item["name"]).lower(), ) ) return rows def summary_line(payload: dict[str, Any]) -> str: totals = payload.get("totals") if isinstance(payload.get("totals"), dict) else {} generated = short_time(payload.get("generated_local") or payload.get("generated_at")) cached = "cached" if payload.get("cached") else "live" return ( f"{generated} {cached} | " f"{as_int(totals.get('usable_accounts'))}/{as_int(totals.get('total_accounts'))} usable | " f"today {format_cost(totals.get('today_cost_usd'))} | " f"{format_count(totals.get('today_tokens'))} tokens | " f"{format_count(totals.get('today_requests'))} req" ) def print_once(payload: dict[str, Any], filter_text: str = "", status_payload: dict[str, Any] | None = None, status_error: str = "") -> None: print(summary_line(payload)) if status_payload or status_error: print(monitor_summary(status_payload or {}, status_error)) print("name provider group daily today tokens req kind 5h 7d reset status availability") for row in normalize_account_rows(payload, filter_text): print( f"{row['name'][:30]:<30} " f"{row['provider']:<9} " f"{row['routing_group']:<6} " f"{row['daily_quota_cell']:<10} " f"{format_cost(row['today_cost_usd']):<10} " f"{format_count(row['today_tokens']):<8} " f"{format_count(row['today_requests']):<5} " f"{row['kind_label']:<9} " f"{row['five_hour']:<9} " f"{row['weekly']:<9} " f"{row['reset']:<11} " f"{row['status']:<10} " f"{monitor_availability(row, status_payload or {})}" ) def run_textual( api_url: str, status_url: str, logs_url: str, logs_token: str, errors_cn_url: str, errors_us_url: str, refresh_seconds: int, logs_refresh_seconds: int, errors_refresh_seconds: int, timeout: int, logs_limit: int, errors_limit: int, errors_time_range: str, version_message: str = "", start_page: str = "dashboard", ) -> int: try: from rich.text import Text from textual.app import App, ComposeResult from textual.screen import Screen from textual.widgets import DataTable, Footer, Header, Input, Static except ImportError: print("Textual is required. Run with: uv run --with textual python sub2api_quota_tui.py", file=sys.stderr) return 2 class DashboardScreen(Screen[None]): AUTO_FOCUS = "#accounts" BINDINGS = [ ("r", "refresh", "Refresh"), ("/", "focus_filter", "Filter"), ("a", "focus_accounts", "Accounts"), ("k", "focus_keys", "Keys"), ("l", "focus_logs", "Logs"), ("e", "focus_errors", "Errors"), ] def __init__(self) -> None: super().__init__(classes="dashboard") self.accounts_payload: dict[str, Any] = {} self.status_payload: dict[str, Any] = {} self.monitor_error = "" self.accounts_error = "" self.logs_payload: dict[str, Any] = {} self.logs_error = "" self.errors_payload: dict[str, Any] = {} self.errors_error = "" self.keys_payload: dict[str, Any] = {} self.keys_error = "" self.account_rows: list[dict[str, Any]] = [] self.key_rows: list[dict[str, Any]] = [] self.log_rows: list[dict[str, Any]] = [] self.error_rows: list[dict[str, Any]] = [] self.account_by_key: dict[str, dict[str, Any]] = {} self.key_by_key: dict[str, dict[str, Any]] = {} self.log_by_key: dict[str, dict[str, Any]] = {} self.error_by_key: dict[str, dict[str, Any]] = {} self.active_table = "accounts" def compose(self) -> ComposeResult: yield Header(show_clock=True) yield Static("", id="summary") yield Input(placeholder="filter", id="filter") yield DataTable(id="accounts") yield DataTable(id="keys") yield DataTable(id="logs") yield DataTable(id="errors") yield Static("", id="detail") yield Static("", id="status") yield Footer() @staticmethod def configure_table(table: DataTable, columns: tuple[str, ...]) -> None: table.cursor_type = "row" table.zebra_stripes = True # Content-sized columns preserve complete values on wide terminals; # narrow terminals use DataTable's horizontal scrolling instead. for label in columns: table.add_column(label) def on_mount(self) -> None: self.configure_table( self.query_one("#accounts", DataTable), ("ACCOUNT", "Group", "Today", "Daily", "5h", "7d", "Avail"), ) self.configure_table( self.query_one("#keys", DataTable), ("KEY", "Today", "Tokens", "Req"), ) self.configure_table( self.query_one("#logs", DataTable), ( "LOG KEY", "Account", "Model", "First", "Duration", "Tok/s", "Input", "Output", "Cache", "Tokens", "Cost", "Time", "Age", ), ) self.configure_table( self.query_one("#errors", DataTable), ("ERR", "Status", "Key", "Account", "Model", "Time", "Age"), ) self.refresh_all(refresh=True) self.set_interval(refresh_seconds, self.refresh_accounts) self.set_interval(logs_refresh_seconds, self.refresh_logs) self.set_interval(errors_refresh_seconds, self.refresh_errors) self.focus_table("accounts") def action_refresh(self) -> None: self.refresh_all(refresh=True) def action_focus_filter(self) -> None: self.query_one("#filter", Input).focus() def action_focus_accounts(self) -> None: self.focus_table("accounts") def action_focus_keys(self) -> None: table = self.query_one("#keys", DataTable) if table.display: self.focus_table("keys") def action_focus_logs(self) -> None: self.focus_table("logs") def action_focus_errors(self) -> None: self.focus_table("errors") def focus_table(self, table_id: str) -> None: self.active_table = table_id self.query_one(f"#{table_id}", DataTable).focus() self.render_active_detail() def on_input_changed(self, event: Input.Changed) -> None: if event.input.id == "filter": self.render_tables() def on_input_submitted(self, event: Input.Submitted) -> None: if event.input.id == "filter": self.focus_table(self.active_table) def refresh_all(self, refresh: bool = False) -> None: self.query_one("#status", Static).update("refreshing accounts, keys, logs, and errors...") self.refresh_accounts(refresh=refresh) self.refresh_logs() self.refresh_errors() def refresh_accounts(self, refresh: bool = False) -> None: self.accounts_error = "" try: self.accounts_payload = fetch_payload(api_url, timeout, refresh=refresh) self.status_payload, self.monitor_error = fetch_optional_payload(status_url, timeout) except Exception as exc: self.accounts_error = str(exc) self.render_accounts() self.render_meta() def refresh_logs(self) -> None: self.logs_error = "" if not str(logs_token or "").strip(): self.logs_payload = {} self.logs_error = "admin token not configured" else: try: self.logs_payload = fetch_logs_payload(logs_url, logs_token, timeout, logs_limit) except Exception as exc: self.logs_error = str(exc) self.render_logs() self.refresh_keys() self.render_meta() def refresh_keys(self) -> None: self.keys_error = "" if 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) except Exception as exc: self.keys_payload = {} self.keys_error = str(exc) self.render_keys() def refresh_errors(self) -> None: self.errors_error = "" if not str(logs_token or "").strip(): self.errors_payload = {} self.errors_error = "admin token not configured" else: try: self.errors_payload = fetch_merged_errors( default_error_sources(errors_cn_url, errors_us_url), logs_token, timeout, errors_limit, errors_time_range, ) except Exception as exc: self.errors_error = str(exc) self.render_errors() self.render_meta() def render_tables(self) -> None: self.render_accounts() self.render_keys() self.render_logs() self.render_errors() self.render_meta() self.render_active_detail() def render_accounts(self) -> None: filter_text = self.query_one("#filter", Input).value self.account_rows = normalize_account_rows(self.accounts_payload, filter_text) table = self.query_one("#accounts", DataTable) table.clear() self.account_by_key = {} for index, row in enumerate(self.account_rows): key = f"account-{row['id']}-{index}" self.account_by_key[key] = row table.add_row( row["name"], row["routing_group"], format_cost(row["today_cost_usd"]), row["daily_quota_cell"], row["five_hour"], row["weekly"], monitor_availability(row, self.status_payload), key=key, ) def render_keys(self) -> None: filter_text = self.query_one("#filter", Input).value self.key_rows = normalize_key_rows(self.keys_payload, filter_text) table = self.query_one("#keys", DataTable) table.clear() self.key_by_key = {} for index, row in enumerate(self.key_rows): key = f"key-{row['id']}-{index}" self.key_by_key[key] = row color = key_color(row["name"]) table.add_row( Text(str(row["name"]), style=color) if color else str(row["name"]), format_cost(row["cost"]), format_count(row["tokens"]), format_count(row["requests"]), key=key, ) table.display = bool(self.key_rows) and self.app.size.height >= 20 def render_logs(self) -> None: filter_text = self.query_one("#filter", Input).value self.log_rows = normalize_log_rows(self.logs_payload, filter_text) table = self.query_one("#logs", DataTable) table.clear() self.log_by_key = {} for index, row in enumerate(self.log_rows): key = f"log-{row['id']}-{index}" self.log_by_key[key] = row color = key_color(row["key"]) table.add_row( Text(row["key"], style=color) if color else row["key"], row["account"], row["model"], format_seconds(row["first_token_ms"]), format_seconds(row["duration_ms"]), format_rate(row["tokens_per_second"]), format_count(row["input_tokens"]), format_count(row["output_tokens"]), format_count(row["cache_tokens"]), format_count(row["tokens"]), format_cost(row["cost"]), row["time"], row["age"], key=key, ) def render_errors(self) -> None: filter_text = self.query_one("#filter", Input).value self.error_rows = normalize_error_rows(self.errors_payload, filter_text) table = self.query_one("#errors", DataTable) table.clear() self.error_by_key = {} for index, row in enumerate(self.error_rows): key = f"error-{row['node']}-{row['id']}-{index}" self.error_by_key[key] = row table.add_row( row["node"], str(row["status_code"]), row["key"], row["account"], row["model"], row["time"], row["age"], key=key, ) def errors_total(self) -> int: sources = self.errors_payload.get("sources") if isinstance(self.errors_payload.get("sources"), dict) else {} totals = [ as_int(info.get("total")) for info in sources.values() if isinstance(info, dict) and info.get("ok") ] return sum(totals) if totals else len(self.errors_payload.get("items") or []) def render_meta(self) -> None: totals = self.accounts_payload.get("totals") if isinstance(self.accounts_payload.get("totals"), dict) else {} account_total = as_int(totals.get("total_accounts")) or len(self.accounts_payload.get("accounts") or []) usable = as_int(totals.get("usable_accounts")) logs_total = as_int(logs_envelope(self.logs_payload).get("total")) if not logs_total: logs_total = len(log_items(self.logs_payload)) key_summary = "keys -" if self.key_rows: top_key = self.key_rows[0] key_summary = f"keys {top_key['name']} {format_cost(top_key['cost'])}" summary = Text() summary.append( f"accounts {len(self.account_rows)}/{account_total} ({usable} usable, {format_cost(totals.get('today_cost_usd'))})", style="green", ) 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(" | ", 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 self.accounts_error: status_bits.append(f"accounts: {self.accounts_error}") else: status_bits.extend( ( monitor_summary(self.status_payload, self.monitor_error), f"source {self.accounts_payload.get('source_name') or '-'}", ) ) shared_admin_error = self.logs_error and self.logs_error == self.errors_error if shared_admin_error: status_bits.append(f"logs/errors: {self.logs_error}") else: if self.logs_error: status_bits.append(f"logs: {self.logs_error}") if self.errors_error: status_bits.append(f"errors: {self.errors_error}") if self.keys_error and self.keys_error != self.logs_error: status_bits.append(f"keys: {self.keys_error}") if not self.errors_error and self.errors_payload: sources = self.errors_payload.get("sources") if isinstance(self.errors_payload.get("sources"), dict) else {} source_bits = [] for node in ("cn", "us"): info = sources.get(node) if isinstance(sources.get(node), dict) else {} if info: source_bits.append(f"{node} {'ok' if info.get('ok') else 'err'}") if source_bits: status_bits.append("errors " + "/".join(source_bits)) self.query_one("#status", Static).update(" | ".join(bit for bit in status_bits if bit) or "ready") def render_active_detail(self) -> None: rows_by_table = { "accounts": self.account_rows, "keys": self.key_rows, "logs": self.log_rows, "errors": self.error_rows, } rows = rows_by_table[self.active_table] if not rows: self.query_one("#detail", Static).update(f"no {self.active_table}") return self.render_detail(self.active_table, rows[0]) def render_detail(self, table_id: str, row: dict[str, Any]) -> None: if table_id == "accounts": detail = ( f"{row['name']} | {row['provider']} | group {row['routing_group']} | {row['kind_label']} | " f"daily {row['daily_quota_cell']} | today {format_cost(row['today_cost_usd'])}, " f"{format_count(row['today_tokens'])} tokens, {format_count(row['today_requests'])} req | " f"{monitor_detail(row, self.status_payload)}" ) error = str(row.get("error") or "").strip() if error: detail += f" | {error}" elif table_id == "keys": detail = ( f"{row['name']} | today {format_cost(row['cost'])} | " f"{format_count(row['tokens'])} tokens | {format_count(row['requests'])} req" ) elif table_id == "logs": detail = log_detail_line(row) else: detail = error_detail_line(row) self.query_one("#detail", Static).update(detail) def on_data_table_row_highlighted(self, event: DataTable.RowHighlighted) -> None: table_id = str(event.control.id or "") if table_id not in {"accounts", "keys", "logs", "errors"}: return if event.control.has_focus: self.active_table = table_id if table_id != self.active_table: return row_maps = { "accounts": self.account_by_key, "keys": self.key_by_key, "logs": self.log_by_key, "errors": self.error_by_key, } row = row_maps[table_id].get(str(event.row_key.value)) if row: self.render_detail(table_id, row) class AccountsScreen(Screen[None]): AUTO_FOCUS = "#accounts" BINDINGS = [ ("r", "refresh", "Refresh"), ("/", "focus_filter", "Filter"), ("l", "show_logs", "Logs"), ("e", "show_errors", "Errors"), ] def __init__(self) -> None: super().__init__() self.payload: dict[str, Any] = {} self.status_payload: dict[str, Any] = {} self.status_error = "" self.keys_payload: dict[str, Any] = {} self.keys_error = "" self.rows: list[dict[str, Any]] = [] self.row_by_key: dict[str, dict[str, Any]] = {} def compose(self) -> ComposeResult: yield Header(show_clock=True) yield Static("", id="summary") yield Input(placeholder="filter", id="filter") yield DataTable(id="accounts") yield DataTable(id="keys") yield Static("", id="detail") yield Static("", id="status") yield Footer() def on_mount(self) -> None: table = self.query_one("#accounts", DataTable) table.cursor_type = "row" table.zebra_stripes = True table.add_columns("Name", "Provider", "Group", "Daily", "Today", "Tokens", "Req", "Kind", "5h", "7d", "Reset", "Status", "Availability") keys_table = self.query_one("#keys", DataTable) keys_table.cursor_type = "row" keys_table.zebra_stripes = True keys_table.add_columns("Key", "Today", "Tokens", "Req") self.refresh_data(refresh=True) self.set_interval(refresh_seconds, self.refresh_data) def action_refresh(self) -> None: self.refresh_data(refresh=True) def action_focus_filter(self) -> None: self.query_one("#filter", Input).focus() def action_show_logs(self) -> None: self.app.switch_screen(LogsScreen()) def action_show_errors(self) -> None: self.app.switch_screen(ErrorsScreen()) def on_input_changed(self, event: Input.Changed) -> None: if event.input.id == "filter": self.render_payload() def on_input_submitted(self, event: Input.Submitted) -> None: if event.input.id == "filter": self.query_one("#accounts", DataTable).focus() def refresh_data(self, refresh: bool = False) -> None: 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.refresh_keys() self.render_payload() self.render_keys() status_bits = [version_message, monitor_summary(self.status_payload, self.status_error), f"source {self.payload.get('source_name') or '-'}"] if self.keys_error: status_bits.append(f"keys: {self.keys_error}") status.update(" | ".join(bit for bit in status_bits if bit) + f" | {api_url}") except Exception as exc: status.update(f"error: {exc}") def refresh_keys(self) -> None: self.keys_error = "" self.keys_payload = {} if not str(logs_token or "").strip(): self.keys_error = "logs token not configured" return try: self.keys_payload = fetch_key_usage_payload(logs_url, logs_token, timeout) except Exception as exc: self.keys_error = str(exc) def render_keys(self) -> None: table = self.query_one("#keys", DataTable) table.clear() rows = normalize_key_rows(self.keys_payload) table.display = bool(rows) for row in rows: color = key_color(row["name"]) table.add_row( Text(str(row["name"]), style=color) if color else str(row["name"]), format_cost(row["cost"]), format_count(row["tokens"]), format_count(row["requests"]), ) def render_payload(self) -> None: filter_text = self.query_one("#filter", Input).value self.rows = normalize_account_rows(self.payload, filter_text) table = self.query_one("#accounts", DataTable) table.clear() self.row_by_key = {} for row in self.rows: key = str(row["id"]) self.row_by_key[key] = row table.add_row( row["name"], row["provider"], row["routing_group"], row["daily_quota_cell"], format_cost(row["today_cost_usd"]), format_count(row["today_tokens"]), format_count(row["today_requests"]), row["kind_label"], row["five_hour"], row["weekly"], row["reset"], row["status"], monitor_availability(row, self.status_payload), key=key, ) self.query_one("#summary", Static).update(summary_line(self.payload)) if self.rows: self.render_detail(self.rows[0]) else: self.query_one("#detail", Static).update("no accounts") def on_data_table_row_highlighted(self, event: DataTable.RowHighlighted) -> None: key = str(event.row_key.value) row = self.row_by_key.get(key) if row: self.render_detail(row) def render_detail(self, row: dict[str, Any]) -> None: raw = row.get("raw") if isinstance(row.get("raw"), dict) else {} detail = ( f"{row['name']} | {row['provider']} | group {row['routing_group']} | {row['kind_label']} | {row['account_type']} | " f"daily {row['daily_quota_cell']} ({format_percent(row['daily_quota_used_percent'])} used, " f"{row['daily_quota_remaining'] if row['daily_quota_remaining'] is not None else '-'} left) | " f"today {format_cost(row['today_cost_usd'])}, {format_count(row['today_tokens'])} tokens, " f"{format_count(row['today_requests'])} req | latest {row['latest_usage_at']} | " f"priority {row['priority'] if row['priority'] is not None else '-'} | " f"{monitor_detail(row, self.status_payload)}" ) error = str(raw.get("error") or "").strip() if error: detail += f" | {error}" self.query_one("#detail", Static).update(detail) class LogsScreen(Screen[None]): AUTO_FOCUS = "#logs" BINDINGS = [ ("r", "refresh", "Refresh"), ("/", "focus_filter", "Filter"), ("a", "show_accounts", "Accounts"), ("e", "show_errors", "Errors"), ] def __init__(self) -> None: super().__init__() self.payload: dict[str, Any] = {} self.rows: list[dict[str, Any]] = [] self.row_by_key: dict[str, dict[str, Any]] = {} def compose(self) -> ComposeResult: yield Header(show_clock=True) yield Static("", id="summary") yield Input(placeholder="filter", id="filter") yield DataTable(id="logs") yield Static("", id="detail") yield Static("", id="status") yield Footer() def on_mount(self) -> None: table = self.query_one("#logs", DataTable) table.cursor_type = "row" table.zebra_stripes = True table.add_columns( "Key", "Account", "Model", "Effort", "Type", "Input", "Output", "Cache", "Tokens", "Cost", "First", "Duration", "Tok/s", "Time", "Age", ) self.refresh_data() self.set_interval(logs_refresh_seconds, self.refresh_data) def action_refresh(self) -> None: self.refresh_data() def action_focus_filter(self) -> None: self.query_one("#filter", Input).focus() def action_show_accounts(self) -> None: self.app.switch_screen(AccountsScreen()) def action_show_errors(self) -> None: self.app.switch_screen(ErrorsScreen()) def on_input_changed(self, event: Input.Changed) -> None: if event.input.id == "filter": self.render_payload() def on_input_submitted(self, event: Input.Submitted) -> None: if event.input.id == "filter": self.query_one("#logs", DataTable).focus() def refresh_data(self) -> None: status = self.query_one("#status", Static) if not str(logs_token or "").strip(): status.update(logs_token_hint()) return status.update("refreshing logs...") try: self.payload = fetch_logs_payload(logs_url, logs_token, timeout, logs_limit) self.render_payload() status_bits = [version_message, f"latest {logs_limit} requests | {logs_url}"] status.update(" | ".join(bit for bit in status_bits if bit)) except Exception as exc: status.update(f"logs error: {exc}") def render_payload(self) -> None: filter_text = self.query_one("#filter", Input).value self.rows = normalize_log_rows(self.payload, filter_text) table = self.query_one("#logs", DataTable) table.clear() self.row_by_key = {} for index, row in enumerate(self.rows): key = f"{row['id']}-{index}" self.row_by_key[key] = row color = key_color(row["key"]) table.add_row( Text(row["key"], style=color) if color else row["key"], row["account"], row["model"], row["effort"], row["type"], format_count(row["input_tokens"]), format_count(row["output_tokens"]), format_count(row["cache_tokens"]), format_count(row["tokens"]), format_cost(row["cost"]), format_seconds(row["first_token_ms"]), format_seconds(row["duration_ms"]), format_rate(row["tokens_per_second"]), row["time"], row["age"], key=key, ) self.query_one("#summary", Static).update(logs_summary_line(self.payload, len(self.rows))) if self.rows: self.query_one("#detail", Static).update(log_detail_line(self.rows[0])) else: self.query_one("#detail", Static).update("no request logs") def on_data_table_row_highlighted(self, event: DataTable.RowHighlighted) -> None: key = str(event.row_key.value) row = self.row_by_key.get(key) if row: self.query_one("#detail", Static).update(log_detail_line(row)) class ErrorsScreen(Screen[None]): AUTO_FOCUS = "#errors" BINDINGS = [ ("r", "refresh", "Refresh"), ("/", "focus_filter", "Filter"), ("a", "show_accounts", "Accounts"), ("l", "show_logs", "Logs"), ] def __init__(self) -> None: super().__init__() self.payload: dict[str, Any] = {} self.rows: list[dict[str, Any]] = [] self.row_by_key: dict[str, dict[str, Any]] = {} def compose(self) -> ComposeResult: yield Header(show_clock=True) yield Static("", id="summary") yield Input(placeholder="filter", id="filter") yield DataTable(id="errors") yield Static("", id="detail") yield Static("", id="status") yield Footer() def on_mount(self) -> None: 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") self.refresh_data() self.set_interval(errors_refresh_seconds, self.refresh_data) def action_refresh(self) -> None: self.refresh_data() def action_focus_filter(self) -> None: self.query_one("#filter", Input).focus() def action_show_accounts(self) -> None: self.app.switch_screen(AccountsScreen()) def action_show_logs(self) -> None: self.app.switch_screen(LogsScreen()) def on_input_changed(self, event: Input.Changed) -> None: if event.input.id == "filter": self.render_payload() def on_input_submitted(self, event: Input.Submitted) -> None: if event.input.id == "filter": self.query_one("#errors", DataTable).focus() def refresh_data(self) -> None: status = self.query_one("#status", Static) if not str(logs_token or "").strip(): status.update(logs_token_hint()) return status.update("refreshing errors from cn + us...") try: self.payload = fetch_merged_errors( default_error_sources(errors_cn_url, errors_us_url), logs_token, timeout, errors_limit, errors_time_range, ) self.render_payload() sources = self.payload.get("sources") if isinstance(self.payload.get("sources"), dict) else {} bits = [] for node in ("cn", "us"): info = sources.get(node) if isinstance(sources.get(node), dict) else {} if not info: continue if info.get("ok"): bits.append(f"{node} ok") else: bits.append(f"{node} {info.get('error') or 'error'}") status_bits = [version_message, f"every {errors_refresh_seconds}s", " | ".join(bits)] status.update(" | ".join(bit for bit in status_bits if bit)) except Exception as exc: status.update(f"errors error: {exc}") def render_payload(self) -> None: filter_text = self.query_one("#filter", Input).value self.rows = normalize_error_rows(self.payload, filter_text) table = self.query_one("#errors", DataTable) table.clear() self.row_by_key = {} for index, row in enumerate(self.rows): key = f"{row['node']}-{row['id']}-{index}" self.row_by_key[key] = row table.add_row( row["node"], str(row["status_code"]), row["key"], row["account"], row["model"], row["phase"], row["type"], row["owner"], row["time"], row["age"], key=key, ) self.query_one("#summary", Static).update(errors_summary_line(self.payload, len(self.rows))) if self.rows: self.query_one("#detail", Static).update(error_detail_line(self.rows[0])) else: self.query_one("#detail", Static).update("no error logs") def on_data_table_row_highlighted(self, event: DataTable.RowHighlighted) -> None: key = str(event.row_key.value) row = self.row_by_key.get(key) if row: self.query_one("#detail", Static).update(error_detail_line(row)) class Sub2APIQuotaApp(App[None]): CSS = """ #summary { height: 1; padding: 0 1; color: $foreground; } #filter { height: 1; border: none; padding: 0 1; } #filter:focus { border: none; } #accounts { height: 1fr; min-height: 3; } #keys { height: auto; max-height: 6; } .dashboard #accounts { background: $success 5%; } .dashboard #accounts > .datatable--header { background: $success 35%; color: $foreground; } .dashboard #keys { height: 1fr; min-height: 0; max-height: 100%; background: $warning 5%; } .dashboard #keys > .datatable--header { background: $warning 35%; color: $foreground; } #logs { height: 1fr; min-height: 3; } .dashboard #logs { background: $secondary 5%; } .dashboard #logs > .datatable--header { background: $secondary 35%; color: $foreground; } #errors { height: 1fr; min-height: 3; } .dashboard #errors { background: $error 5%; } .dashboard #errors > .datatable--header { background: $error 35%; color: $foreground; } #detail { height: 2; padding: 0 1; background: $surface-lighten-1; } #status { height: 1; padding: 0 1; color: $text-muted; } """ BINDINGS = [ ("q", "quit", "Quit"), ] def on_mount(self) -> None: if start_page == "accounts": self.push_screen(AccountsScreen()) elif start_page == "logs": self.push_screen(LogsScreen()) elif start_page == "errors": self.push_screen(ErrorsScreen()) else: self.push_screen(DashboardScreen()) Sub2APIQuotaApp().run() return 0 def logs_token_hint() -> str: return ( "logs token not configured: set SHUSUB2_LOGS_TOKEN or write the Sub2API admin API key to " f"{logs_token_config_file_path()}" ) def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Sub2API quota and daily usage TUI") 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("--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("--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: server4 fixed origin /api/v1/admin/ops/errors; CF sub2apius often blocks non-browser clients)", ) parser.add_argument("--save-config", action="store_true", help="persist --api-url/--logs-url/--logs-token/--errors-*-url 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( "--version-check-timeout", type=int, default=env_int("SHUSUB2_VERSION_CHECK_TIMEOUT", DEFAULT_VERSION_CHECK_TIMEOUT_SECONDS), ) parser.add_argument("--no-version-check", action="store_true", default=os.environ.get("SHUSUB2_NO_VERSION_CHECK", "").strip().lower() in {"1", "true", "yes", "on"}) parser.add_argument( "--refresh-seconds", type=int, default=env_int("SUB2API_QUOTA_TUI_REFRESH_SECONDS", DEFAULT_REFRESH_SECONDS), ) parser.add_argument( "--logs-refresh-seconds", type=int, default=env_int("SHUSUB2_LOGS_REFRESH_SECONDS", DEFAULT_LOGS_REFRESH_SECONDS), ) parser.add_argument( "--errors-refresh-seconds", type=int, default=env_int("SHUSUB2_ERRORS_REFRESH_SECONDS", DEFAULT_ERRORS_REFRESH_SECONDS), ) parser.add_argument( "--logs-limit", type=int, default=env_int("SHUSUB2_LOGS_LIMIT", DEFAULT_LOGS_LIMIT), ) parser.add_argument( "--errors-limit", type=int, default=env_int("SHUSUB2_ERRORS_LIMIT", DEFAULT_ERRORS_LIMIT), ) 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)", ) 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("--logs", action="store_true", help="start on the dedicated request logs page (with --once: print a logs snapshot)") parser.add_argument("--errors", action="store_true", help="start on the dedicated merged cn+us errors page (with --once: print an errors snapshot)") parser.add_argument("--filter", default="", help="initial filter for --once output") return parser 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) 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.logs, args.errors)) if selected_pages > 1: print("choose only one of --accounts, --logs, or --errors", file=sys.stderr) return 2 if args.save_config or args.install: config_path = write_api_url_config(args.api_url) print(f"saved api url to {config_path}") 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) 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) print(f"saved errors cn url to {errors_cn_path}") if str(args.errors_us_url or "").strip(): errors_us_path = write_config_value(errors_us_url_config_file_path(), args.errors_us_url) print(f"saved errors us url to {errors_us_path}") if args.install: print("installing shusub2 with uv tool...") return run_install_command() version_message = check_version_update( args.version_check_url, max(1, args.version_check_timeout), disabled=bool(args.no_version_check), ) if args.once: if version_message: print(version_message) if args.logs: if not str(args.logs_token or "").strip(): 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) return 0 if args.errors: if not str(args.logs_token or "").strip(): 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, args.timeout, errors_limit, errors_time_range, ), args.filter, ) return 0 status_payload, status_error = fetch_optional_payload(status_url, args.timeout) print_once(fetch_payload(args.api_url, args.timeout, refresh=True), args.filter, status_payload, status_error) if str(args.logs_token or "").strip(): print() try: print_keys_once(fetch_key_usage_payload(args.logs_url, args.logs_token, args.timeout)) except Exception as exc: print(f"keys error: {exc}", file=sys.stderr) return 0 start_page = "dashboard" if args.accounts: start_page = "accounts" elif args.logs: start_page = "logs" elif args.errors: start_page = "errors" return run_textual( args.api_url, status_url, args.logs_url, args.logs_token, args.errors_cn_url, args.errors_us_url, max(1, args.refresh_seconds), max(1, args.logs_refresh_seconds), max(1, args.errors_refresh_seconds), max(1, args.timeout), logs_limit, errors_limit, errors_time_range, version_message, start_page=start_page, ) if __name__ == "__main__": raise SystemExit(main())