122 lines
4.7 KiB
Python
122 lines
4.7 KiB
Python
"""Smoke: 豆包 Seedance 视频生成 tool 端到端走通。
|
|
|
|
跑法: .venv/Scripts/python.exe scripts/smoke_seedance.py
|
|
依赖 .env 里 ARK_API_KEY / ZCBOT_DB_URL。**会真的调豆包 API,产生 ~¥1.86 (480p 5s) 费用 + 等 30-90s**。
|
|
|
|
校验:
|
|
1. ArkConfig.load() 拿到 cfg + video.seedance_2_fast 存在
|
|
2. SeedanceTool.execute(prompt=..., resolution='480p', duration=4) 返回 [seedance ...] 文案
|
|
3. videos/<ts>-<rand>.mp4 落盘且大于 0 字节
|
|
4. 同名 .meta.json 存在 + 含 prompt/model_id/cost_cny/tokens/cgt_id 字段
|
|
5. usage_events 多出一行 kind="video",单价 + 分辨率 + 时长 snapshot 在 units jsonb
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
env_file = ROOT / ".env"
|
|
if env_file.exists():
|
|
for line in env_file.read_text(encoding="utf-8").splitlines():
|
|
line = line.strip()
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
k, _, v = line.partition("=")
|
|
os.environ.setdefault(k.strip(), v.strip())
|
|
|
|
from sqlalchemy import text
|
|
|
|
from core.ark_client import ArkConfig
|
|
from core.storage import session_scope
|
|
from core.storage.models import Task, User
|
|
from tools.seedance import SeedanceTool
|
|
|
|
|
|
def main() -> int:
|
|
cfg = ArkConfig.load()
|
|
if cfg is None:
|
|
print("[SKIP] ARK_API_KEY 未设(或 config/media/doubao.yaml 缺失),无法测真接口")
|
|
return 0
|
|
|
|
video_cfg = (cfg.raw.get("video") or {})
|
|
if not video_cfg:
|
|
print("[SKIP] doubao.yaml 无 video 段")
|
|
return 0
|
|
variant_key, variant_cfg = next(iter(video_cfg.items()))
|
|
print(f"[setup] variant={variant_key} model={variant_cfg.get('model_id')}")
|
|
print(f" price_text2video=¥{variant_cfg.get('price_cny_per_mtoken_text2video')}/Mtok")
|
|
|
|
uid = uuid.uuid4()
|
|
tid = uuid.uuid4()
|
|
ws_user = ROOT / "workspace" / "users" / str(uid)
|
|
wd = ws_user / "smoke_seedance"
|
|
wd.mkdir(parents=True, exist_ok=True)
|
|
# 拆两个事务:models 未定 relationship,UOW 不知 User→Task FK 依赖,同一事务里 Task 会先插炸 FK
|
|
with session_scope() as s:
|
|
s.add(User(user_id=uid))
|
|
with session_scope() as s:
|
|
s.add(Task(task_id=tid, user_id=uid, name="smoke_seedance", working_dir=str(wd)))
|
|
|
|
tool = SeedanceTool(
|
|
ark_cfg=cfg,
|
|
video_variant_cfg=variant_cfg,
|
|
variant_key=variant_key,
|
|
working_dir=wd,
|
|
task_id=tid,
|
|
user_id=uid,
|
|
base_dir=wd,
|
|
user_root=ws_user,
|
|
)
|
|
|
|
# 选最便宜档位省钱:480p / 4s / 16:9 → ~¥1.5
|
|
prompt = "一只橙色的小猫从窗台跳下来,水彩风格,镜头平视跟随"
|
|
print(f"[call] prompt={prompt!r} resolution=480p duration=4")
|
|
print(f" (异步任务,等 30-90s 出片,先 submit 后轮询)")
|
|
result = tool.execute(prompt=prompt, resolution="480p", duration=4, ratio="16:9")
|
|
print(f"[tool result]\n{result}\n")
|
|
|
|
if result.startswith("[Error]") or result.startswith("[Cancelled]"):
|
|
print(f"[FAIL] tool 返回 error/cancelled")
|
|
return 2
|
|
|
|
mp4s = list((wd / "videos").glob("*.mp4"))
|
|
assert len(mp4s) == 1, f"videos/*.mp4 应当 1 个,实际 {len(mp4s)}"
|
|
mp4 = mp4s[0]
|
|
assert mp4.stat().st_size > 0, f"{mp4} 大小为 0"
|
|
print(f"[OK] mp4 落盘 {mp4.name} ({mp4.stat().st_size} bytes)")
|
|
|
|
meta_path = mp4.with_suffix(".meta.json")
|
|
assert meta_path.exists(), f"meta 文件不存在 {meta_path}"
|
|
meta = json.loads(meta_path.read_text(encoding="utf-8"))
|
|
for k in ("prompt", "model_id", "resolution", "duration_s", "tokens", "cost_cny", "cgt_id", "ts"):
|
|
assert k in meta, f"meta 缺字段 {k}"
|
|
print(f"[OK] meta 字段齐全: {list(meta.keys())}")
|
|
print(f" tokens={meta['tokens']} cost=¥{meta['cost_cny']} cgt={meta['cgt_id']}")
|
|
|
|
with session_scope() as s:
|
|
rows = s.execute(text(
|
|
"SELECT kind, model_profile, units, cost_cny FROM usage_events "
|
|
"WHERE task_id = :tid"
|
|
), {"tid": str(tid)}).all()
|
|
assert len(rows) == 1, f"usage_events 行数应 1,实际 {len(rows)}"
|
|
row = rows[0]
|
|
assert row.kind == "video", f"kind 应 video,实际 {row.kind}"
|
|
assert row.model_profile == f"doubao.{variant_key}", f"model_profile = {row.model_profile}"
|
|
for k in ("resolution", "duration_s", "tokens", "price_cny_per_mtoken"):
|
|
assert k in row.units, f"units 缺 {k} snapshot"
|
|
print(f"[OK] usage_events: kind={row.kind} model={row.model_profile} cost_cny={row.cost_cny}")
|
|
print(f" units snapshot: {row.units}")
|
|
|
|
print("\n[PASS] smoke_seedance 全部通过")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|