feat(monitoring): 添加 API 请求统计和监控面板

本次提交引入了 API 请求统计功能，并将原“密钥状态”页面重构为功能更全面的“监控面板”。

主要变更包括：

- **数据库与服务层:**
    - 新增 `RequestLog` 数据模型 (`app/database/models.py`)，用于存储 API 请求的详细信息（时间、模型、密钥、成功状态、状态码、耗时）。
    - 在 `app/database/services.py` 中添加 `add_request_log` 和 `get_request_stats` 函数，分别用于记录单次请求和获取时间窗口内的统计数据。
    - 新增 `app/service/stats_service.py`，封装了获取 API 调用统计逻辑。

- **API 请求日志记录:**
    - 在 Gemini (`gemini_chat_service.py`) 和 OpenAI (`openai_chat_service.py`) 聊天服务中，于 API 调用前后添加了 `add_request_log` 调用，以记录请求的成功与否及耗时。

- **前端监控面板:**
    - 将 `/keys` 路由对应的页面 (`keys_status.html`) 从“密钥状态”重构为“监控面板”。
    - 页面顶部新增统计卡片区域，展示：
        - 密钥统计：总数、有效数、无效数。
        - API 调用统计：1分钟内、1小时内、24小时内、本月调用次数。
    - 密钥列表（有效/无效）采用响应式网格布局 (`grid`)，并增加了悬停动效和边框高亮。
    - 优化了有效密钥列表的筛选逻辑，在无匹配项时显示提示信息。
    - 为新的统计卡片和列表项添加了相应的 CSS 样式。
    - 更新了 `keys_status.js` 以支持筛选无结果时的提示。

- **路由与导航:**
    - 在 `app/router/routes.py` 中添加了 `/stats` 端点，用于获取 API 统计数据。
    - 更新了 `config_editor.html` 和 `error_logs.html` 中的导航链接，使其指向新的“监控面板”。

- **日志配置:**
    - 在 `app/log/logger.py` 中，为 `sqlalchemy.exc` 设置了 WARNING 日志级别。

这些更改旨在提供更好的系统可观测性，方便用户监控 API 密钥状态和请求频率。

app/service/chat/gemini_chat_service.py

@@ -2,8 +2,9 @@
 
 import json
 import re
+import datetime # Add datetime import
+import time # Add time import
 from typing import Any, AsyncGenerator, Dict, List
-
 from app.config.config import settings
 from app.domain.gemini_models import GeminiRequest
 from app.handler.response_handler import GeminiResponseHandler
@@ -11,7 +12,7 @@ from app.handler.stream_optimizer import gemini_optimizer
 from app.log.logger import get_gemini_logger
 from app.service.client.api_client import GeminiApiClient
 from app.service.key.key_manager import KeyManager
-from app.database.services import add_error_log
+from app.database.services import add_error_log, add_request_log # Import add_request_log
 
 logger = get_gemini_logger()
 
@@ -147,27 +148,54 @@ class GeminiChatService:
     ) -> Dict[str, Any]:
         """生成内容"""
         payload = _build_payload(model, request)
+        start_time = time.perf_counter()
+        request_datetime = datetime.datetime.now() # Record request time
+        is_success = False
+        status_code = None
+        response = None
+
         try:
             response = await self.api_client.generate_content(payload, model, api_key)
+            # Assuming success if no exception is raised and response is received
+            # The actual status code might be within the response structure or headers,
+            # but api_client doesn't seem to expose it directly here.
+            # We'll assume 200 for success if no exception.
+            is_success = True
+            status_code = 200 # Assume 200 on success
             return self.response_handler.handle_response(response, model, stream=False)
         except Exception as e:
-            logger.error(f"Normal API call failed with error: {str(e)}")
-            error_code = None
+            is_success = False
             error_log_msg = str(e)
-            # 尝试从异常消息中解析状态码
+            logger.error(f"Normal API call failed with error: {error_log_msg}")
+            # Try to parse status code from exception
             match = re.search(r"status code (\d+)", error_log_msg)
             if match:
