6a2632da70
将 Clover 从上层产品包旧仓库中独立出来,建立专属版本控制。 当前状态=纵切片端到端已打通(登录→选品→出文出图→审核→下载包), M1文案质量去套路化已验收。此提交作为后续按核销清单逐条修复的基线。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
95 lines
3.4 KiB
Python
95 lines
3.4 KiB
Python
"""
|
||
端到端生图验证（补齐后质量检查）
|
||
- 真实 DB 解密 key → 真实 AIClients → 真实 plan_image_set → 真实调 apiports 出 6 图
|
||
- 只验证生图链路；文案侧用占位 note，不走文案生成
|
||
用法（容器内）：python3 /app/test/e2e_image_check.py
|
||
"""
|
||
import asyncio
|
||
import os
|
||
import sys
|
||
|
||
sys.path.insert(0, "/app")
|
||
|
||
OUT_DIR = "/app/test/output_e2e"
|
||
REF_IMG = "/app/test/ref_product.png" # 预先拷入的真实参考图
|
||
IMAGE_COUNT = 6
|
||
|
||
|
||
def load_clients_and_product():
|
||
"""用真实 DB 解密 key 构建 clients，并取产品 dict。"""
|
||
from app.core.database import SessionLocal
|
||
from app.models.product import Product
|
||
from app.workers.pipeline_steps import (
|
||
decrypt_user_key, build_clients_and_clear_key, build_product_dict,
|
||
)
|
||
db = SessionLocal()
|
||
try:
|
||
product = db.query(Product).order_by(Product.id.desc()).first()
|
||
if not product:
|
||
raise SystemExit("库里没有产品记录")
|
||
product_dict = build_product_dict(product)
|
||
# 用产品所属 workspace 的 key（user_id 取 product.workspace_id 同号那条）
|
||
plain_key = decrypt_user_key(db, product.workspace_id, product.workspace_id)
|
||
clients = build_clients_and_clear_key(plain_key)
|
||
plain_key = None
|
||
return clients, product_dict
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
def fake_note(product_dict: dict) -> dict:
|
||
"""占位文案 note（不走文案生成，只为生图提供 hook/卖点上下文）。"""
|
||
pts = product_dict.get("selling_points") or ["温和不刺激", "成分干净", "上脸快吸收"]
|
||
return {
|
||
"title": f"{product_dict['name']}用一周的真实变化",
|
||
"content": "姐妹们这个我真的要安利，敏感肌也能用，质地清爽不黏腻，"
|
||
"坚持用上脸状态肉眼可见地稳定了，分享给同样在找平替的你。",
|
||
"tags": ["敏感肌", "护肤平替", "成分党", "种草"],
|
||
"selling_points": pts,
|
||
}
|
||
|
||
|
||
async def main():
|
||
os.makedirs(OUT_DIR, exist_ok=True)
|
||
if not os.path.isfile(REF_IMG):
|
||
raise SystemExit(f"参考图不存在：{REF_IMG}（请先拷入）")
|
||
with open(REF_IMG, "rb") as f:
|
||
ref_bytes = f.read()
|
||
print(f"参考图已加载：{len(ref_bytes)} bytes")
|
||
|
||
clients, product_dict = load_clients_and_product()
|
||
print(f"产品：{product_dict['name']} | 类目：{product_dict['category']} "
|
||
f"| style_tone：{product_dict['style_tone']}")
|
||
print(f"卖点：{product_dict['selling_points']}")
|
||
|
||
note = fake_note(product_dict)
|
||
|
||
from app.services.ai_engine.image_gen import generate_storyboard_images
|
||
results = await generate_storyboard_images(
|
||
client=clients,
|
||
note=note,
|
||
product=product_dict,
|
||
image_count=IMAGE_COUNT,
|
||
reference_images=[ref_bytes],
|
||
)
|
||
|
||
ok = 0
|
||
for i, r in enumerate(results, 1):
|
||
role = r.get("role", f"img{i}")
|
||
if r.get("error"):
|
||
print(f" [{i}] {role} ❌ {r['error']}")
|
||
continue
|
||
b = r.get("image_bytes") or b""
|
||
path = os.path.join(OUT_DIR, f"{i:02d}_{role}.png")
|
||
with open(path, "wb") as f:
|
||
f.write(b)
|
||
ok += 1
|
||
print(f" [{i}] {role} ✅ {len(b)} bytes → {path}")
|
||
|
||
print(f"\n完成：{ok}/{len(results)} 张成功，输出目录 {OUT_DIR}")
|
||
await clients.aclose()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|