Image generation API documentation

const API_BASE = 'https://imgapi.vip/prod-api';
const SUBMIT_PATH = '/tool/imgapi/draw/Async';
const QUERY_PATH = '/tool/gptimage2/query';

const SUPPORTED_MODELS = new Set([
  'gpt-image-2.5',
  'gpt-image-2.5-flare',
  'gpt-image-2.5-sunburst',
  'gpt-image-2',
  'nano-banana-2',
  'nano-banana-pro',
]);
const SUPPORTED_ASPECT_RATIOS = new Set([
  'auto', '1:1', '3:2', '2:3', '16:9', '9:16', '4:3', '3:4',
  '21:9', '9:21', '1:3', '3:1', '2:1', '1:2',
]);
const SUPPORTED_QUALITIES = new Set(['auto', 'low', 'medium', 'high', 'xhigh', 'max']);
const SUPPORTED_RESOLUTIONS = new Set(['1K', '2K', '4K']);
const TRANSIENT_CODES = new Set([408, 425, 429, 500, 502, 503, 504]);

const DEFAULT_SUBMIT_TIMEOUT_MS = 60_000;
const DEFAULT_QUERY_TIMEOUT_MS = 20_000;
const DEFAULT_MAX_POLL_DURATION_MS = 10 * 60_000;
const DEFAULT_INITIAL_POLL_INTERVAL_MS = 2_000;
const DEFAULT_MAX_POLL_INTERVAL_MS = 8_000;

export class ImgApiError extends Error {
  constructor(message, options = {}) {
    super(message, options.cause ? { cause: options.cause } : undefined);
    this.name = 'ImgApiError';
    this.status = options.status ?? 0;
    this.code = options.code ?? null;
    this.retryAfterMs = options.retryAfterMs ?? 0;
    this.payload = options.payload ?? null;
    this.taskId = options.taskId ?? null;
    this.retryable = options.retryable ?? false;
    this.transportError = options.transportError ?? false;
    this.submissionUncertain = options.submissionUncertain ?? false;
  }
}

function isRecord(value) {
  return value !== null && typeof value === 'object' && !Array.isArray(value);
}

function positiveNumber(value, fallback, name) {
  const number = value === undefined ? fallback : Number(value);
  if (!Number.isFinite(number) || number <= 0) {
    throw new TypeError(`${name}  must be greater than  0  as a number`);
  }
  return number;
}

function parseRetryAfterMs(value) {
  const text = String(value ?? '').trim();
  if (!text) return 0;

  const seconds = Number(text);
  if (Number.isFinite(seconds)) return Math.max(0, seconds * 1000);

  const moment = Date.parse(text);
  return Number.isFinite(moment) ? Math.max(0, moment - Date.now()) : 0;
}

function getRetryAfterMs(data, response) {
  const bodyValue = Number(data?.retryAfterMs);
  if (Number.isFinite(bodyValue) && bodyValue > 0) return bodyValue;
  return parseRetryAfterMs(response.headers.get('Retry-After'));
}

function getMessage(data, fallback) {
  for (const key of ['error', 'msg', 'message', 'failure_reason']) {
    const value = data?.[key];
    if (typeof value === 'string' && value.trim()) return value.trim();
  }
  return fallback;
}

function normalizeEnvelope(envelope) {
  if (!isRecord(envelope)) return null;
  return isRecord(envelope.data)
    ? { ...envelope, ...envelope.data }
    : envelope;
}

function createRequestSignal(timeoutMs, externalSignal) {
  const controller = new AbortController();
  let timedOut = false;
  let externalAborted = false;

  const onExternalAbort = () => {
    externalAborted = true;
    controller.abort(externalSignal.reason ?? new Error('Operation cancelled'));
  };

  if (externalSignal?.aborted) {
    onExternalAbort();
  } else if (externalSignal) {
    externalSignal.addEventListener('abort', onExternalAbort, { once: true });
  }

  const timer = setTimeout(() => {
    timedOut = true;
    controller.abort(new Error(`Request timeout（${timeoutMs}ms）`));
  }, timeoutMs);

  return {
    signal: controller.signal,
    timedOut: () => timedOut,
    externalAborted: () => externalAborted,
    cleanup() {
      clearTimeout(timer);
      externalSignal?.removeEventListener('abort', onExternalAbort);
    },
  };
}

function defaultSleep(ms, signal) {
  if (signal?.aborted) {
    return Promise.reject(signal.reason ?? new Error('Operation cancelled'));
  }

  return new Promise((resolve, reject) => {
    const timer = setTimeout(() => {
      signal?.removeEventListener('abort', onAbort);
      resolve();
    }, Math.max(0, ms));

    const onAbort = () => {
      clearTimeout(timer);
      signal?.removeEventListener('abort', onAbort);
      reject(signal.reason ?? new Error('Operation cancelled'));
    };
    signal?.addEventListener('abort', onAbort, { once: true });
  });
}

function isTransient(error) {
  return Boolean(
    error?.transportError
      || TRANSIENT_CODES.has(Number(error?.status))
      || TRANSIENT_CODES.has(Number(error?.code)),
  );
}

function withTaskId(error, taskId) {
  if (error instanceof ImgApiError) {
    if (!error.taskId) error.taskId = taskId;
    return error;
  }
  return new ImgApiError(error?.message || 'Task query failed', { taskId, cause: error });
}

function calculatePollDelay(attempt, initialMs, maxMs, random) {
  const exponential = Math.min(maxMs, initialMs * (1.5 ** Math.min(attempt, 20)));
  const jitterFactor = 0.85 + (Math.max(0, Math.min(1, random())) * 0.3);
  return Math.max(1, Math.round(exponential * jitterFactor));
}

function validateCardKey(cardKey) {
  if (!/^[0-9a-fA-F]{16}$/.test(String(cardKey ?? ''))) {
    throw new Error('Set a valid server-side  16 -character  IMGAPI_CARD_KEY');
  }
  return String(cardKey);
}

function validateGenerateOptions(options = {}) {
  if (!isRecord(options)) throw new TypeError('Generation settings must be an object');

  const model = String(options.model ?? '');
  if (!SUPPORTED_MODELS.has(model)) {
    throw new TypeError(`model  supports only  ${[...SUPPORTED_MODELS].join('、')}`);
  }

  if (typeof options.prompt !== 'string' || !options.prompt.trim()) {
    throw new TypeError('prompt  must not be empty');
  }
  if (options.prompt.length > 10000) {
    throw new TypeError('prompt  at most  10000  characters');
  }

  const aspectRatio = options.aspectRatio ?? 'auto';
  if (!SUPPORTED_ASPECT_RATIOS.has(aspectRatio)) {
    throw new TypeError(`Unsupported  aspectRatio：${aspectRatio}`);
  }

  const quality = options.quality ?? 'auto';
  if (!SUPPORTED_QUALITIES.has(quality)) {
    throw new TypeError(`Unsupported  quality：${quality}`);
  }

  if (['xhigh', 'max'].includes(quality) && model !== 'gpt-image-2.5-sunburst') {
    throw new TypeError('xhigh、max  only supports  gpt-image-2.5-sunburst');
  }
  const resolution = options.resolution ?? '1K';
  if (!SUPPORTED_RESOLUTIONS.has(resolution)) {
    throw new TypeError(`Unsupported  resolution：${resolution}`);
  }

  const urls = options.urls ?? [];
  const files = options.files ?? [];
  if (!Array.isArray(urls)) throw new TypeError('urls  must be an array');
  if (!Array.isArray(files)) throw new TypeError('files  must be an array');
  if (urls.length + files.length > 12) {
    throw new TypeError('urls  and  files  total limit:  12  reference images');
  }

  const normalizedUrls = urls.map((value, index) => {
    if (typeof value !== 'string' || !value.trim()) {
      throw new TypeError(`urls[${index}]  must be  HTTPS  image URL`);
    }
    const text = value.trim();
    let parsed;
    try {
      parsed = new URL(text);
    } catch {
      throw new TypeError(`urls[${index}]  is not a valid  URL`);
    }
    if (parsed.protocol !== 'https:') {
      throw new TypeError(`urls[${index}]  must use  HTTPS`);
    }
    return text;
  });

  files.forEach((file, index) => {
    const isPath = typeof file === 'string' && file.trim();
    const isBlob = typeof Blob !== 'undefined' && file instanceof Blob;
    const isBuffer = typeof Buffer !== 'undefined' && Buffer.isBuffer(file);
    if (!isPath && !isBlob && !isBuffer) {
      throw new TypeError(`files[${index}]  supports local paths only, 、Blob/File  or  Buffer`);
    }
  });

  return {
    model,
    prompt: options.prompt,
    aspectRatio,
    quality,
    resolution,
    urls: normalizedUrls,
    files: [...files],
  };
}

