What causes "可灵查询 API 错误: code={data.get('code')}, message={data.get('message')}" in ATH-MaaS/Pixelle-Video?

Check the code/message: auth errors -> refresh credentials; task-not-found -> verify the task_id and that the endpoint matches how the task was submitted.

ATH-MaaS/Pixelle-Video · error · RuntimeError

可灵查询 API 错误: code={data.get('code')}, message={data.get('mes

Error message

可灵查询 API 错误: code={data.get('code')}, message={data.get('message')}

What it means

The Kling task-status query endpoint returned HTTP 200 but with a non-zero business code. _query_task raises RuntimeError with the API's code/message instead of returning data, so the polling loop in _poll_until_done aborts.

Source

Thrown at pixelle_video/services/api_services/video_kling.py:285

        查询单个任务状态

        Returns:
            API 响应中的 data 字段
        """
        url = f"{self.base_url}/v1/videos/{endpoint}/{task_id}"
        headers = self._auth_headers()

        resp = self._session.get(
            url,
            headers=headers,
            timeout=30,
            proxies=_proxy_dict(self.local_proxy),
        )
        resp.raise_for_status()
        data = resp.json()

        if data.get("code") != 0:
            raise RuntimeError(
                f"可灵查询 API 错误: code={data.get('code')}, message={data.get('message')}"
            )

        return data["data"]

    # ─── 轮询等待 ───

    def _poll_until_done(self, task_id: str, endpoint: str = "image2video") -> dict:
        """
        轮询任务直到完成或失败

        Returns:
            任务结果数据

        Raises:
            RuntimeError: 任务失败
            TimeoutError: 超过最大轮询次数
        """

View on GitHub (pinned to 848b054e4f)

Solutions

Check the code/message: auth errors -> refresh credentials; task-not-found -> verify the task_id and that the endpoint matches how the task was submitted.
Ensure _poll_until_done passes the same endpoint used in _submit_task (image2video if an image was supplied, else text2video).
If the code indicates a transient server error, add retry with backoff around _query_task before failing the whole poll loop.
Start polling promptly after submission so the task isn't purged before the first query.

Example fix

# before
result = self._query_task(task_id, endpoint="image2video")  # task submitted via text2video
# after
endpoint = "image2video" if image_path else "text2video"
result = self._query_task(task_id, endpoint=endpoint)

Defensive patterns

Strategy: retry

Validate before calling

def can_query(task_id: str) -> bool:
    return bool(task_id) and isinstance(task_id, str)  # 并确保 endpoint 与提交时一致
endpoint = "image2video" if used_image else "text2video"

Try / catch

try:
    result = client._query_task(task_id, endpoint=endpoint)
except RuntimeError as e:
    if "任务不存在" in str(e) or "not found" in str(e).lower():
        raise  # task_id/endpoint 错误，重试无意义
    time.sleep(3)  # 瞬时错误，退避后重查
    result = client._query_task(task_id, endpoint=endpoint)

Prevention

Always pass the same endpoint used at submission into every poll
Refresh JWT before long polls so credentials don't expire mid-generation
Query the task promptly after submission to avoid retention cleanup
Log the API code to distinguish auth vs not-found vs transient errors

When it happens

Trigger: GET /v1/videos/{endpoint}/{task_id} returns {"code": <non-zero>, "message": ...} (video_kling.py:284-287) — usually task_id not found (wrong task id or task expired/removed server-side), auth failure on the polling call, or querying against the wrong endpoint (image2video vs text2video mismatch).

Common situations: Polling a task_id generated by a different endpoint type, JWT expired mid-poll after a long generation, task purged by Kling retention policy before polling started, or transient API-side errors surfacing as non-zero codes.

Related errors

AI-assisted analysis of ATH-MaaS/Pixelle-Video@848b054e4f (2026-08-30). Data as JSON: /api/errors/941fe0e075afe127. Report an issue: GitHub.