```

feat(api): 统一图像生成功能至generations接口

- 将gpt-image-2模型统一使用AI_BASE_URL下的generations接口，
  不再区分有无参考图走不同接口
- 在payload中添加response_format字段并移除output_format字段
- 更新图像数据处理逻辑，将image_data转换为image数组格式

refactor(config): 更新图像生成API配置

- 将GPT_IMAGE_GENERATION_API从硬编码URL改为使用AI_BASE_URL动态构建

refactor(generation_service): 简化图像生成请求验证逻辑

- 移除has_reference_images判断逻辑，统一使用generations接口
- 积分模式下gpt-image-2改用PREMIUM_KEY而非专用编辑key
- 移除不必要的image_data变量声明

refactor(task_service): 移除废弃的图像处理函数

- 删除_base64相关处理函数(_decode_base64_image等)
- 移除_minio上传相关函数(_upload_image_bytes_to_minio等)
- 移除gpt-image-2特定处理逻辑(_process_gpt_image_edit)
- 简化图像生成任务处理流程
```

blueprints/api.py

@@ -96,12 +96,12 @@ def generate():
             payload = {
                 "prompt": prompt,
                 "model": model_value,
+                "response_format": "url",
                 "size": data.get("size") or "2048x2048",
                 "quality": "high",
-                "output_format": "png",
             }
             if image_data:
