diff --git a/.env.example b/.env.example index e2d74a2..7f92aaa 100644 --- a/.env.example +++ b/.env.example @@ -34,3 +34,12 @@ STORAGE_BACKEND=local # ── V2：对外地址（插件/前端回写、生成图回调）── APP_BASE_URL=http://127.0.0.1:8800 + +# ── V2.1：套图生图（试算页一键生成 / 单张 AI 生图，provider 按模型名自动路由）── +# 豆包 Seedream（火山方舟） +ARK_API_KEY= +# RightAPI 中转（gpt-image-2 / gpt-image-2-vip / nano-banana 系列） +RIGHTAPI_API_KEY= +# 可选覆盖：默认生图 provider（doubao | tongyi | rightapi）、水印徽章图路径 +# IMAGE_PROVIDER=doubao +# WATERMARK_IMAGE_PATH=server/assets/watermark.jpg diff --git a/server/api/export.py b/server/api/export.py new file mode 100644 index 0000000..9b367e9 --- /dev/null +++ b/server/api/export.py @@ -0,0 +1,121 @@ +"""导出采集图片：把勾选的采集图/生成图打包成 ZIP 下载到本地。 + +平移自 image-suite-studio；路径按 docs/v2.1/api.md §7 定为 POST /api/export/images +（前端 services/suite.ts exportImages 同款契约，注意与 ISS 的 /export-images 不同）。 +ZIP 内部按分组名建子文件夹（主图 / SKU图片 / 详情图 / 手动上传）， +文件名沿用采集 key（main-001 等）+ SKU 规格名；顶层文件夹用商品标题（清洗后）。 +""" +from __future__ import annotations + +import io +import mimetypes +import re +import zipfile + +from fastapi import APIRouter, HTTPException +from fastapi.responses import StreamingResponse +from pydantic import BaseModel, Field + +from api.proxy import guess_referer +from services.storage import download_bytes, local_path + +router = APIRouter(prefix="/api", tags=["export"]) + +_EXT_BY_CTYPE = { + "image/jpeg": ".jpg", + "image/png": ".png", + "image/webp": ".webp", + "image/gif": ".gif", + "image/bmp": ".bmp", +} + + +class ExportImageItem(BaseModel): + url: str + groupName: str = "主图" + variantName: str | None = None + key: str = "" # 采集 key，如 main-001 / sku-002 / upload-001 + + +class ExportImagesRequest(BaseModel): + title: str | None = Field(default=None, description="商品标题，用作 ZIP 顶层文件夹名") + images: list[ExportImageItem] + + +def _clean(name: str) -> str: + """清洗文件夹/文件名非法字符（与插件 cleanFilename 同规则，Windows 兼容）。""" + s = re.sub(r'[<>:"/\\|?*\x00-\x1f]', "_", (name or "").strip()) + s = re.sub(r"\s+", "_", s) + return s.strip(" .")[:80] + + +def _ext(url: str, ctype: str) -> str: + """由 content-type（优先）或 URL 后缀决定扩展名。""" + ctype = ctype.split(";")[0].strip().lower() + if ctype in _EXT_BY_CTYPE: + return _EXT_BY_CTYPE[ctype] + if ctype.startswith("image/"): + return "." + ctype.split("/")[-1] + path = url.split("?")[0].lower() + for ext in (".jpg", ".jpeg", ".png", ".webp", ".gif", ".bmp"): + if path.endswith(ext): + return ".jpg" if ext == ".jpeg" else ext + return ".jpg" + + +def _is_image(url: str, ctype: str) -> bool: + ctype = ctype.split(";")[0].strip().lower() + if ctype.startswith("image/"): + return True + return bool(re.search(r"\.(jpe?g|png|webp|gif|bmp)(\?|$)", url, re.IGNORECASE)) + + +async def _download(url: str) -> tuple[bytes, str]: + """本地 media 文件直读磁盘；远程 URL（源站 CDN / 七牛）带 Referer 下载。""" + path = local_path(url) + if path is not None: + mime = mimetypes.guess_type(path.name)[0] or "image/jpeg" + return path.read_bytes(), mime + return await download_bytes(url, referer=guess_referer(url)) + + +@router.post("/export/images") +async def export_images(req: ExportImagesRequest): + if not req.images: + raise HTTPException(status_code=400, detail="没有可导出的图片") + + root = _clean(req.title) or "采集图片" + buf = io.BytesIO() + used: set[str] = set() + + with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf: + for img in req.images: + try: + data, ctype = await _download(img.url) + except Exception: # noqa: BLE001 + continue # 单张失败不中断整包 + if not _is_image(img.url, ctype): + continue + + ext = _ext(img.url, ctype) + base = _clean(img.key) or "image" + if img.variantName: + base += f"-{_clean(img.variantName)}" + filename = f"{base}{ext}" + if filename in used: # 同名加序号防覆盖 + stem = filename[: -len(ext)] + n = 2 + while f"{stem}-{n}{ext}" in used: + n += 1 + filename = f"{stem}-{n}{ext}" + used.add(filename) + + group = _clean(img.groupName) or "图片" + zf.writestr(f"{root}/{group}/{filename}", data) + + buf.seek(0) + return StreamingResponse( + buf, + media_type="application/zip", + headers={"Content-Disposition": 'attachment; filename="collect.zip"'}, + ) diff --git a/server/api/proxy.py b/server/api/proxy.py new file mode 100644 index 0000000..dcf3387 --- /dev/null +++ b/server/api/proxy.py @@ -0,0 +1,54 @@ +"""图片代理：绕过源站防盗链，供前端 预览与生图参考使用。 + +平移自 image-suite-studio（/api/proxy-image?url=... 形式，docs/v2.1/api.md §7）。 +""" +from __future__ import annotations + +from urllib.parse import urlparse + +from fastapi import APIRouter, HTTPException, Query, Response + +from services.storage import download_bytes + +router = APIRouter(prefix="/api", tags=["proxy"]) + +# 域名片段 → 防盗链所需 Referer +_REFERER_BY_DOMAIN: list[tuple[str, str]] = [ + ("alicdn.com", "https://www.taobao.com"), + ("taobao.com", "https://www.taobao.com"), + ("tmall.com", "https://www.tmall.com"), + ("1688.com", "https://www.1688.com"), + ("ozon.ru", "https://www.ozon.ru"), + ("ozon.kz", "https://www.ozon.ru"), + ("ozon.by", "https://www.ozon.ru"), + ("ozonusercontent.com", "https://www.ozon.ru"), +] + + +def guess_referer(url: str) -> str | None: + host = (urlparse(url).hostname or "").lower() + for frag, referer in _REFERER_BY_DOMAIN: + if frag in host: + return referer + return None + + +@router.get("/proxy-image") +async def proxy_image(url: str = Query(..., description="源站图片 URL")): + scheme = urlparse(url).scheme + if scheme not in ("http", "https"): + raise HTTPException(status_code=400, detail="仅支持 http/https URL") + try: + data, ctype = await download_bytes(url, referer=guess_referer(url)) + except Exception as exc: # noqa: BLE001 + raise HTTPException(status_code=502, detail=f"图片拉取失败: {exc}") from exc + if not ctype.startswith("image/"): + ctype = "image/jpeg" + return Response( + content=data, + media_type=ctype, + headers={ + "Cache-Control": "public, max-age=86400", + "Access-Control-Allow-Origin": "*", + }, + ) diff --git a/server/api/suite.py b/server/api/suite.py new file mode 100644 index 0000000..c1322e7 --- /dev/null +++ b/server/api/suite.py @@ -0,0 +1,320 @@ +"""套图规划 / 一键生成 / 任务查询 / 结果导出 / 单张 AI 生图。 + +按 docs/v2.1/api.md §2-6 实现：引擎平移自 image-suite-studio（services/planner|generator| +tasks|watermark + services/prompts），任务存进程内内存表（重启即失效）。 +生成图回调回写 product_assets(group_key='generated')；product_id 缺省时只落盘不回写。 +注：V2.1 阶段本组接口暂不接鉴权（现有鉴权后续可能重做）。 +""" +from __future__ import annotations + +import io +import mimetypes +import uuid +import zipfile + +from fastapi import APIRouter, BackgroundTasks, HTTPException +from fastapi.responses import StreamingResponse + +from config import get_settings +from db import get_session_factory +from models import Product, ProductAsset +from schemas.suite import ( + PLATFORM_SPECS, + RIGHTAPI_MODELS, + SUPPORTED_TYPES, + TONGYI_MODELS, + ImageEditSingleRequest, + ImageEditSingleResponse, + PlanItemOut, + PlanResponse, + SuiteCreateResponse, + SuiteGenerateRequest, + SuiteOut, + SuitePlanRequest, + TextMaterial, + resolve_provider, +) +from services.generator import _image_size, GENERATORS, run_suite +from services.planner import generate_plan +from services.prompts import build_context, build_prompt, type_name +from services.storage import download_bytes, get_storage, local_path +from services.tasks import IMG_OK, Task, TaskImage, create_task, get_task +from api.proxy import guess_referer + +router = APIRouter(prefix="/api", tags=["suite"]) + + +def texts_to_raw(texts: list[TextMaterial]) -> dict: + """前端组装的文本素材 → prompt 上下文用的 raw dict（后写的覆盖先写的）。""" + raw: dict = {} + for t in texts: + if t.kind == "title" and t.content: + raw["title"] = t.content + elif t.kind == "price" and t.content: + raw["price"] = t.content + elif t.kind == "brand" and t.content: + raw["brand"] = t.content + elif t.kind == "params" and t.pairs: + merged = {p["key"]: p["value"] for p in (raw.get("params") or [])} + for p in t.pairs: + merged.setdefault(p["key"], p["value"]) + raw["params"] = [{"key": k, "value": v} for k, v in merged.items()] + elif t.kind == "selling_point" and t.content: + raw["sellingPoints"] = t.content + elif t.kind == "desc" and t.content: + raw["desc"] = t.content + elif t.kind == "sales" and t.content: + raw["sales"] = t.content + elif t.kind == "shop" and t.content: + raw["shop"] = t.content + return raw + + +def _validate_model(provider_name: str, model: str | None) -> None: + if provider_name == "tongyi" and model and model not in TONGYI_MODELS: + raise HTTPException(status_code=400, detail=f"不支持的模型: {model}（tongyi 支持: {TONGYI_MODELS}）") + if provider_name == "rightapi" and model and model not in RIGHTAPI_MODELS: + raise HTTPException(status_code=400, detail=f"不支持的模型: {model}（rightapi 支持: {RIGHTAPI_MODELS}）") + + +# ── 生成图回写商品素材 ──────────────────────────────────────────────────── + +async def _append_generated_asset(product_id: str, image: TaskImage) -> str | None: + """把一张生成完成的图追加为 product_assets(generated)，并累加 asset_counts。返回 asset_id。""" + from sqlalchemy import func, select + + pid = uuid.UUID(product_id) + async with get_session_factory()() as db: + count = await db.scalar( + select(func.count(ProductAsset.id)).where( + ProductAsset.product_id == pid, + ProductAsset.group_key == "generated", + ) + ) + asset = ProductAsset( + product_id=pid, + group_key="generated", + variant_name=None, + sort_order=count or 0, + type="img", + source_url="", + stored_url=image.url, + status="uploaded", + ) + db.add(asset) + await db.flush() + + product = await db.get(Product, pid) + if product is not None: + counts = dict(product.asset_counts or {}) + counts["generated"] = int(counts.get("generated") or 0) + 1 + product.asset_counts = counts + await db.commit() + return str(asset.id) + + +# ── 出图方案规划 ────────────────────────────────────────────────────────── + +@router.post("/suite/plan", response_model=PlanResponse) +async def plan_suite(req: SuitePlanRequest) -> PlanResponse: + """DeepSeek 根据商品信息生成出图方案（同步接口，数秒级）。""" + product_info = texts_to_raw(req.texts) + if not product_info.get("title"): + raise HTTPException(status_code=400, detail="缺少商品标题，无法规划") + try: + result = await generate_plan( + product_info, req.sku_variants, req.image_stats, req.platform, + requirements=req.requirements, + ) + except RuntimeError as exc: + raise HTTPException(status_code=502, detail=str(exc)) from exc + except Exception as exc: # noqa: BLE001 + raise HTTPException(status_code=502, detail=f"规划失败: {exc}") from exc + return PlanResponse(summary=result["summary"], items=[PlanItemOut(**i) for i in result["items"]]) + + +# ── 一键生成 ───────────────────────────────────────────────────────────── + +@router.post("/suite/generate", response_model=SuiteCreateResponse) +async def generate_suite(req: SuiteGenerateRequest, background: BackgroundTasks) -> SuiteCreateResponse: + if not req.images: + raise HTTPException(status_code=400, detail="未勾选任何图片，无法生成") + + # 生成任务列表：方案优先；无方案时按 types（空则默认四种） + if req.plan: + jobs: list[dict] = [] + for item in req.plan: + if item.count <= 0: + continue + if item.kind not in SUPPORTED_TYPES: + raise HTTPException(status_code=400, detail=f"方案项「{item.title}」的图类型不支持: {item.kind}") + # 同一项多张 → 展开为多任务，第二张起在标题上加序号 + for n in range(item.count): + jobs.append({ + "kind": item.kind, + "title": item.title if item.count == 1 else f"{item.title}{n + 1}", + "detail": item.detail, + "prompt_hint": item.prompt_hint, + "variant_name": item.variant_name, + }) + if not jobs: + raise HTTPException(status_code=400, detail="方案中所有项的数量都是 0") + else: + types = req.types or ["white_bg", "key_features", "lifestyle", "multi_scene"] + bad = [t for t in types if t not in SUPPORTED_TYPES] + if bad: + raise HTTPException(status_code=400, detail=f"不支持的图类型: {bad}") + jobs = [{"kind": t, "title": type_name(t), "detail": "", "prompt_hint": "", "variant_name": None} for t in types] + + if req.platform not in PLATFORM_SPECS: + raise HTTPException(status_code=400, detail=f"不支持的目标平台: {req.platform}（ozon | wb | cn）") + spec = PLATFORM_SPECS[req.platform] + + settings = get_settings() + # 前端只传模型名：已知模型直接路由到对应 provider（如 gpt-image-2-vip → rightapi） + provider_name = resolve_provider(req.model, None, settings.image_provider) + _validate_model(provider_name, req.model) + + product_id = (req.product_id or "").strip() or None + if product_id: + try: + uuid.UUID(product_id) + except ValueError as exc: + raise HTTPException(status_code=400, detail="product_id 不是合法 UUID") from exc + + task = create_task( + status="pending", + platform=req.platform, + lang=spec["lang"], + ratio=spec["ratio"], + style_set=req.style_set, + style_prompt=req.style_prompt, + requirements=req.requirements, + provider=provider_name, + model=req.model, + total=len(jobs), + context=texts_to_raw(req.texts), + plan=jobs, + watermark=req.watermark.model_dump() if req.watermark else None, + # 参考图池：main 组优先，其余组按序补充（variant 绑定靠 variant_name 匹配） + ref_images=[ + { + "url": i.url, + "group_key": i.group_key, + "variant_name": i.variant_name, + } + for i in sorted(req.images, key=lambda x: 0 if x.group_key == "main" else 1) + ], + ) + + # 每张成功即回写 generated 组；未关联商品时仅落存储不回写 + background.add_task(run_suite, task, _append_generated_asset if product_id else None) + return SuiteCreateResponse(suite_id=task.id) + + +# ── 任务查询 / 结果导出 ────────────────────────────────────────────────── + +@router.get("/suites/{suite_id}", response_model=SuiteOut) +async def get_suite(suite_id: str): + task = get_task(suite_id) + if task is None: + raise HTTPException(status_code=404, detail="任务不存在（服务可能已重启），请重新生成") + return SuiteOut( + id=task.id, + status=task.status, + style_set=task.style_set, + platform=task.platform, + lang=task.lang, + ratio=task.ratio, + provider=task.provider, + model=task.model, + total=task.total, + images=[ + {"type_id": i.type_id, "name": i.name, "url": i.url, "status": i.status, "error": i.error} + for i in task.images + ], + error=task.error, + ) + + +@router.get("/suites/{suite_id}/zip") +async def download_suite_zip(suite_id: str): + """把任务内所有成功图打包成 ZIP（中文文件名）。""" + task = get_task(suite_id) + if task is None: + raise HTTPException(status_code=404, detail="任务不存在（服务可能已重启）") + + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf: + seen: set[str] = set() + for i, img in enumerate([i for i in task.images if i.status == IMG_OK]): + path = local_path(img.url or "") + if path is not None: + data, suffix = path.read_bytes(), path.suffix or ".jpg" + else: + try: # 七牛等远程存储：直接拉取公开 URL + data, _ = await download_bytes(img.url or "") + except Exception: # noqa: BLE001 + continue + suffix = ".png" if data[:8] == b"\x89PNG\r\n\x1a\n" else ".jpg" + filename = img.name or img.type_id + if filename in seen: # 同类型多张时加序号防覆盖 + filename = f"{filename}-{i + 1}" + seen.add(filename) + zf.writestr(f"{filename}{suffix}", data) + buf.seek(0) + return StreamingResponse( + buf, + media_type="application/zip", + headers={"Content-Disposition": f"attachment; filename=\"suite-{suite_id}.zip\""}, + ) + + +# ── 单张 AI 生图（V2.1 新增，ISS 无对应实现）────────────────────────────── + +@router.post("/suite/image-edit", response_model=ImageEditSingleResponse) +async def suite_image_edit(req: ImageEditSingleRequest) -> ImageEditSingleResponse: + """轻量单张再生成：原图 + 一句话要求 + 模型选择。同步接口（单张可长达数分钟）。""" + settings = get_settings() + provider_name = resolve_provider(req.model, None, settings.image_provider) + if provider_name not in GENERATORS: + raise HTTPException(status_code=400, detail=f"未知 provider: {provider_name}") + _validate_model(provider_name, req.model) + + spec = {"lang": "ru", "ratio": "3:4"} # 试算页固定 Ozon 规格（俄文图内文案 · 3:4） + model = req.model + is_wan = provider_name == "tongyi" and model.lower().startswith("wan") + + # 复用套图的 prompt 家族：custom 类型装配「用户要求」为最高优先级约束， + # 商品外观仍由参考图锁定（保真语义与一键生成一致）。 + ctx = build_context({"title": "AI 精修"}, fallback_name="product") + prompt = build_prompt( + provider_name, model, "custom", ctx, + style_set=5, lang=spec["lang"], + extra={"title": "AI 精修", "detail": req.prompt, "prompt_hint": ""}, + requirements=req.prompt, + ) + + try: + generator = GENERATORS[provider_name] + data = await generator(prompt, [req.image_url], size=_image_size(provider_name, spec["ratio"], is_wan=is_wan, model=model), model=model) + except RuntimeError as exc: + raise HTTPException(status_code=502, detail=str(exc)) from exc + except Exception as exc: # noqa: BLE001 + raise HTTPException(status_code=502, detail=f"生成失败: {exc}") from exc + + is_png = data[:8] == b"\x89PNG\r\n\x1a\n" + ext = ".png" if is_png else ".jpg" + ctype = "image/png" if is_png else "image/jpeg" + url = await get_storage().save_bytes(data, f"suites/single/{uuid.uuid4().hex}{ext}", ctype) + + asset_id: str | None = None + if req.append and req.product_id: + try: + asset_id = await _append_generated_asset( + req.product_id, + TaskImage(type_id="custom", name="AI生图", url=url, status="ok"), + ) + except Exception: # noqa: BLE001 + pass # 回写失败不影响结果返回，图片已在任务网格可见 + return ImageEditSingleResponse(url=url, asset_id=asset_id) diff --git a/server/assets/watermark.jpg b/server/assets/watermark.jpg new file mode 100644 index 0000000..af3f3bd Binary files /dev/null and b/server/assets/watermark.jpg differ diff --git a/server/config/settings.py b/server/config/settings.py index 4f784e7..b60ffff 100644 --- a/server/config/settings.py +++ b/server/config/settings.py @@ -50,6 +50,33 @@ class Settings(BaseSettings): # ── V2：对外地址（插件/前端回写、生成图回调）── app_base_url: str = "http://127.0.0.1:8800" + # ── V2.1 套图生图（provider 路由与 image-suite-studio 一致）── + image_provider: str = "doubao" # 未知模型时的兜底 provider：doubao | tongyi | rightapi + request_timeout: int = 300 # 单张生图请求超时（秒） + poll_max_wait: int = 600 # 异步任务轮询上限（秒） + # 水印徽章源图（图片水印用，可 .env 覆盖） + watermark_image_path: str = str(Path(__file__).resolve().parents[1] / "assets" / "watermark.jpg") + + # 豆包 / 火山方舟 Seedream + ark_api_key: str = "" + ark_base_url: str = "https://ark.cn-beijing.volces.com/api/v3/images/generations" + ark_image_model: str = "doubao-seedream-4-5-251128" + + # 通义 / DashScope（key 复用上方 dashscope_api_key；dashscope_base_http_api_url 是业务空间专用，与此处无关） + dashscope_model: str = "wan2.7-image-pro" + dashscope_base_url: str = "" # 留空按模型自动选择万象异步/千问同步端点 + + # RightAPI（OpenAI 兼容中转：gpt-image / nano-banana 系列） + rightapi_api_key: str = "" + rightapi_base_url: str = "https://rightapi.ai/draw" + rightapi_image_model: str = "gpt-image-2" + rightapi_max_retries: int = 3 # 429/5xx/超时的重试次数（1 = 不重试） + rightapi_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" + @property def cors_origin_list(self) -> list[str]: if not self.cors_origins.strip(): diff --git a/server/main.py b/server/main.py index 7bbc5c3..d24569e 100644 --- a/server/main.py +++ b/server/main.py @@ -5,7 +5,7 @@ from fastapi.middleware.cors import CORSMiddleware from fastapi.staticfiles import StaticFiles from sqlalchemy import text -from api import ai, auth, categories, collection, fx, image, ozon, products, publish, shops +from api import ai, auth, categories, collection, export, fx, image, ozon, products, proxy, publish, shops, suite from config import get_settings from db import get_engine @@ -37,6 +37,10 @@ app.include_router(fx.router) app.include_router(ai.router) app.include_router(image.router) app.include_router(ozon.router) +# V2.1 套图生图（试算页 04 区块）/ 导出 / 图片代理 +app.include_router(suite.router) +app.include_router(export.router) +app.include_router(proxy.router) @app.on_event("startup") diff --git a/server/requirements.txt b/server/requirements.txt index e70fe71..da387b0 100644 --- a/server/requirements.txt +++ b/server/requirements.txt @@ -14,3 +14,6 @@ alembic>=1.13.0 PyJWT>=2.8.0 cryptography>=42.0.0 qiniu>=7.13.0 + +# V2.1：套图生成（水印合成） +pillow>=10.0.0 diff --git a/server/schemas/suite.py b/server/schemas/suite.py new file mode 100644 index 0000000..0f9a7db --- /dev/null +++ b/server/schemas/suite.py @@ -0,0 +1,160 @@ +"""套图规划 / 一键生成 / 单张 AI 生图 契约（studio/src/services/suite.ts 同源，见 docs/v2.1/api.md）。""" +from __future__ import annotations + +from typing import Literal + +from pydantic import BaseModel, Field + +SUPPORTED_TYPES = [ + "white_bg", "key_features", "selling_pt", "material", + "lifestyle", "multi_scene", "ecommerce_detail", + "size_chart", "sku_collection", "custom", +] + +# 通义（DashScope）生图模型白名单 +TONGYI_MODELS = ["qwen-image-3.0-pro", "wan2.7-image-pro", "wan2.6-image", "wan2.6-t2i"] + +# RightAPI 生图模型白名单（gpt-image 系列 + Google nano-banana 系列，同一中转） +RIGHTAPI_MODELS = [ + "gpt-image-2", + "gpt-image-2-vip", + "nano-banana", + "nano-banana-2", + "nano-banana-2-lite", + "nano-banana-pro", +] + +# 模型 → provider 推断表：前端只传模型名，服务端据此路由（模型名优先于 provider 字段） +MODEL_PROVIDERS: dict[str, str] = { + **{m: "tongyi" for m in TONGYI_MODELS}, + **{m: "rightapi" for m in RIGHTAPI_MODELS}, +} + +# 豆包 Seedream 走火山方舟；模型由 ARK_IMAGE_MODEL 指定（不在下拉白名单内也允许直连豆包） +DOUBAO_MODELS = ["doubao-seedream-4-5-251128"] +MODEL_PROVIDERS.update({m: "doubao" for m in DOUBAO_MODELS}) + + +def resolve_provider(model: str | None, requested: str | None, default: str) -> str: + """已知模型名直接定位 provider；未知模型回退到请求指定的 provider 或默认值。""" + if model and model in MODEL_PROVIDERS: + return MODEL_PROVIDERS[model] + return requested or default + + +# 目标平台 → 文案语言 + 图片比例（平台决定规格，不再单独选语言） +PLATFORM_SPECS: dict[str, dict] = { + "ozon": {"lang": "ru", "ratio": "3:4", "label": "Ozon"}, + "wb": {"lang": "ru", "ratio": "3:4", "label": "Wildberries"}, + "cn": {"lang": "zh", "ratio": "1:1", "label": "中文（国内平台）"}, +} + + +class TextMaterial(BaseModel): + kind: str = Field(..., description="title | params | selling_point | desc | price | brand | sales | shop") + content: str = "" + pairs: list[dict] | None = None # [{key, value}] + + +class WatermarkOptions(BaseModel): + """生成图水印：AI 出图后由服务端后处理合成（与生图模型无关）。右下角，默认文字 xiongmaoyx。""" + + enabled: bool = Field(default=False, description="是否开启水印") + type: Literal["image", "text"] = Field(default="image", description="图片水印 | 文字水印") + text: str = Field(default="xiongmaoyx", description="文字水印内容") + opacity: int = Field(default=30, ge=1, le=100, description="不透明度（%）") + + +class PlanItem(BaseModel): + """出图方案项：一类图 × 数量，可绑定 SKU 规格。""" + kind: str = Field(default="custom", description="图类型（SUPPORTED_TYPES 之一）") + title: str = Field(..., description="方案标题，如「主图·粉色」") + detail: str = Field(default="", description="这张图展示什么（中文）") + prompt_hint: str = Field(default="", description="构图提示（英文，进生图 prompt）") + count: int = Field(default=1, ge=0, le=5) + variant_name: str | None = Field(default=None, description="绑定的 SKU 规格名") + + +class SuitePlanRequest(BaseModel): + """POST /api/suite/plan：DeepSeek 根据商品信息生成出图方案。""" + product_id: str | None = Field(default=None, description="便于日志与上下文定位，可空") + texts: list[TextMaterial] = Field(default_factory=list) + sku_variants: list[str] = Field(default_factory=list, description="带图的 SKU 规格名") + image_stats: dict = Field(default_factory=dict, description="分组图片数量统计 {main: n, ...}") + platform: str = Field(default="ozon") + requirements: str | None = Field(default=None, description="生图要求（最高优先级，规划方案必须遵循）") + + +class PlanItemOut(BaseModel): + kind: str + title: str + detail: str = "" + prompt_hint: str = "" + count: int = 1 + variant_name: str | None = None + + +class PlanResponse(BaseModel): + summary: str = "" + items: list[PlanItemOut] + + +class GenerateImageItem(BaseModel): + url: str = Field(..., description="勾选的参考底图 URL（stored_url 或源站原图）") + group_key: str = Field(default="main", description="main | sku | detail | upload") + variant_name: str | None = Field(default=None, description="SKU 规格名（方案绑定用）") + + +class SuiteGenerateRequest(BaseModel): + """POST /api/suite/generate：提交套图任务，返回 suite_id 后前端轮询。""" + product_id: str | None = Field(default=None, description="用于生成图回写素材（generated 组），可空") + texts: list[TextMaterial] = Field(default_factory=list) + images: list[GenerateImageItem] = Field(default_factory=list, description="勾选的参考底图") + style_set: int = Field(default=1, ge=1, le=5) + style_prompt: str | None = Field(default=None, description="用户改写的风格提示词（覆盖 style_set 模板）") + requirements: str | None = Field(default=None, description="生图要求（最高优先级，强制约束，覆盖其他设定）") + plan: list[PlanItem] | None = Field(default=None, description="出图方案（优先于 types）") + types: list[str] = Field(default_factory=list, description="旧参数：无方案时按类型生成") + platform: str = Field(default="ozon") + model: str | None = Field(default=None, description="生图模型名（按 MODEL_PROVIDERS 路由 provider）") + watermark: WatermarkOptions | None = Field(default=None, description="生成图水印（服务端后处理合成）；关闭时不传") + + +class SuiteCreateResponse(BaseModel): + suite_id: str + + +class SuiteImageOut(BaseModel): + type_id: str + name: str + url: str + status: str + error: str | None = None + + +class SuiteOut(BaseModel): + id: str + status: str + style_set: int + platform: str + lang: str + ratio: str + provider: str + model: str | None = None + total: int = 0 # 计划总张数（进度分母；images 是逐张追加，过程中 length < total） + images: list[SuiteImageOut] + error: str | None = None + + +class ImageEditSingleRequest(BaseModel): + """POST /api/suite/image-edit：单张 AI 生图（试算页每张图的「AI 生图」入口）。""" + product_id: str | None = Field(default=None, description="append=true 时回写到该商品的 generated 组") + image_url: str = Field(..., description="原图（stored_url 或 source_url，远程由服务端带 Referer 代下）") + prompt: str = Field(..., min_length=1, description="用户要求（必填，进生图指令）") + model: str = Field(default="nano-banana-2") + append: bool = Field(default=True, description="结果是否追加为商品素材（generated 组）") + + +class ImageEditSingleResponse(BaseModel): + url: str + asset_id: str | None = None diff --git a/server/services/generator.py b/server/services/generator.py new file mode 100644 index 0000000..ea99bd6 --- /dev/null +++ b/server/services/generator.py @@ -0,0 +1,494 @@ +"""套图生成服务：图像 provider（豆包 Seedream / 通义万相 / RightAPI 中转）+ 任务执行器。 + +平移自 image-suite-studio/services/generator.py，适配点： + - 存储走本仓 services/storage 抽象（本地 data/media 兜底 / 七牛），save_bytes 直接返回可访问 URL； + - 参考图解析复用 storage.local_path 与 storage.download_bytes（防盗链 Referer 由 api/proxy.guess_referer 提供）； + - run_suite 新增可选 on_image_ok 回调：每张成功即通知调用方落库（product_assets/generated）， + 本模块不接触数据库，保持「套图引擎不依赖商品模型」的独立产品化裁剪能力。 +""" +from __future__ import annotations + +import asyncio +import base64 +import logging +import mimetypes +import re +from typing import Awaitable, Callable + +import httpx + +from config import get_settings +from services.prompts import build_context, build_prompt, type_name +from services.tasks import ( + IMG_FAILED, + IMG_OK, + TASK_DONE, + TASK_FAILED, + TASK_PARTIAL, + TASK_RUNNING, + Task, + TaskImage, +) +from services.watermark import apply_watermark +from services.storage import download_bytes, get_storage, local_path + +log = logging.getLogger("suite.generator") + +# 单张成功回调：(image) -> None；由 API 层注入用于回写商品素材 +OnImageOk = Callable[[TaskImage], Awaitable[None]] + + +class ApiError(RuntimeError): + """带 HTTP 状态码的 API 错误（用于区分可重试的网关/限流错误）。""" + + def __init__(self, message: str, status: int = 0): + super().__init__(message) + self.status = status + + +_HTML_TITLE_RE = re.compile(r"]*>(.*?)", re.IGNORECASE | re.DOTALL) + + +def _raise_api_error(resp, provider: str): + """HTTP 错误时抛出带 API 错误码/信息的异常（响应体里有真正的失败原因）。""" + if resp.is_success: + return + text = resp.text or "" + if " 作摘要，避免整段 HTML 进错误信息 + m = _HTML_TITLE_RE.search(text) + detail = (re.sub(r"\s+", " ", m.group(1)).strip() if m else "") or "网关返回 HTML 错误页（上游/CDN 故障）" + 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} — {detail or '无错误详情'}", resp.status_code) + +# 参考图选择：material 用第 2 张（背面/细节），其余用第 1 张（正面） +TYPE_REF_INDEX = { + "material": 1, +} +DEFAULT_REF_COUNT = 2 # 每次生图最多带的参考图数（正面 1 张 + 背面/细节 1 张） + + +def _image_size(provider: str, ratio: str, is_wan: bool = True, model: str = "") -> str: + """平台比例 → provider 尺寸参数。3:4 竖版（Ozon），1:1 方图（国内）。""" + if provider == "doubao": + return "1536x2048" if ratio == "3:4" else "2048x2048" + if provider == "rightapi": + # gpt-image 自定义尺寸约束：16 的倍数、长短边比 ≤ 3:1（1536x2048 合法） + return "1536x2048" if ratio == "3:4" else "2048x2048" + # tongyi：万象与千问的 size 语法相同（* 分隔），档位不同 + # wan2.6 系列总像素限制在 [1280², 1440²]，wan2.7 的 1536*2048/2048*2048 会超限 + if model.startswith("wan2.6"): + return "1152*1536" if ratio == "3:4" else "1440*1440" + if ratio == "3:4": + return "1536*2048" if is_wan else "768*1024" + return "2048*2048" if is_wan else "1024*1024" + +_DOUBAO_ANTI_AI = ( + "authentic real-world photography, natural imperfections, genuine texture, " + "no synthetic look, no CGI quality, no heavy post-processing" +) + +DEFAULT_NEGATIVE_PROMPT = ( + "AI-generated look, artificial, CGI quality, 3D render, synthetic texture, " + "plastic skin, mannequin-like, too perfect, oversaturated, HDR, heavy vignette, " + "low resolution, blurry, deformed, bad anatomy, overexposed, underexposed, grainy, " + "watermark, text distortion, bad typography, overlapping text, cheap look, cartoon" +) + + +# ── 参考图解析 ──────────────────────────────────────────────────────────── + +def _bytes_to_data_uri(data: bytes, mime: str) -> str: + return f"data:{mime};base64,{base64.b64encode(data).decode()}" + + +async def _resolve_ref_bytes(url: str) -> tuple[bytes, str]: + """参考图 URL → (bytes, mime)。本地 media 文件直读磁盘；远程 URL 带 Referer 下载。 + + 生图 API 的服务器无法访问 127.0.0.1，代理 URL 也不能直接透传， + 所以统一在本地解析成原始字节再进请求体（data URI 或 multipart）。 + """ + if url.startswith("data:"): + head, _, b64 = url.partition(",") + mime = head[5:].split(";", 1)[0] or "image/jpeg" + return base64.b64decode(b64), mime + path = local_path(url) + if path is not None: + mime = mimetypes.guess_type(path.name)[0] or "image/jpeg" + return path.read_bytes(), mime + if url.startswith(("http://", "https://")): + from api.proxy import guess_referer + data, ctype = await download_bytes(url, referer=guess_referer(url)) + if not ctype.startswith("image/"): + ctype = "image/jpeg" + return data, ctype + raise FileNotFoundError(f"无法解析参考图: {url}") + + +async def _resolve_ref(url: str) -> str: + data, mime = await _resolve_ref_bytes(url) + return _bytes_to_data_uri(data, mime) + + +# ── Provider：豆包 Seedream（火山方舟）──────────────────────────────────── + +async def generate_doubao(prompt: str, ref_images: list[str], size: str = "2048x2048", model: str | None = None) -> bytes: + s = get_settings() + if not s.ark_api_key: + raise RuntimeError("未配置 ARK_API_KEY（.env）") + body = { + "model": model or s.ark_image_model, + "prompt": prompt.rstrip(". ") + ". " + _DOUBAO_ANTI_AI, + "size": size, + "response_format": "url", + "watermark": False, + "n": 1, + } + if ref_images: + body["image"] = [await _resolve_ref(u) for u in ref_images] + async with httpx.AsyncClient(timeout=s.request_timeout, verify=False) as client: + resp = await client.post( + s.ark_base_url, + headers={"Authorization": f"Bearer {s.ark_api_key}", "Content-Type": "application/json"}, + json=body, + ) + _raise_api_error(resp, "豆包") + img_url = resp.json()["data"][0]["url"] + dl = await client.get(img_url, timeout=s.request_timeout) + dl.raise_for_status() + return dl.content + + +# ── Provider：通义万相 / 千问（DashScope）──────────────────────────────── + +def _is_wan_model(model: str) -> bool: + return model.lower().startswith("wan") + + +def _is_t2i_model(model: str) -> bool: + """纯文生图模型（如 wan2.6-t2i）：不接受参考图，商品一致性只能靠文案描述。""" + return "t2i" in model.lower() + + +async def _tongyi_poll_task(client: httpx.AsyncClient, key: str, task_id: str, max_wait: int) -> str: + poll_url = "https://dashscope.aliyuncs.com/api/v1/tasks/" + task_id + elapsed, interval = 0, 3 + while elapsed < max_wait: + resp = await client.get(poll_url, headers={"Authorization": f"Bearer {key}"}, timeout=30) + resp.raise_for_status() + result = resp.json() + status = result.get("output", {}).get("task_status", "") + if status == "SUCCEEDED": + choices = result["output"].get("choices", []) + if choices: + content = choices[0].get("message", {}).get("content", []) + if content: + return content[0].get("image", "") + results = result["output"].get("results", []) + if results: + return results[0].get("url") or results[0].get("b64_image", "") + raise RuntimeError(f"通义任务成功但无结果: {result}") + if status in ("FAILED", "UNKNOWN"): + raise RuntimeError(f"通义任务失败: {result}") + await asyncio.sleep(interval) + elapsed += interval + interval = min(interval + 2, 10) + raise TimeoutError(f"通义异步任务超时 ({max_wait}s): task_id={task_id}") + + +async def generate_tongyi(prompt: str, ref_images: list[str], size: str = "2048*2048", model: str | None = None) -> bytes: + s = get_settings() + if not s.dashscope_api_key: + raise RuntimeError("未配置 DASHSCOPE_API_KEY（.env）") + model = model or s.dashscope_model + is_wan = _is_wan_model(model) + url = s.dashscope_base_url or ( + "https://dashscope.aliyuncs.com/api/v1/services/aigc/image-generation/generation" + if is_wan + else "https://dashscope.aliyuncs.com/api/v1/services/aigc/multimodal-generation/generation" + ) + + # t2i 模型不接受参考图：content 只有文本，商品一致性依赖 prompt 里的标题/卖点描述 + content: list[dict] = [] + if not _is_t2i_model(model): + content = [{"image": await _resolve_ref(u)} for u in ref_images] + content.append({"text": prompt}) + + params = {"size": size, "n": 1, "watermark": False} + if not is_wan: + params["prompt_extend"] = False + params["negative_prompt"] = DEFAULT_NEGATIVE_PROMPT[:500] + + headers = {"Authorization": f"Bearer {s.dashscope_api_key}", "Content-Type": "application/json"} + if is_wan: + headers["X-DashScope-Async"] = "enable" + + body = {"model": model, "input": {"messages": [{"role": "user", "content": content}]}, "parameters": params} + + async with httpx.AsyncClient(timeout=s.request_timeout, verify=False) as client: + resp = await client.post(url, headers=headers, json=body) + _raise_api_error(resp, "通义") + data = resp.json() + if is_wan: + task_id = data.get("output", {}).get("task_id", "") + if not task_id: + raise RuntimeError(f"通义万象未返回 task_id: {data}") + img_url = await _tongyi_poll_task(client, s.dashscope_api_key, task_id, s.poll_max_wait) + if img_url.startswith("data:") or len(img_url) > 500: + return base64.b64decode(img_url.split(",", 1)[-1] if "," in img_url else img_url) + dl = await client.get(img_url, timeout=s.request_timeout) + dl.raise_for_status() + return dl.content + img_url = data["output"]["choices"][0]["message"]["content"][0]["image"] + dl = await client.get(img_url, timeout=s.request_timeout) + dl.raise_for_status() + return dl.content + + +# ── Provider：RightAPI（gpt-image / nano-banana，OpenAI 兼容中转）────────── + +# 可重试的状态码：中转限流/网关抖动（该中转限流时返回 Cloudflare 502 而非 429） +RETRYABLE_STATUS = {429, 500, 502, 503, 504} + + +async def _rightapi_poll_task(client: httpx.AsyncClient, headers: dict, origin: str, + task_id: str, max_wait: int) -> dict: + """轮询站点级任务查询接口 GET /v1/tasks/{task_id}（不带 /draw 前缀）。 + + 实测要点： + - 完成响应**没有** status:"completed" 字段，完成判定 = 响应里出现 data； + - progress 基本不动（0~2），不能当进度条依据； + - 失败态 = status 为 failed / error / cancelled。 + """ + poll_url = f"{origin}/v1/tasks/{task_id}" + elapsed, interval = 0, 3 + while elapsed < max_wait: + resp = await client.get(poll_url, headers=headers, timeout=30) + _raise_api_error(resp, "RightAPI") + result = resp.json() + status = result.get("status", "") + if status in ("failed", "error", "cancelled"): + err = result.get("error") or {} + raise RuntimeError(f"RightAPI 任务失败: {err.get('message') or result}") + if "data" in result: + return result + await asyncio.sleep(interval) + elapsed += interval + interval = min(interval + 2, 10) + raise TimeoutError(f"RightAPI 异步任务超时 ({max_wait}s): task_id={task_id}") + + +def _rightapi_extract_image(result: dict) -> tuple[str | None, str | None]: + """从轮询完成结果里取 (kind, payload)：kind ∈ url | b64，未取到返回 (None, None)。 + + 完成形状为 Images 协议：{"created":..., "data":[{"url": "..."}]}（实测只见 url）。 + """ + data = result.get("data") or [] + if data: + item = data[0] or {} + url = item.get("url") or "" + if url: + return ("url", url) + b64 = item.get("b64_json") or "" + if b64: + return ("b64", b64) + return (None, None) + + +async def _rightapi_request(s, prompt: str, ref_images: list[str], size: str, model: str) -> bytes: + """RightAPI 各模型：统一走 /v1/images/generations（异步）。 + + 官方协议（2026-07 起统一异步）： + - POST /draw/v1/images/generations，请求体固定带 async:true，参考图放 image 数组（data-URI）； + - 返回 task_id 后轮询 GET /v1/tasks/{task_id}（站点级，不带 /draw）； + - 参数只有 model/prompt/n/size/imageSize/image/async；不传 quality/output_format/input_fidelity。 + 参考图沿用现选图逻辑（≤2 张，image 数组）。单张 1-5 分钟，轮询上限 poll_max_wait 兜底。 + """ + base = s.rightapi_base_url.rstrip("/") + # 任务查询是站点级接口，不带 /draw：从 base 里拆出 origin（https://rightapi.ai/draw → https://rightapi.ai） + origin = base.split("/draw", 1)[0].rstrip("/") or base + headers = {"Authorization": f"Bearer {s.rightapi_api_key}"} + + body = { + "model": model, + "prompt": prompt, + "n": 1, + "size": size, + "async": True, + } + if ref_images: + body["image"] = [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/generations", + headers={**headers, "Content-Type": "application/json"}, + json=body, + ) + _raise_api_error(resp, "RightAPI") + submitted = resp.json() + task_id = submitted.get("task_id") or "" + if task_id: + result = await _rightapi_poll_task(client, headers, origin, task_id, s.poll_max_wait) + else: + # 极端兜底：个别中转可能同步返回 data（文档不保证，但防御处理） + result = submitted + + kind, payload = _rightapi_extract_image(result) + if kind == "b64" and payload: + return base64.b64decode(payload.split(",", 1)[-1] if "," in payload else payload) + if kind == "url" and payload: + dl = await client.get(payload, timeout=s.request_timeout) + dl.raise_for_status() + return dl.content + raise RuntimeError(f"RightAPI 任务完成但没有图片数据: {result}") + + +async def generate_rightapi(prompt: str, ref_images: list[str], size: str = "2048x2048", model: str | None = None) -> bytes: + """带重试的 RightAPI 入口：429/5xx/超时按递增间隔重试。 + + 实测该中转对同 key 连续请求有分钟级冷却（成功一张后紧接着的请求会被网关秒拒 502）， + 60s → 120s → 240s 的退避基本能等到窗口放开。 + """ + s = get_settings() + if not s.rightapi_api_key: + raise RuntimeError("未配置 RIGHTAPI_API_KEY（.env）") + model = model or s.rightapi_image_model + attempts = max(1, s.rightapi_max_retries) + + last_exc: Exception | None = None + for i in range(attempts): + try: + return await _rightapi_request(s, prompt, ref_images, size, model) + except ApiError as exc: + last_exc = exc + if exc.status not in RETRYABLE_STATUS: + raise # 参数错误等不可重试，立即失败 + except (httpx.TimeoutException, httpx.TransportError) as exc: + last_exc = exc # 网络抖动/超时可重试 + if i == attempts - 1: + break + wait = s.rightapi_retry_wait * (2 ** i) + log.warning("RightAPI 第 %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} + + +# ── 任务执行器 ──────────────────────────────────────────────────────────── + +def _order_refs(refs: list[str], type_id: str) -> list[str]: + """参考图槽位选择 + 截断：material 偏好第 2 张，其余用第 1 张。""" + preferred = TYPE_REF_INDEX.get(type_id) + if preferred is not None and len(refs) > preferred: + refs = [refs[preferred]] + [r for i, r in enumerate(refs) if i != preferred] + return refs[:DEFAULT_REF_COUNT] + + +def _refs_for_job(images: list[dict], job: dict) -> list[str]: + """无状态路径：按方案项选参考图。 + + 优先 variant_name 精确匹配（「主图·粉色」用粉色那张 SKU 图）； + 匹配不到则回退 main 组第一张（再退到任意第一张）。 + """ + variant = job.get("variant_name") + if variant: + matched = [i["url"] for i in images if i.get("variant_name") == variant] + if matched: + return matched[:DEFAULT_REF_COUNT] + mains = [i["url"] for i in images if i.get("group_key") == "main"] + others = [i["url"] for i in images if i.get("group_key") != "main"] + pool = mains or others or [i["url"] for i in images] + if not pool: + raise RuntimeError("任务没有参考图") + return _order_refs(pool, job.get("kind", "")) + + +# 串行生成队列：所有用户共享同一批 API key，并发生成会触发中转限流 +# （rightapi 同 key 分钟级冷却）；同一时间只跑一个任务，其余保持 pending 排队。 +_GEN_LOCK = asyncio.Lock() + + +async def _store_image_bytes(task: Task, index: int, data: bytes) -> str: + """生成图字节 → 存储落盘，返回可访问 URL。按 PNG 魔数定扩展名（部分中转不遵守 output_format）。""" + is_png = data[:8] == b"\x89PNG\r\n\x1a\n" + ext = ".png" if is_png else ".jpg" + ctype = "image/png" if is_png else "image/jpeg" + return await get_storage().save_bytes(data, f"suites/{task.id}/{index}{ext}", ctype) + + +async def run_suite(task: Task, on_image_ok: OnImageOk | None = None) -> None: + """后台执行套图任务：排队 → 逐张生成 → 落盘 → 更新内存状态；单张失败不中断。 + + on_image_ok：每张成功落盘后的回调（API 层用于实时回写 product_assets/generated）， + 回调异常只记日志，不影响任务本身。 + """ + settings = get_settings() + provider_name = task.provider or settings.image_provider + generator = GENERATORS.get(provider_name) + if generator is None: + task.status = TASK_FAILED + task.error = f"未知 provider: {provider_name}" + return + + ctx = build_context(task.context or {}, fallback_name="") + model = task.model or { + "tongyi": settings.dashscope_model, + "rightapi": settings.rightapi_image_model, + }.get(provider_name, settings.ark_image_model) + is_wan = provider_name == "tongyi" and _is_wan_model(model) + size = _image_size(provider_name, task.ratio, is_wan=is_wan, model=model) + jobs = [dict(j) for j in task.plan] + + async with _GEN_LOCK: + task.status = TASK_RUNNING + ok, failed = 0, 0 + failures: list[str] = [] + for job in jobs: + type_id = job["kind"] + image = TaskImage(type_id=type_id, name=job.get("title") or type_name(type_id)) + task.images.append(image) + try: + # 提示词按模型家族分发：国产主体参考 / gpt edits 保真 / google 主体保持 + prompt = build_prompt( + provider_name, model, type_id, ctx, task.style_set, task.lang, + extra=job, style_prompt=task.style_prompt, requirements=task.requirements, + ) + refs = _refs_for_job(list(task.ref_images or []), job) + data = await generator(prompt, refs, size=size, model=model) + # 水印：AI 出图返回后、落盘前的后处理（失败不阻断，内部返回原图） + wm = task.watermark or {} + if wm.get("enabled"): + data = apply_watermark(data, wm) + image.url = await _store_image_bytes(task, len(task.images), data) + image.status = IMG_OK + ok += 1 + if on_image_ok is not None and image.url: + try: + await on_image_ok(image) + except Exception: # noqa: BLE001 + log.exception("套图 %s 第 %d 张回写商品素材失败", task.id, len(task.images)) + except Exception as exc: # noqa: BLE001 + log.exception("套图 %s 类型 %s 生成失败", task.id, type_id) + image.status = IMG_FAILED # 默认 pending，失败显式置 failed + image.error = str(exc)[:500] + failures.append(f"{job.get('title') or type_name(type_id)}：{str(exc)[:200]}") + failed += 1 + + task.status = TASK_DONE if failed == 0 else (TASK_PARTIAL if ok > 0 else TASK_FAILED) + if failed: + uniq = list(dict.fromkeys(failures)) # 去重保序 + detail = "；".join(uniq[:6]) + if len(uniq) > 6: + detail += f"；…等共 {failed} 张失败" + if ok == 0: + task.error = f"全部生成失败。{detail}" + else: + task.error = f"部分生成失败（{failed} 张）。{detail}" diff --git a/server/services/planner.py b/server/services/planner.py new file mode 100644 index 0000000..edbf1c9 --- /dev/null +++ b/server/services/planner.py @@ -0,0 +1,227 @@ +"""出图方案规划器：DeepSeek 根据采集的商品信息生成套图方案。 + +方案每项 = 一类图（标题 + 说明 + 生图提示 + 张数 + 可选 SKU 绑定）， +生成时按方案逐张出图；参考图可按 variant_name 精确绑定到对应 SKU 图。 +""" +from __future__ import annotations + +import json +import logging +import re + +import httpx + +from config import get_settings + +log = logging.getLogger("suite.planner") + +# 规划器可选用的图类型（与 prompt.py 的 builder 对应） +ALLOWED_KINDS = [ + "white_bg", "key_features", "selling_pt", "material", + "lifestyle", "multi_scene", "ecommerce_detail", + "size_chart", "sku_collection", "custom", +] + +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。 +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 用英文描述构图要点。 + +## 输出示例（紧凑单行） +{"summary":"三色收纳盒全套图","items":[{"kind":"white_bg","title":"主图·粉色","detail":"粉色SKU白底主视觉","prompt_hint":"front view on white background","count":1,"variant_name":"粉色"}]}""" + + +def _system_prompt_with_requirements(requirements: str | None) -> str: + """把生图要求作为最高优先级约束注入 system prompt（置于规划规则之前）。 + + 不仅声明优先级，还明确要求把要求落地到每个方案项的 prompt_hint， + 避免模型只把要求当作背景信息而不影响输出。 + """ + if not (requirements and requirements.strip()): + return SYSTEM_PROMPT + marker = "\n## 输出硬性约束" + idx = SYSTEM_PROMPT.find(marker) + if idx < 0: + return SYSTEM_PROMPT + req = requirements.strip() + block = ( + "\n## 生图要求（最高优先级，硬性约束，覆盖下方所有规划规则与约束）\n" + + req + + "\n\n" + + "规划方案时，必须把上述生图要求落地到每一项：\n" + + "1. 每个方案项的 prompt_hint 必须融入上述要求的关键约束（如要求纯黑背景，则每个 prompt_hint 都要写明 black background）；\n" + + "2. title / detail 措辞不得与上述要求矛盾；\n" + + "3. 任何规划规则与上述要求冲突时，一律以本生图要求为准。\n" + ) + return SYSTEM_PROMPT[:idx] + block + SYSTEM_PROMPT[idx:] + + +def _normalize_items(raw_items: list, sku_variants: list[str]) -> list[dict]: + """清洗模型输出：kind 白名单、count 钳制、variant 必须真实存在。""" + items: list[dict] = [] + for it in raw_items: + if not isinstance(it, dict): + continue + 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 + try: + count = max(0, min(3, int(it.get("count", 1)))) + except (TypeError, ValueError): + count = 1 + variant = str(it.get("variant_name") or "").strip() or None + if variant and variant not in sku_variants: + variant = None # 幻觉规格：丢弃绑定，回退主图 + items.append({ + "kind": kind, + "title": title, + "detail": str(it.get("detail") or "").strip()[:80], + "prompt_hint": str(it.get("prompt_hint") or "").strip()[:300], + "count": count, + "variant_name": variant, + }) + return items + + +def _repair_truncated(s: str) -> dict | None: + """截断修复：从最后一个完整的 '}' 处截断，剥尾逗号后按括号配平补全闭合。 + + 适用于「items 数组中途被 max_tokens 截断」的场景——截断点在完整对象边界， + 此前的字符串必然已闭合，简单计数配平即可。 + """ + for cut in (m.end() for m in reversed(list(re.finditer(r'\}', s)))): + cand = s[:cut].rstrip().rstrip(',') + opens: list[str] = [] + for ch in cand: + if ch in '{[': + opens.append(ch) + elif ch == '}' and opens and opens[-1] == '{': + opens.pop() + elif ch == ']' and opens and opens[-1] == '[': + opens.pop() + suffix = ''.join('}' if o == '{' else ']' for o in reversed(opens)) + try: + data = json.loads(cand + suffix) + if isinstance(data, dict): + return data + except json.JSONDecodeError: + continue + return None + + +def _extract_json(text: str) -> dict: + """从模型输出提取 JSON：剥离思考块/markdown 围栏，截断时尝试修复。""" + s = (text or '').strip() + # 剥离思考块（思考型模型会把推理过程放进 ） + s = re.sub(r'.*?', '', s, flags=re.S).strip() + # 剥离 markdown 代码围栏 + m = re.search(r'```(?:json)?\s*(.*?)```', s, flags=re.S) + if m: + s = m.group(1).strip() + try: + return json.loads(s) + except json.JSONDecodeError: + pass + start = s.find('{') + if start >= 0: + repaired = _repair_truncated(s[start:]) + if repaired is not None: + log.warning("规划器输出疑似被截断，已自动截断修复（可能丢失末尾部分方案项）") + return repaired + raise ValueError("模型输出无法解析为 JSON") + + +async def generate_plan( + product_info: dict, + sku_variants: list[str], + image_stats: dict, + platform: str, + requirements: str | None = None, +) -> dict: + """调用 DeepSeek 生成方案。返回 {summary, items}。 + + requirements：生图要求，最高优先级注入 system prompt，规划方案必须遵循。 + """ + s = get_settings() + if not s.deepseek_api_key: + raise RuntimeError("未配置 DEEPSEEK_API_KEY（.env）") + + user_payload: dict = { + "商品信息": product_info, # {title, desc, params:[{key,value}], sellingPoints, price} + "SKU规格": sku_variants, # 带图的 SKU 规格名（variant_name 只能从中选） + "图片统计": image_stats, # {main: n, sku: n, detail: n} + "目标平台": platform, # ozon/wb/cn（决定图内文案语言） + } + # 生图要求同时在 user 侧强调（与 system prompt 双重约束），确保模型真正遵循 + if requirements and requirements.strip(): + 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"] + + try: + data = _extract_json(content) + except ValueError as exc: + log.error( + "规划器输出解析失败 finish_reason=%s content[:200]=%s", + 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} diff --git a/server/services/prompts/__init__.py b/server/services/prompts/__init__.py index e69de29..986277c 100644 --- a/server/services/prompts/__init__.py +++ b/server/services/prompts/__init__.py @@ -0,0 +1,45 @@ +"""套图提示词引擎：按模型家族分发，各家族独立封装。 + +不同家族的生图语义差异极大，共用一套提示词会导致语义错配 +（gpt-image-2 按文字重造商品即由此而来），故按家族各自成册： + + alibaba 通义 wan*/qwen*（DashScope）—— 主体参考语义 + doubao 豆包 Seedream（火山方舟）—— 主体参考语义，与通义共用装配 + gpt gpt-image-2 / gpt-image-2-vip（RightAPI）—— /v1/images/edits 编辑语义 + google nano-banana 系列（RightAPI）—— 原生主体保持语义 + +路由规则：provider 为主；rightapi 内再按模型名细分 gpt / google。 +""" +from __future__ import annotations + +from . import alibaba, doubao, google, gpt +from .common import build_context, type_name + +_MODULE_BY_FAMILY = { + "alibaba": alibaba, + "doubao": doubao, + "gpt": gpt, + "google": google, +} + + +def prompt_family(provider: str, model: str | None) -> str: + """(provider, model) → 提示词家族名。""" + if provider == "rightapi": + if (model or "").lower().startswith("nano-banana"): + return "google" + return "gpt" # gpt-image-* 及未知中转模型默认按 edits 语义处理 + if provider == "tongyi": + return "alibaba" + return "doubao" # doubao 及默认 provider + + +def build_prompt(provider: str, model: str | None, type_id: str, ctx: dict, style_set: int, + lang: str, extra: dict | None = None, style_prompt: str | None = None, + requirements: str | None = None) -> str: + """按模型家族构造指定图类型的完整生图 prompt。参数含义见各家族 build_prompt。""" + module = _MODULE_BY_FAMILY[prompt_family(provider, model)] + return module.build_prompt( + type_id, ctx, style_set, lang, + extra=extra, style_prompt=style_prompt, requirements=requirements, + ) diff --git a/server/services/prompts/alibaba.py b/server/services/prompts/alibaba.py new file mode 100644 index 0000000..fc921a4 --- /dev/null +++ b/server/services/prompts/alibaba.py @@ -0,0 +1,169 @@ +"""阿里通义（wan* 万相 / qwen* 千问）提示词：国产"主体参考"语义。 + +生图 API 把参考图当商品锚（subject reference）、prompt 当场景描述， +风格词/文字商品描述不会反噬商品本体，负面清单也可以安全写入 prompt。 +豆包（doubao.py）与此语义一致，直接复用本模块装配。 +""" +from __future__ import annotations + +from .common import ( + STYLE_SETS, TEXT_RENDER, requirements_block, resolve_style, selling_point_lines, +) + +# ── 公共组件（主体参考语义专用）──────────────────────────────────────────── + +QUALITY = ( + "Shot on Sony A7R V with 85mm lens at f/2.0, ultra-detailed, photorealistic, " + "8K commercial image quality, professional retouching." +) + +PRODUCT_REF_LOCK = ( + "CRITICAL: The product must look EXACTLY the same as in the reference image — " + "identical silhouette, proportions, colors, print pattern, stitching and every design detail. " + "Only the background, camera angle, lighting and styling may change. " + "Do not redesign, add or remove any element of the product." +) + +DEFAULT_NEGATIVE_INTENT = ( + "no AI-generated look, no CGI quality, no plastic appearance, no watermark, " + "no distorted text, no deformed product, no extra limbs, no blurry areas" +) + +# ── 各图类型 Prompt ─────────────────────────────────────────────────────── + +def _prompt_white_bg(ctx: dict, style: dict, lang: str) -> str: + return ( + f"E-commerce main product image on pure white background (RGB 255,255,255), " + f"product \"{ctx['title']}\" centered and filling about 85% of the frame, " + f"front view, even shadowless studio lighting with a faint natural contact shadow, " + f"{style['tone']}. No text, no props, no background elements. {QUALITY} {PRODUCT_REF_LOCK}" + ) + +def _prompt_key_features(ctx: dict, style: dict, lang: str) -> str: + sp = selling_point_lines(ctx, lang) or ctx["title"] + return ( + f"E-commerce key-features infographic for product \"{ctx['title']}\", square layout: " + f"product on the left two-thirds ({style['bg']}), right column lists 3 feature callouts " + f"with minimal line icons, thin leader lines pointing to product details. " + f"Feature callouts: {sp}. {style['tone']}. {TEXT_RENDER[lang]} {QUALITY} {PRODUCT_REF_LOCK}" + ) + +def _prompt_selling_pt(ctx: dict, style: dict, lang: str) -> str: + sp = selling_point_lines(ctx, lang, 1) or ctx["title"] + return ( + f"Single-selling-point e-commerce poster for product \"{ctx['title']}\": " + f"hero product close-up at dynamic angle ({style['bg']}), one large bold headline " + f"about \"{sp}\", generous negative space, one small magnified detail circle " + f"highlighting material or craft. {style['tone']}. {TEXT_RENDER[lang]} {QUALITY} {PRODUCT_REF_LOCK}" + ) + +def _prompt_material(ctx: dict, style: dict, lang: str) -> str: + return ( + f"Macro material close-up of product \"{ctx['title']}\": extreme detail shot revealing " + f"fabric weave / surface texture / stitching / finish, shallow depth of field, " + f"raking light across the surface, {style['tone']}. Small caption label in corner. " + f"{TEXT_RENDER[lang]} {QUALITY} {PRODUCT_REF_LOCK}" + ) + +def _prompt_lifestyle(ctx: dict, style: dict, lang: str) -> str: + bg = f" ({style['bg']})" if style.get("bg") else "" + return ( + f"Lifestyle in-context scene for product \"{ctx['title']}\": the product is naturally " + f"used / placed in a real environment{bg}, realistic human-scale surroundings, " + f"soft daylight, authentic candid mood, product remains the clear visual focus. " + f"{style['tone']}. {TEXT_RENDER[lang]} {QUALITY} {PRODUCT_REF_LOCK}" + ) + +def _prompt_multi_scene(ctx: dict, style: dict, lang: str) -> str: + sp = selling_point_lines(ctx, lang) + return ( + f"Triptych multi-scene e-commerce image for product \"{ctx['title']}\": three vertical panels " + f"separated by thin gutters, each panel shows the SAME product in a different usage scene " + f"(e.g. home interior / outdoor street / office desk), consistent color grading across panels. " + f"Panel captions: {sp}. {style['tone']}. {TEXT_RENDER[lang]} {QUALITY} {PRODUCT_REF_LOCK}" + ) + +def _prompt_ecommerce_detail(ctx: dict, style: dict, lang: str) -> str: + sp = selling_point_lines(ctx, lang) or ctx["title"] + params = ctx["params_line"] + return ( + f"E-commerce detail-page hero section for product \"{ctx['title']}\", square layout: " + f"top half is a hero banner with the product at a 3/4 angle ({style['bg']}); " + f"bottom half is a clean spec card listing 3 feature rows with line icons" + + (f" (specs: {params})" if params else "") + + f" and one highlighted row: {sp}. {style['tone']}. {TEXT_RENDER[lang]} {QUALITY} {PRODUCT_REF_LOCK}" + ) + +def _prompt_size_chart(ctx: dict, style: dict, lang: str) -> str: + dims = ctx["params_line"] + return ( + f"Product size chart infographic for \"{ctx['title']}\": product shown in clean front and side views " + f"on light background, with thin measurement annotation lines (arrows) marking length, width and height, " + f"measurement values rendered next to each line" + + (f" (known specs: {dims})" if dims else "") + + f", small caption row, precise technical drawing aesthetic. {style['tone']}. " + f"{TEXT_RENDER[lang]} {QUALITY} {PRODUCT_REF_LOCK}" + ) + +def _prompt_sku_collection(ctx: dict, style: dict, lang: str) -> str: + return ( + f"Colorway collection image for product \"{ctx['title']}\": the SAME product in all its color/variant " + f"options arranged in a neat equal grid (2-4 items per row), each colorway with a small label chip below it, " + f"consistent lighting and scale across all items, clean e-commerce presentation. " + f"{style['tone']}. {TEXT_RENDER[lang]} {QUALITY} {PRODUCT_REF_LOCK}" + ) + +def _prompt_custom(ctx: dict, style: dict, lang: str, extra: dict) -> str: + hint = (extra.get("prompt_hint") or "").strip() + purpose = extra.get("title") or "" + detail = extra.get("detail") or "" + bg = f" {style['bg']} as environment." if style.get("bg") else "" + composed = ( + f"E-commerce marketing image for product \"{ctx['title']}\"" + + (f" — {purpose}" if purpose else "") + + (f": {detail}" if detail else "") + + "." + ) + if hint: + composed += f" Composition: {hint}." + return f"{composed}{bg} {style['tone']}. {TEXT_RENDER[lang]} {QUALITY} {PRODUCT_REF_LOCK}" + +_PROMPT_BUILDERS = { + "white_bg": _prompt_white_bg, + "key_features": _prompt_key_features, + "selling_pt": _prompt_selling_pt, + "material": _prompt_material, + "lifestyle": _prompt_lifestyle, + "multi_scene": _prompt_multi_scene, + "ecommerce_detail": _prompt_ecommerce_detail, + "size_chart": _prompt_size_chart, + "sku_collection": _prompt_sku_collection, +} + + +def build_prompt(type_id: str, ctx: dict, style_set: int, lang: str, extra: dict | None = None, + style_prompt: str | None = None, requirements: str | None = None) -> str: + """构造指定图类型的完整生图 prompt（主体参考语义）。 + + extra: 方案项信息 {title, detail, prompt_hint}——custom 类型必需， + 预设类型也会把 prompt_hint 作为构图补充注入。 + style_prompt: 用户改写的风格提示词，覆盖 style_set 内置模板（tone/bg 整体替换）。 + requirements: 生图要求（最高优先级，强制约束），置于 prompt 最前面， + 声明覆盖一切冲突指令，用户可在此输入强制要求。 + """ + style = resolve_style(style_set, style_prompt) + extra = extra or {} + if type_id == "custom": + prompt = _prompt_custom(ctx, style, lang, extra) + else: + builder = _PROMPT_BUILDERS.get(type_id) + if builder is None: + raise ValueError(f"未知图类型: {type_id}") + prompt = builder(ctx, style, lang) + hint = (extra.get("prompt_hint") or "").strip() + if hint: + prompt = prompt.rstrip(".") + f". Additional composition guidance: {hint}." + req = requirements_block(requirements) + if req: + prompt = f"{req} {prompt}" + return prompt + ". " + DEFAULT_NEGATIVE_INTENT diff --git a/server/services/prompts/common.py b/server/services/prompts/common.py new file mode 100644 index 0000000..6422354 --- /dev/null +++ b/server/services/prompts/common.py @@ -0,0 +1,154 @@ +"""提示词公共层：与模型家族无关的商品上下文、风格模板、图类型名与文案组件。 + +各家族模块（alibaba / doubao / gpt / google）只负责"如何对模型说话"， +商品信息提炼与风格体系统一在这里维护，避免多处漂移。 +""" +from __future__ import annotations + +import re + +# ── 风格模板（与插件端 STYLE_SET_OPTIONS 对应；提示词可被用户在插件里改写覆盖）─── + +STYLE_SETS: dict[int, dict] = { + 1: { + "name": "北欧极简", + "tone": "北欧极简风：浅灰或米白背景，柔和漫射光，低饱和色调，画面留白充足，构图克制干净", + "bg": "", + }, + 2: { + "name": "清新明亮", + "tone": "清新明亮风：明亮的白色到浅蓝渐变背景，高调光线，色彩明快通透，整体轻盈干净", + "bg": "", + }, + 3: { + "name": "高级感深色", + "tone": "高级质感风：深灰或炭黑背景，戏剧性侧光打光，突出商品材质与光泽，沉稳高级", + "bg": "", + }, + 4: { + "name": "暖调生活", + "tone": "温暖生活风：暖米色背景，暖色灯光氛围，温馨的家居质感，亲和力强", + "bg": "", + }, + 5: { + "name": "纯净棚拍", + "tone": "标准电商棚拍：纯色浅背景，均匀的正面柔光，无杂物干扰，商品居中突出", + "bg": "", + }, +} + +# ── 图类型中文名（导出文件名用）─────────────────────────────────────────── + +TYPE_NAMES_ZH: dict[str, str] = { + "white_bg": "白底主图", + "key_features": "核心卖点图", + "selling_pt": "卖点图", + "material": "材质图", + "lifestyle": "场景展示图", + "multi_scene": "多场景拼图", + "ecommerce_detail": "电商详情图", + "size_chart": "尺寸标注图", + "sku_collection": "SKU合集图", + "custom": "创意图", +} + +# ── 图内营销文案渲染规范（各家族共用；语言由平台决定）────────────────────── + +TEXT_RENDER = { + "zh": ( + "Render concise Chinese marketing text inside the image: main headline max 8 Chinese characters, " + "sub-lines max 12 characters each, font is modern clean sans-serif (Source Han Sans style), " + "high legibility, tasteful typography layout, colors harmonized with the composition. " + "No spelling errors, no garbled characters." + ), + "en": ( + "Render concise English marketing text inside the image: headline max 5 words, " + "sub-lines max 8 words each, Helvetica Neue style sans-serif, high legibility, " + "tasteful typography layout, colors harmonized with the composition. No spelling errors." + ), + "ru": ( + "Render concise Russian marketing text inside the image: headline max 4 words, " + "sub-lines max 6 words each, modern clean sans-serif (Inter / PT Sans style), " + "proper Cyrillic typography, high legibility, tasteful layout, colors harmonized with the composition. " + "No spelling errors, no mixed latin/cyrillic gibberish." + ), +} + + +def resolve_style(style_set: int, style_prompt: str | None = None) -> dict: + """用户改写的风格提示词整体覆盖内置模板（tone/bg 整体替换）。""" + if style_prompt and style_prompt.strip(): + return {"name": "custom", "tone": style_prompt.strip(), "bg": ""} + return STYLE_SETS.get(style_set, STYLE_SETS[1]) + + +def requirements_block(requirements: str | None) -> str: + """用户强制要求块：最高优先级、置于提示词最前、覆盖冲突指令（原文保留不翻译）。""" + if requirements and requirements.strip(): + return ( + "STRICT REQUIREMENTS (highest priority, must be followed exactly, " + "override any conflicting instruction): " + + requirements.strip().rstrip(".") + + "." + ) + return "" + + +# ── 商品上下文提炼 ──────────────────────────────────────────────────────── + +def _shorten(text: str, n: int) -> str: + text = re.sub(r"\s+", " ", (text or "")).strip() + return text[:n] + +def _clean_title(title: str) -> str: + """去掉常见堆砌词，让标题更可读。""" + t = _shorten(title, 60) + return re.sub(r"[【【】】\\[\\]|/]", " ", t).strip() + +def build_context(raw: dict, fallback_name: str = "", fallback_desc: str = "") -> dict: + """从采集数据提炼生图上下文：标题、描述行、卖点列表、参数行。 + + raw: {title, desc, price, params: [{key, value}], sellingPoints} + """ + title = _clean_title(raw.get("title") or fallback_name or "product") + desc = _shorten(raw.get("desc") or fallback_desc or "", 200) + + # 卖点：优先显式卖点文本；否则从参数表里挑短而有信息量的键值对 + selling_points: list[dict] = [] + sp_text = raw.get("sellingPoints") or "" + if sp_text: + for chunk in re.split(r"[；;\n·]+|(?= 5: + break + + params_line = "; ".join( + f"{p.get('key')}: {p.get('value')}" for p in (raw.get("params") or [])[:8] + ) + return { + "title": title, + "title_en": title, # 采集源多为中文标题，英文场景直接用原词避免乱翻译 + "desc": desc, + "selling_points": selling_points[:3], + "params_line": params_line, + "price": raw.get("price") or "", + } + +def selling_point_lines(ctx: dict, lang: str, max_n: int = 3) -> str: + """卖点列表 → 单行文案（图内 callout/标题用），无卖点返回空串。""" + sps = ctx["selling_points"][:max_n] + if not sps: + return "" + key = "zh" if lang == "zh" else "en" + return "; ".join(s[key] for s in sps if s.get(key)) + + +def type_name(type_id: str) -> str: + return TYPE_NAMES_ZH.get(type_id, type_id) diff --git a/server/services/prompts/doubao.py b/server/services/prompts/doubao.py new file mode 100644 index 0000000..eeea4c2 --- /dev/null +++ b/server/services/prompts/doubao.py @@ -0,0 +1,9 @@ +"""豆包（火山方舟 Seedream）提示词。 + +豆包与通义同为国产"主体参考"生图模型：参考图即商品锚、prompt 为场景描述， +提示词语义一致，直接复用阿里系装配；差异（去 AI 味后缀）在 generator 层追加。 +独立成文件便于后续按豆包特性分化。 +""" +from __future__ import annotations + +from .alibaba import build_prompt as build_prompt # noqa: F401 主体参考语义与通义共用 diff --git a/server/services/prompts/google.py b/server/services/prompts/google.py new file mode 100644 index 0000000..03cd3a0 --- /dev/null +++ b/server/services/prompts/google.py @@ -0,0 +1,163 @@ +"""Google 图像模型（nano-banana / nano-banana-2 / nano-banana-2-lite / nano-banana-pro）提示词。 + +语义：Gemini 图像编辑 —— 原生主体保持能力强，输入图即"主体 + 底图"， +对自然语言指令遵循好。不套用 GPT 的编辑契约（冗长的拒绝条款反而稀释指令）， +也不用负面清单（无 negative_prompt 参数）。要点： + - 开头一句话钉死"主体 = 第一张图里的商品，逐像素保持"； + - 指令自然语言描述目标画面（场景/排版/文案），不重述商品外观； + - 标题/参数仅作识别背景并声明以图为准。 +""" +from __future__ import annotations + +from .common import TEXT_RENDER, requirements_block, resolve_style, selling_point_lines + +_SUBJECT_LOCK = ( + "SUBJECT LOCK (highest priority): the product in the first image is the subject. " + "Keep it exactly as photographed — same shape, proportions, colors, print/pattern, " + "logo, label and every detail — and place that very product into the result. " + "A second image, when present, is another view of the same product for reference only." +) + +_QUALITY = ( + "OUTPUT: photorealistic commercial e-commerce photography, ultra-detailed, " + "natural light and shadow, professional retouching." +) + +_REMINDER = ( + "Reminder: keep the product exactly as in the first image; change only its surroundings, " + "composition, lighting and overlay graphics." +) + + +def _anchor(ctx: dict) -> str: + """商品文字锚定：仅供识别，明确以图为准（同 gpt 模块，避免文字反噬商品）。""" + line = f"Context (identification only): the product is \"{ctx['title']}\"" + if ctx.get("params_line"): + line += f" ({ctx['params_line']})" + return line + ". The image, not this text, defines the product's appearance." + + +# ── 各图类型指令（自然语言编辑口吻）──────────────────────────────────────── + +def _task_white_bg(ctx: dict, lang: str) -> str: + return ( + "Replace the background of this product photo with seamless pure white (RGB 255,255,255): " + "product centered in front view filling about 85% of the frame, even studio lighting with only " + "a faint natural contact shadow. No props, no added text, no background elements." + ) + +def _task_key_features(ctx: dict, lang: str) -> str: + sp = selling_point_lines(ctx, lang) or ctx["title"] + return ( + "Create a square key-features infographic: the unchanged product on the left two-thirds, " + "a clean right-hand panel with 3 feature callouts using minimal line icons and thin leader " + f"lines pointing at the product. Callout copy: {sp}." + ) + +def _task_selling_pt(ctx: dict, lang: str) -> str: + sp = selling_point_lines(ctx, lang, 1) or ctx["title"] + return ( + "Turn the photo into a single-selling-point poster: hero close-up of the unchanged product at " + f"a dynamic angle, one large bold headline about \"{sp}\", generous negative space, and a small " + "magnified circle zooming into an existing detail of the product." + ) + +def _task_material(ctx: dict, lang: str) -> str: + return ( + "Create an extreme macro close-up of an existing area of the product's surface, showing its " + "true fabric weave / texture / stitching exactly as in the photo; shallow depth of field, " + "raking light, small caption in a corner." + ) + +def _task_lifestyle(ctx: dict, lang: str) -> str: + return ( + "Place the unchanged product into a realistic everyday scene where it would naturally be used: " + "human-scale surroundings, soft daylight, authentic candid mood, the product as the clear visual focus." + ) + +def _task_multi_scene(ctx: dict, lang: str) -> str: + sp = selling_point_lines(ctx, lang) + task = ( + "Build a triptych of three vertical panels separated by thin gutters: each panel shows an " + "identical copy of the product in a different usage scene (home interior / outdoor street / " + "office desk), with consistent color grading across panels." + ) + if sp: + task += f" Panel captions: {sp}." + return task + +def _task_ecommerce_detail(ctx: dict, lang: str) -> str: + sp = selling_point_lines(ctx, lang) or ctx["title"] + params = ctx["params_line"] + return ( + "Compose a square detail-page hero section: top half a hero banner with the unchanged product " + "at a 3/4 angle; bottom half a clean spec card with 3 feature rows and line icons" + + (f" (specs: {params})" if params else "") + + f", one highlighted row: {sp}." + ) + +def _task_size_chart(ctx: dict, lang: str) -> str: + dims = ctx["params_line"] + return ( + "Create a size chart: the unchanged product in clean front and side views on a light background, " + "thin measurement annotation lines (arrows) marking length, width and height with values beside " + "each line" + + (f" (known specs: {dims})" if dims else "") + + ", small caption row, precise technical-drawing aesthetic." + ) + +def _task_sku_collection(ctx: dict, lang: str) -> str: + # 不展开"全部配色"：会凭空造出新商品；只排列同一件的多个副本 + return ( + "Arrange several identical copies of the product in a neat equal grid (2-4 per row) with a small " + "label chip below each copy; identical lighting and scale across copies. Every copy shows this " + "exact product — do not invent other colorways or variants." + ) + +_TASK_BUILDERS = { + "white_bg": (_task_white_bg, False), + "key_features": (_task_key_features, True), + "selling_pt": (_task_selling_pt, True), + "material": (_task_material, True), + "lifestyle": (_task_lifestyle, True), + "multi_scene": (_task_multi_scene, True), + "ecommerce_detail": (_task_ecommerce_detail, True), + "size_chart": (_task_size_chart, True), + "sku_collection": (_task_sku_collection, True), +} + + +def build_prompt(type_id: str, ctx: dict, style_set: int, lang: str, extra: dict | None = None, + style_prompt: str | None = None, requirements: str | None = None) -> str: + """构造指定图类型的 prompt：要求块 → 指令 → 主体锁 → 锚定 → 风格 → 文案 → 画质 → 提醒。""" + style = resolve_style(style_set, style_prompt) + extra = extra or {} + hint = (extra.get("prompt_hint") or "").strip() + + if type_id == "custom": + purpose = extra.get("title") or "" + detail = extra.get("detail") or "" + task = "Create an e-commerce marketing image featuring the product from the first image" + task += f" — {purpose}" if purpose else "" + task += f": {detail}" if detail else "" + task += "." + wants_text = True + else: + entry = _TASK_BUILDERS.get(type_id) + if entry is None: + raise ValueError(f"未知图类型: {type_id}") + builder, wants_text = entry + task = builder(ctx, lang) + if hint: + task += f" Composition guidance: {hint}." + + parts = [p for p in (requirements_block(requirements),) if p] + parts.append(task) + parts.append(_SUBJECT_LOCK) + parts.append(_anchor(ctx)) + parts.append(f"Scene style (scene and background only, never the product): {style['tone']}.") + if wants_text: + parts.append(f"Text overlay (a graphic layer, never printed on the product): {TEXT_RENDER[lang]}") + parts.append(_QUALITY) + parts.append(_REMINDER) + return "\n\n".join(parts) diff --git a/server/services/prompts/gpt.py b/server/services/prompts/gpt.py new file mode 100644 index 0000000..5f8c64a --- /dev/null +++ b/server/services/prompts/gpt.py @@ -0,0 +1,195 @@ +"""GPT 图像模型（gpt-image-2 / gpt-image-2-vip，RightAPI 中转）提示词。 + +语义：/v1/images/edits —— 输入图是"被编辑的照片"，prompt 是编辑指令； +与通义/豆包的"主体参考"语义完全不同：参考图不是商品锚，模型会按文字指令 +重新渲染整张图。此前与国产模型共用场景提示词，再用文字锚定商品并要求输出 +"匹配商品描述"，导致模型把商品改造成营销关键词描述的样子（必现商品被改）。 + +本模块写法原则： + 1. 商品只由 Image 1 定义；标题/参数仅作识别背景并声明"以图为准"， + 绝不要求输出匹配文字描述（那等于授权模型改商品）； + 2. 指令只说"改什么"（背景/场景/排版/文案），不描述商品外观； + 3. 分节精简、首尾重申保真；不用负面清单（gpt 无 negative_prompt 参数， + 罗列畸形反而往上下文植入概念）； + 4. sku 合集 / 多拼图明确"复制同一件商品，禁止发明新配色或变体"。 +""" +from __future__ import annotations + +from .common import TEXT_RENDER, requirements_block, resolve_style, selling_point_lines + +# 保真锁：商品由 Image 1 唯一定义，其余指令一律不得触碰商品本体 +_PRESERVE = ( + "PRESERVE (absolute, overrides every other instruction below): the product shown in Image 1. " + "Reuse the photographed product exactly as it is — identical shape, silhouette, proportions, " + "colors, print/pattern, logo and label text, materials, stitching and surface details. " + "Do not redesign, restyle, recolor, re-pattern, tidy up or substitute the product, " + "and do not let any style or text instruction below alter it. Image 2 is a secondary " + "view of the same product for reference only." +) + +_STYLE = ( + "SCENE STYLE (applies to background, scene, props and lighting only — never to the product): " +) + +_QUALITY = ( + "OUTPUT: photorealistic commercial e-commerce photography, ultra-detailed, " + "natural light and shadow, professional retouching." +) + +_REMINDER = ( + "FINAL CHECK: the product itself must remain exactly as photographed in Image 1 — " + "only its surroundings, composition, lighting and overlay graphics may differ." +) + +# 图内文案：明确是"排版图层"，不落在商品本体上 +_TEXT_SCOPE = ( + "TEXT OVERLAY (a graphic layer on the composition, never printed on the product): " +) + + +def _anchor(ctx: dict) -> str: + """商品文字锚定：仅供识别，明确声明以图为准。 + + 只放标题 + 参数、不放营销描述——描述里的卖点词（"卡通""加固""防水"等） + 在 edits 语义下会被执行到商品上；官逆通道（-vip）参考图被弱化时， + 文字锚定用于帮模型认出"是哪件商品"，而不是"长什么样"。 + """ + line = f"CONTEXT (identification only): the product is \"{ctx['title']}\"" + if ctx.get("params_line"): + line += f" ({ctx['params_line']})" + return ( + line + + ". Image 1 — not this text — defines the product's appearance; " + "if they ever conflict, follow Image 1." + ) + + +# ── 各图类型的编辑指令（只描述改动，不描述商品）──────────────────────────── + +def _task_white_bg(ctx: dict, lang: str) -> str: + return ( + "TASK: Clean up this product photo for a marketplace listing. Replace the entire " + "background with seamless pure white (RGB 255,255,255); recompose with the product " + "centered in front view filling about 85% of the frame; keep only a faint natural " + "contact shadow. No props, no text, no background elements." + ) + +def _task_key_features(ctx: dict, lang: str) -> str: + sp = selling_point_lines(ctx, lang) or ctx["title"] + return ( + "TASK: Feature infographic on a square canvas. Keep the product unchanged on the left " + "two-thirds; build the right third as a clean info panel listing 3 feature callouts with " + f"minimal line icons and thin leader lines pointing at parts of the product. Callout copy: {sp}." + ) + +def _task_selling_pt(ctx: dict, lang: str) -> str: + sp = selling_point_lines(ctx, lang, 1) or ctx["title"] + return ( + "TASK: Single-selling-point poster. Hero close-up of the unchanged product at a dynamic " + f"angle, generous negative space, one large bold headline about \"{sp}\", plus one small " + "magnified circle zooming into an existing detail of the product (zoom only — do not " + "invent details that are not in the photo)." + ) + +def _task_material(ctx: dict, lang: str) -> str: + return ( + "TASK: Material close-up. Zoom tightly into an existing area of the product's surface and " + "show its true texture — fabric weave, surface finish, stitching — exactly as it appears in " + "Image 1; shallow depth of field, raking light across the surface; small caption label in a corner." + ) + +def _task_lifestyle(ctx: dict, lang: str) -> str: + return ( + "TASK: Lifestyle scene. Place the unchanged product into a realistic everyday environment " + "where it would naturally be used: human-scale surroundings, soft daylight, authentic candid " + "mood, matched shadows and color temperature, the product remaining the clear visual focus." + ) + +def _task_multi_scene(ctx: dict, lang: str) -> str: + sp = selling_point_lines(ctx, lang) + task = ( + "TASK: Triptych showcase. Build three vertical panels separated by thin gutters; every panel " + "contains an IDENTICAL copy of the product from Image 1 (do not re-render it differently per " + "panel) placed in a different usage scene (e.g. home interior / outdoor street / office desk), " + "with consistent color grading across panels." + ) + if sp: + task += f" Panel captions: {sp}." + return task + +def _task_ecommerce_detail(ctx: dict, lang: str) -> str: + sp = selling_point_lines(ctx, lang) or ctx["title"] + params = ctx["params_line"] + return ( + "TASK: Detail-page hero section on a square canvas. Top half: hero banner with the unchanged " + "product at a 3/4 angle. Bottom half: clean spec card with 3 feature rows and line icons" + + (f" (specs: {params})" if params else "") + + f", one highlighted row: {sp}." + ) + +def _task_size_chart(ctx: dict, lang: str) -> str: + dims = ctx["params_line"] + return ( + "TASK: Measurement chart. Show the unchanged product in clean front and side views on a light " + "background; overlay thin technical annotation lines (arrows) marking length, width and height " + "with measurement values rendered beside each line" + + (f" (known specs: {dims})" if dims else "") + + "; precise technical-drawing aesthetic, small caption row." + ) + +def _task_sku_collection(ctx: dict, lang: str) -> str: + # 关键差异：不允许像国产模型那样展开"全部配色"——edits 语义下那会凭空造出新商品 + return ( + "TASK: Product lineup. Arrange several IDENTICAL copies of the product from Image 1 in a neat " + "equal grid (2-4 per row) with a small label chip below each copy; identical lighting and scale " + "across copies. Every copy must show this exact product — do NOT invent other colorways, " + "variants or versions." + ) + +_TASK_BUILDERS = { + "white_bg": (_task_white_bg, False), + "key_features": (_task_key_features, True), + "selling_pt": (_task_selling_pt, True), + "material": (_task_material, True), + "lifestyle": (_task_lifestyle, True), + "multi_scene": (_task_multi_scene, True), + "ecommerce_detail": (_task_ecommerce_detail, True), + "size_chart": (_task_size_chart, True), + "sku_collection": (_task_sku_collection, True), +} + + +def build_prompt(type_id: str, ctx: dict, style_set: int, lang: str, extra: dict | None = None, + style_prompt: str | None = None, requirements: str | None = None) -> str: + """构造指定图类型的 edits 语义 prompt：要求块 → 编辑指令 → 保真锁 → 锚定 → 风格 → 文案 → 画质 → 终检。""" + style = resolve_style(style_set, style_prompt) + extra = extra or {} + hint = (extra.get("prompt_hint") or "").strip() + + if type_id == "custom": + purpose = extra.get("title") or "" + detail = extra.get("detail") or "" + task = "TASK: Create an e-commerce marketing image featuring the product from Image 1" + task += f" — {purpose}" if purpose else "" + task += f": {detail}" if detail else "" + task += "." + wants_text = True + else: + entry = _TASK_BUILDERS.get(type_id) + if entry is None: + raise ValueError(f"未知图类型: {type_id}") + builder, wants_text = entry + task = builder(ctx, lang) + if hint: + task += f" Composition guidance: {hint}." + + parts = [p for p in (requirements_block(requirements),) if p] + parts.append(task) + parts.append(_PRESERVE) + parts.append(_anchor(ctx)) + parts.append(f"{_STYLE}{style['tone']}.") + if wants_text: + parts.append(f"{_TEXT_SCOPE}{TEXT_RENDER[lang]}") + parts.append(_QUALITY) + parts.append(_REMINDER) + return "\n\n".join(parts) diff --git a/server/services/storage.py b/server/services/storage.py index 265f36d..bb6c67c 100644 --- a/server/services/storage.py +++ b/server/services/storage.py @@ -103,3 +103,15 @@ def get_storage(): if settings.use_qiniu: return QiniuStorage() return LocalStorage() + + +def local_path(stored_url_or_key: str) -> Path | None: + """stored_url（http…/media/xxx）或 media key → 本地文件路径。 + + 仅本地存储模式下可命中；七牛 URL 不含 /media/，返回 None 由调用方回退远程下载。 + """ + s = stored_url_or_key or "" + if "/media/" in s: + s = s.split("/media/", 1)[1] + p = _LOCAL_ROOT / s + return p if p.is_file() else None diff --git a/server/services/tasks.py b/server/services/tasks.py new file mode 100644 index 0000000..f6fee8e --- /dev/null +++ b/server/services/tasks.py @@ -0,0 +1,70 @@ +"""内存任务注册表：套图生成任务的生命周期与进程一致（重启即新会话）。 + +轮询/导出只服务「当前会话正在跟踪的任务」——前端没有历史记录功能， +任务状态无需跨进程持久化；重启后轮询自然 404，前端提示任务已中断。 +""" +from __future__ import annotations + +import uuid +from dataclasses import dataclass, field + +# 任务状态 +TASK_PENDING = "pending" +TASK_RUNNING = "running" +TASK_DONE = "done" +TASK_PARTIAL = "partial" +TASK_FAILED = "failed" + +# 任务内单张图状态 +IMG_PENDING = "pending" # 生成中（前端据此隐藏占位格，只渲染 ok/failed 终态） +IMG_OK = "ok" +IMG_FAILED = "failed" + + +@dataclass +class TaskImage: + """任务里单张生成图：完成一张追加一条（前端进度 x/y 依赖此语义）。""" + + type_id: str + name: str + status: str = IMG_PENDING # 循环里先建后跑，成功改 ok、失败显式改 failed + url: str = "" + error: str | None = None + + +@dataclass +class Task: + """一次套图生成任务：轮询可见字段 + 仅供 run_suite 消费的执行参数。""" + + id: str + status: str = TASK_PENDING + platform: str = "cn" + lang: str = "zh" + ratio: str = "1:1" + style_set: int = 1 + style_prompt: str | None = None + requirements: str | None = None + provider: str = "" + model: str | None = None + total: int = 0 # 计划总张数（进度分母） + images: list[TaskImage] = field(default_factory=list) + error: str | None = None + # ── 执行参数（不进轮询响应）── + context: dict = field(default_factory=dict) # 采集文本素材（build_context 的输入） + plan: list[dict] = field(default_factory=list) # 展开后的逐张任务 + ref_images: list[dict] = field(default_factory=list) # 参考图池（main 优先） + watermark: dict | None = None # 水印选项（落盘前服务端后处理） + + +# 进程内任务表：asyncio 单事件循环读写，无并发问题；不做淘汰（单会话量级很小） +_TASKS: dict[str, Task] = {} + + +def create_task(**kwargs) -> Task: + task = Task(id=uuid.uuid4().hex, **kwargs) + _TASKS[task.id] = task + return task + + +def get_task(task_id: str) -> Task | None: + return _TASKS.get(task_id) diff --git a/server/services/watermark.py b/server/services/watermark.py new file mode 100644 index 0000000..1eb1bde --- /dev/null +++ b/server/services/watermark.py @@ -0,0 +1,122 @@ +"""生成图水印：AI 出图返回后、落盘前的后处理合成（不经过生图模型）。 + +样式复刻 ozonSeller「图表处理」的默认水印： + - 图片水印：徽章图中心裁方 → 圆形遮罩 → 宽度为图宽 15%，右下角，边距约 1% 图宽； + - 文字水印：字号为图宽 6%（下限 12px），白色填充 + 黑色描边（alpha 0.55， + 描边宽 fontSize/8），加粗无衬线。 +容错原则：字体/资产缺失或合成异常时 log 警告并返回原图，绝不阻断生图。 +""" +from __future__ import annotations + +import io +import logging +from pathlib import Path + +from PIL import Image, ImageDraw, ImageFont + +from config import get_settings + +log = logging.getLogger("suite.watermark") + +# 尺寸比例（与 ozonSeller app.js 常量一致） +BADGE_SCALE = 0.15 # 图片水印直径 / 图宽 +BADGE_MARGIN = 0.01 # 图片水印边距 / 图宽（ozonSeller 固定 10px，按比例更稳） +TEXT_SCALE = 0.06 # 文字字号 / 图宽 +TEXT_MIN_SIZE = 12 +STROKE_ALPHA = 0.55 +STROKE_RATIO = 1 / 8 # 描边宽 / 字号 + +# CJK/西文字体回退链（macOS 本地服务）；命中后模块级缓存 +_FONT_CANDIDATES = [ + "/System/Library/Fonts/PingFang.ttc", + "/System/Library/Fonts/Hiragino Sans GB.ttc", + "/System/Library/Fonts/STHeiti Light.ttc", + "/Library/Fonts/Arial Unicode.ttf", +] +_font_path: str | None = None + + +def _load_font(size: int) -> ImageFont.FreeTypeFont | ImageFont.ImageFont: + global _font_path + if _font_path is None: + _font_path = next((p for p in _FONT_CANDIDATES if Path(p).is_file()), "") + if _font_path: + return ImageFont.truetype(_font_path, size) + log.warning("未找到系统字体（%s），文字水印退化为 Pillow 默认字体，中文可能乱码", _FONT_CANDIDATES) + return ImageFont.load_default(size) if size >= 10 else ImageFont.load_default() + + +def _circular_badge(size: int) -> Image.Image | None: + """徽章资产 → 指定直径的圆形 RGBA 贴片；资产缺失返回 None。""" + path = get_settings().watermark_image_path + try: + badge = Image.open(path).convert("RGBA") + except Exception as exc: # noqa: BLE001 + log.warning("水印图片加载失败（%s），跳过图片水印: %s", path, exc) + return None + side = min(badge.size) # 中心裁方 + left, top = (badge.width - side) // 2, (badge.height - side) // 2 + square = badge.crop((left, top, left + side, top + side)).resize((size, size)) + mask = Image.new("L", (size, size), 0) + ImageDraw.Draw(mask).ellipse((0, 0, size - 1, size - 1), fill=255) + square.putalpha(mask) + return square + + +def _apply_image_watermark(canvas: Image.Image, opacity: float) -> None: + size = max(24, round(canvas.width * BADGE_SCALE)) + badge = _circular_badge(size) + if badge is None: + return + badge.putalpha(badge.getchannel("A").point(lambda a: round(a * opacity))) + margin = max(10, round(canvas.width * BADGE_MARGIN)) + canvas.alpha_composite(badge, (canvas.width - size - margin, canvas.height - size - margin)) + + +def _apply_text_watermark(canvas: Image.Image, text: str, opacity: float) -> None: + text = (text or "").strip() + if not text: + return + font_size = max(TEXT_MIN_SIZE, round(canvas.width * TEXT_SCALE)) + font = _load_font(font_size) + layer = Image.new("RGBA", canvas.size, (0, 0, 0, 0)) + draw = ImageDraw.Draw(layer) + bbox = draw.textbbox((0, 0), text, font=font, stroke_width=max(1, round(font_size * STROKE_RATIO))) + tw, th = bbox[2] - bbox[0], bbox[3] - bbox[1] + if tw >= canvas.width: # 文案比图还宽：按比例缩字号重排一次 + font_size = max(TEXT_MIN_SIZE, round(font_size * canvas.width / tw * 0.94)) + font = _load_font(font_size) + bbox = draw.textbbox((0, 0), text, font=font, stroke_width=max(1, round(font_size * STROKE_RATIO))) + tw, th = bbox[2] - bbox[0], bbox[3] - bbox[1] + margin = max(10, round(canvas.width * BADGE_MARGIN)) + x = canvas.width - tw - margin - bbox[0] + y = canvas.height - th - margin - bbox[1] + stroke = (0, 0, 0, round(255 * STROKE_ALPHA)) + fill = (255, 255, 255, 255) + draw.text((x, y), text, font=font, fill=fill, stroke_width=max(1, round(font_size * STROKE_RATIO)), + stroke_fill=stroke) + layer.putalpha(layer.getchannel("A").point(lambda a: round(a * opacity))) + canvas.alpha_composite(layer) + + +def apply_watermark(data: bytes, opts: dict) -> bytes: + """给图片字节加水印，返回同格式字节；opts: {type, text, opacity(0-100)}。""" + is_png = data[:8] == b"\x89PNG\r\n\x1a\n" + fmt = "PNG" if is_png else "JPEG" + try: + img = Image.open(io.BytesIO(data)) + canvas = img.convert("RGBA") + opacity = min(100, max(1, int(opts.get("opacity") or 30))) / 100 + if opts.get("type") == "text": + _apply_text_watermark(canvas, opts.get("text") or "", opacity) + else: + _apply_image_watermark(canvas, opacity) + out = io.BytesIO() + if fmt == "PNG": + canvas.save(out, format="PNG") + else: + canvas.convert("RGB").save(out, format="JPEG", quality=95) + return out.getvalue() + except Exception: # noqa: BLE001 + log.exception("水印合成失败，返回原图") + return data