import requests
import time
import json
def poll_task_status(task_id, max_attempts=60, interval=3):
"""轮询任务状态直到完成或失败"""
for attempt in range(max_attempts):
response = requests.get(
"https://your-api-domain.com/v1/jobs/recordInfo",
params={"taskId": task_id},
headers={"Authorization": "Bearer YOUR_API_KEY"}
)
result = response.json()
data = result.get("data", {})
state = data.get("state")
print(f"尝试 {attempt + 1}: 状态 = {state}")
if state == "success":
result_json = json.loads(data.get("resultJson", "{}"))
print("✅ 任务完成!")
print("结果:", result_json.get("resultUrls"))
return result_json
if state == "fail":
fail_msg = data.get("failMsg", "未知错误")
print(f"❌ 任务失败: {fail_msg}")
raise Exception(fail_msg)
# 仍在处理中,等待下次轮询
time.sleep(interval)
raise Exception("任务在最大尝试次数后超时")
# 使用示例
try:
results = poll_task_status("task_z-image_1765174270120")
print("生成的内容 URL:", results.get("resultUrls"))
except Exception as e:
print(f"错误: {e}")