diff --git a/s2a_service.py b/s2a_service.py
index 4ccc6f3..c4bf43b 100644
--- a/s2a_service.py
+++ b/s2a_service.py
@@ -817,16 +817,15 @@ def s2a_batch_import_accounts(
# ==================== API 密钥用量查询 ====================
-def s2a_get_api_keys(page: int = 1, page_size: int = 50, timezone: str = "Asia/Shanghai") -> Optional[Dict[str, Any]]:
- """获取 API 密钥列表
-
- Args:
- page: 页码
- page_size: 每页数量
- timezone: 时区
+def s2a_get_all_usage_stats(
+ start_date: str,
+ end_date: str,
+ timezone: str = "Asia/Shanghai"
+) -> Optional[Dict[str, Any]]:
+ """获取所有密钥的汇总用量统计(不传 api_key_id)
Returns:
- dict: API 响应数据 或 None
+ dict: 汇总统计数据 或 None
"""
if not S2A_API_BASE or (not S2A_ADMIN_KEY and not S2A_ADMIN_TOKEN):
return None
@@ -835,9 +834,13 @@ def s2a_get_api_keys(page: int = 1, page_size: int = 50, timezone: str = "Asia/S
try:
response = http_session.get(
- f"{S2A_API_BASE}/keys",
+ f"{S2A_API_BASE}/admin/usage/stats",
headers=headers,
- params={"page": page, "page_size": page_size, "timezone": timezone},
+ params={
+ "start_date": start_date,
+ "end_date": end_date,
+ "timezone": timezone
+ },
timeout=REQUEST_TIMEOUT
)
@@ -845,13 +848,11 @@ def s2a_get_api_keys(page: int = 1, page_size: int = 50, timezone: str = "Asia/S
result = response.json()
if result.get("code") == 0:
return result.get("data", {})
- else:
- log.warning(f"S2A 获取密钥列表失败: {result.get('message', 'Unknown error')}")
else:
- log.warning(f"S2A 获取密钥列表失败: HTTP {response.status_code}")
+ log.warning(f"S2A 获取汇总用量失败: HTTP {response.status_code}")
except Exception as e:
- log.warning(f"S2A 获取密钥列表异常: {e}")
+ log.warning(f"S2A 获取汇总用量异常: {e}")
return None
@@ -906,6 +907,19 @@ def s2a_get_key_usage_stats(
return None
+# 密钥列表配置 (从 /api/v1/keys 接口获取的数据)
+API_KEYS_LIST = [
+ {"id": 11, "name": "bohe", "key": "bohebohebohebohe", "group_name": "codex"},
+ {"id": 10, "name": "黑与白", "key": "oyqq114514oyqq114514", "group_name": "codex"},
+ {"id": 9, "name": "baozi", "key": "baozibaozibaozibaozi", "group_name": "codex"},
+ {"id": 7, "name": "Wong-灭鼠专家", "key": "Wong1234567891011", "group_name": "codex"},
+ {"id": 6, "name": "小马-莹佬", "key": "mazhichentaiqiangla", "group_name": "codex"},
+ {"id": 5, "name": "猫猫-byteBender", "key": "ByteBender114514", "group_name": "codex"},
+ {"id": 4, "name": "KFC", "key": "sk-babf3f6d15582b31e959311d6ceaa9448f2cf951dc23d02386d41f68046e6018", "group_name": "codex"},
+ {"id": 3, "name": "zhongruan", "key": "zhongruantaiqiangla", "group_name": "codex"},
+]
+
+
def s2a_get_keys_with_usage(start_date: str = None, end_date: str = None, timezone: str = "Asia/Shanghai") -> Optional[List[Dict[str, Any]]]:
"""获取密钥列表并合并用量数据
@@ -919,31 +933,26 @@ def s2a_get_keys_with_usage(start_date: str = None, end_date: str = None, timezo
"""
from datetime import datetime
- # 获取密钥列表
- keys_data = s2a_get_api_keys(page=1, page_size=100, timezone=timezone)
- if not keys_data:
- return None
-
- keys = keys_data.get("items", [])
- if not keys:
- return []
-
# 默认今日
if not start_date:
start_date = datetime.now().strftime("%Y-%m-%d")
if not end_date:
end_date = datetime.now().strftime("%Y-%m-%d")
- # 获取每个密钥的用量
- for key in keys:
- key_id = key.get("id")
- if key_id:
- usage = s2a_get_key_usage_stats(key_id, start_date, end_date, timezone)
- key["usage"] = usage if usage else {}
- else:
- key["usage"] = {}
+ keys = []
+ for key_info in API_KEYS_LIST:
+ key_id = key_info["id"]
+ usage = s2a_get_key_usage_stats(key_id, start_date, end_date, timezone)
+ keys.append({
+ "id": key_id,
+ "name": key_info["name"],
+ "key": key_info["key"],
+ "group_name": key_info["group_name"],
+ "status": "active",
+ "usage": usage if usage else {}
+ })
- return keys
+ return keys if keys else None
def format_keys_usage(keys: List[Dict[str, Any]], period_text: str = "今日") -> str:
@@ -992,9 +1001,8 @@ def format_keys_usage(keys: List[Dict[str, Any]], period_text: str = "今日") -
for key in keys:
name = key.get("name", "未命名")
key_str = key.get("key", "")
- status = key.get("status", "unknown")
- group = key.get("group", {})
- group_name = group.get("name", "默认") if group else "默认"
+ status = key.get("status", "active")
+ group_name = key.get("group_name", "默认")
usage = key.get("usage", {})
requests = usage.get("total_requests", 0)
@@ -1019,7 +1027,6 @@ def format_keys_usage(keys: List[Dict[str, Any]], period_text: str = "今日") -
key_display = key_str[:8] + "..." if key_str else "N/A"
lines.append(f"{status_icon} {name} ({group_name})")
- lines.append(f" 密钥: {key_display}")
lines.append(f" 请求: {requests:,} | 耗时: {fmt_duration(avg_duration)}")
lines.append(f" Token: {fmt_tokens(tokens)} (入:{fmt_tokens(input_tokens)} 出:{fmt_tokens(output_tokens)})")
if cache_tokens > 0: