feat: AI 生图、图片下载优化

server/api/suite.py

@@ -28,6 +28,7 @@ from schemas.suite import (
     SuiteGenerateRequest,
     SuiteOut,
     SuitePlanRequest,
+    api_model,
     resolve_provider,
 )
 from services.generator import _image_size, GENERATORS, run_suite
@@ -98,9 +99,10 @@ async def generate_suite(req: SuiteGenerateRequest, background: BackgroundTasks)
     spec = PLATFORM_SPECS[req.platform]
 
     settings = get_settings()
-    # 前端只传模型名：已知模型直接路由到对应 provider（如 gpt-image-2-vip → rightapi）
+    # 前端只传模型名：已知模型直接路由到对应 provider（如 gpt-image-2-vip → rightapi、gpt-image-2@pcore → pcore）
     provider_name = resolve_provider(req.model, None, settings.image_provider)
     validate_model(provider_name, req.model)
+    model = api_model(req.model)  # 剥 @provider 后缀，供应商侧只认裸模型名
 
     product_id = (req.product_id or "").strip() or None
     if product_id:
@@ -118,7 +120,7 @@ async def generate_suite(req: SuiteGenerateRequest, background: BackgroundTasks)
         style_prompt=req.style_prompt,
         requirements=req.requirements,
         provider=provider_name,
-        model=req.model,
+        model=model,
         total=len(jobs),
         context=texts_to_raw(req.texts),
         plan=jobs,
@@ -214,7 +216,7 @@ async def suite_image_edit(req: ImageEditSingleRequest) -> ImageEditSingleRespon
     validate_model(provider_name, req.model)
 
     spec = {"lang": "ru", "ratio": "3:4"}  # 试算页固定 Ozon 规格（俄文图内文案 · 3:4）
-    model = req.model
+    model = api_model(req.model)  # 剥 @provider 后缀，供应商侧只认裸模型名
     is_wan = provider_name == "tongyi" and model.lower().startswith("wan")
 
     # 复用套图的 prompt 家族：custom 类型装配「用户要求」为最高优先级约束，

server/config/settings.py

@@ -72,6 +72,12 @@ class Settings(BaseSettings):
     rightapi_max_retries: int = 3       # 429/5xx/超时的重试次数（1 = 不重试）
     rightapi_retry_wait: int = 60       # 重试基础等待秒数，按 60→120→240 递增
 
+    # Pcore（OpenAI 兼容中转：pcore.ai，异步图片任务；模型值带 @pcore 后缀路由到此）
+    pcore_api_key: str = ""
+    pcore_base_url: str = "https://pcore.ai"
+    pcore_max_retries: int = 3          # 429/5xx/超时的重试次数（1 = 不重试）
+    pcore_retry_wait: int = 60          # 重试基础等待秒数，按 60→120→240 递增
+
     # DeepSeek 出图方案规划器（key 复用上方 deepseek_api_key；文案生成的模型走 config/models.yaml，互不影响）
     deepseek_base_url: str = "https://api.deepseek.com/v1"
     deepseek_model: str = "deepseek-v4-flash"

server/schemas/suite.py

@@ -24,15 +24,25 @@ RIGHTAPI_MODELS = [
     "nano-banana-pro",
 ]
 
-# 模型 → provider 推断表：前端只传模型名，服务端据此路由（模型名优先于 provider 字段）
-MODEL_PROVIDERS: dict[str, str] = {
-    **{m: "tongyi" for m in TONGYI_MODELS},
-    **{m: "rightapi" for m in RIGHTAPI_MODELS},
-}
+# Pcore 生图模型白名单（pcore.ai 中转；同一模型名与 RightAPI 重名，用 @pcore 后缀区分路由）
+PCORE_MODELS = ["gpt-image-2"]
 
 # 豆包 Seedream 走火山方舟；模型由 ARK_IMAGE_MODEL 指定（不在下拉白名单内也允许直连豆包）
 DOUBAO_MODELS = ["doubao-seedream-4-5-251128"]
