feat: 迁移 ISS 服务端
This commit is contained in:
Joey
@@ -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"'},
|
||||
)
|
||||
@@ -0,0 +1,54 @@
|
||||
"""图片代理：绕过源站防盗链，供前端 <img> 预览与生图参考使用。
|
||||
|
||||
平移自 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": "*",
|
||||
},
|
||||
)
|
||||
@@ -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)
|
||||
Binary file not shown.
|
After Width: | Height: | Size: 14 KiB |
@@ -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():
|
||||
|
||||
+5
-1
@@ -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")
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
@@ -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"<title[^>]*>(.*?)</title>", re.IGNORECASE | re.DOTALL)
|
||||
|
||||
|
||||
def _raise_api_error(resp, provider: str):
|
||||
"""HTTP 错误时抛出带 API 错误码/信息的异常（响应体里有真正的失败原因）。"""
|
||||
if resp.is_success:
|
||||
return
|
||||
text = resp.text or ""
|
||||
if "<html" in text[:300].lower() or text.lstrip()[:15].lower().startswith("<!doctype"):
|
||||
# Cloudflare/网关错误页：取 <title> 作摘要，避免整段 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}"
|
||||
@@ -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()
|
||||
# 剥离思考块（思考型模型会把推理过程放进 <think>）
|
||||
s = re.sub(r'<think>.*?</think>', '', 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}
|
||||
@@ -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,
|
||||
)
|
||||
|
||||
@@ -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
|
||||
@@ -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·]+|(?<!\d)\.(?!\d)", sp_text):
|
||||
c = _shorten(chunk, 20)
|
||||
if c and len(selling_points) < 5:
|
||||
selling_points.append({"zh": c, "en": c})
|
||||
if not selling_points:
|
||||
for p in (raw.get("params") or [])[:12]:
|
||||
k, v = _shorten(p.get("key", ""), 10), _shorten(str(p.get("value", "")), 16)
|
||||
if k and v and k.lower() not in {"货号", "sku", "isbn", "上架时间"}:
|
||||
selling_points.append({"zh": f"{k} {v}", "en": f"{k} {v}"})
|
||||
if len(selling_points) >= 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)
|
||||
@@ -0,0 +1,9 @@
|
||||
"""豆包（火山方舟 Seedream）提示词。
|
||||
|
||||
豆包与通义同为国产"主体参考"生图模型：参考图即商品锚、prompt 为场景描述，
|
||||
提示词语义一致，直接复用阿里系装配；差异（去 AI 味后缀）在 generator 层追加。
|
||||
独立成文件便于后续按豆包特性分化。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from .alibaba import build_prompt as build_prompt # noqa: F401 主体参考语义与通义共用
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user