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

feat: infer status url and check updates

Browse Source
This commit is contained in:
yunyaozhou
2026-06-09 16:45:17 +08:00
parent 3b9d092c06
commit 01aa4a6012
5 changed files with 123 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+13 -6
View File
@@ -9,25 +9,23 @@ Run once from a public Git repo with `uvx`:
```bash
uvx --from git+https://gitea.shujk.top/shujakuin/shusub2.git shusub2 \
--api-url https://example.com/api/tui/accounts
--api-url https://codex.server2.shujk.top/1232131231313123/api/tui/accounts
```
Install as a user command:
```bash
uv tool install git+https://gitea.shujk.top/shujakuin/shusub2.git
shusub2 --api-url https://example.com/api/tui/accounts
shusub2 --api-url https://codex.server2.shujk.top/1232131231313123/api/tui/accounts
```
Configure default API and optional status URLs for `shusub2`:
Configure the default public API URL for `shusub2`:
```bash
mkdir -p ~/.config/shusub2
chmod 700 ~/.config/shusub2
printf '%s\n' 'https://example.com/api/tui/accounts' > ~/.config/shusub2/api-url
printf '%s\n' 'https://example.com/api/status' > ~/.config/shusub2/status-url
printf '%s\n' 'https://codex.server2.shujk.top/1232131231313123/api/tui/accounts' > ~/.config/shusub2/api-url
chmod 600 ~/.config/shusub2/api-url
chmod 600 ~/.config/shusub2/status-url
shusub2
```
@@ -38,6 +36,9 @@ Environment variables override the config file:
- `SHUSUB2_API_URL_FILE`
- `SHUSUB2_STATUS_URL`
- `SHUSUB2_STATUS_URL_FILE`
- `SHUSUB2_VERSION_CHECK_URL`
- `SHUSUB2_VERSION_CHECK_TIMEOUT`
- `SHUSUB2_NO_VERSION_CHECK`
## Local Development
@@ -64,6 +65,12 @@ The TUI reads `/api/tui/accounts` and can optionally read `sub2api-status`
`/api/status` for channel monitor health. It does not need SSH, database
access, API keys, OAuth credentials, or plaintext env files.
When `--status-url` is omitted, `shusub2` infers a sibling `/api/status` URL
from account URLs ending in `/api/tui/accounts`, so the public Codex endpoint
automatically enables monitor availability. On startup it also checks the
public Gitea repo for a newer package version and prints a short upgrade hint
when one is available.
Columns are ordered for scanning inside zellij:
```text
pyproject.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "shusub2"
version = "0.1.5"
version = "0.1.6"
description = "Terminal UI for Sub2API account quota and daily usage"
readme = "README.md"
requires-python = ">=3.11"
sub2api_quota_tui.py
+91 -4
View File
@@ -4,8 +4,10 @@ from __future__ import annotations
import argparse
import datetime as dt
import importlib.metadata
import json
import os
import re
import sys
import urllib.parse
import urllib.request
@@ -13,14 +15,19 @@ from pathlib import Path
from typing import Any
APP_NAME = "shusub2"
FALLBACK_VERSION = "0.1.6"
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_VERSION_CHECK_URL = "https://gitea.shujk.top/shujakuin/shusub2/raw/branch/main/pyproject.toml"
DEFAULT_REFRESH_SECONDS = 60
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"
def env_int(name: str, default: int, *, minimum: int = 1) -> int:
@@ -61,6 +68,71 @@ def default_status_url() -> str:
)
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() == "":
@@ -422,7 +494,7 @@ def print_once(payload: dict[str, Any], filter_text: str = "", status_payload: d
)
def run_textual(api_url: str, status_url: str, refresh_seconds: int, timeout: int) -> int:
def run_textual(api_url: str, status_url: str, refresh_seconds: int, timeout: int, version_message: str = "") -> int:
try:
from textual.app import App, ComposeResult
from textual.widgets import DataTable, Footer, Header, Input, Static
@@ -486,7 +558,7 @@ def run_textual(api_url: str, status_url: str, refresh_seconds: int, timeout: in
self.payload = fetch_payload(api_url, timeout, refresh=refresh)
self.status_payload, self.status_error = fetch_optional_payload(status_url, timeout)
self.render_payload()
status_bits = [monitor_summary(self.status_payload, self.status_error), f"source {self.payload.get('source_name') or '-'}"]
status_bits = [version_message, monitor_summary(self.status_payload, self.status_error), f"source {self.payload.get('source_name') or '-'}"]
status.update(" | ".join(bit for bit in status_bits if bit) + f" | {api_url}")
except Exception as exc:
status.update(f"error: {exc}")
@@ -552,6 +624,13 @@ 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("--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,
@@ -565,11 +644,19 @@ def build_parser() -> argparse.ArgumentParser:
def main(argv: list[str] | None = None) -> int:
args = build_parser().parse_args(argv)
status_url = str(args.status_url or "").strip() or inferred_status_url(args.api_url)
version_message = check_version_update(
args.version_check_url,
max(1, args.version_check_timeout),
disabled=bool(args.no_version_check),
)
if args.once:
status_payload, status_error = fetch_optional_payload(args.status_url, args.timeout)
if version_message:
print(version_message)
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)
return 0
return run_textual(args.api_url, args.status_url, max(1, args.refresh_seconds), max(1, args.timeout))
return run_textual(args.api_url, status_url, max(1, args.refresh_seconds), max(1, args.timeout), version_message)
if __name__ == "__main__":
tests/test_payload.py
+17
View File
@@ -86,6 +86,23 @@ class Sub2APIQuotaTUITests(unittest.TestCase):
self.assertEqual([row["name"] for row in mod.normalize_account_rows(payload, "sfast")], ["beta"])
self.assertEqual([row["name"] for row in mod.normalize_account_rows(payload, "anthropic")], ["alpha"])
def test_infers_public_status_url_from_accounts_url(self) -> None:
mod = load_module()
self.assertEqual(
mod.inferred_status_url("https://codex.server2.shujk.top/1232131231313123/api/tui/accounts"),
"https://codex.server2.shujk.top/1232131231313123/api/status",
)
self.assertEqual(mod.inferred_status_url("https://example.com/nope"), "")
def test_version_update_message_only_for_newer_versions(self) -> None:
mod = load_module()
self.assertEqual(mod.latest_version_from_text('name = "shusub2"\nversion = "0.1.7"\n'), "0.1.7")
self.assertIn("0.1.6 -> 0.1.7", mod.version_update_message("0.1.7", "0.1.6"))
self.assertEqual(mod.version_update_message("0.1.6", "0.1.6"), "")
self.assertEqual(mod.version_update_message("0.1.5", "0.1.6"), "")
def test_once_output_is_name_first_and_includes_daily_quota(self) -> None:
mod = load_module()
payload = {
uv.lock
Generated
+1 -1
View File
@@ -85,7 +85,7 @@ wheels = [
[[package]]
name = "shusub2"
version = "0.1.5"
version = "0.1.6"
source = { editable = "." }
dependencies = [
{ name = "textual" },