feat: add sub2apicn request logs page

- new logs page (press l / a to switch, or start with --logs) reading the latest 100 requests from the Sub2API admin usage API on sub2apicn - columns: Key, Account, Model, Type, Tokens, Cost, Latency, Time with a per-row detail line (token buckets, actual cost, first-token latency) - refreshes every 60s by default (--logs-refresh-seconds, --logs-limit) - admin API key read from SHUSUB2_LOGS_TOKEN or ~/.config/shusub2/logs-token; --save-config now persists logs-url/logs-token too - --once --logs prints a one-shot logs snapshot Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-21 02:18:47 +08:00
parent 6e7831da7c
commit b68d801024
5 changed files with 614 additions and 32 deletions
@@ -17,12 +17,17 @@ from typing import Any


 APP_NAME = "shusub2"
-FALLBACK_VERSION = "0.1.7"
+FALLBACK_VERSION = "0.2.0"
 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"
 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_LOGS_LIMIT = 100
 DEFAULT_TIMEOUT_SECONDS = 10
 DEFAULT_VERSION_CHECK_TIMEOUT_SECONDS = 2
 MONITOR_OK_STATUSES = {"operational", "ok", "success"}
@@ -70,26 +75,55 @@ def default_status_url() -> str:
    )


+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 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")
-    path = config_file_path()
-    path.parent.mkdir(parents=True, exist_ok=True)
-    try:
-        path.parent.chmod(0o700)
-    except OSError:
-        pass
-    path.write_text(value + "\n", encoding="utf-8")
-    try:
-        path.chmod(0o600)
-    except OSError:
-        pass
-    return path
+    return write_config_value(config_file_path(), value)


 def run_install_command() -> int:
@@ -259,6 +293,166 @@ def fetch_optional_payload(url: str, timeout: int) -> tuple[dict[str, Any], str]
        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_total_tokens(item: dict[str, Any]) -> int:
+    return (
+        as_int(item.get("input_tokens"))
+        + as_int(item.get("output_tokens"))
+        + as_int(item.get("cache_creation_tokens"))
+        + as_int(item.get("cache_read_tokens"))
+    )
+
+
+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()
+        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,
+            "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")),
+            "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")),
+            "created_at": str(item.get("created_at") or ""),
+            "time": short_time(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, 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']}"
+    first_token = f"{row['first_token_ms']}ms" if row["first_token_ms"] > 0 else "-"
+    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"latency {format_latency(row['duration_ms'])} first {first_token}"
+    )
+    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                     type    tokens   cost       latency   time")
+    for row in rows:
+        print(
+            f"{row['key'][:20]:<21} "
+            f"{row['account'][:20]:<21} "
+            f"{row['model'][:24]:<25} "
+            f"{row['type']:<7} "
+            f"{format_count(row['tokens']):<8} "
+            f"{format_cost(row['cost']):<10} "
+            f"{format_latency(row['duration_ms']):<9} "
+            f"{row['time']}"
+        )
+
+
 def kind_label(kind: Any) -> str:
    return {
        "quota_limited": "quota",
@@ -526,26 +720,32 @@ 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, version_message: str = "") -> int:
+def run_textual(
+    api_url: str,
+    status_url: str,
+    logs_url: str,
+    logs_token: str,
+    refresh_seconds: int,
+    logs_refresh_seconds: int,
+    timeout: int,
+    logs_limit: int,
+    version_message: str = "",
+    start_page: str = "accounts",
+) -> int:
    try:
        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 Sub2APIQuotaApp(App[None]):
-        CSS = """
-        #summary { height: 1; padding: 0 1; color: $accent; }
-        #filter { height: 3; }
-        #accounts { height: 1fr; }
-        #detail { height: 3; padding: 0 1; border-top: solid $panel; }
-        #status { height: 1; padding: 0 1; color: $text-muted; }
-        """
+    class AccountsScreen(Screen[None]):
+        AUTO_FOCUS = "#accounts"
        BINDINGS = [
-            ("q", "quit", "Quit"),
            ("r", "refresh", "Refresh"),
            ("/", "focus_filter", "Filter"),
+            ("l", "show_logs", "Logs"),
        ]

        def __init__(self) -> None:
