如何高效监控Design Automation API的WorkItem状态并触发后续操作？

一、最优方案：用Webhook实现主动推送 #

Design Automation支持通过Webhook在WorkItem状态变更时主动通知你的服务，不用再被动轮询，这是效率最高的方式。

配置步骤 #

1. 搭建一个可公开访问的HTTP端点，用来接收Design Automation的回调请求
2. 创建WorkItem时，通过callback参数指定这个端点，同时设置要监听的事件（比如只监听任务完成的onComplete事件）

代码示例 #

Python（FastAPI接收回调）

先写个简单的FastAPI服务处理回调：

from fastapi import FastAPI, Request
import json

app = FastAPI()

@app.post("/forge-webhook")
async def handle_webhook(request: Request):
    payload = await request.json()
    workitem_id = payload["id"]
    status = payload["status"]
    
    if status == "success":
        report_url = payload["reportUrl"]
        # 这里可以添加下载日志、输出.txt的逻辑
        print(f"WorkItem {workitem_id} 完成，日志地址：{report_url}")
    elif status == "failed":
        print(f"WorkItem {workitem_id} 执行失败")
    
    return {"status": "ok"}

创建WorkItem时带上callback配置：

import requests

headers = {"Authorization": "Bearer YOUR_ACCESS_TOKEN", "Content-Type": "application/json"}
create_workitem_body = {
    "activityId": "YOUR_ACTIVITY_ID",
    "arguments": {
        # 这里填你的任务参数配置
    },
    "callback": {
        "url": "https://你的公网域名/forge-webhook",
        "events": ["onComplete"],
        "headers": {"X-Secret-Token": "自定义验证令牌"} # 可选，用来验证请求来源合法性
    }
}

response = requests.post("https://developer.api.autodesk.com/da/us-east/v3/workitems", headers=headers, json=create_workitem_body)
workitem_info = response.json()

JavaScript（Express接收回调）

先搭建Express服务：

const express = require('express');
const bodyParser = require('body-parser');
const app = express();
app.use(bodyParser.json());

app.post('/forge-webhook', (req, res) => {
    const payload = req.body;
    const workitemId = payload.id;
    const status = payload.status;

    if (status === 'success') {
        const reportUrl = payload.reportUrl;
        console.log(`WorkItem ${workitemId} 完成，日志地址：${reportUrl}`);
        // 这里添加下载日志和输出文件的逻辑
    } else if (status === 'failed') {
        console.log(`WorkItem ${workitemId} 执行失败`);
    }

    res.status(200).send({ status: 'ok' });
});

app.listen(3000, () => {
    console.log('Webhook服务运行在3000端口');
});

创建WorkItem时配置callback：

const fetch = require('node-fetch');

async function createWorkItem() {
    const headers = {
        'Authorization': 'Bearer YOUR_ACCESS_TOKEN',
        'Content-Type': 'application/json'
    };
    const body = {
        activityId: 'YOUR_ACTIVITY_ID',
        arguments: {
            // 任务参数配置
        },
        callback: {
            url: 'https://你的公网域名/forge-webhook',
            events: ['onComplete'],
            headers: { 'X-Secret-Token': '自定义验证令牌' }
        }
    };

    const response = await fetch('https://developer.api.autodesk.com/da/us-east/v3/workitems', {
        method: 'POST',
        headers,
        body: JSON.stringify(body)
    });
    const workitemInfo = await response.json();
    console.log('WorkItem创建成功，ID：', workitemInfo.id);
}

createWorkItem();

二、无Webhook时：优化轮询策略 #

如果没法用Webhook（比如没有公网端点），可以把固定间隔轮询改成指数退避策略，减少无效请求：

Python实现（指数退避） #

import time
import requests
import json

def get_workitem_status(workitem_id, headers):
    url = f"https://developer.api.autodesk.com/da/us-east/v3/workitems/{workitem_id}"
    response = requests.get(url, headers=headers)
    return response.json()

workitem_id = "你的WorkItemID"
status = "inprogress"
initial_delay = 60  # 初始等待1分钟
max_delay = 300     # 最长等待5分钟
current_delay = initial_delay

while status == "inprogress":
    workitem_data = get_workitem_status(workitem_id, headers)
    status = workitem_data.get('status')
    
    if status == "inprogress":
        time.sleep(current_delay)
        # 下次等待时间翻倍，不超过max_delay
        current_delay = min(current_delay * 2, max_delay)
    else:
        if status == "success":
            report_url = workitem_data["reportUrl"]
            print(f"任务完成，日志地址：{report_url}")
            # 处理下载输出.txt和日志的逻辑
        else:
            print(f"任务失败，状态：{status}")

JavaScript实现（指数退避） #

const fetch = require('node-fetch');

async function getWorkitemStatus(workitemId, headers) {
    const url = `https://developer.api.autodesk.com/da/us-east/v3/workitems/${workitemId}`;
    const response = await fetch(url, { headers });
    return response.json();
}

async function pollWorkitem(workitemId, headers) {
    let status = 'inprogress';
    let initialDelay = 60 * 1000; // 1分钟
    let maxDelay = 5 * 60 * 1000; // 5分钟
    let currentDelay = initialDelay;

    while (status === 'inprogress') {
        const workitemData = await getWorkitemStatus(workitemId, headers);
        status = workitemData.status;

        if (status === 'inprogress') {
            await new Promise(resolve => setTimeout(resolve, currentDelay));
            currentDelay = Math.min(currentDelay * 2, maxDelay);
        } else {
            if (status === 'success') {
                const reportUrl = workitemData.reportUrl;
                console.log(`任务完成，日志地址：${reportUrl}`);
                // 处理下载逻辑
            } else {
                console.log(`任务失败，状态：${status}`);
            }
        }
    }
}

// 使用示例
const workitemId = "你的WorkItemID";
const headers = { 'Authorization': 'Bearer YOUR_ACCESS_TOKEN' };
pollWorkitem(workitemId, headers);

三、获取输出文件和日志 #

任务成功后：

1. 日志文件直接通过返回的reportUrl下载即可
2. 输出.txt文件需要在创建Activity时，在arguments里配置output参数，指定存储位置（比如OSS或S3），任务完成后通过对应URL下载。示例配置：

"arguments": {
    "outputTxt": {
        "url": "s3://你的存储桶/output.txt",
        "headers": { "Authorization": "Bearer S3_TOKEN" },
        "verb": "put"
    }
}

内容的提问来源于stack exchange，提问作者Hung Nguyen