feat(ai): enforce direct image edit contract

app/ai.py

@@ -136,12 +136,16 @@ def validate_direct_generation_config(
     errors = []
     for category, label in required:
         try:
-            _role_model(
+            model = _role_model(
                 category,
                 ai_cfg.get("default_%s_model" % category),
                 models_path,
                 models=models,
             )
+            if category == "image":
+                compatibility_error = appconfig.image_model_config_error(model)
+                if compatibility_error:
+                    raise AIError(compatibility_error)
         except Exception as exc:
             errors.append("%s模型%s" % (label, str(exc)))
     if errors:
@@ -345,45 +349,28 @@ def gen_cover(
     quality = _jpg_quality(jpg_quality if jpg_quality is not None else ai_cfg.get("jpg_quality", 90))
     attempts = _attempt_count(ai_cfg, retry)
 
-    api_type = model.get("api_type", "auto")
     _notify_step(on_step, "cover_build_request")
-    if api_type == "images_edits":
-        body, content_type = _image_edit_body(model, cover_prompt, old_cover_path, resolution)
-        _notify_step(on_step, "cover_request")
-        data = _call_with_retry(
+    compatibility_error = appconfig.image_model_config_error(model)
+    if compatibility_error:
+        raise AIError(compatibility_error)
+    body, content_type = _image_edit_body(model, cover_prompt, [old_cover_path], resolution)
+    _notify_step(on_step, "cover_request")
+    data = _call_with_retry(
+        model,
+        body,
+        cfg,
+        attempts,
+        request_kind="multipart",
+        content_type=content_type,
+        on_retry=lambda attempt, total_attempts, exc: _notify_retry(
+            on_step,
+            "cover_request",
+            attempt,
+            total_attempts,
+            exc,
             model,
-            body,
-            cfg,
-            attempts,
-            request_kind="multipart",
-            content_type=content_type,
-            on_retry=lambda attempt, total_attempts, exc: _notify_retry(
-                on_step,
-                "cover_request",
-                attempt,
-                total_attempts,
-                exc,
-                model,
-            ),
-        )
-    else:
-        payload = _image_chat_payload(model, cover_prompt, old_cover_path, resolution)
-        _notify_step(on_step, "cover_request")
-        data = _call_with_retry(
-            model,
-            payload,
-            cfg,
-            attempts,
-            request_kind="json",
-            on_retry=lambda attempt, total_attempts, exc: _notify_retry(
-                on_step,
-                "cover_request",
-                attempt,
-                total_attempts,
-                exc,
-                model,
-            ),
-        )
+        ),
+    )
 
     _notify_step(on_step, "cover_parse_response")
     image_bytes = _extract_image_bytes(data, model, cfg)
@@ -1624,7 +1611,7 @@ def _debug_cmhub_image_url_enabled():
 
 
 def _extract_cmhub_image_url(data, base_url):
-    candidate = _find_image_ref(data)
+    candidate = _find_cmhub_image_ref(data)
     if not candidate:
         return ""
     return _normalize_cmhub_image_url(candidate, base_url)
@@ -2554,32 +2541,34 @@ def _chat_payload(model, messages):
     return payload
 
 
-def _image_chat_payload(model, cover_prompt, old_cover_path, resolution):
-    prompt = "%s\n\n目标分辨率：%s。" % (str(cover_prompt or "").strip(), resolution)
-    content = [
-        {"type": "text", "text": prompt.strip()},
-        {
-            "type": "image_url",
-            "image_url": {"url": _image_data_url(old_cover_path)},
-        },
-    ]
-    return _chat_payload(model, [{"role": "user", "content": content}])
-
-
-def _image_edit_body(model, cover_prompt, old_cover_path, resolution):
-    fields = {
+def _image_edit_body(model, cover_prompt, image_paths, resolution):
+    if isinstance(image_paths, (str, bytes, os.PathLike)):
+        image_paths = [image_paths]
+    image_paths = list(image_paths or [])
+    if not image_paths:
+        raise AIError("图片编辑请求至少需要一张本地参考图")
+    fields = copy.deepcopy(model.get("extra_body", {}))
+    fields.update({
         "model": model["model"],
         "prompt": str(cover_prompt or ""),
         "size": _resolution_size_text(resolution),
-    }
-    fields.update(copy.deepcopy(model.get("extra_body", {})))
-    files = {
-        "image": (
-            os.path.basename(old_cover_path),
-            open(old_cover_path, "rb").read(),
-            mimetypes.guess_type(old_cover_path)[0] or "application/octet-stream",
+        "n": "1",
+    })
+    files = []
+    for image_path in image_paths:
+        path = os.fspath(image_path)
+        with open(path, "rb") as fh:
+            data = fh.read()
+        files.append(
+            (
+                "image[]",
+                (
+                    os.path.basename(path),
+                    data,
+                    mimetypes.guess_type(path)[0] or "application/octet-stream",
+                ),
+            )
         )
-    }
     return _multipart_body(fields, files)
 
 
@@ -2595,7 +2584,8 @@ def _multipart_body(fields, files):
                 b"\r\n",
             ]
         )
-    for name, file_info in files.items():
+    file_items = files.items() if isinstance(files, dict) else files
+    for name, file_info in file_items:
         filename, data, content_type = file_info
         chunks.extend(
             [
@@ -2656,15 +2646,38 @@ def _content_text(content):
 def _extract_image_bytes(data, model, config):
     image_ref = _find_image_ref(data)
     if not image_ref:
-        raise AIError("AI 返回中没有图片数据")
+        raise AIError("AI 图片响应不符合 OpenAI 图片编辑接口")
     if image_ref.startswith("data:"):
         return _decode_data_url(image_ref)
     if _looks_base64(image_ref):
         return base64.b64decode(image_ref)
+    parts = urllib.parse.urlsplit(image_ref)
+    if parts.scheme not in {"http", "https"} or not parts.netloc:
+        raise AIError("AI 图片地址只允许 http/https")
     return _download_image(image_ref, model, config)
 
 
 def _find_image_ref(value):
+    """Read only the fixed OpenAI Images API response fields for direct calls."""
+
+    if not isinstance(value, dict):
+        return None
+    data = value.get("data")
+    if not isinstance(data, list):
+        return None
+    for item in data:
+        if not isinstance(item, dict):
+            continue
+        for key in ("b64_json", "url"):
+            candidate = item.get(key)
+            if isinstance(candidate, str) and candidate.strip():
+                return candidate.strip()
+    return None
+
+
+def _find_cmhub_image_ref(value):
+    """Read the default gateway's documented and legacy image response shapes."""
+
     if isinstance(value, dict):
         for key in ("b64_json", "base64", "image_base64", "image", "url"):
             candidate = value.get(key)
@@ -2678,7 +2691,7 @@ def _find_image_ref(value):
             if isinstance(candidate, str):
                 return candidate
         if isinstance(image_url, list):
-            candidate = _find_image_ref_from_list(image_url)
+            candidate = _find_cmhub_image_ref_from_list(image_url)
             if candidate:
                 return candidate
         for key in ("image_urls", "urls"):
@@ -2686,28 +2699,28 @@ def _find_image_ref(value):
             if isinstance(candidate, str) and candidate.strip():
                 return candidate.strip()
             if isinstance(candidate, list):
-                found = _find_image_ref_from_list(candidate)
+                found = _find_cmhub_image_ref_from_list(candidate)
                 if found:
                     return found
         for key in ("result", "data", "choices", "output", "content", "images", "files"):
-            candidate = _find_image_ref(value.get(key))
+            candidate = _find_cmhub_image_ref(value.get(key))
             if candidate:
                 return candidate
         message = value.get("message")
         if message is not None:
-            candidate = _find_image_ref(message)
+            candidate = _find_cmhub_image_ref(message)
             if candidate:
                 return candidate
     elif isinstance(value, list):
-        return _find_image_ref_from_list(value)
+        return _find_cmhub_image_ref_from_list(value)
     return None
 
 
-def _find_image_ref_from_list(values):
+def _find_cmhub_image_ref_from_list(values):
     for item in values:
         if isinstance(item, str) and item.strip():
             return item.strip()
-        candidate = _find_image_ref(item)
+        candidate = _find_cmhub_image_ref(item)
         if candidate:
             return candidate
     return None