@@ -579,10 +779,17 @@ def run_textual(api_url: str, status_url: str, refresh_seconds: int, timeout: in
        def action_focus_filter(self) -> None:
            self.query_one("#filter", Input).focus()

+        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("#accounts", DataTable).focus()
+
        def refresh_data(self, refresh: bool = False) -> None:
            status = self.query_one("#status", Static)
            status.update("refreshing...")
@@ -648,6 +855,119 @@ def run_textual(api_url: str, status_url: str, refresh_seconds: int, timeout: in
                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"),
+        ]
+
+        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", "Type", "Tokens", "Cost", "Latency", "Time")
+            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 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 not configured: set SHUSUB2_LOGS_TOKEN or write the Sub2API admin API key to "
+                    f"{logs_token_config_file_path()}"
+                )
+                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
+                table.add_row(
+                    row["key"],
+                    row["account"],
+                    row["model"],
+                    row["type"],
+                    format_count(row["tokens"]),
+                    format_cost(row["cost"]),
+                    format_latency(row["duration_ms"]),
+                    row["time"],
+                    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 Sub2APIQuotaApp(App[None]):
+        CSS = """
+        #summary { height: 1; padding: 0 1; color: $accent; }
+        #filter { height: 3; }
+        #accounts { height: 1fr; }
+        #logs { height: 1fr; }
+        #detail { height: 3; padding: 0 1; border-top: solid $panel; }
+        #status { height: 1; padding: 0 1; color: $text-muted; }
+        """
+        BINDINGS = [
+            ("q", "quit", "Quit"),
+        ]
+
+        def on_mount(self) -> None:
+            self.push_screen(LogsScreen() if start_page == "logs" else AccountsScreen())
+
    Sub2APIQuotaApp().run()
    return 0

@@ -656,8 +976,10 @@ 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("--save-config", action="store_true", help="persist --api-url to ~/.config/shusub2/api-url before running")
-    parser.add_argument("--install", action="store_true", help="persist --api-url, install shusub2 as a uv tool, then exit")
+    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 the logs page (prefer SHUSUB2_LOGS_TOKEN or ~/.config/shusub2/logs-token)")
+    parser.add_argument("--save-config", action="store_true", help="persist --api-url/--logs-url/--logs-token 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",
@@ -670,8 +992,19 @@ def build_parser() -> argparse.ArgumentParser:
        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(
+        "--logs-limit",
+        type=int,
+        default=env_int("SHUSUB2_LOGS_LIMIT", DEFAULT_LOGS_LIMIT),
+    )
    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("--logs", action="store_true", help="start on the request logs page (with --once: print a logs snapshot)")
    parser.add_argument("--filter", default="", help="initial filter for --once output")
    return parser

@@ -679,9 +1012,16 @@ 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)
+    logs_limit = max(1, min(args.logs_limit, 1000))
    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 args.install:
        print("installing shusub2 with uv tool...")
        return run_install_command()
@@ -693,10 +1033,24 @@ def main(argv: list[str] | None = None) -> int:
    if args.once:
        if version_message:
            print(version_message)
+        if args.logs:
+            print_logs_once(fetch_logs_payload(args.logs_url, args.logs_token, args.timeout, logs_limit), 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)
        return 0
-    return run_textual(args.api_url, status_url, max(1, args.refresh_seconds), max(1, args.timeout), version_message)
+    return run_textual(
+        args.api_url,
+        status_url,
+        args.logs_url,
+        args.logs_token,
+        max(1, args.refresh_seconds),
+        max(1, args.logs_refresh_seconds),
+        max(1, args.timeout),
+        logs_limit,
+        version_message,
+        start_page="logs" if args.logs else "accounts",
+    )


 if __name__ == "__main__":