| """商品视觉 API 路由（Product Visual） |
| |
| 对齐 Xiaoye AI 商品图工作台能力： |
| - POST /api/product-visual/tasks 创建商品素材包/详情页任务 |
| - GET /api/product-visual/tasks 查询任务列表（?ids=a,b） |
| - GET /api/product-visual/tasks/{task_id} 查询单个任务 |
| - POST /api/product-visual/tasks/{task_id}/items/{seq}/retry 单张重试 |
| |
| 请求体（multipart/form-data）： |
| - client_task_id: str |
| - prompt: str 商品描述/卖点/规格 |
| - scenario: asset_pack | detail_page |
| - platform: auto | taobao_tmall | jd | pdd | douyin | xiaohongshu | amazon | tiktok_shop | shopee |
| - output_count: 6 | 8 | 10 | 12 |
| - model: str (默认 gpt-image-2) |
| - image: 文件（一张或多张商品图，字段名 image，可重复） |
| """ |
|
|
| from __future__ import annotations |
|
|
| from fastapi import APIRouter, Header, HTTPException, Query, Request, Response |
| from fastapi.concurrency import run_in_threadpool |
|
|
| from api.image_inputs import parse_image_edit_request, read_image_sources |
| from api.support import require_identity, resolve_image_base_url |
| from services.product_visual_service import PLATFORM_SPECS, product_visual_service |
|
|
|
|
| def _clean(value: object, default: str = "") -> str: |
| return str(value if value is not None else default).strip() |
|
|
|
|
| def _extract_product_visual_fields(payload: dict[str, object]) -> dict[str, object]: |
| """从请求载荷中提取商品视觉参数，非法值回退默认。""" |
| scenario = _clean(payload.get("scenario"), "asset_pack").lower() |
| if scenario not in {"asset_pack", "detail_page"}: |
| scenario = "asset_pack" |
| platform = _clean(payload.get("platform"), "auto").lower() |
| if platform not in PLATFORM_SPECS: |
| platform = "auto" |
| try: |
| output_count = int(payload.get("output_count") or 8) |
| except (TypeError, ValueError): |
| output_count = 8 |
| if output_count not in {6, 8, 10, 12}: |
| output_count = 8 |
| return {"scenario": scenario, "platform": platform, "output_count": output_count} |
|
|
|
|
| def create_router() -> APIRouter: |
| router = APIRouter() |
|
|
| @router.post("/api/product-visual/tasks") |
| async def create_product_visual_task( |
| request: Request, |
| authorization: str | None = Header(default=None), |
| ): |
| identity = require_identity(authorization) |
| payload, image_sources, _mask_sources = await parse_image_edit_request(request) |
| client_task_id = str(payload.get("client_task_id") or "").strip() |
| if not client_task_id: |
| raise HTTPException(status_code=400, detail={"error": "client_task_id is required"}) |
| prompt = str(payload.get("prompt") or "").strip() |
| if not prompt: |
| raise HTTPException(status_code=400, detail={"error": "prompt is required"}) |
| fields = _extract_product_visual_fields(payload) |
| scenario = fields["scenario"] |
| platform = fields["platform"] |
| output_count = fields["output_count"] |
| model = str(payload.get("model") or "gpt-image-2").strip() |
| mode = str(payload.get("mode") or "concurrent").strip().lower() |
| images = await read_image_sources(image_sources) |
| try: |
| return await run_in_threadpool( |
| product_visual_service.submit, |
| identity, |
| client_task_id=client_task_id, |
| scenario=scenario, |
| platform=platform, |
| output_count=output_count, |
| prompt=prompt, |
| images=images, |
| model=model, |
| mode=mode, |
| base_url=resolve_image_base_url(request), |
| ) |
| except ValueError as exc: |
| raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc |
|
|
| @router.get("/api/product-visual/tasks") |
| async def list_product_visual_tasks( |
| ids: str = Query(default=""), |
| authorization: str | None = Header(default=None), |
| ): |
| identity = require_identity(authorization) |
| task_ids = [item.strip() for item in ids.split(",") if item.strip()] |
| return await run_in_threadpool(product_visual_service.list_tasks, identity, task_ids) |
|
|
| @router.get("/api/product-visual/tasks/{task_id}") |
| async def get_product_visual_task( |
| task_id: str, |
| authorization: str | None = Header(default=None), |
| ): |
| identity = require_identity(authorization) |
| task = await run_in_threadpool(product_visual_service.get_task, identity, task_id) |
| if task is None: |
| raise HTTPException(status_code=404, detail={"error": "商品视觉任务不存在"}) |
| return task |
|
|
| @router.get("/api/product-visual/tasks/{task_id}/source-images/{index}") |
| async def get_product_visual_source_image( |
| task_id: str, |
| index: int, |
| authorization: str | None = Header(default=None), |
| ): |
| identity = require_identity(authorization) |
| source = await run_in_threadpool( |
| product_visual_service.get_source_image, |
| identity, |
| task_id, |
| index, |
| ) |
| if source is None: |
| raise HTTPException(status_code=404, detail={"error": "参考原图不存在"}) |
| content, media_type, filename = source |
| return Response( |
| content=content, |
| media_type=media_type, |
| headers={ |
| "Cache-Control": "private, max-age=86400", |
| "Content-Disposition": f'inline; filename="{filename}"', |
| }, |
| ) |
|
|
| @router.post("/api/product-visual/tasks/{task_id}/items/{seq}/retry") |
| async def retry_product_visual_item( |
| task_id: str, |
| seq: int, |
| request: Request, |
| authorization: str | None = Header(default=None), |
| ): |
| identity = require_identity(authorization) |
| try: |
| return await run_in_threadpool( |
| product_visual_service.retry_item, |
| identity, |
| task_id, |
| seq, |
| resolve_image_base_url(request), |
| ) |
| except ValueError as exc: |
| raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc |
|
|
| @router.post("/api/product-visual/tasks/{task_id}/items/{seq}/iterate") |
| async def iterate_product_visual_item( |
| task_id: str, |
| seq: int, |
| request: Request, |
| authorization: str | None = Header(default=None), |
| ): |
| identity = require_identity(authorization) |
| try: |
| payload = await request.json() |
| except Exception: |
| raise HTTPException(status_code=400, detail={"error": "请求体必须是 JSON"}) from None |
| if not isinstance(payload, dict): |
| raise HTTPException(status_code=400, detail={"error": "请求体必须是 JSON"}) from None |
| revision = _clean(payload.get("revision")) |
| if not revision: |
| raise HTTPException(status_code=400, detail={"error": "修改意见不能为空"}) |
| mask_data_url = _clean(payload.get("mask")) |
| try: |
| return await run_in_threadpool( |
| product_visual_service.iterate_item, |
| identity, |
| task_id, |
| seq, |
| revision, |
| resolve_image_base_url(request), |
| mask_data_url=mask_data_url, |
| ) |
| except ValueError as exc: |
| raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc |
|
|
| return router |
|
|
