feat: 实现 OpenAI 兼容 API 端点和批量代理删除

新增与 OpenAI 规范兼容的 API 端点： - `/openai/v1/models` - `/openai/v1/chat/completions` (支持流式传输、重试和密钥切换) - `/openai/v1/embeddings` - `/openai/v1/images/generations` 包含： - 在 `app/router/openai_compatiable_routes.py` 中新增路由。 - `OpenAICompatiableService` 用于处理请求逻辑、日志记录和错误管理。 - 更新 `OpenaiApiClient` 以支持新方法和代理使用。 - 修改 `app/domain/openai_models.py` 以实现兼容性。 - 为新 API 添加专用日志记录器 (`openai_compatible`)。 - 为新路由 (`/openai`, `/api/version/check`) 添加认证中间件豁免。增强配置编辑器 UI： - 在 `app/static/js/config_editor.js` 和 `app/templates/config_editor.html` 中添加批量代理删除功能。
2026-09-05 23:56:37 +08:00 · 2025-04-30 20:39:47 +08:00
parent 7da9110704
commit 83ce50975a
9 changed files with 562 additions and 22 deletions
@@ -47,9 +47,9 @@ class GeminiApiClient(ApiClient):
        proxy_to_use = None
        if settings.PROXIES:
            proxy_to_use = random.choice(settings.PROXIES)
-            logger.info(f"using proxy: {proxy_to_use}")
+            logger.info(f"Using proxy: {proxy_to_use}")
            
-        async with httpx.AsyncClient(timeout=timeout, proxy=proxy_to_use) as client: # 修改：直接传递代理字符串
+        async with httpx.AsyncClient(timeout=timeout, proxy=proxy_to_use) as client:
            url = f"{self.base_url}/models/{model}:generateContent?key={api_key}"
            response = await client.post(url, json=payload)
            if response.status_code != 200:
@@ -64,9 +64,9 @@ class GeminiApiClient(ApiClient):
        proxy_to_use = None
        if settings.PROXIES:
            proxy_to_use = random.choice(settings.PROXIES)
-            logger.info(f"using proxy: {proxy_to_use}")
+            logger.info(f"Using proxy: {proxy_to_use}")

-        async with httpx.AsyncClient(timeout=timeout, proxy=proxy_to_use) as client: # 修改：直接传递代理字符串
+        async with httpx.AsyncClient(timeout=timeout, proxy=proxy_to_use) as client:
            url = f"{self.base_url}/models/{model}:streamGenerateContent?alt=sse&key={api_key}"
            async with client.stream(method="POST", url=url, json=payload) as response:
                if response.status_code != 200:
@@ -75,3 +75,96 @@ class GeminiApiClient(ApiClient):
                    raise Exception(f"API call failed with status code {response.status_code}, {error_msg}")
                async for line in response.aiter_lines():
                    yield line
