"""商品视觉 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: # noqa: BLE001 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