function mimeTypeFromFilename(filename) {
  const lower = filename.toLowerCase();
  if (lower.endsWith('.png')) return 'image/png';
  if (lower.endsWith('.jpg') || lower.endsWith('.jpeg')) return 'image/jpeg';
  if (lower.endsWith('.webp')) return 'image/webp';
  if (lower.endsWith('.gif')) return 'image/gif';
  if (lower.endsWith('.avif')) return 'image/avif';
  return 'application/octet-stream';
}

async function appendLocalFile(form, file, index) {
  if (typeof file === 'string') {
    const [{ readFile }, { basename }] = await Promise.all([
      import('node:fs/promises'),
      import('node:path'),
    ]);
    const filename = basename(file);
    const bytes = await readFile(file);
    form.append('files', new Blob([bytes], { type: mimeTypeFromFilename(filename) }), filename);
    return;
  }

  if (typeof Buffer !== 'undefined' && Buffer.isBuffer(file)) {
    form.append('files', new Blob([file]), `reference-${index + 1}`);
    return;
  }

  const filename = typeof file.name === 'string' && file.name
    ? file.name
    : `reference-${index + 1}`;
  form.append('files', file, filename);
}

async function createGenerateRequest(options, cardKey) {
  const fields = {
    key: cardKey,
    model: options.model,
    prompt: options.prompt,
    aspectRatio: options.aspectRatio,
    quality: options.quality,
    resolution: options.resolution,
  };

  if (options.files.length === 0) {
    return {
      body: { ...fields, urls: options.urls },
      isFormData: false,
    };
  }

  const form = new FormData();
  for (const [name, value] of Object.entries(fields)) {
    form.append(name, String(value));
  }
  for (const url of options.urls) form.append('urls', url);
  for (const [index, file] of options.files.entries()) {
    await appendLocalFile(form, file, index);
  }

  return { body: form, isFormData: true };
}

function extractTaskId(data) {
  const taskId = data?.id ?? data?.task_id ?? data?.taskId;
  return taskId === undefined || taskId === null || String(taskId).trim() === ''
    ? null
    : String(taskId);
}

export function createImgApiClient({
  cardKey = process.env.IMGAPI_CARD_KEY,
  fetchImpl = globalThis.fetch,
  sleepImpl = defaultSleep,
  now = () => performance.now(),
  random = Math.random,
} = {}) {
  const key = validateCardKey(cardKey);
  if (typeof fetchImpl !== 'function') throw new Error('Current  Node.js  environment does not support  fetch，use  Node.js 18+');
  if (typeof sleepImpl !== 'function' || typeof now !== 'function' || typeof random !== 'function') {
    throw new TypeError('fetch/sleep/now/random  configuration is invalid');
  }

  async function post(path, body, {
    isFormData = false,
    timeoutMs,
    signal,
  } = {}) {
    const timeout = positiveNumber(timeoutMs, DEFAULT_QUERY_TIMEOUT_MS, 'timeoutMs');
    const requestSignal = createRequestSignal(timeout, signal);
    let response;
    let raw;

    try {
      response = await fetchImpl(`${API_BASE}${path}`, {
        method: 'POST',
        headers: isFormData
          ? { Accept: 'application/json' }
          : { Accept: 'application/json', 'Content-Type': 'application/json' },
        body: isFormData ? body : JSON.stringify(body),
        signal: requestSignal.signal,
      });
      raw = await response.text();
    } catch (cause) {
      if (requestSignal.externalAborted()) {
        throw new ImgApiError('Operation cancelled', { cause });
      }
      const message = requestSignal.timedOut()
        ? `Request timeout（${timeout}ms）`
        : `Network request failed：${cause?.message || 'unknown error'}`;
      throw new ImgApiError(message, {
        cause,
        retryable: true,
        transportError: true,
      });
    } finally {
      requestSignal.cleanup();
    }

    let envelope = {};
    if (raw?.trim()) {
      try {
        envelope = JSON.parse(raw);
      } catch (cause) {
        throw new ImgApiError(
          response.ok
            ? 'The API returned an unreadable non- JSON  response'
            : `Request failed（HTTP ${response.status}，response is not  JSON）`,
          {
            status: response.status,
            retryable: response.status >= 500,
            cause,
          },
        );
      }
    }

    const data = normalizeEnvelope(envelope);
    if (!data) {
      throw new ImgApiError('API returned  JSON  is not an object', {
        status: response.status,
        retryable: response.status >= 500,
      });
    }

    const hasBusinessCode = Object.hasOwn(data, 'code');
    const businessFailed = data.ok === false
      || (hasBusinessCode && Number(data.code) !== 200);

    if (!response.ok || businessFailed) {
      const status = response.status;
      const code = data.code ?? null;
      throw new ImgApiError(
        getMessage(data, `Request failed（HTTP ${status}）`),
        {
          status,
          code,
          retryAfterMs: getRetryAfterMs(data, response),
          payload: data,
          retryable: TRANSIENT_CODES.has(Number(status)) || TRANSIENT_CODES.has(Number(code)),
        },
      );
    }

    return data;
  }

  async function submitImageTask(options, {
    signal,
    submitTimeoutMs = DEFAULT_SUBMIT_TIMEOUT_MS,
    onTaskCreated,
  } = {}) {
    const normalized = validateGenerateOptions(options);
    const request = await createGenerateRequest(normalized, key);
    let submitted;

    try {
      submitted = await post(SUBMIT_PATH, request.body, {
        isFormData: request.isFormData,
        timeoutMs: submitTimeoutMs,
        signal,
      });
    } catch (error) {
      const status = Number(error?.status || 0);
      const uncertain = Boolean(error?.transportError || status === 408 || status >= 500);
      if (uncertain) {
        throw new ImgApiError(
          `Submission uncertain：${error.message}；Do not automatically resubmit: this may create duplicate tasks or charges`,
          {
            status: error.status,
            code: error.code,
            payload: error.payload,
            retryAfterMs: error.retryAfterMs,
            cause: error,
            submissionUncertain: true,
          },
        );
      }
      throw error;
    }

    const status = String(submitted.status ?? '').toLowerCase();
    const taskId = extractTaskId(submitted);

    if (status === 'succeeded') {
      if (typeof submitted.image !== 'string' || !submitted.image.trim()) {
        throw new ImgApiError('Task status is  succeeded， but the response has no  image', {
          taskId,
          payload: submitted,
        });
      }
      return { taskId, status, image: submitted.image, raw: submitted };
    }

    if (status === 'failed' || status === 'refunded') {
      throw new ImgApiError(getMessage(submitted, 'Generation failed'), {
        taskId,
        payload: submitted,
      });
    }

    if (!taskId) {
      throw new ImgApiError('The submission did not return a Task ID', { payload: submitted });
    }

    if (onTaskCreated) {
      try {
        await onTaskCreated(taskId, submitted);
      } catch (cause) {
        throw new ImgApiError(
          'The task was submitted, but Save the task ID  failed. Use the error  taskId  to resume polling; do not resubmit',
          { taskId, payload: submitted, cause },
        );
      }
    }

    return {
      taskId,
      status: status || 'submitted',
      image: null,
      raw: submitted,
    };
  }

  async function queryImageTask(taskId, {
    signal,
    queryTimeoutMs = DEFAULT_QUERY_TIMEOUT_MS,
  } = {}) {
    const id = String(taskId ?? '').trim();
    if (!id) throw new TypeError('taskId  must not be empty');

    try {
      return await post(QUERY_PATH, { key, id }, {
        timeoutMs: queryTimeoutMs,
        signal,
      });
    } catch (error) {
      throw withTaskId(error, id);
    }
  }

  async function waitForImageTask(taskId, {
    signal,
    queryTimeoutMs = DEFAULT_QUERY_TIMEOUT_MS,
    maxPollDurationMs = DEFAULT_MAX_POLL_DURATION_MS,
    initialPollIntervalMs = DEFAULT_INITIAL_POLL_INTERVAL_MS,
    maxPollIntervalMs = DEFAULT_MAX_POLL_INTERVAL_MS,
    onPoll,
  } = {}) {
    const id = String(taskId ?? '').trim();
    if (!id) throw new TypeError('taskId  must not be empty');

    const maxDuration = positiveNumber(maxPollDurationMs, DEFAULT_MAX_POLL_DURATION_MS, 'maxPollDurationMs');
    const initialInterval = positiveNumber(initialPollIntervalMs, DEFAULT_INITIAL_POLL_INTERVAL_MS, 'initialPollIntervalMs');
    const maxInterval = positiveNumber(maxPollIntervalMs, DEFAULT_MAX_POLL_INTERVAL_MS, 'maxPollIntervalMs');
    if (maxInterval < initialInterval) {
      throw new TypeError('maxPollIntervalMs  must not be less than  initialPollIntervalMs');
    }

    const startedAt = now();
    let attempt = 0;
    let nextDelayMs = initialInterval;

    while (now() - startedAt < maxDuration) {
      const remainingBeforeSleep = maxDuration - (now() - startedAt);
      try {
        await sleepImpl(Math.min(nextDelayMs, remainingBeforeSleep), signal);
      } catch (cause) {
        throw new ImgApiError(
          'Local waiting stopped. The server task may still be running. Use  taskId  to resume polling',
          { taskId: id, cause },
        );
      }

      if (now() - startedAt >= maxDuration) break;

      let queried;
      try {
        queried = await queryImageTask(id, { signal, queryTimeoutMs });
      } catch (error) {
        if (!isTransient(error)) throw withTaskId(error, id);
        attempt += 1;
        nextDelayMs = error.retryAfterMs > 0
          ? error.retryAfterMs
          : calculatePollDelay(attempt, initialInterval, maxInterval, random);
        await onPoll?.({
          taskId: id,
          attempt,
          status: 'retrying',
          nextDelayMs,
          error,
        });
        continue;
      }

      const status = String(queried.status ?? '').toLowerCase();
      await onPoll?.({ taskId: id, attempt, status, response: queried });

      if (status === 'succeeded') {
        if (typeof queried.image !== 'string' || !queried.image.trim()) {
          throw new ImgApiError('Task status is  succeeded， but the response has no  image', {
            taskId: id,
            payload: queried,
          });
        }
        return { taskId: id, status, image: queried.image, raw: queried };
      }

      if (status === 'failed' || status === 'refunded') {
        throw new ImgApiError(getMessage(queried, 'Generation failed'), {
          taskId: id,
          payload: queried,
        });
      }

      if (status !== 'submitted' && status !== 'processing') {
        throw new ImgApiError(`The API returned an unknown Task status：${status || 'empty'}`, {
          taskId: id,
          payload: queried,
        });
      }

      attempt += 1;
      nextDelayMs = calculatePollDelay(attempt, initialInterval, maxInterval, random);
    }

    throw new ImgApiError(
      'Generation is taking longer. The server task may still be running. Save  taskId  and resume polling later',
      { taskId: id },
    );
  }

  async function generateImage(options, waitOptions = {}) {
    const submitted = await submitImageTask(options, waitOptions);
    if (submitted.image) return submitted;
    return waitForImageTask(submitted.taskId, waitOptions);
  }

  return {
    submitImageTask,
    queryImageTask,
    waitForImageTask,
    generateImage,
  };
}