+
+
+class OpenaiApiClient(ApiClient):
+    """OpenAI API客户端"""
+
+    def __init__(self, base_url: str, timeout: int = DEFAULT_TIMEOUT):
+        self.base_url = base_url
+        self.timeout = timeout
+        
+    async def get_models(self, api_key: str) -> Dict[str, Any]:
+        timeout = httpx.Timeout(self.timeout, read=self.timeout)
+        async with httpx.AsyncClient(timeout=timeout) as client:
+            url = f"{self.base_url}/openai/models"
+            headers = {"Authorization": f"Bearer {api_key}"}
+            response = await client.get(url, headers=headers)
+            if response.status_code != 200:
+                error_content = response.text
+                raise Exception(f"API call failed with status code {response.status_code}, {error_content}")
+            return response.json()
+
+    async def generate_content(self, payload: Dict[str, Any], api_key: str) -> Dict[str, Any]:
+        timeout = httpx.Timeout(self.timeout, read=self.timeout)
+        
+        proxy_to_use = None
+        if settings.PROXIES:
+            proxy_to_use = random.choice(settings.PROXIES)
+            logger.info(f"Using proxy: {proxy_to_use}")
+
+        async with httpx.AsyncClient(timeout=timeout, proxy=proxy_to_use) as client:
+            url = f"{self.base_url}/openai/chat/completions"
+            headers = {"Authorization": f"Bearer {api_key}"}
+            response = await client.post(url, json=payload, headers=headers)
+            if response.status_code != 200:
+                error_content = response.text
+                raise Exception(f"API call failed with status code {response.status_code}, {error_content}")
+            return response.json()
+
+    async def stream_generate_content(self, payload: Dict[str, Any], api_key: str) -> AsyncGenerator[str, None]:
+        timeout = httpx.Timeout(self.timeout, read=self.timeout)
+        
+        proxy_to_use = None
+        if settings.PROXIES:
+            proxy_to_use = random.choice(settings.PROXIES)
+            logger.info(f"Using proxy: {proxy_to_use}")
+
+        async with httpx.AsyncClient(timeout=timeout, proxy=proxy_to_use) as client:
+            url = f"{self.base_url}/openai/chat/completions"
+            headers = {"Authorization": f"Bearer {api_key}"}
+            async with client.stream(method="POST", url=url, json=payload, headers=headers) as response:
+                if response.status_code != 200:
+                    error_content = await response.aread()
+                    error_msg = error_content.decode("utf-8")
+                    raise Exception(f"API call failed with status code {response.status_code}, {error_msg}")
+                async for line in response.aiter_lines():
+                    yield line
+    
+    async def create_embeddings(self, input: str, model: str, api_key: str) -> Dict[str, Any]:
+        timeout = httpx.Timeout(self.timeout, read=self.timeout)
+        
+        proxy_to_use = None
+        if settings.PROXIES:
+            proxy_to_use = random.choice(settings.PROXIES)
+            logger.info(f"Using proxy: {proxy_to_use}")
+
+        async with httpx.AsyncClient(timeout=timeout, proxy=proxy_to_use) as client:
+            url = f"{self.base_url}/openai/embeddings"
+            headers = {"Authorization": f"Bearer {api_key}"}
+            payload = {
+                "input": input,
+                "model": model,
+            }
+            response = await client.post(url, json=payload, headers=headers)
+            if response.status_code != 200:
+                error_content = response.text
+                raise Exception(f"API call failed with status code {response.status_code}, {error_content}")
+            return response.json()
+                    
+    async def generate_images(self, payload: Dict[str, Any], api_key: str) -> Dict[str, Any]:
+        timeout = httpx.Timeout(self.timeout, read=self.timeout)
+        
+        proxy_to_use = None
+        if settings.PROXIES:
+            proxy_to_use = random.choice(settings.PROXIES)
+            logger.info(f"Using proxy: {proxy_to_use}")
+
+        async with httpx.AsyncClient(timeout=timeout, proxy=proxy_to_use) as client:
+            url = f"{self.base_url}/openai/images/generations"
+            headers = {"Authorization": f"Bearer {api_key}"}
+            response = await client.post(url, json=payload, headers=headers)
+            if response.status_code != 200:
+                error_content = response.text
+                raise Exception(f"API call failed with status code {response.status_code}, {error_content}")
+            return response.json()
@@ -0,0 +1,197 @@
+
+import datetime
+import json
+import re
+import time
+from typing import Any, AsyncGenerator, Dict, Union
+
+from app.config.config import settings
+from app.database.services import (
+    add_error_log,
+    add_request_log,
+)
+from app.domain.openai_models import ChatRequest, ImageGenerationRequest
+from app.service.client.api_client import OpenaiApiClient
+from app.service.key.key_manager import KeyManager
+from app.log.logger import get_openai_compatible_logger
+
+logger = get_openai_compatible_logger()
+
+class OpenAICompatiableService:
+
+    def __init__(self, base_url: str, key_manager: KeyManager = None):
+        self.key_manager = key_manager
+        self.base_url = base_url
+        self.api_client = OpenaiApiClient(base_url, settings.TIME_OUT)
+        
+    async def get_models(self, api_key: str) -> Dict[str, Any]:
+        return await self.api_client.get_models(api_key)
+
+    async def create_chat_completion(
+        self,
+        request: ChatRequest,
+        api_key: str,
+    ) -> Union[Dict[str, Any], AsyncGenerator[str, None]]:
+        """创建聊天完成"""
+        request_dict = request.model_dump()
+        # 移除值为null的
+        request_dict = {k: v for k, v in request_dict.items() if v is not None}
+        del request_dict["top_k"] # 删除top_k参数，目前不支持该参数
+        if request.stream:
+            return self._handle_stream_completion(request.model, request_dict, api_key)
+        return await self._handle_normal_completion(request.model, request_dict, api_key)
+
+    async def generate_images(
+        self,
+        request: ImageGenerationRequest,
+    ) -> Dict[str, Any]:
+        """生成图片"""
+        request_dict = request.model_dump()
+        # 移除值为null的
+        request_dict = {k: v for k, v in request_dict.items() if v is not None}
+        api_key = settings.PAID_KEY
+        return await self.api_client.generate_images(request_dict, api_key)
+
+    async def create_embeddings(
+        self,
+        input_text: str,
+        model: str,
+        api_key: str,
+    ) -> Dict[str, Any]:
+        """创建嵌入"""
+        return await self.api_client.create_embeddings(input_text, model, api_key)
+
+    async def _handle_normal_completion(
+        self, model: str, request: dict, 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(request, api_key)
+            is_success = True
+            status_code = 200
+            return response
+        except Exception as e:
+            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:
+                status_code = int(match.group(1))
+            else:
+                status_code = 500
+
+            await add_error_log(
+                gemini_key=api_key,
+                model_name=model,
+                error_type="openai-compatiable-non-stream",
+                error_log=error_log_msg,
+                error_code=status_code,
+                request_msg=request,
+            )
+            raise e
+        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, api_key: str
+    ) -> AsyncGenerator[str, None]:
+        """处理流式聊天完成，添加重试逻辑"""
+        retries = 0
+        max_retries = settings.MAX_RETRIES
+        is_success = False
+        status_code = None
+        final_api_key = api_key
+
+        while retries < max_retries:
+            start_time = time.perf_counter()
+            request_datetime = datetime.datetime.now()
+            current_attempt_key = api_key
+            final_api_key = current_attempt_key
+            try:
+                async for line in self.api_client.stream_generate_content(
+                    payload, current_attempt_key
+                ):
+                    if line.startswith("data:"):
+                        # print(line)
+                        yield line + "\n\n"
+                logger.info("Streaming completed successfully")
+                is_success = True
+                status_code = 200
+                break  # 成功后退出循环
+            except Exception as e:
+                retries += 1
+                is_success = False
+                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
+
+                # Log error to error log table
+                await add_error_log(
+                    gemini_key=current_attempt_key,
+                    model_name=model,
+                    error_type="openai-compatiable-stream",
+                    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 
+                else:
+                    logger.error("KeyManager not available for retry logic.")
+                    break 
+
+                if retries >= max_retries:
+                    logger.error(f"Max retries ({max_retries}) reached for streaming.")
+                    break
+            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,
+                    is_success=is_success,
+                    status_code=status_code,
+                    latency_ms=latency_ms,
+                    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"
+                    
+