feat: add /api/keys endpoint — masked API keys with usage data (rolling/weekly/monthly)

gateway/scripts/dashboard.py

@@ -1190,6 +1190,7 @@ def api_rdp_toggle():
 # ════════════════════════════════════════════════════════════
 _USAGE_STATS_FILE = TEMP_DIR / "usage_stats.json"
 _USAGE_COLLECTOR_SCRIPT = _SCRIPT_DIR / "usage_collector.py"
+_USAGE_MONITOR_DIR = _SCRIPT_DIR / "usage_monitor"
 _usage_collector_running = False
 
 @app.route("/api/usage")
@@ -1234,6 +1235,71 @@ def api_usage_refresh():
     return jsonify({"ok": True, "message": "collection triggered (~10s), poll /api/usage shortly"})
 
 
+# ── Key Registry ──────────────────────────────────────────────
+_ACCOUNTS_FILE = _USAGE_MONITOR_DIR / "accounts.json"
+
+@app.route("/api/keys")
+def api_keys():
+    """返回所有可用 API key（masked 首尾）及各自用量。
+    合并 accounts.json（key 原文）和 usage_stats.json（用量数据）。
+    """
+    try:
+        # 读 key 原文
+        if not _ACCOUNTS_FILE.exists():
+            return jsonify({"ok": False, "error": "accounts.json not found", "keys": []})
+        with open(str(_ACCOUNTS_FILE), "r", encoding="utf-8") as f:
+            accounts_data = json.load(f)
+        accounts = accounts_data.get("accounts", [])
+
+        # 读用量数据
+        usage_by_kid = {}
+        if _USAGE_STATS_FILE.exists():
+            with open(str(_USAGE_STATS_FILE), "r", encoding="utf-8") as f:
+                usage_data = json.load(f)
+            for a in usage_data.get("accounts", []):
+                usage_by_kid[a.get("key_id", "")] = a
+
+        # 合并
+        keys_out = []
+        for a in accounts:
+            kid = a.get("key_id", "")
+            raw_key = a.get("key", "")
+            # Mask: 前8后4，不足12字符则不mask
+            if len(raw_key) > 12:
+                masked = raw_key[:8] + "..." + raw_key[-4:]
+            else:
+                masked = raw_key  # 短 key 不 mask
+
+            entry = {
+                "key_id": kid,
+                "label": a.get("label", kid),
+                "key_masked": masked,
+            }
+            # 附加用量
+            usage = usage_by_kid.get(kid)
+            if usage:
+                entry["workspace_id"] = usage.get("workspace_id", "")
+                for period in ["rolling", "weekly", "monthly"]:
+                    p = usage.get(period)
+                    if p:
+                        entry[period] = {
+                            "usage_percent": p.get("usage_percent"),
+                            "reset_in_sec": p.get("reset_in_sec"),
+                            "status": p.get("status", "unknown"),
+                        }
+                entry["session_expired"] = usage.get("session_expired", False)
+                entry["last_update"] = usage.get("last_update_iso", "")
+            keys_out.append(entry)
+
+        return jsonify({
+            "ok": True,
+            "count": len(keys_out),
+            "keys": keys_out,
+        })
+    except Exception as e:
+        return jsonify({"ok": False, "error": str(e), "keys": []})
+
+
 # ── Auto-collect timer (every 5 min, keeps cookies alive) ──
 _USAGE_AUTO_INTERVAL = 300