"""
imgAPI  resilient synchronous  Python  client. Dependencies：
    pip install "httpx>=0.27,<1"

Security notes：
- CardKey  is read only from a server environment variable or constructor parameter, never browser code。
- Submit a task is never retried automatically; a timeout or interrupted response may still have created a task。
- After a query failure, reuse the original  task_id，do not resubmit generation。
- For local files, ，HTTPX  uses  multipart/form-data  streaming uploads。

Python：3.10+
"""

from __future__ import annotations

import json
import inspect
import math
import mimetypes
import os
import random
import re
import time
from contextlib import ExitStack
from dataclasses import dataclass
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
from pathlib import Path
from types import TracebackType
from typing import Any, Callable, Iterable, Mapping, Sequence, TypeAlias
from urllib.parse import urlsplit

import httpx


API_BASE = "https://imgapi.vip/prod-api"
GENERATE_PATH = "/tool/imgapi/draw/Async"
QUERY_PATH = "/tool/gptimage2/query"

SUPPORTED_MODELS = frozenset({
    "gpt-image-2.5",
    "gpt-image-2.5-flare",
    "gpt-image-2.5-sunburst",
    "gpt-image-2",
    "nano-banana-2",
    "nano-banana-pro",
})
SUPPORTED_QUALITIES = frozenset({"auto", "low", "medium", "high", "xhigh", "max"})
SUPPORTED_RESOLUTIONS = frozenset({"1K", "2K", "4K"})

MAX_PROMPT_LENGTH = 10000
MAX_REFERENCE_IMAGES = 12
DEFAULT_POLL_DURATION_SECONDS = 10 * 60
DEFAULT_MAX_POLL_ATTEMPTS = 300

PENDING_STATUSES = frozenset({"submitted", "processing", "pending", "queued", "running"})
SUCCEEDED_STATUSES = frozenset({"succeeded", "success", "completed"})
FAILED_STATUSES = frozenset({"failed", "refunded", "cancelled", "canceled"})

RETRYABLE_QUERY_HTTP_STATUSES = frozenset({408, 425, 429, 500, 502, 503, 504})
AMBIGUOUS_SUBMISSION_HTTP_STATUSES = frozenset({408, 425, 500, 502, 503, 504})

PathInput: TypeAlias = str | os.PathLike[str]
TaskCreatedCallback: TypeAlias = Callable[..., None]
StatusCallback: TypeAlias = Callable[["TaskSnapshot"], None]
CancelCallback: TypeAlias = Callable[[], bool]


class ImgApiError(Exception):
    """Base class for all  imgAPI  client exceptions。"""

    def __init__(
        self,
        message: str,
        *,
        status: int = 0,
        code: Any = None,
        retry_after_ms: int = 0,
        payload: Mapping[str, Any] | None = None,
        task_id: str | None = None,
        submission_uncertain: bool = False,
        definitely_not_sent: bool = False,
    ) -> None:
        super().__init__(message)
        self.status = int(status or 0)
        self.code = code
        self.retry_after_ms = max(0, int(retry_after_ms or 0))
        self.payload = dict(payload or {})
        self.task_id = task_id
        self.submission_uncertain = bool(submission_uncertain)
        self.definitely_not_sent = bool(definitely_not_sent)


class ValidationError(ImgApiError):
    """Invalid call parameters。"""


class TransportError(ImgApiError):
    """Network, pool or transport error。"""


class ProtocolError(ImgApiError):
    """HTTP  succeeded but the response violates the API contract。"""


class ApiResponseError(ImgApiError):
    """HTTP  or business status indicates failure。"""


class SubmissionUncertainError(ImgApiError):
    """Submission uncertain; do not resubmit automatically to avoid duplicate tasks or charges。"""


class TaskFailedError(ImgApiError):
    """The generation task failed, was refunded or was Cancel。"""


class TaskPersistenceError(ImgApiError):
    """The server returned  task_id， but the caller could not persist that  ID。"""


class TaskCallbackError(ImgApiError):
    """Task status callback failed。"""


class PollTimeoutError(ImgApiError):
    """Polling limit reached; the task may still be running on the server。"""


