tripocli/tripo/features/motion.py

"""文字生成动作 — text to motion（AI 动画），并把动作应用到角色。

    tripocli motion -p "让角色挥手打招呼" --duration 5 --wait          # 生成动作，得到 motion_asset_id
    tripocli motion -p "让角色走一段" --project <pid> --wait           # 生成并直接应用到角色
    tripocli motion --list                                            # 列出我的动作资产
    tripocli motion -p "..." --segment "3:跳一下" --segment "2:落地"    # 多阶段动作

实测流程（在动画页「AI 动画」标签里点生成抓到）：

    POST /v2/studio/motion/generate
    {"segments": [{"duration_seconds": 5, "prompt": "让角色挥手打招呼"}]}
      → {task_id, status:"queued"}

    POST /v2/studio/motion/get_task   {"task_id": "..."}
      → 轮询直到 status="success"，取 asset_id

    POST /v2/studio/operation/retarget_model
    {"motion_asset_id": "<asset_id>", "model_version":"default",
     "project_id": "...", "rig_type": "biped"}      → 把动作套到模型上

其它已验证的 motion 接口：`/motion/list_assets`（我的动作资产）、`/motion/get_asset`、
`/motion/list_active_tasks`（正在跑的任务）。
"""

import os
import time

from . import _common, animate
from ..errors import TripoAPIError, TripoTimeout

NAME = "motion"
HELP = "文字生成动作：用一句话生成动画并（可选）应用到已绑定的角色"


def add_arguments(parser):
    parser.add_argument("-p", "--prompt", help="动作描述，如「让角色挥手打招呼」")
    parser.add_argument("--duration", type=float, default=5, help="动作时长（秒，默认 5）")
    parser.add_argument(
        "--segment", action="append",
        help="多阶段动作：\"时长:描述\"，可重复（如 --segment \"3:跳一下\" --segment \"2:落地\"）",
    )
    parser.add_argument("--list", action="store_true", help="列出我的动作资产后退出")
    parser.add_argument("--project", help="生成后直接应用到该项目（需已绑定骨骼）")
    parser.add_argument("--type", default="biped", help="骨骼类型，默认 biped（配合 --project 使用）")
    parser.add_argument("-w", "--wait", action="store_true", help="等待动作生成完成")
    parser.add_argument("-o", "--output", help="应用动作后的模型保存路径（配合 --project）")
    parser.add_argument("--timeout", type=int, default=300, help="等待超时秒数（默认 300）")


def _segments(args):
    if args.segment:
        out = []
        for raw in args.segment:
            if ":" in raw:
                duration, text = raw.split(":", 1)
                out.append({"duration_seconds": float(duration), "prompt": text.strip()})
            else:
                out.append({"duration_seconds": args.duration, "prompt": raw.strip()})
        return out
    if not args.prompt:
        return []
    return [{"duration_seconds": args.duration, "prompt": args.prompt}]


def _poll_task(client, task_id, timeout):
    deadline = time.time() + timeout
    while time.time() < deadline:
        data = client.get_data("POST", "/v2/studio/motion/get_task", body={"task_id": task_id}) or {}
        status = data.get("status")
        if status == "success":
            return data
        if status in ("failed", "banned", "expired"):
            raise TripoAPIError(f"动作生成失败: status={status} error={data.get('error')}")
        time.sleep(3)
    raise TripoTimeout(f"动作生成在 {timeout}s 内未完成（task_id={task_id}）")


def _list_assets(ctx):
    data = ctx.client.get_data("POST", "/v2/studio/motion/list_assets", body={}) or {}
    assets = data.get("assets") or []
    ctx.out.data(
        data,
        human=lambda: [
            ctx.out.line(f"共 {len(assets)} 个动作资产"),
            *[ctx.out.line(
                f"   {a.get('asset_id') or a.get('id')}  [{a.get('action_type') or a.get('category') or ''}]  "
                f"{a.get('title') or a.get('name') or a.get('prompt') or ''}"
            ) for a in assets],
        ],
    )
    return data


def run(ctx):
    args = ctx.args

    if args.list:
        return _list_assets(ctx)

    segments = _segments(args)
    if not segments:
        raise ValueError("需要 -p/--prompt 描述动作（或用 --segment 组合多阶段动作，--list 查看已有资产）")

    ctx.out.line("🎬 生成动作: " + " | ".join(f"{s['duration_seconds']}s {s['prompt']}" for s in segments))
    task = ctx.client.get_data("POST", "/v2/studio/motion/generate", body={"segments": segments}) or {}
    task_id = task.get("task_id")
    if not task_id:
        raise TripoAPIError(f"motion/generate 未返回 task_id: {task}")

    if not args.wait:
        ctx.out.data(
            task,
            human=lambda: (
                ctx.out.line(f"✅ 已提交 task_id={task_id}"),
                ctx.out.line(f"取结果: tripocli motion --wait ... 或轮询 /motion/get_task"),
            ),
        )
        return task

    result = _poll_task(ctx.client, task_id, args.timeout)
    asset_id = result.get("asset_id")

    payload = {"task": result, "motion_asset_id": asset_id}
    if args.project and asset_id:
        ctx.out.line(f"🦴 应用到项目 {args.project} ...")
        body = {
            "motion_asset_id": asset_id,
            "model_version": "default",
            "project_id": args.project,
            "rig_type": args.type,
        }
        applied = ctx.client.get_data("POST", "/v2/studio/operation/retarget_model", body=body) or {}
        op_id = applied.get("operator_id")
        if op_id:
            _common.wait_and_report(ctx, op_id, timeout=900)
        entry = animate._retarget_entry(ctx.client, args.project, op_id) or {}
        model_url = entry.get("model_url")
        output_path = None
        if args.output and model_url:
            ctx.client.download(model_url, args.output)
            output_path = os.path.abspath(args.output)
        payload["applied"] = {"result": applied, "model_url": model_url, "output_file": output_path}

    ctx.out.data(
        payload,
        human=lambda: (
            ctx.out.line(f"🎉 动作已生成 motion_asset_id={asset_id}"),
            ctx.out.line(f"   应用到角色: tripocli animate -p <project_id> --motion-asset {asset_id} --wait"),
        ),
    )
    return payload