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

feat: compress shusub2 polling

Browse Source
This commit is contained in:
yunyaozhou
2026-07-27 21:39:04 +08:00
parent 4d8d117fb1
commit 162ce03c4e
5 changed files with 81 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+7 -5
View File
@@ -83,9 +83,11 @@ Environment variables override the config file:
`shusub2` opens a single dashboard with Accounts, Keys, Logs, and Errors
tables stacked vertically. Press `a`, `k`, `l`, or `e` to focus a table, `/` to
filter all four tables, and `r` to refresh all data. The two-line detail area
follows the selected row. Accounts, Keys, Logs, and Errors use green, yellow,
magenta, and red section styling respectively.
filter all four tables, and `r` to refresh all data immediately. Automatic Accounts,
Keys/Logs, and Errors refresh defaults to every five minutes. The client requests
gzip-compressed JSON and transparently decodes it when the upstream supports it.
The two-line detail area follows the selected row. Accounts, Keys, Logs, and
Errors use green, yellow, magenta, and red section styling respectively.
The dashboard sizes every column from the fetched content so wide terminals show
complete keys, accounts, and models. On a narrow terminal, a table scrolls
@@ -112,7 +114,7 @@ The middle dashboard table shows request logs, and `l` focuses it.
`shusub2 --once --logs` prints one logs snapshot to stdout.
The logs page reads the latest requests (default 100, `--logs-limit`) from the
Sub2API admin usage API and refreshes every 60 seconds by default
Sub2API admin usage API and refreshes every five minutes by default
(`--logs-refresh-seconds` / `SHUSUB2_LOGS_REFRESH_SECONDS`). Columns:
```text
@@ -168,7 +170,7 @@ parallel, then merges them by `created_at`:
backend, but Cloudflare often returns 1010 for non-browser clients, so the TUI
defaults to the fixed server4 origin while still labeling rows as `us`.
The dashboard and dedicated errors page refresh every 60 seconds by default
The dashboard and dedicated errors page refresh every five minutes by default
(`--errors-refresh-seconds` / `SHUSUB2_ERRORS_REFRESH_SECONDS`). Query window
defaults to `24h` (`--errors-time-range` / `SHUSUB2_ERRORS_TIME_RANGE`); each source uses
`page=1` and `page_size` from `--errors-limit` (default 100, max 500).
pyproject.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "shusub2"
version = "0.2.10"
version = "0.2.11"
description = "Terminal UI for Sub2API account quota and daily usage"
readme = "README.md"
requires-python = ">=3.11"
sub2api_quota_tui.py
+27 -24
View File
@@ -4,6 +4,7 @@ from __future__ import annotations
import argparse
import datetime as dt
import gzip
import importlib.metadata
import json
import os
@@ -19,7 +20,7 @@ from typing import Any
APP_NAME = "shusub2"
FALLBACK_VERSION = "0.2.10"
FALLBACK_VERSION = "0.2.11"
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"
@@ -33,9 +34,9 @@ DEFAULT_ERRORS_US_URL = "https://sub2api.server4.shujk.top:19857/api/v1/admin/op
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_REFRESH_SECONDS = 300
DEFAULT_LOGS_REFRESH_SECONDS = 300
DEFAULT_ERRORS_REFRESH_SECONDS = 300
DEFAULT_LOGS_LIMIT = 100
DEFAULT_ERRORS_LIMIT = 100
DEFAULT_ERRORS_TIME_RANGE = "24h"
@@ -332,15 +333,26 @@ def add_refresh_param(api_url: str, refresh: bool) -> str:
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"))
def decode_json_response(response: Any, error_message: str) -> dict[str, Any]:
raw = response.read()
content_encoding = str(response.headers.get("Content-Encoding", "")).lower()
if "gzip" in {value.strip() for value in content_encoding.split(",")}:
raw = gzip.decompress(raw)
data = json.loads(raw.decode("utf-8"))
if not isinstance(data, dict):
raise RuntimeError("API did not return a JSON object")
raise RuntimeError(error_message)
return data
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", "Accept-Encoding": "gzip"},
)
with urllib.request.urlopen(req, timeout=timeout) as response:
return decode_json_response(response, "API did not return a JSON object")
def fetch_optional_payload(url: str, timeout: int) -> tuple[dict[str, Any], str]:
if not str(url or "").strip():
return {}, ""
@@ -359,15 +371,12 @@ def logs_request_url(logs_url: str, limit: int) -> str:
def fetch_logs_payload(logs_url: str, token: str, timeout: int, limit: int = DEFAULT_LOGS_LIMIT) -> dict[str, Any]:
headers = {"Accept": "application/json"}
headers = {"Accept": "application/json", "Accept-Encoding": "gzip"}
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
return decode_json_response(response, "logs API did not return a JSON object")
def logs_envelope(payload: dict[str, Any]) -> dict[str, Any]:
@@ -465,7 +474,7 @@ def admin_api_base(logs_url: str) -> str:
def fetch_admin_json(url: str, token: str, timeout: int, body: dict[str, Any] | None = None) -> dict[str, Any]:
headers = {"Accept": "application/json"}
headers = {"Accept": "application/json", "Accept-Encoding": "gzip"}
if str(token or "").strip():
headers["x-api-key"] = str(token).strip()
data = None
@@ -474,10 +483,7 @@ def fetch_admin_json(url: str, token: str, timeout: int, body: dict[str, Any] |
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
return decode_json_response(response, "admin API did not return a JSON object")
def payload_data(payload: dict[str, Any]) -> dict[str, Any]:
@@ -683,15 +689,12 @@ def fetch_errors_payload(
limit: int = DEFAULT_ERRORS_LIMIT,
time_range: str = DEFAULT_ERRORS_TIME_RANGE,
) -> dict[str, Any]:
headers = {"Accept": "application/json"}
headers = {"Accept": "application/json", "Accept-Encoding": "gzip"}
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
return decode_json_response(response, "errors API did not return a JSON object")
def error_items(payload: dict[str, Any]) -> list[dict[str, Any]]:
tests/test_payload.py
+45
View File
@@ -105,6 +105,51 @@ class Sub2APIQuotaTUITests(unittest.TestCase):
self.assertEqual(mod.version_update_message("0.1.7", "0.1.7"), "")
self.assertEqual(mod.version_update_message("0.1.6", "0.1.7"), "")
def test_gzip_json_requests_decode_responses_and_request_compression(self) -> None:
mod = load_module()
payload = {"ok": True}
compressed = mod.gzip.compress(mod.json.dumps(payload).encode("utf-8"))
requests = []
class FakeResponse:
headers = {"Content-Encoding": "gzip"}
def __enter__(self):
return self
def __exit__(self, exc_type, exc, traceback):
return False
def read(self):
return compressed
old_urlopen = mod.urllib.request.urlopen
mod.urllib.request.urlopen = lambda request, timeout: (requests.append(request) or FakeResponse())
try:
self.assertEqual(mod.fetch_payload("https://example.test/accounts", 1), payload)
self.assertEqual(mod.fetch_logs_payload("https://example.test/usage", "test-token", 1), payload)
self.assertEqual(mod.fetch_admin_json("https://example.test/admin", "test-token", 1, body={"ids": [1]}), payload)
self.assertEqual(mod.fetch_errors_payload("https://example.test/errors", "test-token", 1), payload)
finally:
mod.urllib.request.urlopen = old_urlopen
class PlainResponse:
headers = {}
def read(self):
return mod.json.dumps(payload).encode("utf-8")
self.assertEqual(mod.decode_json_response(PlainResponse(), "plain JSON failed"), payload)
self.assertEqual(len(requests), 4)
self.assertTrue(all(request.get_header("Accept-encoding") == "gzip" for request in requests))
def test_default_refresh_intervals_are_five_minutes(self) -> None:
mod = load_module()
self.assertEqual(mod.DEFAULT_REFRESH_SECONDS, 300)
self.assertEqual(mod.DEFAULT_LOGS_REFRESH_SECONDS, 300)
self.assertEqual(mod.DEFAULT_ERRORS_REFRESH_SECONDS, 300)
def test_write_api_url_config_uses_config_file_env(self) -> None:
mod = load_module()
uv.lock
Generated
+1 -1
View File
@@ -85,7 +85,7 @@ wheels = [
[[package]]
name = "shusub2"
version = "0.2.10"
version = "0.2.11"
source = { editable = "." }
dependencies = [
{ name = "textual" },