错误码参考

{
  "id": "task-unified-1772618027-cmeisy8h",
  "object": "image.generation.task",
  "status": "failed",
  "model": "gemini-3.1-flash-image-preview",
  "progress": 0,
  "error": {
    "code": "content_policy_violation",
    "message": "Content policy violation.\nYour request was blocked by safety filters..."
  }
}

import requests
import time

def poll_task_with_retry(task_id, api_key, max_retries=3):
    """带自动重试的任务轮询"""
    headers = {"Authorization": f"Bearer {api_key}"}

    for attempt in range(max_retries):
        resp = requests.get(
            f"https://api.evolink.ai/v1/tasks/{task_id}",
            headers=headers
        )
        data = resp.json()

        if data["status"] == "completed":
            return data["results"]

        if data["status"] == "failed":
            error = data.get("error", {})
            code = error.get("code", "unknown_error")
            message = error.get("message", "")

            # 客户端错误 — 不可重试，需要修改请求
            if code in [
                "content_policy_violation",
                "invalid_parameters",
                "image_processing_error",
                "image_dimension_mismatch",
            ]:
                raise Exception(f"Client error [{code}]: {message}")

            # 服务端错误 — 可以重试
            if code in [
                "generation_failed_no_content",
                "service_error",
                "generation_timeout",
                "resource_exhausted",
                "quota_exceeded",
            ]:
                if attempt < max_retries - 1:
                    wait = 2 ** attempt * 5  # 5s, 10s, 20s
                    time.sleep(wait)
                    continue
                raise Exception(f"Server error [{code}]: {message}")

            raise Exception(f"Error [{code}]: {message}")

        # 任务仍在处理中
        time.sleep(3)

    raise Exception("Max polling attempts exceeded")