-                payload["images"] = [{"image_url": img} for img in image_data]
+                payload["image"] = [img.split(",", 1)[1] if "," in img else img for img in image_data]
         else:
             payload = {
                 "prompt": prompt,

config.py

@@ -40,7 +40,7 @@ class Config:
     CHAT_API = f"{AI_BASE_URL}/v1/chat/completions"
     VIDEO_GEN_API = f"{AI_BASE_URL}/v2/videos/generations"
     VIDEO_POLL_API = f"{AI_BASE_URL}/v2/videos/generations/{{task_id}}"
-    GPT_IMAGE_GENERATION_API = "https://nas.4x4g.com:8317/v1/images/generations"
+    GPT_IMAGE_GENERATION_API = f"{AI_BASE_URL}/v1/images/generations"
     GPT_IMAGE_EDIT_API = "https://nas.4x4g.com:8317/v1/images/edits"
     GPT_IMAGE_EDIT_KEY = "sk-vDmmYNzoe7UjWskKx"
     

services/generation_service.py

@@ -34,15 +34,13 @@ def validate_generation_request(user, data):
     model_value = data.get("model")
     is_gemini_flash_image_preview = model_value == "gemini-3.1-flash-image-preview"
     is_gpt_image_edit = model_value == "gpt-image-2"
-    image_data = data.get("image_data", [])
-    has_reference_images = bool(image_data)
 
     if isinstance(is_premium, str):
         is_premium = is_premium.lower() in ("1", "true", "yes", "on")
 
-    # gpt-image-2 有参考图走 edits，无参考图走 generations；其它模型沿用现有 generations 接口
+    # gpt-image-2 统一走 AI_BASE_URL 下的 generations 接口，参考图放在 image 数组中。
     target_api = (
-        (Config.GPT_IMAGE_EDIT_API if has_reference_images else Config.GPT_IMAGE_GENERATION_API)
+        Config.GPT_IMAGE_GENERATION_API
         if is_gpt_image_edit
         else Config.AI_API
     )
@@ -62,9 +60,9 @@ def validate_generation_request(user, data):
         if user.points <= 0:
             return None, None, 0, False, "可用积分已耗尽，请充值或切换至自定义 Key 模式"
 
-        # 积分模式下，gpt-image-2 使用固定的专用 key
+        # 积分模式下，gpt-image-2 跟随 AI_BASE_URL 的内置高级线路。
         if is_gpt_image_edit:
-            api_key = Config.GPT_IMAGE_EDIT_KEY
+            api_key = Config.PREMIUM_KEY
         elif is_premium:
             api_key = (
                 Config.GEMINI_FLASH_IMAGE_PREMIUM_KEY
@@ -77,7 +75,7 @@ def validate_generation_request(user, data):
             api_key = Config.TRIAL_KEY
 
         target_api = (
-            (Config.GPT_IMAGE_EDIT_API if has_reference_images else Config.GPT_IMAGE_GENERATION_API)
+            Config.GPT_IMAGE_GENERATION_API
             if is_gpt_image_edit
             else Config.TRIAL_API
         )

services/task_service.py

@@ -4,8 +4,6 @@ import json
 import requests
 import io
 import time
-import base64
-import binascii
 import threading
 from urllib.parse import quote
 from requests.exceptions import ConnectionError, ConnectTimeout
@@ -48,199 +46,6 @@ def _extract_error_detail(data, default="未知错误"):
     return default
 
 
-def _guess_image_extension(content_type, fallback=".png"):
-    mapping = {
-        "image/png": ".png",
-        "image/jpeg": ".jpg",
-        "image/jpg": ".jpg",
-        "image/webp": ".webp",
-    }
-    return mapping.get((content_type or "").lower(), fallback)
-
-
-def _put_object_bytes(object_name, content, content_type):
-    s3_client.put_object(
-        Bucket=Config.MINIO["bucket"],
-        Key=object_name,
-        Body=content,
-        ContentType=content_type,
-    )
-
-
-def _upload_image_bytes_to_minio(content, content_type="image/png"):
-    ext = _guess_image_extension(content_type)
-    base_filename = f"gen-{uuid.uuid4().hex}"
-    full_filename = f"{base_filename}{ext}"
-    thumb_filename = f"{base_filename}-thumb.jpg"
-
-    try:
-        _put_object_bytes(full_filename, content, content_type)
-    except Exception as upload_e:
-        raise Exception(f"MinIO 上传原图失败: {upload_e}")
-
-    full_url = f"{Config.MINIO['public_url']}{quote(full_filename)}"
-    thumb_url = full_url
-
-    try:
-        img = Image.open(io.BytesIO(content))
-        if img.mode in ("RGBA", "P"):
-            img = img.convert("RGB")
-
-        w, h = img.size
-        if w > 400:
-            ratio = 400 / float(w)
-            img.thumbnail((400, int(h * ratio)), Image.Resampling.LANCZOS)
-
-        thumb_io = io.BytesIO()
-        img.save(thumb_io, format="JPEG", quality=80, optimize=True)
-        thumb_io.seek(0)
-
-        _put_object_bytes(thumb_filename, thumb_io.getvalue(), "image/jpeg")
-        thumb_url = f"{Config.MINIO['public_url']}{quote(thumb_filename)}"
-    except Exception as thumb_e:
-        system_logger.warning(f"缩略图生成失败: {thumb_e}")
-
-    return {"url": full_url, "thumb": thumb_url}
-
-
-def _decode_base64_image(image_value):
-    raw_value = image_value or ""
-    content_type = "image/png"
-    if raw_value.startswith("data:") and "," in raw_value:
-        header, raw_value = raw_value.split(",", 1)
-        if ";" in header:
-            content_type = header[5:].split(";", 1)[0] or content_type
-
-    # 清理换行和空白，兼容部分服务端返回的分段 Base64
-    raw_value = "".join(raw_value.strip().split())
-    padding = (-len(raw_value)) % 4
-    if padding:
-        raw_value += "=" * padding
-
-    try:
-        decoded = base64.b64decode(raw_value, validate=True)
-    except binascii.Error:
-        try:
-            decoded = base64.urlsafe_b64decode(raw_value)
-        except Exception as decode_e:
-            raise Exception(f"Base64 解码失败: {decode_e}")
-
-    if not decoded:
-        raise Exception("Base64 解码失败: 结果为空")
-
-    # 提前验证图片内容，避免把损坏字节流误判成 MinIO 上传问题
-    try:
-        img = Image.open(io.BytesIO(decoded))
-        img.verify()
-    except Exception as verify_e:
-        raise Exception(f"Base64 已解码，但图片数据无效: {verify_e}")
-
-    return decoded, content_type
-
-
-def _extract_b64_images(result):
-    images = []
-    for item in result.get("data", []) or []:
-        if not isinstance(item, dict):
-            continue
-        if item.get("b64_json"):
-            images.append(item["b64_json"])
-        elif item.get("image_base64"):
-            images.append(item["image_base64"])
-    return images
-
-
-def _extract_image_urls(result):
-    urls = []
-    for item in result.get("data", []) or []:
-        if isinstance(item, dict) and item.get("url"):
-            urls.append(item["url"])
-    return urls
-
-
-def _process_gpt_image_edit(app, user_id, task_id, payload, api_key, target_api, cost, use_trial):
-    headers = {
-        "Authorization": f"Bearer {api_key}",
-        "Content-Type": "application/json",
-    }
-    # gpt-image-2 的 edits 接口走站内地址，直连比经过公共代理更稳定
-    response = requests.post(
-        target_api,
-        json=payload,
-        headers=headers,
-        timeout=Config.PROXY_TIMEOUT_GENERATION,
-    )
-
-    if response.status_code != 200:
-        try:
-            error_data = response.json()
-        except Exception:
-            error_data = {"message": response.text}
-        raise Exception(f"gpt-image-2 请求失败: {_extract_error_detail(error_data)}")
-
-    result = response.json()
-    b64_images = _extract_b64_images(result)
-    if not b64_images:
-        url_images = _extract_image_urls(result)
-        if not url_images:
-            raise Exception("gpt-image-2 未返回可用图片数据")
-
-        processed_data = [{"url": image_url, "thumb": image_url} for image_url in url_images]
-        new_record = GenerationRecord(
-            user_id=user_id,
-            prompt=payload.get("prompt"),
-            model=payload.get("model"),
-            cost=cost,
-            image_urls=json.dumps(processed_data),
-        )
-        db.session.add(new_record)
-        db.session.commit()
-
-        system_logger.info("gpt-image-2 生图任务完成", user_id=user_id, task_id=task_id, model=payload.get("model"))
-        redis_client.setex(
-            f"task:{task_id}",
-            3600,
-            json.dumps({"status": "complete", "urls": url_images, "record_id": new_record.id}),
-        )
-        return
-
-    processed_data = []
-    final_urls = []
-    output_format = (payload.get("output_format") or "png").lower()
-    default_content_type = f"image/{output_format}" if output_format in ("png", "jpeg", "jpg", "webp") else "image/png"
-
-    for b64_image in b64_images:
-        try:
-            image_bytes, content_type = _decode_base64_image(b64_image)
-        except Exception as decode_e:
-            raise Exception(f"gpt-image-2 返回的 Base64 图片无效: {decode_e}")
-        if content_type == "image/png" and default_content_type != "image/png":
-            content_type = default_content_type
-        try:
-            uploaded = _upload_image_bytes_to_minio(image_bytes, content_type)
-        except Exception as upload_e:
-            raise Exception(f"gpt-image-2 结果已生成，但同步到 MinIO 失败: {upload_e}")
-        processed_data.append(uploaded)
-        final_urls.append(uploaded["url"])
-
-    new_record = GenerationRecord(
-        user_id=user_id,
-        prompt=payload.get("prompt"),
-        model=payload.get("model"),
-        cost=cost,
-        image_urls=json.dumps(processed_data),
-    )
-    db.session.add(new_record)
-    db.session.commit()
-
-    system_logger.info("gpt-image-2 生图任务完成", user_id=user_id, task_id=task_id, model=payload.get("model"))
-    redis_client.setex(
-        f"task:{task_id}",
-        3600,
-        json.dumps({"status": "complete", "urls": final_urls, "record_id": new_record.id}),
-    )
-
-
 def sync_images_background(app, record_id, raw_urls):
     """后台同步图片至 MinIO，并生成缩略图，带重试机制"""
     with app.app_context():
@@ -324,10 +129,6 @@ def process_image_generation(app, user_id, task_id, payload, api_key, target_api
         redis_client.setex(f"task:{task_id}", 3600, json.dumps({"status": "processing", "message": "任务已提交，正在排队处理..."}))
         try:
             # 1. 提交异步请求 (带重试机制)
-            if payload.get('model') == 'gpt-image-2':
-                _process_gpt_image_edit(app, user_id, task_id, payload, api_key, target_api, cost, use_trial)
-                return
-
             submit_resp = None
             last_error = None
             active_api = target_api