-MODEL_PROVIDERS.update({m: "doubao" for m in DOUBAO_MODELS})
+
+# 模型 → provider 推断表：前端只传模型名，服务端据此路由（模型名优先于 provider 字段）
+# 同名模型走多家中转时，前端传「api_model@provider」形式（如 gpt-image-2@pcore）
+MODEL_PROVIDERS: dict[str, str] = {
+    **{m: "tongyi" for m in TONGYI_MODELS},
+    **{m: "rightapi" for m in RIGHTAPI_MODELS},
+    **{m: "doubao" for m in DOUBAO_MODELS},
+    **{f"{m}@pcore": "pcore" for m in PCORE_MODELS},
+}
+
+
+def api_model(model: str | None) -> str | None:
+    """前端模型值 → 发给供应商 API 的模型名：'gpt-image-2@pcore' → 'gpt-image-2'。"""
+    return model.split("@", 1)[0] if model and "@" in model else model
 
 
 def resolve_provider(model: str | None, requested: str | None, default: str) -> str:

server/services/generator.py

@@ -61,9 +61,11 @@ def _raise_api_error(resp, provider: str):
         raise ApiError(f"{provider} API HTTP {resp.status_code} — {detail}", resp.status_code)
     try:
         body = resp.json()
-        detail = f"{body.get('code', '')}: {body.get('message', '')}".strip(': ')
     except Exception:  # noqa: BLE001
-        detail = text[:200]
+        raise ApiError(f"{provider} API HTTP {resp.status_code} — {text[:200]}", resp.status_code)
+    # 兼容两种错误形状：顶层 {code, message}（豆包/通义）与嵌套 {error: {code, message}}（new_api 系中转）
+    err = body.get("error") if isinstance(body.get("error"), dict) else body
+    detail = f"{err.get('code', '')}: {err.get('message', '')}".strip(': ')
     raise ApiError(f"{provider} API HTTP {resp.status_code} — {detail or '无错误详情'}", resp.status_code)
 
 # 参考图选择：material 用第 2 张（背面/细节），其余用第 1 张（正面）
@@ -80,6 +82,9 @@ def _image_size(provider: str, ratio: str, is_wan: bool = True, model: str = "")
     if provider == "rightapi":
         # gpt-image 自定义尺寸约束：16 的倍数、长短边比 ≤ 3:1（1536x2048 合法）
         return "1536x2048" if ratio == "3:4" else "2048x2048"
+    if provider == "pcore":
+        # pcore 的 size 是枚举（默认 1024x1024），不接受自定义尺寸；用 gpt-image 枚举：竖版 1024x1536（近似 3:4）
+        return "1024x1536" if ratio == "3:4" else "1024x1024"
     # tongyi：万象与千问的 size 语法相同（* 分隔），档位不同
     # wan2.6 系列总像素限制在 [1280², 1440²]，wan2.7 的 1536*2048/2048*2048 会超限
     if model.startswith("wan2.6"):