-                error_code = int(match.group(1))
-            
+                status_code = int(match.group(1))
+            else:
+                status_code = 500 # Default to 500 if parsing fails
+
+            # Log error to error log table
             await add_error_log(
                 gemini_key=api_key,
                 model_name=model,
                 error_type="gemini_chat_service",
                 error_log=error_log_msg,
-                error_code=error_code,
+                error_code=status_code,
                 request_msg=payload
             )
-            raise e # 重新抛出异常，以便上层处理
+            raise e # Re-throw exception for upstream handling
+        finally:
+            end_time = time.perf_counter()
+            latency_ms = int((end_time - start_time) * 1000)
+            # Log request to request log table
+            await add_request_log(
+                model_name=model,
+                api_key=api_key,
+                is_success=is_success,
+                status_code=status_code,
+                latency_ms=latency_ms,
+                request_time=request_datetime
+            )
 
     async def stream_generate_content(
         self, model: str, request: GeminiRequest, api_key: str
@@ -176,60 +204,96 @@ class GeminiChatService:
         retries = 0
         max_retries = settings.MAX_RETRIES
         payload = _build_payload(model, request)
-        while retries < max_retries:
-            try:
-                async for line in self.api_client.stream_generate_content(
-                    payload, model, api_key
-                ):
-                    # print(line)
-                    if line.startswith("data:"):
-                        line = line[6:]
-                        response_data = self.response_handler.handle_response(
-                            json.loads(line), model, stream=True
-                        )
-                        text = self._extract_text_from_response(response_data)
-                        # 如果有文本内容，且开启了流式输出优化器，则使用流式输出优化器处理
-                        if text and settings.STREAM_OPTIMIZER_ENABLED:
-                            # 使用流式输出优化器处理文本输出
-                            async for (
-                                optimized_chunk
-                            ) in gemini_optimizer.optimize_stream_output(
-                                text,
-                                lambda t: self._create_char_response(response_data, t),
-                                lambda c: "data: " + json.dumps(c) + "\n\n",
-                            ):
-                                yield optimized_chunk
-                        else:
-                            # 如果没有文本内容（如工具调用等），整块输出
-                            yield "data: " + json.dumps(response_data) + "\n\n"
-                logger.info("Streaming completed successfully")
-                break
-            except Exception as e:
-                retries += 1
-                error_log_msg = str(e)
-                logger.warning(
-                    f"Streaming API call failed with error: {error_log_msg}. Attempt {retries} of {max_retries}"
-                )
-                # 解析错误信息并记录到数据库
-                error_code = None
-                match = re.search(r"status code (\d+)", error_log_msg)
-                if match:
-                    error_code = int(match.group(1))
-                
-                await add_error_log(
-                    gemini_key=api_key,
-                    model_name=model,
-                    error_log=error_log_msg,
-                    error_code=error_code,
-                    request_msg=payload
-                )
-                
-                # 尝试切换 API Key
-                api_key = await self.key_manager.handle_api_failure(api_key,retries)
-                if api_key:
-                    logger.info(f"Switched to new API key: {api_key}")
-                if retries >= max_retries:
-                    logger.error(
-                        f"Max retries ({max_retries}) reached for streaming. Raising error"
+        start_time = time.perf_counter() # Record start time before loop
+        request_datetime = datetime.datetime.now()
+        is_success = False
+        status_code = None
+        final_api_key = api_key # Store the initial key
+
+        try:
+            while retries < max_retries:
+                current_attempt_key = api_key # Key used for this attempt
+                final_api_key = current_attempt_key # Update final key used
+                try:
+                    async for line in self.api_client.stream_generate_content(
+                        payload, model, current_attempt_key
+                    ):
+                        # print(line)
+                        if line.startswith("data:"):
+                            line = line[6:]
+                            response_data = self.response_handler.handle_response(
+                                json.loads(line), model, stream=True
+                            )
+                            text = self._extract_text_from_response(response_data)
+                            # 如果有文本内容，且开启了流式输出优化器，则使用流式输出优化器处理
+                            if text and settings.STREAM_OPTIMIZER_ENABLED:
+                                # 使用流式输出优化器处理文本输出
+                                async for (
+                                    optimized_chunk
+                                ) in gemini_optimizer.optimize_stream_output(
+                                    text,
+                                    lambda t: self._create_char_response(response_data, t),
+                                    lambda c: "data: " + json.dumps(c) + "\n\n",
+                                ):
+                                    yield optimized_chunk
+                            else:
+                                # 如果没有文本内容（如工具调用等），整块输出
+                                yield "data: " + json.dumps(response_data) + "\n\n"
+                    logger.info("Streaming completed successfully")
+                    is_success = True
+                    status_code = 200 # Assume 200 on success
+                    break # Exit loop on success
+                except Exception as e:
+                    retries += 1
+                    is_success = False # Mark as failed for this attempt
+                    error_log_msg = str(e)
+                    logger.warning(
+                        f"Streaming API call failed with error: {error_log_msg}. Attempt {retries} of {max_retries}"
                     )
-                    break
+                    # Parse error code for logging
+                    match = re.search(r"status code (\d+)", error_log_msg)
+                    if match:
+                        status_code = int(match.group(1))
+                    else:
+                        status_code = 500 # Default if parsing fails
+
+                    # Log error to error log table
+                    await add_error_log(
+                        gemini_key=current_attempt_key, # Log key used for this failed attempt
+                        model_name=model,
+                        error_log=error_log_msg,
+                        error_code=status_code,
+                        request_msg=payload
+                    )
+
+                    # Attempt to switch API Key
+                    api_key = await self.key_manager.handle_api_failure(current_attempt_key, retries)
+                    if api_key:
+                        logger.info(f"Switched to new API key: {api_key}")
+                    else: # No more keys or retries exceeded by handle_api_failure logic
+                         logger.error(f"No valid API key available after {retries} retries.")
+                         break # Exit loop if no key available
+
+                    if retries >= max_retries:
+                        logger.error(
+                            f"Max retries ({max_retries}) reached for streaming."
+                        )
+                        break # Exit loop after max retries
+        finally:
+            # Log the final outcome of the streaming request
+            end_time = time.perf_counter()
+            latency_ms = int((end_time - start_time) * 1000)
+            await add_request_log(
+                model_name=model,
+                api_key=final_api_key, # Log the last key used
+                is_success=is_success, # Log the final success status
+                status_code=status_code, # Log the last known status code
+                latency_ms=latency_ms, # Log total time including retries
+                request_time=request_datetime
+            )
+            # If the loop finished due to failure, ensure an exception is raised if not already handled
+            if not is_success and retries >= max_retries:
+                 # We need to raise an exception here if the loop exited due to max retries failure
+                 # However, the original code structure doesn't explicitly raise here after the loop.
+                 # For now, we just log. Consider raising HTTPException if needed.
+                 pass

app/service/chat/openai_chat_service.py

@@ -2,6 +2,8 @@
 
 import json
 import re
+import datetime # Add datetime import
+import time # Add time import
 from copy import deepcopy
 from typing import Any, AsyncGenerator, Dict, List, Optional, Union
 
@@ -14,7 +16,7 @@ from app.log.logger import get_openai_logger
 from app.service.client.api_client import GeminiApiClient
 from app.service.image.image_create_service import ImageCreateService
 from app.service.key.key_manager import KeyManager
-from app.database.services import add_error_log
+from app.database.services import add_error_log, add_request_log # Import add_request_log
 
 logger = get_openai_logger()
 
@@ -191,28 +193,49 @@ class OpenAIChatService:
         self, model: str, payload: Dict[str, Any], api_key: str
     ) -> Dict[str, Any]:
         """处理普通聊天完成"""
+        start_time = time.perf_counter()
+        request_datetime = datetime.datetime.now()
+        is_success = False
+        status_code = None
+        response = None
         try:
             response = await self.api_client.generate_content(payload, model, api_key)
+            is_success = True
+            status_code = 200 # Assume 200 on success
             return self.response_handler.handle_response(
                 response, model, stream=False, finish_reason="stop"
             )
         except Exception as e:
-            logger.error(f"Normal API call failed with error: {str(e)}")
-            error_code = None
+            is_success = False
             error_log_msg = str(e)
-            # 尝试从异常消息中解析状态码
+            logger.error(f"Normal API call failed with error: {error_log_msg}")
+            # Try to parse status code from exception
             match = re.search(r"status code (\d+)", error_log_msg)
             if match:
-                error_code = int(match.group(1))
-            
+                status_code = int(match.group(1))
+            else:
+                status_code = 500 # Default if parsing fails
+
             await add_error_log(
-                gemini_key=api_key,
+                gemini_key=api_key, # Note: Parameter name is gemini_key in add_error_log
                 model_name=model,
+                error_type="openai_chat_service", # Indicate service type
                 error_log=error_log_msg,
-                error_code=error_code,
+                error_code=status_code,
                 request_msg=payload
             )
-            raise e # 重新抛出异常，以便上层处理
+            raise e # Re-throw exception
+        finally:
+            end_time = time.perf_counter()
+            latency_ms = int((end_time - start_time) * 1000)
+            await add_request_log(
+                model_name=model,
+                api_key=api_key,
+                is_success=is_success,
+                status_code=status_code,
+                latency_ms=latency_ms,
+                request_time=request_datetime
+            )
 
     async def _handle_stream_completion(
         self, model: str, payload: Dict[str, Any], api_key: str
@@ -220,77 +243,114 @@ class OpenAIChatService:
         """处理流式聊天完成，添加重试逻辑"""
         retries = 0
         max_retries = settings.MAX_RETRIES
-        while retries < max_retries:
-            try:
-                tool_call_flag = False
-                async for line in self.api_client.stream_generate_content(
-                    payload, model, api_key
-                ):
-                    # print(line)
-                    if line.startswith("data:"):
-                        chunk = json.loads(line[6:])
-                        openai_chunk = self.response_handler.handle_response(
-                            chunk, model, stream=True, finish_reason=None
-                        )
-                        if openai_chunk:
-                            # 提取文本内容
-                            text = self._extract_text_from_openai_chunk(openai_chunk)
-                            if text and settings.STREAM_OPTIMIZER_ENABLED:
-                                # 使用流式输出优化器处理文本输出
-                                async for (
-                                    optimized_chunk
-                                ) in openai_optimizer.optimize_stream_output(
-                                    text,
-                                    lambda t: self._create_char_openai_chunk(
-                                        openai_chunk, t
-                                    ),
-                                    lambda c: f"data: {json.dumps(c)}\n\n",
-                                ):
-                                    yield optimized_chunk
-                            else:
-                                # 如果没有文本内容（如工具调用等），整块输出
-                                if "tool_calls" in json.dumps(openai_chunk):
-                                    tool_call_flag = True
-                                yield f"data: {json.dumps(openai_chunk)}\n\n"
-                if tool_call_flag:
-                    yield f"data: {json.dumps(self.response_handler.handle_response({}, model, stream=True, finish_reason='tool_calls'))}\n\n"
-                else:
-                    yield f"data: {json.dumps(self.response_handler.handle_response({}, model, stream=True, finish_reason='stop'))}\n\n"
-                yield "data: [DONE]\n\n"
-                logger.info("Streaming completed successfully")
-                break  # 成功后退出循环
-            except Exception as e:
-                retries += 1
-                error_log_msg = str(e)
-                logger.warning(
-                    f"Streaming API call failed with error: {error_log_msg}. Attempt {retries} of {max_retries}"
-                )
-                # 解析错误信息并记录到数据库
-                error_code = None
-                match = re.search(r"status code (\d+)", error_log_msg)
-                if match:
-                    error_code = int(match.group(1))
-                print(model)
-                await add_error_log(
-                    gemini_key=api_key,
-                    model_name=model,
-                    error_type="openai_chat_service",
-                    error_log=error_log_msg,
-                    error_code=error_code,
-                    request_msg=payload
-                )
-                
-                # 尝试切换 API Key
-                api_key = await self.key_manager.handle_api_failure(api_key,retries)
-                if api_key:
-                    logger.info(f"Switched to new API key: {api_key}")
-                if retries >= max_retries:
-                    logger.error(
-                        f"Max retries ({max_retries}) reached for streaming. Raising error"
-                    )
-                    yield f"data: {json.dumps({'error': 'Streaming failed after retries'})}\n\n"
+        start_time = time.perf_counter() # Record start time before loop
+        request_datetime = datetime.datetime.now()
+        is_success = False
+        status_code = None
+        final_api_key = api_key # Store the initial key
+
+        try:
+            while retries < max_retries:
+                current_attempt_key = api_key # Key used for this attempt
+                final_api_key = current_attempt_key # Update final key used
+                try:
+                    tool_call_flag = False
+                    async for line in self.api_client.stream_generate_content(
+                        payload, model, current_attempt_key
+                    ):
+                        # print(line)
+                        if line.startswith("data:"):
+                            chunk = json.loads(line[6:])
+                            openai_chunk = self.response_handler.handle_response(
+                                chunk, model, stream=True, finish_reason=None
+                            )
+                            if openai_chunk:
+                                # 提取文本内容
+                                text = self._extract_text_from_openai_chunk(openai_chunk)
+                                if text and settings.STREAM_OPTIMIZER_ENABLED:
+                                    # 使用流式输出优化器处理文本输出
+                                    async for (
+                                        optimized_chunk
+                                    ) in openai_optimizer.optimize_stream_output(
+                                        text,
+                                        lambda t: self._create_char_openai_chunk(
+                                            openai_chunk, t
+                                        ),
+                                        lambda c: f"data: {json.dumps(c)}\n\n",
+                                    ):
+                                        yield optimized_chunk
+                                else:
+                                    # 如果没有文本内容（如工具调用等），整块输出
+                                    if "tool_calls" in json.dumps(openai_chunk):
+                                        tool_call_flag = True
+                                    yield f"data: {json.dumps(openai_chunk)}\n\n"
+                    if tool_call_flag:
+                        yield f"data: {json.dumps(self.response_handler.handle_response({}, model, stream=True, finish_reason='tool_calls'))}\n\n"
+                    else:
+                        yield f"data: {json.dumps(self.response_handler.handle_response({}, model, stream=True, finish_reason='stop'))}\n\n"
                     yield "data: [DONE]\n\n"
-                    break
+                    logger.info("Streaming completed successfully")
+                    is_success = True
+                    status_code = 200 # Assume 200 on success
+                    break  # 成功后退出循环
+                except Exception as e:
+                    retries += 1
+                    is_success = False # Mark as failed for this attempt
+                    error_log_msg = str(e)
+                    logger.warning(
+                        f"Streaming API call failed with error: {error_log_msg}. Attempt {retries} of {max_retries}"
+                    )
+                    # Parse error code for logging
+                    match = re.search(r"status code (\d+)", error_log_msg)
+                    if match:
+                        status_code = int(match.group(1))
+                    else:
+                        status_code = 500 # Default if parsing fails
+
+                    # Log error to error log table
+                    await add_error_log(
+                        gemini_key=current_attempt_key, # Note: Parameter name is gemini_key
+                        model_name=model,
+                        error_type="openai_chat_service", # Indicate service type
+                        error_log=error_log_msg,
+                        error_code=status_code,
+                        request_msg=payload
+                    )
+
+                    # Attempt to switch API Key
+                    # Ensure key_manager is available (might need adjustment if not always passed)
+                    if self.key_manager:
+                        api_key = await self.key_manager.handle_api_failure(current_attempt_key, retries)
+                        if api_key:
+                            logger.info(f"Switched to new API key: {api_key}")
+                        else:
+                            logger.error(f"No valid API key available after {retries} retries.")
+                            break # Exit loop if no key available
+                    else:
+                         logger.error("KeyManager not available for retry logic.")
+                         break # Exit loop if key manager is missing
+
+                    if retries >= max_retries:
+                        logger.error(
+                            f"Max retries ({max_retries}) reached for streaming."
+                        )
+                        break # Exit loop after max retries
+        finally:
+            # Log the final outcome of the streaming request
+            end_time = time.perf_counter()
+            latency_ms = int((end_time - start_time) * 1000)
+            await add_request_log(
+                model_name=model,
+                api_key=final_api_key, # Log the last key used
+                is_success=is_success, # Log the final success status
+                status_code=status_code, # Log the last known status code
+                latency_ms=latency_ms, # Log total time including retries
+                request_time=request_datetime
+            )
+            # If the loop finished due to failure, yield error and DONE
+            if not is_success and retries >= max_retries:
+                 yield f"data: {json.dumps({'error': 'Streaming failed after retries'})}\n\n"
+                 yield "data: [DONE]\n\n"
 
     async def create_image_chat_completion(
         self,

app/service/stats_service.py

@@ -0,0 +1,69 @@
+# app/service/stats_service.py
+
+import datetime
+from sqlalchemy import select, func
+
+from app.database.connection import database
+from app.database.models import RequestLog
+from app.log.logger import get_stats_logger
+
+logger = get_stats_logger()
+
+async def get_calls_in_last_seconds(seconds: int) -> int:
+    """获取过去 N 秒内的调用次数 (包括成功和失败)"""
+    try:
+        cutoff_time = datetime.datetime.now() - datetime.timedelta(seconds=seconds)
+        query = select(func.count(RequestLog.id)).where(
+            RequestLog.request_time >= cutoff_time
+        )
+        count_result = await database.fetch_one(query)
+        return count_result[0] if count_result else 0
+    except Exception as e:
+        logger.error(f"Failed to get calls in last {seconds} seconds: {e}")
+        return 0 # Return 0 on error
+
+async def get_calls_in_last_minutes(minutes: int) -> int:
+    """获取过去 N 分钟内的调用次数 (包括成功和失败)"""
+    return await get_calls_in_last_seconds(minutes * 60)
+
+async def get_calls_in_last_hours(hours: int) -> int:
+    """获取过去 N 小时内的调用次数 (包括成功和失败)"""
+    return await get_calls_in_last_seconds(hours * 3600)
+
+async def get_calls_in_current_month() -> int:
+    """获取当前自然月内的调用次数 (包括成功和失败)"""
+    try:
+        now = datetime.datetime.now()
+        start_of_month = now.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
+        query = select(func.count(RequestLog.id)).where(
+            RequestLog.request_time >= start_of_month
+        )
+        count_result = await database.fetch_one(query)
+        return count_result[0] if count_result else 0
+    except Exception as e:
+        logger.error(f"Failed to get calls in current month: {e}")
+        return 0 # Return 0 on error
+
+async def get_api_usage_stats() -> dict:
+    """获取所有需要的 API 使用统计数据"""
+    try:
+        calls_1m = await get_calls_in_last_minutes(1)
+        calls_1h = await get_calls_in_last_hours(1)
+        calls_24h = await get_calls_in_last_hours(24)
+        calls_month = await get_calls_in_current_month()
+
+        return {
+            "calls_1m": calls_1m,
+            "calls_1h": calls_1h,
+            "calls_24h": calls_24h,
+            "calls_month": calls_month,
+        }
+    except Exception as e:
+        logger.error(f"Failed to get API usage stats: {e}")
+        # Return default values on error
+        return {
+            "calls_1m": 0,
+            "calls_1h": 0,
+            "calls_24h": 0,
+            "calls_month": 0,
+        }