Files
tripocli/tripo/features/task.py
T
Mrpzx 977f97995a tripocli v2.0.0:Tripo AI 3D 生成与加工 CLI(面向 Agent 与开发者,纯 Python 标准库、零第三方依赖)
覆盖文生 3D、图生 3D、多视图、批量生成,以及重拓扑、绑定、动画、智能分割、
纹理、PBR、智能 UV、部件补全、格式导出与下载。

代码组织:一个功能一个模块(tripo/features/),入口薄壳 tripocli.py。
2026-10-13 19:22:54 +08:00

68 lines
2.3 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""任务进度 — poll or block on an operator.
tripocli task status <operator_id>
tripocli task wait <operator_id> -p <project_id> -o out.glb
"""
import os
from . import _common
TIMEOUT_DEFAULT = 900
def _status(ctx):
items = ctx.client.progress([ctx.args.operator_id])
if not items:
raise ValueError(f"未找到任务 {ctx.args.operator_id}")
task = items[0]
ctx.out.data(
task,
human=lambda: (
ctx.out.line(f"状态 : {task.get('status')}"),
ctx.out.line(f"进度 : {task.get('progress')}%"),
ctx.out.line(f"剩余 : {task.get('left_time')}s"),
),
)
return task
def _wait(ctx):
args = ctx.args
task = _common.wait_and_report(ctx, args.operator_id, timeout=args.timeout)
output_path = None
model_url = None
if args.project:
model_url, _detail = _common.fetch_model_url(ctx, args.project)
if args.output and model_url:
ctx.client.download(model_url, args.output)
output_path = os.path.abspath(args.output)
payload = {"task": task, "model_url": model_url, "output_file": output_path}
ctx.out.data(
payload,
human=lambda: (
ctx.out.line(f"✅ 任务状态: {task.get('status')}"),
ctx.out.line(f"模型地址: {model_url}") if model_url else ctx.out.line(""),
ctx.out.line(f"已保存: {output_path}") if output_path else ctx.out.line(""),
),
)
return payload
def register(subparsers):
parser = subparsers.add_parser("task", help="任务进度查询与等待")
sub = parser.add_subparsers(dest="task_action", required=True)
status = sub.add_parser("status", help="查询任务状态")
status.add_argument("operator_id", help="operator_id / task_id")
status.set_defaults(func=_status)
wait = sub.add_parser("wait", help="阻塞等待任务完成(可顺带下载)")
wait.add_argument("operator_id", help="operator_id / task_id")
wait.add_argument("-p", "--project", help="项目 ID,用于取最终模型地址")
wait.add_argument("-o", "--output", help="下载保存路径")
wait.add_argument("--timeout", type=int, default=TIMEOUT_DEFAULT, help=f"超时秒数(默认 {TIMEOUT_DEFAULT})")
wait.set_defaults(func=_wait)