Method _post_json

backend/generators/image_api_client.py:52–106

(self, payload: Dict[str, Any], label: str)

        return response.content

    def _post_json(self, payload: Dict[str, Any], label: str) -> Dict[str, Any]:
        url = f"{self.policy.base_url}{self.policy.endpoint_type}"
        request_payload = dict(payload)
        if self.policy.response_format and "response_format" not in request_payload:
            request_payload["response_format"] = self.policy.response_format

        downgraded_response_format = False
        transient_attempt = 0

        while True:
            try:
                logger.debug(f"{label} 请求: {url}")
                response = self.session.post(
                    url,
                    headers=self._headers(),
                    json=request_payload,
                    timeout=self.timeout,
                )
            except (
                requests.exceptions.SSLError,
                requests.exceptions.ConnectionError,
                requests.exceptions.Timeout,
            ) as exc:
                if transient_attempt >= 2:
                    raise Exception(f"{label} 连接失败: {str(exc)}") from exc
                self._sleep_for_retry(transient_attempt)
                transient_attempt += 1
                continue

            if response.status_code == 200:
                return response.json()

            error_detail = response.text[:500]
            if (
                "response_format" in request_payload
                and not downgraded_response_format
                and _is_unknown_response_format(error_detail)
            ):
                logger.warning("上游不支持 response_format，移除后重试一次")
                request_payload.pop("response_format", None)
                downgraded_response_format = True
                continue

            if response.status_code == 429 or response.status_code >= 500:
                if transient_attempt < 2:
                    self._sleep_for_retry(transient_attempt, response)
                    transient_attempt += 1
                    continue

            logger.error(f"{label} 请求失败: status={response.status_code}, error={error_detail}")
            raise Exception(
                f"{label} 请求失败 (状态码: {response.status_code})\n"
                f"错误详情: {error_detail}\n"
                f"请求地址: {url}"
            )

    def _headers(self) -> Dict[str, str]:
        return {