"""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:`smoke_seedance.py seedance_2_pro`;不传走 yaml 第一个
    want_variant = sys.argv[1] if len(sys.argv) > 1 else ""
    if want_variant:
        if want_variant not in video_cfg:
            print(f"[FAIL] variant {want_variant!r} 不在 yaml video 段, 已知: {list(video_cfg.keys())}")
            return 2
        variant_key, variant_cfg = want_variant, video_cfg[want_variant]
    else:
        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())