feat(stats): 添加密钥使用详情统计功能

新增功能允许用户在 Keys 状态页面点击“详情”按钮,查看指定 API 密钥在过去 24 小时内按模型分类的请求次数统计。

主要变更包括:

后端:
- 新增 `app/router/stats_routes.py`,包含 `/api/key-usage-details/{key}` API 端点用于获取密钥使用详情。
- 重构 `app/service/stats_service.py`,将统计相关函数封装到 `StatsService` 类中,并添加 `get_key_usage_details_last_24h` 方法。
- 在 `app/router/routes.py` 中注册新的 `stats_routes`,并更新对 `stats_service` 的调用方式以使用类实例。
- 更新 `app/log/logger.py` 添加 `get_scheduler_routes` 日志记录器,并在 `app/router/scheduler_routes.py` 中使用它。

前端:
- 在 `app/templates/keys_status.html` 中为每个有效和无效密钥列表项添加“详情”按钮。
- 在 `app/templates/keys_status.html` 中添加用于显示密钥使用详情的模态框 HTML 结构。
- 在 `app/static/js/keys_status.js` 中添加 JavaScript 函数 (`showKeyUsageDetails`, `closeKeyUsageDetailsModal`, `renderKeyUsageDetails`) 来处理按钮点击事件、调用后端 API、控制模态框显示/隐藏以及渲染获取到的统计数据。
This commit is contained in:
snaily
2025-04-20 01:41:22 +08:00
parent eb311de0c2
commit c99e090ea9
7 changed files with 361 additions and 107 deletions
+8 -5
View File
@@ -8,9 +8,9 @@ from fastapi.templating import Jinja2Templates
from app.core.security import verify_auth_token
from app.log.logger import get_routes_logger
from app.router import gemini_routes, openai_routes, config_routes, log_routes, scheduler_routes # 新增导入
from app.router import gemini_routes, openai_routes, config_routes, log_routes, scheduler_routes, stats_routes # 新增导入 stats_routes
from app.service.key.key_manager import get_key_manager_instance
from app.service.stats_service import get_api_usage_stats, get_api_call_details # <-- Import stats service and details function
from app.service.stats_service import StatsService
logger = get_routes_logger()
@@ -32,6 +32,7 @@ def setup_routers(app: FastAPI) -> None:
app.include_router(config_routes.router)
app.include_router(log_routes.router)
app.include_router(scheduler_routes.router) # 新增包含 scheduler 路由
app.include_router(stats_routes.router) # 包含 stats API 路由
# 添加页面路由
setup_page_routes(app)
@@ -92,8 +93,8 @@ def setup_page_routes(app: FastAPI) -> None:
valid_key_count = len(keys_status["valid_keys"])
invalid_key_count = len(keys_status["invalid_keys"])
# Get API usage stats
api_stats = await get_api_usage_stats()
stats_service = StatsService()
api_stats = await stats_service.get_api_usage_stats()
logger.info(f"API stats retrieved: {api_stats}")
logger.info(f"Keys status retrieved successfully. Total keys: {total_keys}")
@@ -180,7 +181,9 @@ def setup_api_stats_routes(app: FastAPI) -> None:
return {"error": "Unauthorized"}, 401
logger.info(f"Fetching API call details for period: {period}")
details = await get_api_call_details(period)
# Use the service instance here as well
stats_service = StatsService() # Create an instance
details = await stats_service.get_api_call_details(period)
return details
except ValueError as e:
logger.warning(f"Invalid period requested for API stats details: {period} - {str(e)}")
+2 -2
View File
@@ -7,9 +7,9 @@ from fastapi.responses import JSONResponse
from app.core.security import verify_auth_token # 导入 verify_auth_token
from app.scheduler.key_checker import start_scheduler, stop_scheduler
from app.log.logger import get_routes_logger # 使用路由日志记录器
from app.log.logger import get_scheduler_routes # 使用路由日志记录器
logger = get_routes_logger()
logger = get_scheduler_routes()
router = APIRouter(
prefix="/api/scheduler",
+60
View File
@@ -0,0 +1,60 @@
from fastapi import APIRouter, Depends, HTTPException, Request
from starlette import status
from app.core.security import verify_auth_token
from app.service.stats_service import StatsService
from app.log.logger import get_stats_logger # 使用路由日志记录器
logger = get_stats_logger()
# 认证检查的辅助函数
async def verify_token(request: Request):
auth_token = request.cookies.get("auth_token")
if not auth_token or not verify_auth_token(auth_token):
logger.warning("Unauthorized access attempt to scheduler API")
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Not authenticated",
headers={"WWW-Authenticate": "Bearer"},
)
router = APIRouter(
prefix="/api",
tags=["stats"],
dependencies=[Depends(verify_token)] # Assuming API routes need authentication
)
stats_service = StatsService()
@router.get("/key-usage-details/{key}",
summary="获取指定密钥最近24小时的模型调用次数",
description="根据提供的 API 密钥,返回过去24小时内每个模型被调用的次数统计。")
async def get_key_usage_details(key: str):
"""
Retrieves the model usage count for a specific API key within the last 24 hours.
Args:
key: The API key to get usage details for.
Returns:
A dictionary with model names as keys and their call counts as values.
Example: {"gemini-pro": 10, "gemini-1.5-pro-latest": 5}
Raises:
HTTPException: If an error occurs during data retrieval.
"""
try:
usage_details = await stats_service.get_key_usage_details_last_24h(key)
if usage_details is None:
# Handle case where key might be valid but has no recent usage,
# or if the service layer explicitly returns None for other reasons.
# Returning an empty dict is usually fine for the frontend.
return {}
return usage_details
except Exception as e:
# Log the exception details here if needed
print(f"Error fetching key usage details for key {key[:4]}...: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"获取密钥使用详情时出错: {e}"
)