class PollCancelledError(ImgApiError):
    """The caller stopped waiting; the server task may still be running。"""


class QueryRetryExhaustedError(ImgApiError):
    """Too many consecutive query failures; the original task can be queried later。"""


@dataclass(frozen=True, slots=True)
class SubmissionResult:
    task_id: str | None
    status: str
    image_urls: tuple[str, ...]
    raw: Mapping[str, Any]

    @property
    def image_url(self) -> str | None:
        return self.image_urls[0] if self.image_urls else None

    @property
    def image(self) -> str | None:
        """Compatibility alias。"""
        return self.image_url

    @property
    def is_complete(self) -> bool:
        return self.status in SUCCEEDED_STATUSES and bool(self.image_urls)


@dataclass(frozen=True, slots=True)
class TaskSnapshot:
    task_id: str
    status: str
    image_urls: tuple[str, ...]
    raw: Mapping[str, Any]

    @property
    def image_url(self) -> str | None:
        return self.image_urls[0] if self.image_urls else None

    @property
    def image(self) -> str | None:
        """Compatibility alias。"""
        return self.image_url


@dataclass(frozen=True, slots=True)
class GenerationResult:
    task_id: str | None
    status: str
    image_urls: tuple[str, ...]
    attempts: int
    elapsed_seconds: float
    raw: Mapping[str, Any]

    @property
    def image_url(self) -> str:
        if not self.image_urls:
            raise ProtocolError("The generation result has no image URL", task_id=self.task_id)
        return self.image_urls[0]

    @property
    def image(self) -> str:
        """Compatibility alias。"""
        return self.image_url


@dataclass(frozen=True, slots=True)
class _ValidatedOptions:
    model: str
    prompt: str
    aspect_ratio: str
    quality: str
    resolution: str
    urls: tuple[str, ...]
    files: tuple[Path, ...]


