mirror of
https://github.com/snailyp/gemini-balance.git
synced 2026-05-31 05:09:46 +08:00
7da9110704
为应用程序添加了通过代理服务器访问 Gemini API 的功能。
主要变更包括：
* **配置**:
* 在 `.env.example` 和 `app/config/config.py` 中添加了 `PROXIES` 配置项，允许用户指定一个或多个 HTTP 或 SOCKS5 代理服务器列表。
* 更新 `README.md` 以包含关于代理配置的说明。
* **后端**:
* 修改 `app/service/client/api_client.py` 中的 `GeminiApiClient`，使其在发起请求时能从配置的 `PROXIES` 列表中随机选择一个代理使用。
* 添加了 `app/log/logger.py` 中的 `get_api_client_logger`，用于记录 API 客户端（包括代理使用）的相关日志。
* **前端**:
* 在 `app/templates/config_editor.html` 配置编辑器页面添加了代理列表的显示区域和“添加代理”按钮。
* 实现了用于批量添加代理的模态框 UI。
* 在 `app/static/js/config_editor.js` 中添加了处理代理列表显示、打开/关闭模态框以及处理批量添加代理（包括提取、去重和更新 UI）的 JavaScript 逻辑。
* 确保在初始化配置时为 `PROXIES` 设置默认空列表。
此功能使得用户可以在需要通过代理访问外部网络的环境下使用该应用。
78 lines
3.2 KiB
Python
78 lines
3.2 KiB
Python
# app/services/chat/api_client.py
|
|
|
|
from typing import Dict, Any, AsyncGenerator
|
|
import httpx
|
|
import random
|
|
from abc import ABC, abstractmethod
|
|
from app.config.config import settings
|
|
from app.log.logger import get_api_client_logger
|
|
from app.core.constants import DEFAULT_TIMEOUT
|
|
|
|
logger = get_api_client_logger()
|
|
|
|
class ApiClient(ABC):
|
|
"""API客户端基类"""
|
|
|
|
@abstractmethod
|
|
async def generate_content(self, payload: Dict[str, Any], model: str, api_key: str) -> Dict[str, Any]:
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def stream_generate_content(self, payload: Dict[str, Any], model: str, api_key: str) -> AsyncGenerator[str, None]:
|
|
pass
|
|
|
|
|
|
class GeminiApiClient(ApiClient):
|
|
"""Gemini API客户端"""
|
|
|
|
def __init__(self, base_url: str, timeout: int = DEFAULT_TIMEOUT):
|
|
self.base_url = base_url
|
|
self.timeout = timeout
|
|
|
|
def _get_real_model(self, model: str) -> str:
|
|
if model.endswith("-search"):
|
|
model = model[:-7]
|
|
if model.endswith("-image"):
|
|
model = model[:-6]
|
|
if model.endswith("-non-thinking"):
|
|
model = model[:-13]
|
|
if "-search" in model and "-non-thinking" in model:
|
|
model = model[:-20]
|
|
return model
|
|
|
|
async def generate_content(self, payload: Dict[str, Any], model: str, api_key: str) -> Dict[str, Any]:
|
|
timeout = httpx.Timeout(self.timeout, read=self.timeout)
|
|
model = self._get_real_model(model)
|
|
|
|
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}/models/{model}:generateContent?key={api_key}"
|
|
response = await client.post(url, json=payload)
|
|
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], model: str, api_key: str) -> AsyncGenerator[str, None]:
|
|
timeout = httpx.Timeout(self.timeout, read=self.timeout)
|
|
model = self._get_real_model(model)
|
|
|
|
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}/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:
|
|
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
|