feat: 完成AI模型清单后端
- 扩展 appconfig 读写 config/ai_models.json - 支持模型增删改、分类过滤、key 打码展示和 get_model 明文返回 - 增加模型唯一性、类别保底和测试连接错误返回 - 更新任务看板、模块合约、当前状态和进度记录
This commit is contained in:
chengma
+256
-3
@@ -1,16 +1,21 @@
|
||||
"""Application-level configuration for cmshopee.
|
||||
|
||||
This module owns `config.json`, which stores local app settings and AI role/
|
||||
generation parameters. AI provider definitions and API keys are intentionally
|
||||
kept out of this file; T-005 will own `config/ai_models.json`.
|
||||
This module owns `config.json` plus `config/ai_models.json`. `config.json`
|
||||
stores local app settings and AI role/generation parameters. AI provider
|
||||
definitions and local plaintext API keys live in ignored `config/ai_models.json`.
|
||||
"""
|
||||
|
||||
import copy
|
||||
import json
|
||||
import os
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
|
||||
|
||||
CONFIG_PATH = "config.json"
|
||||
AI_MODELS_PATH = os.path.join("config", "ai_models.json")
|
||||
CATEGORIES = {"text", "image"}
|
||||
API_TYPES = {"chat", "images_edits", "auto"}
|
||||
|
||||
DEFAULT_CONFIG = {
|
||||
"chrome_path": r"C:\Program Files\Google\Chrome\Application\chrome.exe",
|
||||
@@ -37,6 +42,35 @@ DEFAULT_CONFIG = {
|
||||
},
|
||||
}
|
||||
|
||||
DEFAULT_AI_MODELS_CONFIG = {
|
||||
"models": [
|
||||
{
|
||||
"name": "GPT-5.5 文本",
|
||||
"category": "text",
|
||||
"enabled": True,
|
||||
"url": "",
|
||||
"model": "",
|
||||
"api_key": "",
|
||||
"api_type": "chat",
|
||||
"connect_timeout_seconds": 30,
|
||||
"timeout_seconds": 0,
|
||||
"extra_body": {},
|
||||
},
|
||||
{
|
||||
"name": "Nano Banana 2",
|
||||
"category": "image",
|
||||
"enabled": True,
|
||||
"url": "https://api.vectorengine.ai/v1/chat/completions",
|
||||
"model": "gemini-3.1-flash-image-preview",
|
||||
"api_key": "",
|
||||
"api_type": "auto",
|
||||
"connect_timeout_seconds": 30,
|
||||
"timeout_seconds": 0,
|
||||
"extra_body": {},
|
||||
},
|
||||
]
|
||||
}
|
||||
|
||||
SECRET_FIELD_NAMES = {"api_key", "apikey", "key", "token", "password"}
|
||||
|
||||
|
||||
@@ -50,6 +84,12 @@ def default_config() -> dict:
|
||||
return copy.deepcopy(DEFAULT_CONFIG)
|
||||
|
||||
|
||||
def default_ai_models_config() -> dict:
|
||||
"""Return a new copy of the default AI model list."""
|
||||
|
||||
return copy.deepcopy(DEFAULT_AI_MODELS_CONFIG)
|
||||
|
||||
|
||||
def _deep_merge(defaults, loaded):
|
||||
if not isinstance(defaults, dict):
|
||||
return copy.deepcopy(loaded) if loaded is not None else copy.deepcopy(defaults)
|
||||
@@ -163,3 +203,216 @@ def response_timeout(config=None) -> int:
|
||||
if resolution not in timeouts:
|
||||
raise ConfigError(f"未配置分辨率 {resolution} 的返回超时")
|
||||
return int(timeouts[resolution])
|
||||
|
||||
|
||||
def _normalize_ai_model(model):
|
||||
if not isinstance(model, dict):
|
||||
raise ConfigError("AI 模型定义必须是对象")
|
||||
normalized = {
|
||||
"name": str(model.get("name", "")).strip(),
|
||||
"category": str(model.get("category", "")).strip(),
|
||||
"enabled": bool(model.get("enabled", True)),
|
||||
"url": str(model.get("url", "")).strip(),
|
||||
"model": str(model.get("model", "")).strip(),
|
||||
"api_key": str(model.get("api_key", "")),
|
||||
"api_type": str(model.get("api_type", "auto")).strip() or "auto",
|
||||
"connect_timeout_seconds": int(model.get("connect_timeout_seconds", 30) or 30),
|
||||
"timeout_seconds": int(model.get("timeout_seconds", 0) or 0),
|
||||
"extra_body": copy.deepcopy(model.get("extra_body", {})),
|
||||
}
|
||||
if not normalized["name"]:
|
||||
raise ConfigError("AI 模型 name 不能为空")
|
||||
if normalized["category"] not in CATEGORIES:
|
||||
raise ConfigError("AI 模型 category 必须是 text 或 image")
|
||||
if normalized["api_type"] not in API_TYPES:
|
||||
raise ConfigError("AI 模型 api_type 必须是 chat、images_edits 或 auto")
|
||||
if normalized["connect_timeout_seconds"] <= 0:
|
||||
raise ConfigError("AI 模型 connect_timeout_seconds 必须大于 0")
|
||||
if normalized["timeout_seconds"] < 0:
|
||||
raise ConfigError("AI 模型 timeout_seconds 不能小于 0")
|
||||
if not isinstance(normalized["extra_body"], dict):
|
||||
raise ConfigError("AI 模型 extra_body 必须是对象")
|
||||
return normalized
|
||||
|
||||
|
||||
def _normalize_ai_models_config(config):
|
||||
models = config.get("models") if isinstance(config, dict) else None
|
||||
if not isinstance(models, list):
|
||||
raise ConfigError("config/ai_models.json 必须包含 models 列表")
|
||||
normalized = {"models": [_normalize_ai_model(model) for model in models]}
|
||||
_assert_unique_model_names(normalized["models"])
|
||||
_assert_required_categories(normalized["models"])
|
||||
return normalized
|
||||
|
||||
|
||||
def _assert_unique_model_names(models):
|
||||
names = [model["name"] for model in models]
|
||||
duplicates = sorted({name for name in names if names.count(name) > 1})
|
||||
if duplicates:
|
||||
raise ConfigError(f"AI 模型 name 重复: {', '.join(duplicates)}")
|
||||
|
||||
|
||||
def _assert_required_categories(models):
|
||||
enabled_categories = {
|
||||
model["category"] for model in models if model.get("enabled", True)
|
||||
}
|
||||
missing = sorted(CATEGORIES - enabled_categories)
|
||||
if missing:
|
||||
raise ConfigError(
|
||||
"AI 模型清单至少需要启用一个 text 和一个 image 模型，缺少: "
|
||||
+ ", ".join(missing)
|
||||
)
|
||||
|
||||
|
||||
def _model_index(models, name):
|
||||
for index, model in enumerate(models):
|
||||
if model["name"] == name:
|
||||
return index
|
||||
raise ConfigError(f"AI 模型不存在: {name}")
|
||||
|
||||
|
||||
def _mask_api_key(api_key):
|
||||
if not api_key:
|
||||
return ""
|
||||
if len(api_key) <= 8:
|
||||
return "***"
|
||||
return f"{api_key[:4]}***{api_key[-4:]}"
|
||||
|
||||
|
||||
def _public_model(model):
|
||||
public = copy.deepcopy(model)
|
||||
public["api_key"] = _mask_api_key(public.get("api_key", ""))
|
||||
public["api_key_set"] = bool(model.get("api_key"))
|
||||
return public
|
||||
|
||||
|
||||
def save_ai_models_config(config, path=AI_MODELS_PATH) -> dict:
|
||||
"""Persist AI model definitions, including local plaintext API keys."""
|
||||
|
||||
normalized = _normalize_ai_models_config(config)
|
||||
directory = os.path.dirname(os.path.abspath(path))
|
||||
if directory:
|
||||
os.makedirs(directory, exist_ok=True)
|
||||
with open(path, "w", encoding="utf-8") as fh:
|
||||
json.dump(normalized, fh, ensure_ascii=False, indent=2)
|
||||
fh.write("\n")
|
||||
return normalized
|
||||
|
||||
|
||||
def load_ai_models_config(path=AI_MODELS_PATH) -> dict:
|
||||
"""Load AI model definitions, writing defaults first if missing."""
|
||||
|
||||
if not os.path.exists(path):
|
||||
return save_ai_models_config(default_ai_models_config(), path=path)
|
||||
with open(path, "r", encoding="utf-8") as fh:
|
||||
try:
|
||||
loaded = json.load(fh)
|
||||
except json.JSONDecodeError as exc:
|
||||
raise ConfigError(f"AI 模型清单不是有效 JSON: {path}") from exc
|
||||
return _normalize_ai_models_config(loaded)
|
||||
|
||||
|
||||
def list_ai_models(category=None, path=AI_MODELS_PATH, reveal_api_key=False):
|
||||
"""Return AI models, optionally filtered by category."""
|
||||
|
||||
if category is not None and category not in CATEGORIES:
|
||||
raise ConfigError("category 必须是 text 或 image")
|
||||
models = load_ai_models_config(path)["models"]
|
||||
filtered = [
|
||||
copy.deepcopy(model)
|
||||
for model in models
|
||||
if category is None or model["category"] == category
|
||||
]
|
||||
if reveal_api_key:
|
||||
return filtered
|
||||
return [_public_model(model) for model in filtered]
|
||||
|
||||
|
||||
def add_ai_model(model, path=AI_MODELS_PATH) -> None:
|
||||
config = load_ai_models_config(path)
|
||||
normalized = _normalize_ai_model(model)
|
||||
if any(item["name"] == normalized["name"] for item in config["models"]):
|
||||
raise ConfigError(f"AI 模型 name 已存在: {normalized['name']}")
|
||||
config["models"].append(normalized)
|
||||
save_ai_models_config(config, path=path)
|
||||
|
||||
|
||||
def update_ai_model(model_name, path=AI_MODELS_PATH, **fields) -> None:
|
||||
if not fields:
|
||||
return
|
||||
config = load_ai_models_config(path)
|
||||
index = _model_index(config["models"], model_name)
|
||||
updated = copy.deepcopy(config["models"][index])
|
||||
updated.update(fields)
|
||||
normalized = _normalize_ai_model(updated)
|
||||
if normalized["name"] != model_name and any(
|
||||
model["name"] == normalized["name"] for model in config["models"]
|
||||
):
|
||||
raise ConfigError(f"AI 模型 name 已存在: {normalized['name']}")
|
||||
config["models"][index] = normalized
|
||||
save_ai_models_config(config, path=path)
|
||||
|
||||
|
||||
def delete_ai_model(name, path=AI_MODELS_PATH) -> None:
|
||||
config = load_ai_models_config(path)
|
||||
index = _model_index(config["models"], name)
|
||||
remaining = config["models"][:index] + config["models"][index + 1 :]
|
||||
_assert_required_categories(remaining)
|
||||
save_ai_models_config({"models": remaining}, path=path)
|
||||
|
||||
|
||||
def get_model(name, path=AI_MODELS_PATH) -> dict:
|
||||
"""Return the model definition including api_key. Callers must not log it."""
|
||||
|
||||
models = load_ai_models_config(path)["models"]
|
||||
return copy.deepcopy(models[_model_index(models, name)])
|
||||
|
||||
|
||||
def _test_request_payload(model):
|
||||
if model["api_type"] == "images_edits":
|
||||
payload = {"model": model["model"], "prompt": "ping"}
|
||||
payload.update(model.get("extra_body", {}))
|
||||
return payload
|
||||
payload = {
|
||||
"model": model["model"],
|
||||
"messages": [{"role": "user", "content": "ping"}],
|
||||
}
|
||||
payload.update(model.get("extra_body", {}))
|
||||
return payload
|
||||
|
||||
|
||||
def test_ai_model(name, path=AI_MODELS_PATH) -> dict:
|
||||
"""Send a minimal request to the configured model endpoint."""
|
||||
|
||||
try:
|
||||
model = get_model(name, path=path)
|
||||
missing = [
|
||||
field
|
||||
for field in ("url", "model", "api_key")
|
||||
if not str(model.get(field, "")).strip()
|
||||
]
|
||||
if missing:
|
||||
return {"ok": False, "error": "模型缺少字段: " + ", ".join(missing)}
|
||||
data = json.dumps(_test_request_payload(model), ensure_ascii=False).encode(
|
||||
"utf-8"
|
||||
)
|
||||
request = urllib.request.Request(
|
||||
model["url"],
|
||||
data=data,
|
||||
headers={
|
||||
"Authorization": "Bearer " + model["api_key"],
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
method="POST",
|
||||
)
|
||||
with urllib.request.urlopen(
|
||||
request, timeout=int(model["connect_timeout_seconds"])
|
||||
) as response:
|
||||
response.read(1024)
|
||||
return {"ok": 200 <= response.status < 300, "status": response.status}
|
||||
except urllib.error.HTTPError as exc:
|
||||
return {"ok": False, "status": exc.code, "error": f"HTTP {exc.code}"}
|
||||
except urllib.error.URLError as exc:
|
||||
return {"ok": False, "error": str(exc.reason)}
|
||||
except Exception as exc:
|
||||
return {"ok": False, "error": str(exc)}
|
||||
|
||||
Reference in New Issue
Block a user