class ImgApiClient:
    """
    imgAPI  synchronous client. Reuse an instance across requests to reuse  TCP/TLS  connections. If an external  http_client， is supplied, this class does not close it。
    """

    def __init__(
        self,
        card_key: str | None = None,
        *,
        http_client: httpx.Client | None = None,
        submit_timeout: httpx.Timeout | float | None = None,
        query_timeout: httpx.Timeout | float | None = None,
        poll_initial_interval: float = 2.0,
        poll_max_interval: float = 8.0,
        poll_backoff_factor: float = 1.35,
        poll_jitter_ratio: float = 0.15,
        max_consecutive_query_errors: int = 30,
        max_retry_after_seconds: float = 60.0,
        sleep_func: Callable[[float], None] = time.sleep,
        monotonic_func: Callable[[], float] = time.monotonic,
        random_func: Callable[[], float] = random.random,
        sleep_fn: Callable[[float], None] | None = None,
        monotonic_fn: Callable[[], float] | None = None,
        random_fn: Callable[[], float] | None = None,
    ) -> None:
        resolved_key = card_key or os.getenv("IMGAPI_CARD_KEY") or ""
        if not re.fullmatch(r"[0-9a-fA-F]{16}", resolved_key):
            raise ValidationError(
                "IMGAPI_CARD_KEY  must be  16 -character hexadecimal string stored only in a server environment variable"
            )

        if poll_initial_interval < 0:
            raise ValidationError("poll_initial_interval  must not be less than  0")
        if poll_max_interval < poll_initial_interval:
            raise ValidationError("poll_max_interval  must not be less than  poll_initial_interval")
        if poll_backoff_factor < 1:
            raise ValidationError("poll_backoff_factor  must not be less than  1")
        if not 0 <= poll_jitter_ratio <= 1:
            raise ValidationError("poll_jitter_ratio  must be between  0  and  1  inclusive")
        if max_consecutive_query_errors < 1:
            raise ValidationError("max_consecutive_query_errors  must be greater than  0")
        if max_retry_after_seconds < 0:
            raise ValidationError("max_retry_after_seconds  must not be less than  0")

        self._card_key = resolved_key
        self._submit_timeout = submit_timeout or httpx.Timeout(
            connect=15.0,
            read=60.0,
            write=120.0,
            pool=10.0,
        )
        self._query_timeout = query_timeout or httpx.Timeout(
            connect=10.0,
            read=15.0,
            write=15.0,
            pool=10.0,
        )
        self._poll_initial_interval = float(poll_initial_interval)
        self._poll_max_interval = float(poll_max_interval)
        self._poll_backoff_factor = float(poll_backoff_factor)
        self._poll_jitter_ratio = float(poll_jitter_ratio)
        self._max_consecutive_query_errors = int(max_consecutive_query_errors)
        self._max_retry_after_seconds = float(max_retry_after_seconds)
        # *_fn  is a legacy parameter name retained for compatibility。
        self._sleep = sleep_fn or sleep_func
        self._monotonic = monotonic_fn or monotonic_func
        self._random = random_fn or random_func

        self._owns_http_client = http_client is None
        self._client = http_client or httpx.Client(
            base_url=API_BASE,
            headers={
                "Accept": "application/json",
                "User-Agent": "imgapi-python-client/1.0",
            },
            follow_redirects=False,
            limits=httpx.Limits(
                max_connections=100,
                max_keepalive_connections=20,
                keepalive_expiry=30.0,
            ),
            timeout=self._query_timeout,
        )
        self._closed = False

    def __enter__(self) -> "ImgApiClient":
        self._ensure_open()
        return self

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc_value: BaseException | None,
        traceback: TracebackType | None,
    ) -> None:
        self.close()

    def close(self) -> None:
        if self._closed:
            return
        if self._owns_http_client:
            self._client.close()
        self._closed = True

    def submit_image_task(
        self,
        options: Mapping[str, Any] | None = None,
        *,
        model: str | None = None,
        prompt: str | None = None,
        aspect_ratio: str | None = None,
        quality: str | None = None,
        resolution: str | None = None,
        urls: Sequence[str] | None = None,
        files: Sequence[PathInput] | None = None,
        on_task_created: TaskCreatedCallback | None = None,
        **legacy_options: Any,
    ) -> SubmissionResult:
        """
        Submit one generation task without retrying automatically. The original API parameter  aspectRatio  is also accepted, but do not pass
        aspect_ratio  and  aspectRatio  with different values。
        """
        self._ensure_open()
        merged = self._merge_generate_options(
            options,
            model=model,
            prompt=prompt,
            aspect_ratio=aspect_ratio,
            quality=quality,
            resolution=resolution,
            urls=urls,
            files=files,
            legacy_options=legacy_options,
        )
        options = self._validate_options(
            model=merged["model"],
            prompt=merged["prompt"],
            aspect_ratio=merged["aspect_ratio"],
            quality=merged["quality"],
            resolution=merged["resolution"],
            urls=merged["urls"],
            files=merged["files"],
        )

        fields = {
            "key": self._card_key,
            "model": options.model,
            "prompt": options.prompt,
            "aspectRatio": options.aspect_ratio,
            "quality": options.quality,
            "resolution": options.resolution,
        }

        try:
            if options.files:
                response = self._submit_multipart(fields, options.urls, options.files)
            else:
                response = self._client.post(
                    f"{API_BASE}{GENERATE_PATH}",
                    json={**fields, "urls": list(options.urls)},
                    timeout=self._submit_timeout,
                    follow_redirects=False,
                )
        except (httpx.ConnectError, httpx.ConnectTimeout, httpx.PoolTimeout) as exc:
            # No usable connection was established; the request was likely not sentApply server 。
            raise TransportError(
                f"Submit a task prevented connection：{self._safe_exception_text(exc)}",
                submission_uncertain=False,
                definitely_not_sent=True,
            ) from exc
        except httpx.TransportError as exc:
            # An interrupted write or response timeout may still mean the server received the request。
            raise SubmissionUncertainError(
                "Submission uncertain: the server may have created a task. Do not resubmit automatically；"
                f"Check task or usage records first. Transport error：{self._safe_exception_text(exc)}",
                submission_uncertain=True,
            ) from exc

        try:
            payload = self._decode_response(response)
        except ProtocolError as exc:
            # Received an  HTTP  success response without confirmation of task creation; treat submission as uncertain。
            raise SubmissionUncertainError(
                "Unrecognized successful submission response; the server may have created a task。"
                "Do not resubmit automatically. Check task or usage records first。",
                status=exc.status,
                code=exc.code,
                payload=exc.payload,
                submission_uncertain=True,
            ) from exc
        except ApiResponseError as exc:
            if (
                self._numeric_status(exc.status) in AMBIGUOUS_SUBMISSION_HTTP_STATUSES
                or self._numeric_status(exc.code) in AMBIGUOUS_SUBMISSION_HTTP_STATUSES
            ):
                raise SubmissionUncertainError(
                    "The submission returned an ambiguous gateway or / server error. A task may have been created，"
                    "Do not resubmit automatically. Check task records first。",
                    status=exc.status,
                    code=exc.code,
                    retry_after_ms=exc.retry_after_ms,
                    payload=exc.payload,
                    task_id=self._extract_task_id(exc.payload),
                    submission_uncertain=True,
                ) from exc
            raise

        status = self._normalize_status(payload.get("status"))
        task_id = self._extract_task_id(payload)
        image_urls = self._extract_image_urls(payload)

        if status in FAILED_STATUSES:
            raise TaskFailedError(
                self._failure_message(payload, "Generation submission failed"),
                status=response.status_code,
                code=payload.get("code"),
                payload=payload,
                task_id=task_id,
            )

        if status in SUCCEEDED_STATUSES or (not status and image_urls):
            if not image_urls:
                raise ProtocolError(
                    "Submission returned success without an image URL",
                    status=response.status_code,
                    code=payload.get("code"),
                    payload=payload,
                    task_id=task_id,
                )
            result = SubmissionResult(
                task_id=task_id,
                status="succeeded",
                image_urls=image_urls,
                raw=payload,
            )
            if task_id:
                self._run_task_created_callback(on_task_created, task_id, payload)
            return result

        if not task_id:
            raise ProtocolError(
                "The submission did not return a Task ID",
                status=response.status_code,
                code=payload.get("code"),
                payload=payload,
            )

        if status and status not in PENDING_STATUSES:
            raise ProtocolError(
                f"Submission returned an unknown Task status：{status}",
                status=response.status_code,
                code=payload.get("code"),
                payload=payload,
                task_id=task_id,
            )

        result = SubmissionResult(
            task_id=task_id,
            status=status or "submitted",
            image_urls=(),
            raw=payload,
        )
        self._run_task_created_callback(on_task_created, task_id, payload)
        return result

    def query_image_task(self, task_id: str) -> TaskSnapshot:
        """Query an existing task without creating a new one。"""
        self._ensure_open()
        normalized_task_id = self._validate_task_id(task_id)

        try:
            response = self._client.post(
                f"{API_BASE}{QUERY_PATH}",
                json={"key": self._card_key, "id": normalized_task_id},
                timeout=self._query_timeout,
                follow_redirects=False,
            )
        except httpx.TransportError as exc:
            raise TransportError(
                f"Network error while querying the task：{self._safe_exception_text(exc)}",
                task_id=normalized_task_id,
            ) from exc

        try:
            payload = self._decode_response(response)
        except ImgApiError as exc:
            exc.task_id = normalized_task_id
            raise

        status = self._normalize_status(payload.get("status"))
        image_urls = self._extract_image_urls(payload)

        if not status and image_urls:
            status = "succeeded"
        if not status:
            raise ProtocolError(
                "The query did not return Task status",
                status=response.status_code,
                code=payload.get("code"),
                payload=payload,
                task_id=normalized_task_id,
            )

        if status in SUCCEEDED_STATUSES:
            if not image_urls:
                raise ProtocolError(
                    "Task succeeded but the query response has no  image  image URL",
                    status=response.status_code,
                    code=payload.get("code"),
                    payload=payload,
                    task_id=normalized_task_id,
                )
            status = "succeeded"
        elif status in FAILED_STATUSES:
            status = "failed" if status not in {"refunded"} else "refunded"
        elif status in PENDING_STATUSES:
            # Retain the returned  pending  status for the caller。
            pass
        else:
            raise ProtocolError(
                f"The query returned an unknown Task status：{status}",
                status=response.status_code,
                code=payload.get("code"),
                payload=payload,
                task_id=normalized_task_id,
            )

        return TaskSnapshot(
            task_id=normalized_task_id,
            status=status,
            image_urls=image_urls,
            raw=payload,
        )

    def wait_for_image_task(
        self,
        task_id: str,
        *,
        max_poll_duration: float | None = None,
        max_poll_attempts: int = DEFAULT_MAX_POLL_ATTEMPTS,
        on_status: StatusCallback | None = None,
        should_cancel: CancelCallback | None = None,
        max_poll_duration_seconds: float | None = None,
        initial_poll_interval_seconds: float | None = None,
        max_poll_interval_seconds: float | None = None,
    ) -> GenerationResult:
        """Use the original  task_id  for polling; suitable for restoring a saved task after restart。"""
        self._ensure_open()
        normalized_task_id = self._validate_task_id(task_id)
        if max_poll_duration is not None and max_poll_duration_seconds is not None:
            if float(max_poll_duration) != float(max_poll_duration_seconds):
                raise ValidationError(
                    "max_poll_duration  and  max_poll_duration_seconds  must not be supplied with different values"
                )
        selected_duration = (
            max_poll_duration_seconds
            if max_poll_duration_seconds is not None
            else max_poll_duration
        )
        if selected_duration is None:
            selected_duration = DEFAULT_POLL_DURATION_SECONDS

        initial_interval = (
            self._poll_initial_interval
            if initial_poll_interval_seconds is None
            else self._validate_non_negative_finite(
                initial_poll_interval_seconds,
                "initial_poll_interval_seconds",
            )
        )
        maximum_interval = (
            self._poll_max_interval
            if max_poll_interval_seconds is None
            else self._validate_non_negative_finite(
                max_poll_interval_seconds,
                "max_poll_interval_seconds",
            )
        )
        if maximum_interval < initial_interval:
            raise ValidationError(
                "max_poll_interval_seconds  must not be less than  initial_poll_interval_seconds"
            )

        duration, attempts_limit = self._validate_poll_limits(
            selected_duration,
            max_poll_attempts,
        )

        started_at = self._monotonic()
        deadline = started_at + duration
        attempts = 0
        consecutive_query_errors = 0
        retry_after_ms = 0

        while attempts < attempts_limit:
            self._raise_if_cancelled(normalized_task_id, should_cancel)
            remaining = deadline - self._monotonic()
            if remaining <= 0:
                break

            delay = self._next_poll_delay(
                attempt_index=attempts,
                retry_after_ms=retry_after_ms,
                initial_interval=initial_interval,
                maximum_interval=maximum_interval,
            )
            if delay > 0:
                self._sleep(min(delay, remaining))

            self._raise_if_cancelled(normalized_task_id, should_cancel)
            if deadline - self._monotonic() <= 0:
                break

            attempts += 1
            retry_after_ms = 0

            try:
                snapshot = self.query_image_task(normalized_task_id)
            except ApiResponseError as exc:
                failed_status = self._normalize_status(exc.payload.get("status"))
                if failed_status in FAILED_STATUSES:
                    raise TaskFailedError(
                        self._failure_message(exc.payload, "Generation failed"),
                        status=exc.status,
                        code=exc.code,
                        payload=exc.payload,
                        task_id=normalized_task_id,
                    ) from exc

                if self._is_retryable_query_error(exc):
                    consecutive_query_errors += 1
                    retry_after_ms = exc.retry_after_ms
                    self._raise_if_query_errors_exhausted(
                        normalized_task_id,
                        consecutive_query_errors,
                        exc,
                    )
                    continue
                exc.task_id = normalized_task_id
                raise
            except TransportError as exc:
                consecutive_query_errors += 1
                self._raise_if_query_errors_exhausted(
                    normalized_task_id,
                    consecutive_query_errors,
                    exc,
                )
                continue

            consecutive_query_errors = 0
            self._run_status_callback(on_status, snapshot)

            if snapshot.status == "succeeded":
                return GenerationResult(
                    task_id=normalized_task_id,
                    status="succeeded",
                    image_urls=snapshot.image_urls,
                    attempts=attempts,
                    elapsed_seconds=max(0.0, self._monotonic() - started_at),
                    raw=snapshot.raw,
                )

            if snapshot.status in {"failed", "refunded"}:
                raise TaskFailedError(
                    self._failure_message(snapshot.raw, "Generation failed"),
                    code=snapshot.raw.get("code"),
                    payload=snapshot.raw,
                    task_id=normalized_task_id,
                )

        elapsed = max(0.0, self._monotonic() - started_at)
        raise PollTimeoutError(
            f"Polling stopped（elapsed  {elapsed:.1f}  seconds, queries:  {attempts}  times）。"
            f"The task may still be running. Keep Task ID {normalized_task_id}， and resume polling later。",
            task_id=normalized_task_id,
        )

    def generate_image(
        self,
        options: Mapping[str, Any] | None = None,
        *,
        model: str | None = None,
        prompt: str | None = None,
        aspect_ratio: str | None = None,
        quality: str | None = None,
        resolution: str | None = None,
        urls: Sequence[str] | None = None,
        files: Sequence[PathInput] | None = None,
        on_task_created: TaskCreatedCallback | None = None,
        on_status: StatusCallback | None = None,
        should_cancel: CancelCallback | None = None,
        max_poll_duration: float | None = None,
        max_poll_attempts: int = DEFAULT_MAX_POLL_ATTEMPTS,
        max_poll_duration_seconds: float | None = None,
        initial_poll_interval_seconds: float | None = None,
        max_poll_interval_seconds: float | None = None,
        **legacy_options: Any,
    ) -> GenerationResult:
        """Submit one task and wait; never automatically retry submission。"""
        started_at = self._monotonic()
        merged = self._merge_generate_options(
            options,
            model=model,
            prompt=prompt,
            aspect_ratio=aspect_ratio,
            quality=quality,
            resolution=resolution,
            urls=urls,
            files=files,
            legacy_options=legacy_options,
        )

        submitted = self.submit_image_task(
            merged,
            on_task_created=on_task_created,
        )

        if submitted.is_complete:
            return GenerationResult(
                task_id=submitted.task_id,
                status="succeeded",
                image_urls=submitted.image_urls,
                attempts=0,
                elapsed_seconds=max(0.0, self._monotonic() - started_at),
                raw=submitted.raw,
            )

        if not submitted.task_id:
            raise ProtocolError("Task incomplete and no  task_id", payload=submitted.raw)

        result = self.wait_for_image_task(
            submitted.task_id,
            max_poll_duration=max_poll_duration,
            max_poll_attempts=max_poll_attempts,
            on_status=on_status,
            should_cancel=should_cancel,
            max_poll_duration_seconds=max_poll_duration_seconds,
            initial_poll_interval_seconds=initial_poll_interval_seconds,
            max_poll_interval_seconds=max_poll_interval_seconds,
        )
        return GenerationResult(
            task_id=result.task_id,
            status=result.status,
            image_urls=result.image_urls,
            attempts=result.attempts,
            elapsed_seconds=max(0.0, self._monotonic() - started_at),
            raw=result.raw,
        )

    def _submit_multipart(
        self,
        fields: Mapping[str, str],
        urls: tuple[str, ...],
        paths: tuple[Path, ...],
    ) -> httpx.Response:
        with ExitStack() as stack:
            multipart_files: list[tuple[str, tuple[str, Any, str]]] = []
            for path in paths:
                handle = stack.enter_context(path.open("rb"))
                content_type = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
                multipart_files.append(
                    (
                        "files",
                        (self._safe_filename(path.name), handle, content_type),
                    )
                )

            form_data: dict[str, Any] = dict(fields)
            if urls:
                # HTTPX  encodes the list as repeated  form-data  fields。
                form_data["urls"] = list(urls)

            return self._client.post(
                f"{API_BASE}{GENERATE_PATH}",
                data=form_data,
                files=multipart_files,
                timeout=self._submit_timeout,
                follow_redirects=False,
            )

    def _decode_response(self, response: httpx.Response) -> dict[str, Any]:
        retry_after_ms = self._parse_retry_after_ms(
            response.headers.get("Retry-After", ""),
            None,
        )

        try:
            decoded = response.json()
        except (json.JSONDecodeError, UnicodeDecodeError, ValueError) as exc:
            preview = self._safe_response_preview(response.content)
            if 200 <= response.status_code < 300:
                raise ProtocolError(
                    f"API response is not valid  JSON：{preview}",
                    status=response.status_code,
                ) from exc
            raise ApiResponseError(
                f"Request failed（HTTP {response.status_code}）， and response is not valid  JSON：{preview}",
                status=response.status_code,
                retry_after_ms=retry_after_ms,
            ) from exc

        if not isinstance(decoded, Mapping):
            if 200 <= response.status_code < 300:
                raise ProtocolError(
                    "API  JSON  response must be an object",
                    status=response.status_code,
                )
            raise ApiResponseError(
                f"Request failed（HTTP {response.status_code}）",
                status=response.status_code,
                retry_after_ms=retry_after_ms,
            )

        envelope = dict(decoded)
        wrapped = envelope.get("data")
        normalized = dict(envelope)
        if isinstance(wrapped, Mapping):
            normalized.update(wrapped)

        retry_after_ms = self._parse_retry_after_ms(
            response.headers.get("Retry-After", ""),
            normalized.get("retryAfterMs"),
        )

        code = normalized.get("code")
        business_failed = "code" in normalized and not self._is_success_code(code)
        failed = (
            not 200 <= response.status_code < 300
            or normalized.get("ok") is False
            or business_failed
        )

        safe_payload = self._redact_payload(normalized)
        if failed:
            message = self._failure_message(
                safe_payload,
                f"Request failed（HTTP {response.status_code}）",
            )
            raise ApiResponseError(
                message,
                status=response.status_code,
                code=code,
                retry_after_ms=retry_after_ms,
                payload=safe_payload,
            )

        return safe_payload

    def _validate_options(
        self,
        *,
        model: str,
        prompt: str,
        aspect_ratio: str,
        quality: str,
        resolution: str,
        urls: Sequence[str] | None,
        files: Sequence[PathInput] | None,
    ) -> _ValidatedOptions:
        if model not in SUPPORTED_MODELS:
            allowed = "、".join(sorted(SUPPORTED_MODELS))
            raise ValidationError(f"model  supports only ：{allowed}")

        if not isinstance(prompt, str) or not prompt.strip():
            raise ValidationError("prompt  must be a non-empty string")
        if len(prompt) > MAX_PROMPT_LENGTH:
            raise ValidationError(f"prompt  at most  {MAX_PROMPT_LENGTH}  characters")

        if not isinstance(aspect_ratio, str) or not aspect_ratio.strip():
            raise ValidationError("aspectRatio  must be a non-empty string")
        normalized_ratio = aspect_ratio.strip()
        if normalized_ratio != "auto" and not re.fullmatch(
            r"[1-9]\d{0,2}:[1-9]\d{0,2}",
            normalized_ratio,
        ):
            raise ValidationError("aspectRatio  must be  auto  or a ratio such as  1:1、16:9  ")

        if quality in {"xhigh", "max"} and model != "gpt-image-2.5-sunburst":
            raise ValidationError("xhigh、max  only supports  gpt-image-2.5-sunburst")
        if quality not in SUPPORTED_QUALITIES:
            allowed = "、".join(sorted(SUPPORTED_QUALITIES))
            raise ValidationError(f"quality  supports only ：{allowed}")
        if resolution not in SUPPORTED_RESOLUTIONS:
            allowed = "、".join(sorted(SUPPORTED_RESOLUTIONS))
            raise ValidationError(f"resolution  supports only ：{allowed}")

        normalized_urls = self._normalize_urls(urls)
        normalized_files = self._normalize_files(files)
        if len(normalized_urls) + len(normalized_files) > MAX_REFERENCE_IMAGES:
            raise ValidationError(
                f"urls  and  files  total limit:  {MAX_REFERENCE_IMAGES}  reference images"
            )

        return _ValidatedOptions(
            model=model,
            prompt=prompt,
            aspect_ratio=normalized_ratio,
            quality=quality,
            resolution=resolution,
            urls=normalized_urls,
            files=normalized_files,
        )

    def _normalize_urls(self, urls: Sequence[str] | None) -> tuple[str, ...]:
        values = self._normalize_sequence(urls, "urls")
        normalized: list[str] = []
        for index, value in enumerate(values):
            if not isinstance(value, str) or not value.strip():
                raise ValidationError(f"urls[{index}]  must be a non-empty string")
            url = value.strip()
            parsed = urlsplit(url)
            if parsed.scheme.lower() != "https" or not parsed.hostname:
                raise ValidationError(f"urls[{index}]  must be a complete  HTTPS  image URL")
            if parsed.username or parsed.password:
                raise ValidationError(f"urls[{index}] Do not include credentials in  URL  ")
            normalized.append(url)
        return tuple(normalized)

    def _normalize_files(self, files: Sequence[PathInput] | None) -> tuple[Path, ...]:
        values = self._normalize_sequence(files, "files")
        normalized: list[Path] = []
        for index, value in enumerate(values):
            if not isinstance(value, (str, os.PathLike)):
                raise ValidationError(f"files[{index}]  must be a local file path")
            path = Path(value).expanduser()
            if not path.exists():
                raise ValidationError(f"files[{index}]  does not exist：{path}")
            if not path.is_file():
                raise ValidationError(f"files[{index}]  is not a regular file：{path}")
            try:
                with path.open("rb"):
                    pass
            except OSError as exc:
                raise ValidationError(f"files[{index}]  cannot be read：{path}") from exc
            normalized.append(path)
        return tuple(normalized)

    @staticmethod
    def _normalize_sequence(value: Any, field_name: str) -> list[Any]:
        if value is None:
            return []
        if isinstance(value, (str, bytes, bytearray, os.PathLike, Mapping)):
            raise ValidationError(f"{field_name}  must be a list, not a string or object")
        try:
            return list(value)
        except TypeError as exc:
            raise ValidationError(f"{field_name}  must be a list") from exc

    @staticmethod
    def _validate_task_id(task_id: str) -> str:
        if not isinstance(task_id, str) or not task_id.strip():
            raise ValidationError("task_id  must be a non-empty string")
        normalized = task_id.strip()
        if len(normalized) > 512:
            raise ValidationError("task_id  has an invalid length")
        return normalized

    @staticmethod
    def _validate_poll_limits(duration: float, attempts: int) -> tuple[float, int]:
        try:
            normalized_duration = float(duration)
        except (TypeError, ValueError, OverflowError) as exc:
            raise ValidationError("max_poll_duration  must be a valid number") from exc
        if not math.isfinite(normalized_duration) or normalized_duration <= 0:
            raise ValidationError("max_poll_duration  must be greater than  0")
        if not isinstance(attempts, int) or isinstance(attempts, bool) or attempts <= 0:
            raise ValidationError("max_poll_attempts  must be greater than  0  as an integer")
        return normalized_duration, attempts

    def _next_poll_delay(
        self,
        *,
        attempt_index: int,
        retry_after_ms: int,
        initial_interval: float,
        maximum_interval: float,
    ) -> float:
        if retry_after_ms > 0:
            return min(retry_after_ms / 1000.0, self._max_retry_after_seconds)

        base = min(
            initial_interval * (self._poll_backoff_factor ** attempt_index),
            maximum_interval,
        )
        if base <= 0 or self._poll_jitter_ratio <= 0:
            return max(0.0, base)

        random_value = min(1.0, max(0.0, float(self._random())))
        jitter_factor = 1.0 + ((2.0 * random_value - 1.0) * self._poll_jitter_ratio)
        return max(0.0, base * jitter_factor)

    def _is_retryable_query_error(self, exc: ApiResponseError) -> bool:
        status = self._numeric_status(exc.status)
        code = self._numeric_status(exc.code)
        return (
            status in RETRYABLE_QUERY_HTTP_STATUSES
            or code in RETRYABLE_QUERY_HTTP_STATUSES
        )

    def _raise_if_query_errors_exhausted(
        self,
        task_id: str,
        consecutive_errors: int,
        last_error: ImgApiError,
    ) -> None:
        if consecutive_errors < self._max_consecutive_query_errors:
            return
        raise QueryRetryExhaustedError(
            f"Consecutive query failures:  {consecutive_errors} . Keep Task ID {task_id}， and resume polling later。"
            f"Last error：{last_error}",
            status=last_error.status,
            code=last_error.code,
            retry_after_ms=last_error.retry_after_ms,
            payload=last_error.payload,
            task_id=task_id,
        ) from last_error

    def _raise_if_cancelled(
        self,
        task_id: str,
        should_cancel: CancelCallback | None,
    ) -> None:
        if should_cancel is None:
            return
        try:
            cancelled = bool(should_cancel())
        except Exception as exc:
            raise TaskCallbackError(
                f"Cancel check callback failed：{exc}",
                task_id=task_id,
            ) from exc
        if cancelled:
            raise PollCancelledError(
                f"Waiting stopped. The server task may still be running. Keep Task ID {task_id}。",
                task_id=task_id,
            )

    @staticmethod
    def _run_task_created_callback(
        callback: TaskCreatedCallback | None,
        task_id: str,
        raw: Mapping[str, Any],
    ) -> None:
        if callback is None:
            return
        try:
            # The callback accepts  (task_id, raw)， and also supports the legacy single argument  task_id。
            try:
                signature = inspect.signature(callback)
            except (TypeError, ValueError):
                callback(task_id, dict(raw))
            else:
                try:
                    signature.bind(task_id, raw)
                except TypeError:
                    signature.bind(task_id)
                    callback(task_id)
                else:
                    callback(task_id, dict(raw))
        except Exception as exc:
            raise TaskPersistenceError(
                "The server created a task but Save the task ID  failed. Do not resubmit；"
                f"Immediately Save the task ID {task_id}  and resume polling later. Original error：{exc}",
                task_id=task_id,
            ) from exc

    @staticmethod
    def _validate_non_negative_finite(value: Any, field_name: str) -> float:
        try:
            normalized = float(value)
        except (TypeError, ValueError, OverflowError) as exc:
            raise ValidationError(f"{field_name}  must be a valid number") from exc
        if not math.isfinite(normalized) or normalized < 0:
            raise ValidationError(f"{field_name}  must be greater than or equal to  0  as a finite number")
        return normalized

    @staticmethod
    def _merge_generate_options(
        options: Mapping[str, Any] | None,
        *,
        model: str | None,
        prompt: str | None,
        aspect_ratio: str | None,
        quality: str | None,
        resolution: str | None,
        urls: Sequence[str] | None,
        files: Sequence[PathInput] | None,
        legacy_options: Mapping[str, Any],
    ) -> dict[str, Any]:
        if options is None:
            merged: dict[str, Any] = {}
        elif isinstance(options, Mapping):
            merged = dict(options)
        else:
            raise ValidationError("options  must be a dictionary")

        extra = dict(legacy_options)
        for key, value in extra.items():
            if key in merged and merged[key] != value:
                raise ValidationError(f"Parameter  {key}  was supplied more than once with different values")
            merged[key] = value

        explicit = {
            "model": model,
            "prompt": prompt,
            "aspect_ratio": aspect_ratio,
            "quality": quality,
            "resolution": resolution,
            "urls": urls,
            "files": files,
        }
        for key, value in explicit.items():
            if value is None:
                continue
            if key in merged and merged[key] != value:
                raise ValidationError(f"Parameter  {key}  was supplied more than once with different values")
            merged[key] = value

        if "aspectRatio" in merged:
            legacy_ratio = merged.pop("aspectRatio")
            if "aspect_ratio" in merged and merged["aspect_ratio"] != legacy_ratio:
                raise ValidationError("aspect_ratio  and  aspectRatio  must not be supplied with different values")
            merged["aspect_ratio"] = legacy_ratio

        allowed = {
            "model",
            "prompt",
            "aspect_ratio",
            "quality",
            "resolution",
            "urls",
            "files",
        }
        unknown = sorted(set(merged) - allowed)
        if unknown:
            raise ValidationError(f"Unsupported parameters：{'、'.join(unknown)}")

        return {
            "model": merged.get("model"),
            "prompt": merged.get("prompt"),
            "aspect_ratio": merged.get("aspect_ratio", "auto"),
            "quality": merged.get("quality", "auto"),
            "resolution": merged.get("resolution", "1K"),
            "urls": merged.get("urls"),
            "files": merged.get("files"),
        }

    @staticmethod
    def _run_status_callback(
        callback: StatusCallback | None,
        snapshot: TaskSnapshot,
    ) -> None:
        if callback is None:
            return
        try:
            callback(snapshot)
        except Exception as exc:
            raise TaskCallbackError(
                f"Task status callback failed：{exc}。Keep Task ID {snapshot.task_id}。",
                task_id=snapshot.task_id,
            ) from exc

    @staticmethod
    def _resolve_aspect_ratio(aspect_ratio: str, legacy_options: dict[str, Any]) -> str:
        if "aspectRatio" not in legacy_options:
            return aspect_ratio
        legacy_value = legacy_options.pop("aspectRatio")
        if aspect_ratio != "auto" and legacy_value != aspect_ratio:
            raise ValidationError("aspect_ratio  and  aspectRatio  must not be supplied with different values")
        return legacy_value

    @staticmethod
    def _reject_unknown_options(options: Mapping[str, Any]) -> None:
        if options:
            names = "、".join(sorted(options))
            raise ValidationError(f"Unsupported parameters：{names}")

    def _parse_retry_after_ms(self, header_value: str, body_value: Any) -> int:
        try:
            milliseconds = float(body_value or 0)
            if math.isfinite(milliseconds) and milliseconds > 0:
                return int(milliseconds)
        except (TypeError, ValueError, OverflowError):
            pass

        text = str(header_value or "").strip()
        if not text:
            return 0
        try:
            seconds = float(text)
            if math.isfinite(seconds):
                return max(0, int(seconds * 1000))
            return 0
        except (TypeError, ValueError, OverflowError):
            pass

        try:
            moment = parsedate_to_datetime(text)
            if moment.tzinfo is None:
                moment = moment.replace(tzinfo=timezone.utc)
            delta = (moment - datetime.now(timezone.utc)).total_seconds()
            if not math.isfinite(delta):
                return 0
            return max(0, int(delta * 1000))
        except (TypeError, ValueError, OverflowError):
            return 0

    @staticmethod
    def _normalize_status(value: Any) -> str:
        return str(value or "").strip().lower()

    @staticmethod
    def _extract_task_id(payload: Mapping[str, Any]) -> str | None:
        for name in ("id", "task_id", "taskId"):
            value = payload.get(name)
            if value is not None and str(value).strip():
                return str(value).strip()
        return None

    @staticmethod
    def _extract_image_urls(payload: Mapping[str, Any]) -> tuple[str, ...]:
        candidates: list[Any] = []
        for name in ("image", "images", "image_url", "imageUrl"):
            if name in payload:
                candidates.append(payload[name])

        results: list[str] = []

        def add(value: Any) -> None:
            if isinstance(value, str):
                text = value.strip()
                if text and text not in results:
                    results.append(text)
                return
            if isinstance(value, Mapping):
                for key in ("url", "image", "image_url", "imageUrl"):
                    if key in value:
                        add(value[key])
                return
            if isinstance(value, Iterable) and not isinstance(value, (str, bytes, bytearray)):
                for item in value:
                    add(item)

        for candidate in candidates:
            add(candidate)
        return tuple(results)

    def _redact_payload(self, value: Mapping[str, Any]) -> dict[str, Any]:
        def redact(item: Any, key_name: str = "") -> Any:
            lowered = key_name.lower().replace("_", "")
            if lowered in {"key", "cardkey", "apikey", "authorization"}:
                return "***REDACTED***"
            if isinstance(item, Mapping):
                return {str(key): redact(val, str(key)) for key, val in item.items()}
            if isinstance(item, list):
                return [redact(entry) for entry in item]
            if isinstance(item, tuple):
                return tuple(redact(entry) for entry in item)
            if isinstance(item, str) and self._card_key in item:
                return item.replace(self._card_key, "***REDACTED***")
            return item

        result = redact(value)
        return dict(result) if isinstance(result, Mapping) else {}

    @staticmethod
    def _failure_message(payload: Mapping[str, Any], fallback: str) -> str:
        for name in ("error", "msg", "message", "failure_reason"):
            value = payload.get(name)
            if value is not None and str(value).strip():
                return str(value).strip()
        return fallback

    @staticmethod
    def _safe_filename(name: str) -> str:
        cleaned = str(name).replace('"', "_").replace("\r", "_").replace("\n", "_")
        cleaned = cleaned.replace("/", "_").replace("\\", "_")
        return cleaned or "reference-image"

    def _safe_exception_text(self, exc: BaseException) -> str:
        text = str(exc) or exc.__class__.__name__
        return text.replace(self._card_key, "***REDACTED***")

    def _safe_response_preview(self, raw: bytes, limit: int = 300) -> str:
        preview = raw[:limit].decode("utf-8", errors="replace")
        preview = preview.replace(self._card_key, "***REDACTED***")
        suffix = "…" if len(raw) > limit else ""
        return repr(preview + suffix)

    @staticmethod
    def _is_success_code(code: Any) -> bool:
        try:
            return int(code) == 200
        except (TypeError, ValueError, OverflowError):
            return False

    @staticmethod
    def _numeric_status(value: Any) -> int:
        try:
            return int(value)
        except (TypeError, ValueError, OverflowError):
            return 0

    def _ensure_open(self) -> None:
        if self._closed:
            raise RuntimeError("ImgApiClient  is closed")