@@ -379,7 +384,110 @@ async def generate_rightapi(prompt: str, ref_images: list[str], size: str = "204
     raise last_exc  # type: ignore[misc]
 
 
-GENERATORS = {"doubao": generate_doubao, "tongyi": generate_tongyi, "rightapi": generate_rightapi}
+# ── Provider：Pcore（pcore.ai，OpenAI 兼容异步图片协议）──────────────────
+
+async def _pcore_poll_task(client: httpx.AsyncClient, headers: dict, base: str,
+                           kind: str, task_id: str, max_wait: int) -> dict:
+    """轮询 Pcore 任务：GET {base}/v1/images/{kind}/{task_id}，kind ∈ generations | edits。
+
+    状态机：queued / in_progress → 继续；completed / succeeded → 返回；
+    failed / error / cancelled → 抛错。
+    """
+    poll_url = f"{base}/v1/images/{kind}/{task_id}"
+    elapsed, interval = 0, 3
+    while elapsed < max_wait:
+        resp = await client.get(poll_url, headers=headers, timeout=30)
+        _raise_api_error(resp, "Pcore")
+        result = resp.json()
+        status = result.get("status", "")
+        if status in ("failed", "error", "cancelled"):
+            err = result.get("error") or {}
+            raise RuntimeError(f"Pcore 任务失败: {err.get('message') or result}")
+        if status in ("completed", "succeeded"):
+            return result
+        await asyncio.sleep(interval)
+        elapsed += interval
+        interval = min(interval + 2, 10)
+    raise TimeoutError(f"Pcore 异步任务超时 ({max_wait}s): task_id={task_id}")
+
+
+async def _pcore_request(s, prompt: str, ref_images: list[str], size: str, model: str) -> bytes:
+    """Pcore：无参考图走 /v1/images/generations，带参考图走 /v1/images/edits（JSON，data-URI）。
+
+    两者均为异步：提交返回任务 id → 轮询对应资源路径 → 完成后 data[0].url 取图。
+    完成形状与 RightAPI 同为 Images 协议，复用 _rightapi_extract_image。
+    """
+    base = s.pcore_base_url.rstrip("/")
+    headers = {"Authorization": f"Bearer {s.pcore_api_key}"}
+
+    kind = "edits" if ref_images else "generations"
+    body: dict = {"model": model, "prompt": prompt, "n": 1, "size": size, "async": True}
+    if ref_images:
+        # pcore 的 edits 要求 images 为 [{image_url: ...}] 对象数组（实测纯字符串报 images[].image_url is required）
+        body["images"] = [{"image_url": await _resolve_ref(u)} for u in ref_images]
+
+    async with httpx.AsyncClient(timeout=max(s.request_timeout, 600), verify=False) as client:
+        resp = await client.post(
+            f"{base}/v1/images/{kind}",
+            headers={**headers, "Content-Type": "application/json"},
+            json=body,
+        )
+        _raise_api_error(resp, "Pcore")
+        submitted = resp.json()
+        # new_api 系中转常见「HTTP 200 + 错误体」，需主动检查 error 字段
+        if isinstance(submitted, dict) and submitted.get("error"):
+            err = submitted["error"] or {}
+            raise RuntimeError(f"Pcore 提交失败: {err.get('message') or submitted}")
+        task_id = submitted.get("task_id") or submitted.get("id") or ""
+        if not task_id:
+            raise RuntimeError(f"Pcore 提交未返回任务 id: {submitted}")
+        result = await _pcore_poll_task(client, headers, base, kind, task_id, s.poll_max_wait)
+
+        img_kind, payload = _rightapi_extract_image(result)
+        if img_kind == "b64" and payload:
+            return base64.b64decode(payload.split(",", 1)[-1] if "," in payload else payload)
+        if img_kind == "url" and payload:
+            dl = await client.get(payload, timeout=s.request_timeout)
+            dl.raise_for_status()
+            return dl.content
+        raise RuntimeError(f"Pcore 任务完成但没有图片数据: {result}")
+
+
+async def generate_pcore(prompt: str, ref_images: list[str], size: str = "1536x2048", model: str | None = None) -> bytes:
+    """带重试的 Pcore 入口：429/5xx/超时按递增间隔重试（与 RightAPI 同策略）。"""
+    s = get_settings()
+    if not s.pcore_api_key:
+        raise RuntimeError("未配置 PCORE_API_KEY（.env）")
+    model = model or "gpt-image-2"
+    attempts = max(1, s.pcore_max_retries)
+
+    last_exc: Exception | None = None
+    for i in range(attempts):
+        try:
+            return await _pcore_request(s, prompt, ref_images, size, model)
+        except ApiError as exc:
+            last_exc = exc
+            if exc.status not in RETRYABLE_STATUS:
+                raise
+            # 通道不存在（模型未开通）重试无意义，立即失败
+            if "no available channel" in str(exc).lower() or "model_not_found" in str(exc).lower():
+                raise
+        except (httpx.TimeoutException, httpx.TransportError) as exc:
+            last_exc = exc  # 网络抖动/超时可重试
+        if i == attempts - 1:
+            break
+        wait = s.pcore_retry_wait * (2 ** i)
+        log.warning("Pcore 第 %d/%d 次请求失败（%s），%ds 后重试", i + 1, attempts, last_exc, wait)
+        await asyncio.sleep(wait)
+    raise last_exc  # type: ignore[misc]
+
+
+GENERATORS = {
+    "doubao": generate_doubao,
+    "tongyi": generate_tongyi,
+    "rightapi": generate_rightapi,
+    "pcore": generate_pcore,
+}
 
 
 # ── 任务执行器 ────────────────────────────────────────────────────────────

server/services/planner.py

@@ -22,23 +22,39 @@ ALLOWED_KINDS = [
     "size_chart", "sku_collection", "custom",
 ]
 
+# 方案总张数安全上限（仅防模型跑飞，不干预正常规划；生图按张计费）：
+# 默认 AI 自主决定张数（prompt 建议 8 张左右），上限 15；
+# 「生图要求」明确写了「N 张」时按用户指定的来，上限放宽到 24。
+PLAN_SAFETY_MAX = 15
+PLAN_REQUIREMENTS_MAX = 24
+
+
+def _total_budget(requirements: str | None) -> int:
+    """从生图要求提取「N 张」作为总张数预算；未指定时用默认安全上限。"""
+    if requirements:
+        m = re.search(r"(\d+)\s*张", requirements)
+        if m:
+            return min(max(int(m.group(1)), 1), PLAN_REQUIREMENTS_MAX)
+    return PLAN_SAFETY_MAX
+
 SYSTEM_PROMPT = """你是一名资深电商视觉策划。根据商品信息规划一套电商套图的出图方案。
 
 ## 输出硬性约束（违反即失败）
 1. 输出必须是**单行紧凑 JSON**：无换行、无缩进、无空格填充、无注释、无 markdown 围栏。
 2. 顶层只有 summary 和 items 两个字段；每个 item 严格只有 kind/title/detail/prompt_hint/count/variant_name 六个字段，不得增删。
 3. 文本长度上限（中文字符/英文单词数）：summary ≤ 25 字；title ≤ 8 字；detail ≤ 20 字；prompt_hint ≤ 15 个英文词。超限必须删减，不得省略号截断。
-4. count 默认 1，仅当该类图确需多个变体时才 >1，最大 3。总张数 8-15。
+4. count 默认 1，仅当该类图确需多个变体时才 >1，最大 3。总张数由你根据商品复杂度与信息量自主决定，不设硬性数量，但建议 8 张左右；若「生图要求」明确指定了张数，必须严格按指定张数输出。
 5. variant_name 只能从「SKU规格」列表原样照抄；没有绑定就输出 null。
 
 ## 规划规则
-1. SKU 主图：每个带图 SKU 出 1 张独立主图（kind=white_bg），variant_name 填对应规格名；单 SKU 出 1 张（variant_name=null）。
-2. 场景图（kind=lifestyle）：按核心使用场景出 2-4 张，每张聚焦一个场景。
-3. 细节图（kind=material 或 custom）：按关键细节/材质/结构出 2-3 张，每张聚焦一个卖点。
-4. 尺寸标注图（kind=size_chart）：参数含长宽高/尺寸时出 1 张。
-5. SKU 合集图（kind=sku_collection）：SKU >1 时出 1 张。
-6. kind 枚举：white_bg / key_features / selling_pt / material / lifestyle / multi_scene / ecommerce_detail / size_chart / sku_collection / custom。
-7. title 用中文短语（如「主图·粉色」「浴室壁挂」）；detail 中文说明这张图展示什么；prompt_hint 用英文描述构图要点。
+1. 先保证 SKU 主图（规则 2），其余类型（场景/细节/尺寸/合集等）按商品特点自行取舍并分配张数，追求信息效率，不为凑数出图。
+2. SKU 主图：每个带图 SKU 出 1 张独立主图（kind=white_bg），variant_name 填对应规格名；单 SKU 出 1 张（variant_name=null）。
+3. 场景图（kind=lifestyle）：按核心使用场景出 1-3 张，每张聚焦一个场景。
+4. 细节图（kind=material 或 custom）：按关键细节/材质/结构出 1-2 张，每张聚焦一个卖点。
+5. 尺寸标注图（kind=size_chart）：参数含长宽高/尺寸时出 1 张。
+6. SKU 合集图（kind=sku_collection）：SKU >1 时出 1 张。
+7. kind 枚举：white_bg / key_features / selling_pt / material / lifestyle / multi_scene / ecommerce_detail / size_chart / sku_collection / custom。
+8. title 用中文短语（如「主图·粉色」「浴室壁挂」）；detail 中文说明这张图展示什么；prompt_hint 用英文描述构图要点。
 
 ## 输出示例（紧凑单行）
 {"summary":"三色收纳盒全套图","items":[{"kind":"white_bg","title":"主图·粉色","detail":"粉色SKU白底主视觉","prompt_hint":"front view on white background","count":1,"variant_name":"粉色"}]}"""
@@ -78,9 +94,8 @@ def _normalize_items(raw_items: list, sku_variants: list[str]) -> list[dict]:
         kind = str(it.get("kind") or "custom")
         if kind not in ALLOWED_KINDS:
             kind = "custom"
-        title = str(it.get("title") or "").strip()[:20]
-        if not title:
-            continue
+        # title 缺失不再丢弃条目：从 detail / kind 兜底合成，避免整份方案被判空
+        title = str(it.get("title") or "").strip()[:20] or str(it.get("detail") or "").strip()[:12] or kind
         try:
             count = max(0, min(3, int(it.get("count", 1))))
         except (TypeError, ValueError):
@@ -173,55 +188,68 @@ async def generate_plan(
         user_payload["生图要求（最高优先级，必须体现在每个方案项中）"] = requirements.strip()
     user_content = json.dumps(user_payload, ensure_ascii=False)
 
-    async with httpx.AsyncClient(timeout=90, verify=False) as client:
-        resp = await client.post(
-            f"{s.deepseek_base_url.rstrip('/')}/chat/completions",
-            headers={"Authorization": f"Bearer {s.deepseek_api_key}", "Content-Type": "application/json"},
-            json={
-                "model": s.deepseek_model,
-                "messages": [
-                    {"role": "system", "content": _system_prompt_with_requirements(requirements)},
-                    {"role": "user", "content": user_content},
-                ],
-                "response_format": {"type": "json_object"},
-                "temperature": 0.3,
-                "max_tokens": 8000,
-            },
-        )
-        resp.raise_for_status()
-        body = resp.json()
-        message = body["choices"][0]["message"]
-        finish_reason = body["choices"][0].get("finish_reason", "")
-        usage = body.get("usage") or {}
-        log.info(
-            "规划器 token 用量: prompt=%s completion=%s finish=%s",
-            usage.get("prompt_tokens", "?"), usage.get("completion_tokens", "?"), finish_reason,
-        )
-        content = message.get("content") or ""
-        # 思考型输出：content 为空时从 reasoning_content 里捞
-        if not content.strip() and message.get("reasoning_content"):
-            content = message["reasoning_content"]
+    last_error: RuntimeError | ValueError | None = None
+    for attempt in range(2):  # 解析失败/空方案自动重试一次
+        async with httpx.AsyncClient(timeout=90, verify=False) as client:
+            resp = await client.post(
+                f"{s.deepseek_base_url.rstrip('/')}/chat/completions",
+                headers={"Authorization": f"Bearer {s.deepseek_api_key}", "Content-Type": "application/json"},
+                json={
+                    "model": s.deepseek_model,
+                    "messages": [
+                        {"role": "system", "content": _system_prompt_with_requirements(requirements)},
+                        {"role": "user", "content": user_content},
+                    ],
+                    "response_format": {"type": "json_object"},
+                    "temperature": 0.3,
+                    "max_tokens": 8000,
+                    # deepseek-v4 是思维链模型：不禁用 thinking 会把 max_tokens 烧在推理上，
+                    # 正文为空 → 只能拿 reasoning_content 兜底解析 → 频发「解析失败/未返回有效方案项」
+                    "thinking": {"type": "disabled"},
+                },
+            )
+            resp.raise_for_status()
+            body = resp.json()
+            message = body["choices"][0]["message"]
+            finish_reason = body["choices"][0].get("finish_reason", "")
+            usage = body.get("usage") or {}
+            log.info(
+                "规划器 token 用量: prompt=%s completion=%s finish=%s",
+                usage.get("prompt_tokens", "?"), usage.get("completion_tokens", "?"), finish_reason,
+            )
+            content = message.get("content") or ""
+            # 思考型输出：content 为空时从 reasoning_content 里捞
+            if not content.strip() and message.get("reasoning_content"):
+                content = message["reasoning_content"]
 
-    try:
-        data = _extract_json(content)
-    except ValueError as exc:
-        log.error(
-            "规划器输出解析失败 finish_reason=%s content[:200]=%s",
-            finish_reason, content[:200],
+        try:
+            data = _extract_json(content)
+        except ValueError as exc:
+            last_error = exc
+            log.error(
+                "规划器第 %d 次输出解析失败 finish_reason=%s content[:200]=%s",
+                attempt + 1, finish_reason, content[:200],
+            )
+            continue
+
+        items = _normalize_items(data.get("items") or [], sku_variants)
+        if items:
+            # 总量保护：只防跑飞不干预正常规划。生图要求指定「N 张」时按 N 放宽上限
+            budget = _total_budget(requirements)
+            total = sum(i["count"] for i in items)
+            while total > budget and items:
+                last = items[-1]
+                if last["count"] > 1:
+                    last["count"] -= 1
+                else:
+                    items.pop()
+                total = sum(i["count"] for i in items)
+            return {"summary": str(data.get("summary") or "").strip()[:100], "items": items}
+
+        last_error = RuntimeError("规划器未返回有效方案项")
+        log.warning(
+            "规划器第 %d 次返回无有效方案项 finish_reason=%s content[:200]=%s",
+            attempt + 1, finish_reason, content[:200],
         )
-        raise RuntimeError("规划器输出解析失败") from exc
 
-    items = _normalize_items(data.get("items") or [], sku_variants)
-    if not items:
-        raise RuntimeError("规划器未返回有效方案项")
-    # 总量保护：超过 18 张时按比例截断
-    total = sum(i["count"] for i in items)
-    while total > 18 and items:
-        last = items[-1]
-        if last["count"] > 1:
-            last["count"] -= 1
-        else:
-            items.pop()
-        total = sum(i["count"] for i in items)
-
-    return {"summary": str(data.get("summary") or "").strip()[:100], "items": items}
+    raise last_error or RuntimeError("规划器未返回有效方案项")

server/services/prompts/__init__.py

@@ -25,7 +25,7 @@ _MODULE_BY_FAMILY = {
 
 def prompt_family(provider: str, model: str | None) -> str:
     """(provider, model) → 提示词家族名。"""
-    if provider == "rightapi":
+    if provider in ("rightapi", "pcore"):
         if (model or "").lower().startswith("nano-banana"):
             return "google"
         return "gpt"  # gpt-image-* 及未知中转模型默认按 edits 语义处理

server/services/suite_service.py

@@ -38,12 +38,15 @@ def texts_to_raw(texts: list[TextMaterial]) -> dict:
 
 def validate_model(provider_name: str, model: str | None) -> None:
     """按 provider 校验模型名白名单，不合法抛 ValueError。"""
-    from schemas.suite import RIGHTAPI_MODELS, TONGYI_MODELS
+    from schemas.suite import PCORE_MODELS, RIGHTAPI_MODELS, TONGYI_MODELS, api_model
 
-    if provider_name == "tongyi" and model and model not in TONGYI_MODELS:
-        raise ValueError(f"不支持的模型: {model}（tongyi 支持: {TONGYI_MODELS}）")
-    if provider_name == "rightapi" and model and model not in RIGHTAPI_MODELS:
-        raise ValueError(f"不支持的模型: {model}（rightapi 支持: {RIGHTAPI_MODELS}）")
+    m = api_model(model)  # 'gpt-image-2@pcore' → 'gpt-image-2'
+    if provider_name == "tongyi" and m and m not in TONGYI_MODELS:
+        raise ValueError(f"不支持的模型: {m}（tongyi 支持: {TONGYI_MODELS}）")
+    if provider_name == "rightapi" and m and m not in RIGHTAPI_MODELS:
+        raise ValueError(f"不支持的模型: {m}（rightapi 支持: {RIGHTAPI_MODELS}）")
+    if provider_name == "pcore" and m and m not in PCORE_MODELS:
+        raise ValueError(f"不支持的模型: {m}（pcore 支持: {PCORE_MODELS}）")
 
 
 async def append_generated_asset(