Losita/mai-bot
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Ruff Fix & format

Browse Source
This commit is contained in:
墨梓柒
2025-11-29 14:38:42 +08:00
parent d7932595e8
commit 3935ce817e
31 changed files with 678 additions and 684 deletions
Download Patch File Download Diff File Expand all files Collapse all files

113
src/webui/model_routes.py
View File

@@ -43,25 +43,27 @@ def _normalize_url(url: str) -> str:
def _parse_openai_response(data: dict) -> list[dict]:
"""
解析 OpenAI 格式的模型列表响应
格式: { "data": [{ "id": "gpt-4", "object": "model", ... }] }
"""
models = []
if "data" in data and isinstance(data["data"], list):
for model in data["data"]:
if isinstance(model, dict) and "id" in model:
models.append({
"id": model["id"],
"name": model.get("name") or model["id"],
"owned_by": model.get("owned_by", ""),
})
models.append(
{
"id": model["id"],
"name": model.get("name") or model["id"],
"owned_by": model.get("owned_by", ""),
}
)
return models
def _parse_gemini_response(data: dict) -> list[dict]:
"""
解析 Gemini 格式的模型列表响应
格式: { "models": [{ "name": "models/gemini-pro", "displayName": "Gemini Pro", ... }] }
"""
models = []
@@ -72,11 +74,13 @@ def _parse_gemini_response(data: dict) -> list[dict]:
model_id = model["name"]
if model_id.startswith("models/"):
model_id = model_id[7:] # 去掉 "models/" 前缀
models.append({
"id": model_id,
"name": model.get("displayName") or model_id,
"owned_by": "google",
})
models.append(
{
"id": model_id,
"name": model.get("displayName") or model_id,
"owned_by": "google",
}
)
return models
@@ -89,55 +93,54 @@ async def _fetch_models_from_provider(
) -> list[dict]:
"""
从提供商 API 获取模型列表
Args:
base_url: 提供商的基础 URL
api_key: API 密钥
endpoint: 获取模型列表的端点
parser: 响应解析器类型 ('openai' | 'gemini')
client_type: 客户端类型 ('openai' | 'gemini')
Returns:
模型列表
"""
url = f"{_normalize_url(base_url)}{endpoint}"
# 根据客户端类型设置请求头
headers = {}
params = {}
if client_type == "gemini":
# Gemini 使用 URL 参数传递 API Key
params["key"] = api_key
else:
# OpenAI 兼容格式使用 Authorization 头
headers["Authorization"] = f"Bearer {api_key}"
try:
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.get(url, headers=headers, params=params)
response.raise_for_status()
data = response.json()
except httpx.TimeoutException:
raise HTTPException(status_code=504, detail="请求超时，请稍后重试")
except httpx.TimeoutException as e:
raise HTTPException(status_code=504, detail="请求超时，请稍后重试") from e
except httpx.HTTPStatusError as e:
# 注意：使用 502 Bad Gateway 而不是原始的 401/403，
# 因为前端的 fetchWithAuth 会把 401 当作 WebUI 认证失败处理
if e.response.status_code == 401:
raise HTTPException(status_code=502, detail="API Key 无效或已过期")
raise HTTPException(status_code=502, detail="API Key 无效或已过期") from e
elif e.response.status_code == 403:
raise HTTPException(status_code=502, detail="没有权限访问模型列表，请检查 API Key 权限")
raise HTTPException(status_code=502, detail="没有权限访问模型列表，请检查 API Key 权限") from e
elif e.response.status_code == 404:
raise HTTPException(status_code=502, detail="该提供商不支持获取模型列表")
raise HTTPException(status_code=502, detail="该提供商不支持获取模型列表") from e
else:
raise HTTPException(
status_code=502,
detail=f"上游服务请求失败 ({e.response.status_code}): {e.response.text[:200]}"
)
status_code=502, detail=f"上游服务请求失败 ({e.response.status_code}): {e.response.text[:200]}"
) from e
except Exception as e:
logger.error(f"获取模型列表失败: {e}")
raise HTTPException(status_code=500, detail=f"获取模型列表失败: {str(e)}")
raise HTTPException(status_code=500, detail=f"获取模型列表失败: {str(e)}") from e
# 根据解析器类型解析响应
if parser == "openai":
return _parse_openai_response(data)
@@ -150,26 +153,26 @@ async def _fetch_models_from_provider(
def _get_provider_config(provider_name: str) -> Optional[dict]:
"""
从 model_config.toml 获取指定提供商的配置
Args:
provider_name: 提供商名称
Returns:
提供商配置，如果未找到则返回 None
"""
config_path = os.path.join(CONFIG_DIR, "model_config.toml")
if not os.path.exists(config_path):
return None
try:
with open(config_path, "r", encoding="utf-8") as f:
config_data = tomlkit.load(f)
providers = config_data.get("api_providers", [])
for provider in providers:
if provider.get("name") == provider_name:
return dict(provider)
return None
except Exception as e:
logger.error(f"读取提供商配置失败: {e}")
@@ -184,23 +187,23 @@ async def get_provider_models(
):
"""
获取指定提供商的可用模型列表
通过提供商名称查找配置，然后请求对应的模型列表端点
"""
# 获取提供商配置
provider_config = _get_provider_config(provider_name)
if not provider_config:
raise HTTPException(status_code=404, detail=f"未找到提供商: {provider_name}")
base_url = provider_config.get("base_url")
api_key = provider_config.get("api_key")
client_type = provider_config.get("client_type", "openai")
if not base_url:
raise HTTPException(status_code=400, detail="提供商配置缺少 base_url")
if not api_key:
raise HTTPException(status_code=400, detail="提供商配置缺少 api_key")
# 获取模型列表
models = await _fetch_models_from_provider(
base_url=base_url,
@@ -209,7 +212,7 @@ async def get_provider_models(
parser=parser,
client_type=client_type,
)
return {
"success": True,
"models": models,
@@ -236,7 +239,7 @@ async def get_models_by_url(
parser=parser,
client_type=client_type,
)
return {
"success": True,
"models": models,
@@ -251,11 +254,11 @@ async def test_provider_connection(
):
"""
测试提供商连接状态
分两步测试：
1. 网络连通性测试：向 base_url 发送请求，检查是否能连接
2. API Key 验证（可选）：如果提供了 api_key，尝试获取模型列表验证 Key 是否有效
返回：
- network_ok: 网络是否连通
- api_key_valid: API Key 是否有效（仅在提供 api_key 时返回）
@@ -263,11 +266,11 @@ async def test_provider_connection(
- error: 错误信息（如果有）
"""
import time
base_url = _normalize_url(base_url)
if not base_url:
raise HTTPException(status_code=400, detail="base_url 不能为空")
result = {
"network_ok": False,
"api_key_valid": None,
@@ -275,7 +278,7 @@ async def test_provider_connection(
"error": None,
"http_status": None,
}
# 第一步：测试网络连通性
try:
start_time = time.time()
@@ -283,11 +286,11 @@ async def test_provider_connection(
# 尝试 GET 请求 base_url（不需要 API Key）
response = await client.get(base_url)
latency = (time.time() - start_time) * 1000
result["network_ok"] = True
result["latency_ms"] = round(latency, 2)
result["http_status"] = response.status_code
except httpx.ConnectError as e:
result["error"] = f"连接失败：无法连接到服务器 ({str(e)})"
return result
@@ -300,7 +303,7 @@ async def test_provider_connection(
except Exception as e:
result["error"] = f"未知错误：{str(e)}"
return result
# 第二步：如果提供了 API Key，验证其有效性
if api_key:
try:
@@ -313,7 +316,7 @@ async def test_provider_connection(
# 尝试获取模型列表
models_url = f"{base_url}/models"
response = await client.get(models_url, headers=headers)
if response.status_code == 200:
result["api_key_valid"] = True
elif response.status_code in (401, 403):
@@ -322,12 +325,12 @@ async def test_provider_connection(
else:
# 其他状态码，可能是端点不支持，但 Key 可能是有效的
result["api_key_valid"] = None
except Exception as e:
# API Key 验证失败不影响网络连通性结果
logger.warning(f"API Key 验证失败: {e}")
result["api_key_valid"] = None
return result
@@ -342,10 +345,10 @@ async def test_provider_connection_by_name(
model_config_path = os.path.join(CONFIG_DIR, "model_config.toml")
if not os.path.exists(model_config_path):
raise HTTPException(status_code=404, detail="配置文件不存在")
with open(model_config_path, "r", encoding="utf-8") as f:
config = tomlkit.load(f)
# 查找提供商
providers = config.get("api_providers", [])
provider = None
@@ -353,15 +356,15 @@ async def test_provider_connection_by_name(
if p.get("name") == provider_name:
provider = p
break
if not provider:
raise HTTPException(status_code=404, detail=f"未找到提供商: {provider_name}")
base_url = provider.get("base_url", "")
api_key = provider.get("api_key", "")
if not base_url:
raise HTTPException(status_code=400, detail="提供商配置缺少 base_url")
# 调用测试接口
return await test_provider_connection(base_url=base_url, api_key=api_key if api_key else None)