def generate_image(
    options: Mapping[str, Any],
    *,
    card_key: str | None = None,
    on_task_created: TaskCreatedCallback | None = None,
    on_status: StatusCallback | None = None,
    max_poll_duration_seconds: float = DEFAULT_POLL_DURATION_SECONDS,
) -> str:
    """
    Compatibility helper returning the first image  URL。

    In production, own and reuse an  ImgApiClient  instance rather than creating a connection pool per image。
    """
    with ImgApiClient(card_key=card_key) as client:
        result = client.generate_image(
            options,
            on_task_created=on_task_created,
            on_status=on_status,
            max_poll_duration_seconds=max_poll_duration_seconds,
        )
    return result.image_url


def main() -> None:
    """Local example. In production, replace  on_task_created  with persistent storage。"""

    def save_task_id(task_id: str) -> None:
        # In production, persist the task ID before polling。
        print(f"Task ID：{task_id}")

    def show_status(snapshot: TaskSnapshot) -> None:
        print(f"Task status：{snapshot.status}")

    try:
        with ImgApiClient() as client:
            result = client.generate_image(
                model="gpt-image-2",
                prompt="A shiba inu wearing sunglasses, sitting on a beach with a soda, cyberpunk style",
                aspect_ratio="1:1",
                quality="auto",
                resolution="1K",
                urls=[],
                files=[],
                on_task_created=save_task_id,
                on_status=show_status,
            )

        print("Generation succeeded：")
        for image_url in result.image_urls:
            print(image_url)
    except SubmissionUncertainError as exc:
        print(f"Submission uncertain：{exc}")
        raise SystemExit(2) from exc
    except ImgApiError as exc:
        print(f"Generation failed：{exc}")
        if exc.task_id:
            print(f"Keep Task ID：{exc.task_id}")
        raise SystemExit(1) from exc


if __name__ == "__main__":
    main()