feat(settings): support gateway source switching

app/ai.py

@@ -64,6 +64,88 @@ CMHUB_HTTP_POOL_SIZE = 32
 
 _CMHUB_SESSION = None
 _CMHUB_SESSION_LOCK = threading.Lock()
+_RUNTIME_AI_SNAPSHOT_KEY = "_cmshopee_ai_runtime"
+
+
+def freeze_runtime_config(
+    config=None,
+    *,
+    cmhub_config_path=appconfig.CMHUB_CONFIG_PATH,
+    models_path=appconfig.AI_MODELS_PATH,
+    include_cmhub=None,
+    include_direct_models=None,
+):
+    """Return an in-memory AI configuration snapshot for one worker run.
+
+    Application config does not contain the default gateway key and direct model
+    definitions are stored in a separate file. Copy only the values used by the
+    worker into its private config so saving settings cannot change its endpoint,
+    model or credential. Callers must never persist this result.
+    """
+
+    source = appconfig.load_config() if config is None else config
+    configured_cmhub_path = source.get("cmhub_config_path")
+    configured_models_path = source.get("ai_models_path")
+    if configured_cmhub_path and (
+        not cmhub_config_path or cmhub_config_path == appconfig.CMHUB_CONFIG_PATH
+    ):
+        cmhub_config_path = configured_cmhub_path
+    if configured_models_path and (
+        not models_path or models_path == appconfig.AI_MODELS_PATH
+    ):
+        models_path = configured_models_path
+    cmhub_config_path = cmhub_config_path or appconfig.cmhub_config_file_path(source)
+    models_path = models_path or appconfig.ai_models_config_path(source)
+    snapshot = copy.deepcopy(source)
+    backend = appconfig.ai_backend(source)
+    if include_cmhub is None:
+        include_cmhub = backend == "cmhub"
+    if include_direct_models is None:
+        include_direct_models = backend == "direct"
+    runtime = {}
+    if include_cmhub:
+        runtime["cmhub_api_key"] = appconfig.get_cmhub_api_key(path=cmhub_config_path)
+    if include_direct_models:
+        runtime["direct_models"] = appconfig.list_ai_models(
+            path=models_path,
+            reveal_api_key=True,
+        )
+    snapshot[_RUNTIME_AI_SNAPSHOT_KEY] = runtime
+    return snapshot
+
+
+def validate_direct_generation_config(
+    config,
+    generate_mode,
+    *,
+    models_path=appconfig.AI_MODELS_PATH,
+):
+    """Fail before a direct batch starts when its selected model is unusable."""
+
+    cfg = appconfig.load_config() if config is None else config
+    ai_cfg = appconfig.ai_config(cfg)
+    if _ai_backend(ai_cfg) != "direct":
+        return
+    mode = appconfig.normalize_generate_mode(generate_mode)
+    required = []
+    if appconfig.generate_mode_includes_title(mode):
+        required.append(("text", "标题"))
+    if appconfig.generate_mode_includes_cover(mode):
+        required.append(("image", "封面"))
+    models = _runtime_direct_models(cfg)
+    errors = []
+    for category, label in required:
+        try:
+            _role_model(
+                category,
+                ai_cfg.get("default_%s_model" % category),
+                models_path,
+                models=models,
+            )
+        except Exception as exc:
+            errors.append("%s模型%s" % (label, str(exc)))
+    if errors:
+        raise AIError("；".join(errors))
 
 
 def _cmhub_session():
@@ -134,7 +216,12 @@ def gen_title(
             on_event=on_event,
         )
     _notify_step(on_step, "load_text_model")
-    model = _role_model("text", ai_cfg.get("default_text_model"), models_path)
+    model = _role_model(
+        "text",
+        ai_cfg.get("default_text_model"),
+        models_path,
+        models=_runtime_direct_models(cfg),
+    )
     _notify_step(on_step, "title_build_request")
     payload = _chat_payload(
         model,
@@ -248,7 +335,12 @@ def gen_cover(
             on_event=on_event,
         )
     _notify_step(on_step, "load_image_model")
-    model = _role_model("image", ai_cfg.get("default_image_model"), models_path)
+    model = _role_model(
+        "image",
+        ai_cfg.get("default_image_model"),
+        models_path,
+        models=_runtime_direct_models(cfg),
+    )
     resolution = str(resolution or ai_cfg.get("resolution", "1k"))
     quality = _jpg_quality(jpg_quality if jpg_quality is not None else ai_cfg.get("jpg_quality", 90))
     attempts = _attempt_count(ai_cfg, retry)
@@ -1435,7 +1527,11 @@ def _vision_cmhub_error(exc):
 
 def _cmhub_runtime(config, operation, cmhub_config_path):
     hub = appconfig.cmhub_config(config)
-    api_key = appconfig.get_cmhub_api_key(path=cmhub_config_path)
+    runtime = _runtime_ai_snapshot(config)
+    if runtime is not None and "cmhub_api_key" in runtime:
+        api_key = str(runtime.get("cmhub_api_key") or "")
+    else:
+        api_key = appconfig.get_cmhub_api_key(path=cmhub_config_path)
     operation_config = {
         "title": ("title_alias", "生文别名"),
         "image": ("image_alias", "生图别名"),
@@ -2171,10 +2267,33 @@ def _assert_public_ip(value):
 def _redact_cmhub(text, api_key):
     return appconfig.redact_secrets(text, [api_key])
 
-def _role_model(category, name, models_path):
+def _runtime_ai_snapshot(config):
+    if not isinstance(config, dict):
+        return None
+    runtime = config.get(_RUNTIME_AI_SNAPSHOT_KEY)
+    return runtime if isinstance(runtime, dict) else None
+
+
+def _runtime_direct_models(config):
+    runtime = _runtime_ai_snapshot(config)
+    if runtime is None or "direct_models" not in runtime:
+        return None
+    models = runtime.get("direct_models")
+    return copy.deepcopy(models) if isinstance(models, list) else []
+
+
+def _role_model(category, name, models_path, *, models=None):
     if not name:
         raise AIError("未配置默认 %s 模型" % category)
-    model = appconfig.get_model(name, path=models_path)
+    if models is None:
+        model = appconfig.get_model(name, path=models_path)
+    else:
+        model = next(
+            (dict(item) for item in models if str(item.get("name") or "") == str(name)),
+            None,
+        )
+        if model is None:
+            raise AIError("AI 模型不存在: %s" % name)
     if model.get("category") != category:
         raise AIError("模型 %s 不是 %s 类别" % (name, category))
     if not model.get("enabled", True):