feat: 初始化项目，并且接近完成 ozon 部分
This commit is contained in:
Joey
@@ -0,0 +1,316 @@
|
||||
"""套图生成服务：图像 provider（豆包 Seedream / 通义万相）+ 任务执行器。
|
||||
|
||||
Provider 调用方式移植自 ecommerce-image-suite/scripts/generate.py：
|
||||
- doubao：火山方舟 images/generations，同步返回 URL；参考图走 image 字段（data URI）
|
||||
- tongyi：wan* 万象模型走异步任务轮询；qwen* 走同步 multimodal-generation
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import base64
|
||||
import logging
|
||||
import mimetypes
|
||||
from uuid import UUID
|
||||
|
||||
import httpx
|
||||
from sqlalchemy import select
|
||||
|
||||
from config import get_settings
|
||||
from db import get_session_factory
|
||||
from models import Product, ProductAsset, Suite, SuiteImage, SUITE_RUNNING, SUITE_DONE, SUITE_PARTIAL, SUITE_FAILED, STATUS_OK, STATUS_FAILED
|
||||
from services import storage
|
||||
from services.prompt import build_prompt, build_context, type_name
|
||||
|
||||
log = logging.getLogger("suite.generator")
|
||||
|
||||
# 参考图选择：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) -> str:
|
||||
"""平台比例 → provider 尺寸参数。3:4 竖版（Ozon/WB），1:1 方图（国内）。"""
|
||||
if provider == "doubao":
|
||||
return "1536x2048" if ratio == "3:4" else "2048x2048"
|
||||
# tongyi：万象与千问的 size 语法相同（* 分隔），档位不同
|
||||
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(url: str) -> str:
|
||||
"""参考图 URL → data URI。本地 media 文件直读磁盘；远程 URL 带 Referer 下载。
|
||||
|
||||
生图 API 的服务器无法访问 127.0.0.1，代理 URL 也不能直接透传，
|
||||
所以统一在本地解析成 base64 data URI 再进请求体。
|
||||
"""
|
||||
if url.startswith("data:"):
|
||||
return url
|
||||
path = storage.local_path(url)
|
||||
if path is not None:
|
||||
mime = mimetypes.guess_type(path.name)[0] or "image/jpeg"
|
||||
return _bytes_to_data_uri(path.read_bytes(), mime)
|
||||
if url.startswith(("http://", "https://")):
|
||||
from api.proxy import guess_referer
|
||||
data, ctype = await storage.download_bytes(url, referer=guess_referer(url))
|
||||
if not ctype.startswith("image/"):
|
||||
ctype = "image/jpeg"
|
||||
return _bytes_to_data_uri(data, ctype)
|
||||
raise FileNotFoundError(f"无法解析参考图: {url}")
|
||||
|
||||
|
||||
# ── Provider：豆包 Seedream（火山方舟）────────────────────────────────────
|
||||
|
||||
async def generate_doubao(prompt: str, ref_images: list[str], size: str = "2048x2048") -> bytes:
|
||||
s = get_settings()
|
||||
if not s.ark_api_key:
|
||||
raise RuntimeError("未配置 ARK_API_KEY（.env）")
|
||||
body = {
|
||||
"model": 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,
|
||||
)
|
||||
resp.raise_for_status()
|
||||
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")
|
||||
|
||||
|
||||
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") -> bytes:
|
||||
s = get_settings()
|
||||
if not s.dashscope_api_key:
|
||||
raise RuntimeError("未配置 DASHSCOPE_API_KEY（.env）")
|
||||
is_wan = _is_wan_model(s.dashscope_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"
|
||||
)
|
||||
|
||||
content: list[dict] = [{"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": s.dashscope_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)
|
||||
resp.raise_for_status()
|
||||
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
|
||||
|
||||
|
||||
GENERATORS = {"doubao": generate_doubao, "tongyi": generate_tongyi}
|
||||
|
||||
|
||||
# ── 任务执行器 ────────────────────────────────────────────────────────────
|
||||
|
||||
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", ""))
|
||||
|
||||
|
||||
async def _select_ref_images(db, product_id: UUID, type_id: str) -> list[str]:
|
||||
"""商品路径：主图组前几张。转存完成的用本地文件，未完成的直接用源站 URL。"""
|
||||
assets = (await db.scalars(
|
||||
select(ProductAsset).where(
|
||||
ProductAsset.product_id == product_id,
|
||||
ProductAsset.group_key == "main",
|
||||
ProductAsset.type == "img",
|
||||
).order_by(ProductAsset.sort_order)
|
||||
)).all()
|
||||
refs = [a.stored_url or a.source_url for a in assets if (a.stored_url or a.source_url)]
|
||||
if not refs:
|
||||
raise RuntimeError("商品没有可用参考图（未采集主图）")
|
||||
return _order_refs(refs, type_id)
|
||||
|
||||
|
||||
async def run_suite(suite_id: str) -> None:
|
||||
"""后台执行套图任务：逐张生成 → 落盘 → 记录；单张失败不中断。
|
||||
|
||||
两条路径：
|
||||
- 无状态（product_id 为空）：上下文与参考图来自请求自带的 context / ref_images
|
||||
- 商品路径（兼容旧流程）：从 product + product_assets 取
|
||||
"""
|
||||
settings = get_settings()
|
||||
async with get_session_factory()() as db:
|
||||
suite = await db.get(Suite, UUID(suite_id))
|
||||
if suite is None:
|
||||
return
|
||||
|
||||
product = None
|
||||
if suite.product_id:
|
||||
product = await db.get(Product, suite.product_id)
|
||||
if product is None:
|
||||
suite.status = SUITE_FAILED
|
||||
suite.error = "商品不存在"
|
||||
await db.commit()
|
||||
return
|
||||
|
||||
suite.status = SUITE_RUNNING
|
||||
await db.commit()
|
||||
|
||||
provider_name = suite.provider or settings.image_provider
|
||||
generator = GENERATORS.get(provider_name)
|
||||
if generator is None:
|
||||
suite.status = SUITE_FAILED
|
||||
suite.error = f"未知 provider: {provider_name}"
|
||||
await db.commit()
|
||||
return
|
||||
|
||||
raw = suite.context if not product else (product.raw or {})
|
||||
ctx = build_context(raw or {}, fallback_name=product.name if product else "")
|
||||
size = _image_size(provider_name, suite.ratio, is_wan=_is_wan_model(settings.dashscope_model))
|
||||
|
||||
# 任务列表：方案（逐张）优先，旧路径按 types
|
||||
if suite.plan:
|
||||
jobs = [dict(j) for j in suite.plan]
|
||||
else:
|
||||
jobs = [
|
||||
{"kind": t, "title": type_name(t), "detail": "", "prompt_hint": "", "variant_name": None}
|
||||
for t in (suite.types or [])
|
||||
]
|
||||
|
||||
ok, failed = 0, 0
|
||||
for job in jobs:
|
||||
type_id = job["kind"]
|
||||
image_row = SuiteImage(
|
||||
suite_id=suite.id,
|
||||
type_id=type_id,
|
||||
name=job.get("title") or type_name(type_id),
|
||||
status=STATUS_FAILED,
|
||||
)
|
||||
db.add(image_row)
|
||||
await db.flush()
|
||||
try:
|
||||
prompt = build_prompt(type_id, ctx, suite.style_set, suite.lang, extra=job)
|
||||
if product:
|
||||
refs = await _select_ref_images(db, product.id, type_id)
|
||||
else:
|
||||
refs = _refs_for_job(list(suite.ref_images or []), job)
|
||||
data = await generator(prompt, refs, size=size)
|
||||
key = storage.write_bytes(data, key_prefix=f"suites/{suite.id}", ext=".jpg")
|
||||
image_row.stored_url = storage.public_url(key)
|
||||
image_row.status = STATUS_OK
|
||||
ok += 1
|
||||
except Exception as exc: # noqa: BLE001
|
||||
log.exception("套图 %s 类型 %s 生成失败", suite_id, type_id)
|
||||
image_row.error = str(exc)[:500]
|
||||
failed += 1
|
||||
await db.commit()
|
||||
|
||||
suite.status = SUITE_DONE if failed == 0 else (SUITE_PARTIAL if ok > 0 else SUITE_FAILED)
|
||||
if failed and not ok:
|
||||
suite.error = "全部生成失败，请检查 API Key / 参考图"
|
||||
from datetime import datetime, timezone
|
||||
suite.finished_at = datetime.now(timezone.utc)
|
||||
if product:
|
||||
product.stage = "generated" # 商品路径才有的阶段升级
|
||||
await db.commit()
|
||||
@@ -0,0 +1,134 @@
|
||||
"""出图方案规划器：DeepSeek 根据采集的商品信息生成套图方案。
|
||||
|
||||
方案每项 = 一类图（标题 + 说明 + 生图提示 + 张数 + 可选 SKU 绑定），
|
||||
生成时按方案逐张出图；参考图可按 variant_name 精确绑定到对应 SKU 图。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
|
||||
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. SKU 主图：商品有多个带图 SKU（颜色/款式）时，每个 SKU 出 1 张独立主图（kind=white_bg），
|
||||
并在 variant_name 里填对应的 SKU 规格名（必须来自「SKU规格」列表，原样照抄）；
|
||||
单 SKU 商品出 1 张主图即可（variant_name 留空）。
|
||||
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。其他创意图用 custom。
|
||||
7. 总张数控制在 8-15 张；每项 count 为 1-3。
|
||||
8. title 用中文短语（≤8字，如「主图·粉色」「浴室壁挂场景」）；detail 用中文说明这张图要展示什么（≤40字）；
|
||||
prompt_hint 用英文描述构图（角度/布局/光线要点，≤60 words），供生图模型使用。
|
||||
|
||||
## 输出格式（严格 JSON，不要多余文字）
|
||||
{
|
||||
"summary": "整体思路一句话",
|
||||
"items": [
|
||||
{"kind": "white_bg", "title": "主图·粉色", "detail": "粉色SKU白底主视觉", "prompt_hint": "front view on pure white background", "count": 1, "variant_name": "粉色"}
|
||||
]
|
||||
}"""
|
||||
|
||||
|
||||
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
|
||||
|
||||
|
||||
async def generate_plan(
|
||||
product_info: dict,
|
||||
sku_variants: list[str],
|
||||
image_stats: dict,
|
||||
platform: str,
|
||||
) -> dict:
|
||||
"""调用 DeepSeek 生成方案。返回 {summary, items}。"""
|
||||
s = get_settings()
|
||||
if not s.deepseek_api_key:
|
||||
raise RuntimeError("未配置 DEEPSEEK_API_KEY（.env）")
|
||||
|
||||
user_content = json.dumps({
|
||||
"商品信息": 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（决定图内文案语言）
|
||||
}, ensure_ascii=False)
|
||||
|
||||
async with httpx.AsyncClient(timeout=60, 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},
|
||||
{"role": "user", "content": user_content},
|
||||
],
|
||||
"response_format": {"type": "json_object"},
|
||||
"temperature": 0.3,
|
||||
"max_tokens": 2000,
|
||||
},
|
||||
)
|
||||
resp.raise_for_status()
|
||||
content = resp.json()["choices"][0]["message"]["content"]
|
||||
|
||||
try:
|
||||
data = json.loads(content)
|
||||
except json.JSONDecodeError as exc:
|
||||
log.error("规划器输出不是合法 JSON: %s", 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,290 @@
|
||||
"""套图 Prompt 引擎。
|
||||
|
||||
借鉴 ecommerce-image-suite 的动态 Prompt 架构，浓缩为：
|
||||
- 7 种图类型 × 5 套视觉风格模板
|
||||
- 公共组件：QUALITY（画质）/ PRODUCT_REF_LOCK（商品一致性锁）/ TEXT_RENDER（图内文案规范）
|
||||
- 卖点从采集的参数表/卖点文本自动提炼
|
||||
|
||||
核心原则：所有图严格保持商品一致性（same silhouette, same print, same color），
|
||||
只允许改变背景 / 角度 / 光线 / 排版。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
|
||||
# ── 风格模板（与插件端 STYLE_SET_OPTIONS 对应）─────────────────────────────
|
||||
|
||||
STYLE_SETS: dict[int, dict] = {
|
||||
1: {
|
||||
"name": "经典商拍",
|
||||
"tone": "premium commercial e-commerce photography, clean soft studio lighting, "
|
||||
"gentle gradient background, catalog-grade presentation, refined and trustworthy",
|
||||
"bg": "light neutral studio backdrop with soft vignette",
|
||||
},
|
||||
2: {
|
||||
"name": "生活杂志",
|
||||
"tone": "editorial lifestyle magazine aesthetic, natural window light, "
|
||||
"cozy lived-in atmosphere, muted film tones, candid storytelling",
|
||||
"bg": "warm lifestyle home setting with plants and textured fabrics",
|
||||
},
|
||||
3: {
|
||||
"name": "极简高冷",
|
||||
"tone": "minimalist high-end aesthetic, vast negative space, single directional light, "
|
||||
"cool grey palette, architectural calm, quiet luxury",
|
||||
"bg": "seamless light grey studio background with subtle shadow",
|
||||
},
|
||||
4: {
|
||||
"name": "活力爆款",
|
||||
"tone": "vibrant high-conversion e-commerce style, punchy saturated accents, "
|
||||
"energetic composition, bold contrast, promotional poster energy",
|
||||
"bg": "bright colorful gradient backdrop with dynamic geometric shapes",
|
||||
},
|
||||
5: {
|
||||
"name": "暗调质感",
|
||||
"tone": "dark moody premium product photography, dramatic rim lighting, "
|
||||
"deep charcoal background, rich texture detail, luxurious atmosphere",
|
||||
"bg": "matte black background with soft spotlight and subtle smoke haze",
|
||||
},
|
||||
}
|
||||
|
||||
# ── 图类型中文名（导出文件名用）───────────────────────────────────────────
|
||||
|
||||
TYPE_NAMES_ZH: dict[str, str] = {
|
||||
"white_bg": "白底主图",
|
||||
"key_features": "核心卖点图",
|
||||
"selling_pt": "卖点图",
|
||||
"material": "材质图",
|
||||
"lifestyle": "场景展示图",
|
||||
"multi_scene": "多场景拼图",
|
||||
"ecommerce_detail": "电商详情图",
|
||||
"size_chart": "尺寸标注图",
|
||||
"sku_collection": "SKU合集图",
|
||||
"custom": "创意图",
|
||||
}
|
||||
|
||||
# ── 公共组件 ──────────────────────────────────────────────────────────────
|
||||
|
||||
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."
|
||||
)
|
||||
|
||||
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."
|
||||
),
|
||||
}
|
||||
|
||||
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"
|
||||
)
|
||||
|
||||
# ── 商品上下文提炼 ────────────────────────────────────────────────────────
|
||||
|
||||
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 _sp_lines(ctx: dict, lang: str, max_n: int = 3) -> str:
|
||||
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))
|
||||
|
||||
# ── 各图类型 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 = _sp_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 = _sp_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:
|
||||
return (
|
||||
f"Lifestyle in-context scene for product \"{ctx['title']}\": the product is naturally "
|
||||
f"used / placed in a real environment ({style['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 = _sp_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 = _sp_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 ""
|
||||
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} {style['tone']}. {style['bg']} as environment. {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) -> str:
|
||||
"""构造指定图类型的完整生图 prompt。
|
||||
|
||||
extra: 方案项信息 {title, detail, prompt_hint}——custom 类型必需，
|
||||
预设类型也会把 prompt_hint 作为构图补充注入。
|
||||
"""
|
||||
style = STYLE_SETS.get(style_set, STYLE_SETS[1])
|
||||
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}."
|
||||
return prompt + ". " + DEFAULT_NEGATIVE_INTENT
|
||||
|
||||
|
||||
def type_name(type_id: str) -> str:
|
||||
return TYPE_NAMES_ZH.get(type_id, type_id)
|
||||
@@ -0,0 +1,70 @@
|
||||
"""本地文件存储：落 data/media/，由 FastAPI /media 静态托管。"""
|
||||
from __future__ import annotations
|
||||
|
||||
import mimetypes
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
|
||||
import httpx
|
||||
|
||||
from config import get_settings
|
||||
|
||||
|
||||
def media_root() -> Path:
|
||||
root = Path(get_settings().data_dir) / "media"
|
||||
root.mkdir(parents=True, exist_ok=True)
|
||||
return root
|
||||
|
||||
|
||||
def public_url(key: str) -> str:
|
||||
"""media key → 可访问 URL。"""
|
||||
settings = get_settings()
|
||||
return f"{settings.app_base_url.rstrip('/')}/media/{key}"
|
||||
|
||||
|
||||
def _ext_from_url_or_type(hint: str, content_type: str = "") -> str:
|
||||
if content_type:
|
||||
ctype = content_type.split(";")[0].strip().lower()
|
||||
mapping = {
|
||||
"image/jpeg": ".jpg", "image/png": ".png", "image/webp": ".webp",
|
||||
"image/gif": ".gif", "image/bmp": ".bmp", "video/mp4": ".mp4",
|
||||
}
|
||||
if ctype in mapping:
|
||||
return mapping[ctype]
|
||||
ext = mimetypes.guess_extension(hint.split("?")[0].lower()) or ".jpg"
|
||||
return ".jpg" if ext == ".jpe" else ext
|
||||
|
||||
|
||||
def write_bytes(data: bytes, key_prefix: str = "", ext: str = ".jpg") -> str:
|
||||
"""写文件，返回 media key（相对 media 根的路径）。"""
|
||||
key = f"{key_prefix + '/' if key_prefix else ''}{uuid.uuid4().hex}{ext}"
|
||||
path = media_root() / key
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_bytes(data)
|
||||
return key
|
||||
|
||||
|
||||
async def download_bytes(url: str, referer: str | None = None, timeout: float = 60.0) -> tuple[bytes, str]:
|
||||
"""下载远程字节。返回 (bytes, content_type)。"""
|
||||
headers = {"Referer": referer} if referer else {}
|
||||
headers.setdefault("User-Agent", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)")
|
||||
async with httpx.AsyncClient(timeout=timeout, follow_redirects=True, verify=False) as client:
|
||||
resp = await client.get(url, headers=headers)
|
||||
resp.raise_for_status()
|
||||
ctype = (resp.headers.get("content-type") or "application/octet-stream").split(";")[0].strip()
|
||||
return resp.content, ctype
|
||||
|
||||
|
||||
async def save_from_url(url: str, key_prefix: str = "", referer: str | None = None) -> str:
|
||||
data, ctype = await download_bytes(url, referer)
|
||||
key = write_bytes(data, key_prefix, _ext_from_url_or_type(url, ctype))
|
||||
return public_url(key)
|
||||
|
||||
|
||||
def local_path(stored_url_or_key: str) -> Path | None:
|
||||
"""stored_url（http.../media/xxx）或 key → 本地文件路径。"""
|
||||
s = stored_url_or_key
|
||||
if "/media/" in s:
|
||||
s = s.split("/media/", 1)[1]
|
||||
p = media_root() / s
|
||||
return p if p.exists() else None
|
||||
Reference in New Issue
Block a user