ImgAPI 生图 API 接入文档

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.1',
  'nano-banana-2',
  'nano-banana-pro',
  'seedream-5.0-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} 必须是大于 0 的数字`);
  }
  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('操作已取消'));
  };

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

  const timer = setTimeout(() => {
    timedOut = true;
    controller.abort(new Error(`请求超时（${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('操作已取消'));
  }

  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('操作已取消'));
    };
    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 || '查询任务失败', { 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('请在服务端设置合法的 16 位 IMGAPI_CARD_KEY');
  }
  return String(cardKey);
}

function validateGenerateOptions(options = {}) {
  if (!isRecord(options)) throw new TypeError('生图参数必须是对象');

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

  if (typeof options.prompt !== 'string' || !options.prompt.trim()) {
    throw new TypeError('prompt 不能为空');
  }
  if (options.prompt.length > 10000) {
    throw new TypeError('prompt 最多 10000 个字符');
  }

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

  if (model === 'nano-banana-2.1' && !['auto', '1:1', '3:2', '2:3', '16:9', '9:16', '4:3', '3:4'].includes(aspectRatio)) throw new TypeError('Banana 2.1 不支持该 aspectRatio');
  const quality = options.quality ?? 'auto';
  if (model === 'nano-banana-2.1' && quality !== 'auto') throw new TypeError('Banana 2.1 quality 仅支持 auto');
  if (!SUPPORTED_QUALITIES.has(quality)) {
    throw new TypeError(`不支持的 quality：${quality}`);
  }

  if (['xhigh', 'max'].includes(quality) && model !== 'gpt-image-2.5-sunburst') {
    throw new TypeError('xhigh、max 仅支持 gpt-image-2.5-sunburst');
  }
  const resolution = options.resolution ?? (['seedream-5.0-pro', 'nano-banana-2.1'].includes(model) ? '2K' : '1K');
  if (!SUPPORTED_RESOLUTIONS.has(resolution)) {
    throw new TypeError(`不支持的 resolution：${resolution}`);
  }

  if (model === 'seedream-5.0-pro' && resolution === '4K') {
    throw new TypeError('Seedream 5.0 Pro 仅支持 1K、2K');
  }

  const officialChannel = options.officialChannel ?? 'false';
  if (!['false', 'true', 'auto'].includes(officialChannel)) throw new TypeError('officialChannel 必须是字符串 false、true、auto');
  if (officialChannel !== 'false' && !['gpt-image-2.5-flare', 'gpt-image-2.5-sunburst'].includes(model)) throw new TypeError('官方渠道仅支持 Flare / Sunburst');

  let webhook;
  if (options.webhook != null) {
    if (typeof options.webhook !== 'string' || !options.webhook.trim()) {
      throw new TypeError('webhook 必须是有效的 HTTP/HTTPS 回调地址');
    }
    webhook = options.webhook.trim();
    let parsed;
    try { parsed = new URL(webhook); } catch {
      throw new TypeError('webhook 必须是有效的 HTTP/HTTPS 回调地址');
    }
    if (!['http:', 'https:'].includes(parsed.protocol) || !parsed.hostname || parsed.username || parsed.password || parsed.hash) {
      throw new TypeError('webhook 必须是 HTTP/HTTPS 地址，且不包含用户凭据或片段');
    }
  }

  const urls = options.urls ?? [];
  if (!Array.isArray(urls)) throw new TypeError('urls 必须是数组');
  const referenceLimit = model === 'nano-banana-2.1' ? 14 : model === 'seedream-5.0-pro' ? 10 : 12;
  if (urls.length > referenceLimit) {
    throw new TypeError(`urls 最多 ${referenceLimit} 张参考图`);
  }

  const normalizedUrls = urls.map((value, index) => {
    if (typeof value !== 'string' || !value.trim()) {
      throw new TypeError(`urls[${index}] 必须是 HTTPS 图片地址`);
    }
    const text = value.trim();
    let parsed;
    try {
      parsed = new URL(text);
    } catch {
      throw new TypeError(`urls[${index}] 不是合法 URL`);
    }
    if (parsed.protocol !== 'https:') {
      throw new TypeError(`urls[${index}] 必须使用 HTTPS`);
    }
    return text;
  });

  let mask;
  if (options.mask !== undefined) {
    if (typeof options.mask !== 'string' || !options.mask.trim()) {
      throw new TypeError('mask 必须是完整的 HTTP/HTTPS 蒙版图片地址');
    }
    if (!normalizedUrls.length) throw new TypeError('使用 mask 时必须同时传入 urls 原图');
    mask = options.mask.trim();
    let parsed;
    try { parsed = new URL(mask); } catch {
      throw new TypeError('mask 必须是完整的 HTTP/HTTPS 蒙版图片地址');
    }
    if (!['http:', 'https:'].includes(parsed.protocol) || !parsed.hostname || parsed.username || parsed.password || parsed.hash) {
      throw new TypeError('mask 必须是 HTTP/HTTPS 地址，且不包含用户凭据或片段');
    }
  }

  return {
    model,
    prompt: options.prompt,
    aspectRatio,
    quality,
    resolution,
    officialChannel,
    webhook,
    urls: normalizedUrls,
    mask,
  };
}

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

  if (options.webhook !== undefined) fields.webhook = options.webhook;
  if (options.mask !== undefined) fields.mask = options.mask;

  return {
    body: { ...fields, urls: options.urls },
    isFormData: false,
  };
}

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('当前 Node.js 环境不支持 fetch，请使用 Node.js 18+');
  if (typeof sleepImpl !== 'function' || typeof now !== 'function' || typeof random !== 'function') {
    throw new TypeError('fetch/sleep/now/random 配置无效');
  }

  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('操作已取消', { cause });
      }
      const message = requestSignal.timedOut()
        ? `请求超时（${timeout}ms）`
        : `网络请求失败：${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
            ? '接口返回了无法解析的非 JSON 响应'
            : `请求失败（HTTP ${response.status}，响应不是 JSON）`,
          {
            status: response.status,
            retryable: response.status >= 500,
            cause,
          },
        );
      }
    }

    const data = normalizeEnvelope(envelope);
    if (!data) {
      throw new ImgApiError('接口返回的 JSON 不是对象', {
        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, `请求失败（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(
          `提交结果不确定：${error.message}；禁止自动重新提交，以免产生重复任务或重复扣费`,
          {
            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('任务状态为 succeeded，但响应缺少 image', {
          taskId,
          payload: submitted,
        });
      }
      return { taskId, status, image: submitted.image, raw: submitted };
    }

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

    if (!taskId) {
      throw new ImgApiError('提交接口未返回任务 ID', { payload: submitted });
    }

    if (onTaskCreated) {
      try {
        await onTaskCreated(taskId, submitted);
      } catch (cause) {
        throw new ImgApiError(
          '任务已经提交，但保存任务 ID 失败；请使用错误中的 taskId 继续查询，禁止重新提交',
          { 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 不能为空');

    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 不能为空');

    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 不能小于 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(
          '已停止本地等待；服务端任务可能仍在运行，请使用 taskId 继续查询',
          { 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('任务状态为 succeeded，但响应缺少 image', {
            taskId: id,
            payload: queried,
          });
        }
        return { taskId: id, status, image: queried.image, raw: queried };
      }

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

      if (status !== 'submitted' && status !== 'processing') {
        throw new ImgApiError(`接口返回了未知任务状态：${status || 'empty'}`, {
          taskId: id,
          payload: queried,
        });
      }

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

    throw new ImgApiError(
      '生成时间较长；服务端任务可能仍在运行，请保存 taskId 并稍后继续查询',
      { 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 生产增强版同步 Python 客户端。

依赖：
    pip install "httpx>=0.27,<1"

安全原则：
- CardKey 只从服务端环境变量或构造参数读取，不写入浏览器代码。
- 提交任务不会自动重试；提交超时或响应中断时，服务端可能已经创建任务。
- 查询失败时始终复用原 task_id，不重新提交生成任务。
- 参考图通过 urls 传入 HTTPS 图片链接。

Python：3.10+
"""

from __future__ import annotations

import json
import inspect
import math
import os
import random
import re
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
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.1",
    "nano-banana-2",
    "nano-banana-pro",
    "seedream-5.0-pro",
})
SUPPORTED_QUALITIES = frozenset({"auto", "low", "medium", "high", "xhigh", "max"})
SUPPORTED_RESOLUTIONS = frozenset({"1K", "2K", "4K"})

MAX_PROMPT_LENGTH = 10000
REFERENCE_LIMITS = {"nano-banana-2.1": 14, "seedream-5.0-pro": 10}
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})

TaskCreatedCallback: TypeAlias = Callable[..., None]
StatusCallback: TypeAlias = Callable[["TaskSnapshot"], None]
CancelCallback: TypeAlias = Callable[[], bool]


class ImgApiError(Exception):
    """所有 ImgAPI 客户端异常的基类。"""

    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):
    """调用参数不符合要求。"""


class TransportError(ImgApiError):
    """网络、连接池或底层传输错误。"""


class ProtocolError(ImgApiError):
    """HTTP 成功，但服务端响应结构不符合接口约定。"""


class ApiResponseError(ImgApiError):
    """HTTP 或业务状态明确返回失败。"""


class SubmissionUncertainError(ImgApiError):
    """提交结果不确定；禁止自动重新提交，以免重复任务或重复扣费。"""


class TaskFailedError(ImgApiError):
    """生成任务明确失败、退款或取消。"""


class TaskPersistenceError(ImgApiError):
    """服务端已经返回 task_id，但调用方未能持久化该 ID。"""


class TaskCallbackError(ImgApiError):
    """任务状态回调执行失败。"""


class PollTimeoutError(ImgApiError):
    """达到轮询时长或次数上限，任务可能仍在服务端执行。"""


class PollCancelledError(ImgApiError):
    """调用方停止等待，任务可能仍在服务端执行。"""


class QueryRetryExhaustedError(ImgApiError):
    """查询连续失败次数过多，但原任务仍可稍后继续查询。"""


@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:
        """兼容旧调用名称。"""
        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:
        """兼容旧调用名称。"""
        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("生成结果中没有图片地址", task_id=self.task_id)
        return self.image_urls[0]

    @property
    def image(self) -> str:
        """兼容旧调用名称。"""
        return self.image_url


@dataclass(frozen=True, slots=True)
class _ValidatedOptions:
    model: str
    prompt: str
    aspect_ratio: str
    quality: str
    resolution: str
    official_channel: str
    urls: tuple[str, ...]
    webhook: str | None
    mask: str | None


class ImgApiClient:
    """
    ImgAPI 同步客户端。

    推荐将一个实例复用于多个请求，以复用 TCP/TLS 连接。
    若传入外部 http_client，本类不会替调用方关闭它。
    """

    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 必须是 16 位十六进制字符串，且只能保存在服务端环境变量中"
            )

        if poll_initial_interval < 0:
            raise ValidationError("poll_initial_interval 不能小于 0")
        if poll_max_interval < poll_initial_interval:
            raise ValidationError("poll_max_interval 不能小于 poll_initial_interval")
        if poll_backoff_factor < 1:
            raise ValidationError("poll_backoff_factor 不能小于 1")
        if not 0 <= poll_jitter_ratio <= 1:
            raise ValidationError("poll_jitter_ratio 必须在 0 到 1 之间")
        if max_consecutive_query_errors < 1:
            raise ValidationError("max_consecutive_query_errors 必须大于 0")
        if max_retry_after_seconds < 0:
            raise ValidationError("max_retry_after_seconds 不能小于 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 是上一版代码使用的参数名，继续兼容，避免替换客户端后改动调用方。
        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,
        official_channel: str | None = None,
        webhook: str | None = None,
        urls: Sequence[str] | None = None,
        mask: str | None = None,
        on_task_created: TaskCreatedCallback | None = None,
        **legacy_options: Any,
    ) -> SubmissionResult:
        """
        提交一次生成任务，不自动重试。

        为兼容接口原字段名，也接受 aspectRatio 关键字，但不能同时传
        aspect_ratio 和 aspectRatio 两个不同值。
        """
        self._ensure_open()
        merged = self._merge_generate_options(
            options,
            model=model,
            prompt=prompt,
            aspect_ratio=aspect_ratio,
            quality=quality,
            resolution=resolution,
            official_channel=official_channel,
            webhook=webhook,
            urls=urls,
            mask=mask,
            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"],
            official_channel=merged["official_channel"],
            webhook=merged["webhook"],
            urls=merged["urls"],
            mask=merged["mask"],
        )

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

        if options.webhook is not None:
            fields["webhook"] = options.webhook
        if options.mask is not None:
            fields["mask"] = options.mask

        try:
            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:
            # 尚未建立可用连接，通常可以确认请求未送达应用服务器。
            raise TransportError(
                f"提交任务时无法建立连接：{self._safe_exception_text(exc)}",
                submission_uncertain=False,
                definitely_not_sent=True,
            ) from exc
        except httpx.TransportError as exc:
            # 写入中断、读取响应超时等情况下，服务端可能已经收到请求。
            raise SubmissionUncertainError(
                "提交结果不确定：服务端可能已经创建任务。禁止自动重新提交；"
                f"请先检查后台任务或调用记录。底层错误：{self._safe_exception_text(exc)}",
                submission_uncertain=True,
            ) from exc

        try:
            payload = self._decode_response(response)
        except ProtocolError as exc:
            # 已收到 HTTP 成功响应，但无法确认是否已经创建任务，仍需按结果不确定处理。
            raise SubmissionUncertainError(
                "提交接口返回了无法识别的成功响应，服务端可能已经创建任务。"
                "禁止自动重新提交，请先检查后台任务或调用记录。",
                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(
                    "提交接口返回了可能存在歧义的网关/服务器错误。服务端可能已经创建任务，"
                    "禁止自动重新提交，请先检查任务记录。",
                    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, "生成任务提交失败"),
                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(
                    "提交接口返回成功状态，但没有返回图片地址",
                    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(
                "提交接口未返回任务 ID",
                status=response.status_code,
                code=payload.get("code"),
                payload=payload,
            )

        if status and status not in PENDING_STATUSES:
            raise ProtocolError(
                f"提交接口返回了未知任务状态：{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:
        """查询一个已存在的任务；此方法不会创建新任务。"""
        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"查询任务时发生网络错误：{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(
                "查询接口没有返回任务状态",
                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(
                    "任务已成功，但查询接口缺少 image 图片地址",
                    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:
            # 保留官方返回的 pending 状态，方便上层展示。
            pass
        else:
            raise ProtocolError(
                f"查询接口返回了未知任务状态：{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:
        """使用原 task_id 轮询，适合恢复进程重启前保存的任务。"""
        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 与 max_poll_duration_seconds 不能同时传入不同值"
                )
        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 不能小于 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, "生成失败"),
                        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, "生成失败"),
                    code=snapshot.raw.get("code"),
                    payload=snapshot.raw,
                    task_id=normalized_task_id,
                )

        elapsed = max(0.0, self._monotonic() - started_at)
        raise PollTimeoutError(
            f"轮询已停止（耗时 {elapsed:.1f} 秒、查询 {attempts} 次）。"
            f"任务可能仍在生成，请保留任务 ID {normalized_task_id}，稍后继续查询。",
            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,
        official_channel: str | None = None,
        webhook: str | None = None,
        urls: Sequence[str] | None = None,
        mask: str | 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:
        """提交一次任务并等待完成；提交阶段永不自动重试。"""
        started_at = self._monotonic()
        merged = self._merge_generate_options(
            options,
            model=model,
            prompt=prompt,
            aspect_ratio=aspect_ratio,
            quality=quality,
            resolution=resolution,
            official_channel=official_channel,
            webhook=webhook,
            urls=urls,
            mask=mask,
            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_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 _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"接口返回的不是有效 JSON：{preview}",
                    status=response.status_code,
                ) from exc
            raise ApiResponseError(
                f"请求失败（HTTP {response.status_code}），且响应不是有效 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(
                    "接口 JSON 响应必须是对象",
                    status=response.status_code,
                )
            raise ApiResponseError(
                f"请求失败（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"请求失败（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,
        official_channel: str,
        webhook: str | None,
        urls: Sequence[str] | None,
        mask: str | None,
    ) -> _ValidatedOptions:
        if model not in SUPPORTED_MODELS:
            allowed = "、".join(sorted(SUPPORTED_MODELS))
            raise ValidationError(f"model 只支持：{allowed}")

        if not isinstance(prompt, str) or not prompt.strip():
            raise ValidationError("prompt 必须是非空字符串")
        if len(prompt) > MAX_PROMPT_LENGTH:
            raise ValidationError(f"prompt 最多 {MAX_PROMPT_LENGTH} 个字符")

        if not isinstance(aspect_ratio, str) or not aspect_ratio.strip():
            raise ValidationError("aspectRatio 必须是非空字符串")
        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 必须是 auto 或类似 1:1、16:9 的比例")

        if model == "nano-banana-2.1":
            if normalized_ratio not in {"auto", "1:1", "3:2", "2:3", "16:9", "9:16", "4:3", "3:4"}:
                raise ValidationError("Banana 2.1 不支持该 aspectRatio")
            if quality != "auto":
                raise ValidationError("Banana 2.1 quality 仅支持 auto")

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

        if model == "seedream-5.0-pro" and resolution == "4K":
            raise ValidationError("Seedream 5.0 Pro 仅支持 1K、2K")

        if official_channel not in ("false", "true", "auto"):
            raise ValidationError("officialChannel 必须是字符串 false、true、auto")
        if official_channel != "false" and model not in {"gpt-image-2.5-flare", "gpt-image-2.5-sunburst"}:
            raise ValidationError("官方渠道仅支持 Flare / Sunburst")

        if webhook is not None:
            if not isinstance(webhook, str) or not webhook.strip():
                raise ValidationError("webhook 必须是有效的 HTTP/HTTPS 回调地址")
            webhook = webhook.strip()
            try:
                callback = urlsplit(webhook)
                valid = callback.scheme in {"http", "https"} and bool(callback.hostname)
                valid = valid and not callback.username and not callback.password and not callback.fragment
                callback.port
            except ValueError:
                valid = False
            if not valid:
                raise ValidationError("webhook 必须是 HTTP/HTTPS 地址，且不包含用户凭据或片段")

        normalized_urls = self._normalize_urls(urls)
        if mask is not None:
            if not isinstance(mask, str) or not mask.strip():
                raise ValidationError("mask 必须是完整的 HTTP/HTTPS 蒙版图片地址")
            if not normalized_urls:
                raise ValidationError("使用 mask 时必须同时传入 urls 原图")
            mask = mask.strip()
            try:
                parsed_mask = urlsplit(mask)
                valid_mask = parsed_mask.scheme in {"http", "https"} and bool(parsed_mask.hostname)
                valid_mask = valid_mask and not parsed_mask.username and not parsed_mask.password and not parsed_mask.fragment
            except ValueError:
                valid_mask = False
            if not valid_mask:
                raise ValidationError("mask 必须是 HTTP/HTTPS 地址，且不包含用户凭据或片段")
        reference_limit = REFERENCE_LIMITS.get(model, 12)
        if len(normalized_urls) > reference_limit:
            raise ValidationError(
                f"urls 最多 {reference_limit} 张参考图"
            )

        return _ValidatedOptions(
            model=model,
            prompt=prompt,
            aspect_ratio=normalized_ratio,
            quality=quality,
            resolution=resolution,
            official_channel=official_channel,
            webhook=webhook,
            urls=normalized_urls,
            mask=mask,
        )

    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}] 必须是非空字符串")
            url = value.strip()
            parsed = urlsplit(url)
            if parsed.scheme.lower() != "https" or not parsed.hostname:
                raise ValidationError(f"urls[{index}] 必须是完整的 HTTPS 图片地址")
            if parsed.username or parsed.password:
                raise ValidationError(f"urls[{index}] 不允许在 URL 中携带用户名或密码")
            normalized.append(url)
        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} 必须是列表，不能直接传字符串或对象")
        try:
            return list(value)
        except TypeError as exc:
            raise ValidationError(f"{field_name} 必须是列表") 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 必须是非空字符串")
        normalized = task_id.strip()
        if len(normalized) > 512:
            raise ValidationError("task_id 长度异常")
        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 必须是有效数字") from exc
        if not math.isfinite(normalized_duration) or normalized_duration <= 0:
            raise ValidationError("max_poll_duration 必须大于 0")
        if not isinstance(attempts, int) or isinstance(attempts, bool) or attempts <= 0:
            raise ValidationError("max_poll_attempts 必须是大于 0 的整数")
        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_errors} 次。请保留任务 ID {task_id}，稍后继续查询。"
            f"最后错误：{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"取消检查回调执行失败：{exc}",
                task_id=task_id,
            ) from exc
        if cancelled:
            raise PollCancelledError(
                f"已停止等待。服务端任务可能仍在运行，请保留任务 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:
            # 新版回调可接收 (task_id, raw)，同时兼容旧版只接收 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(
                "服务端已经创建任务，但保存任务 ID 失败。禁止重新提交；"
                f"请立即保存任务 ID {task_id} 并稍后恢复查询。原始错误：{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} 必须是有效数字") from exc
        if not math.isfinite(normalized) or normalized < 0:
            raise ValidationError(f"{field_name} 必须是大于或等于 0 的有限数字")
        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,
        official_channel: str | None,
        webhook: str | None,
        urls: Sequence[str] | None,
        mask: str | 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 必须是字典")

        extra = dict(legacy_options)
        for key, value in extra.items():
            if key in merged and merged[key] != value:
                raise ValidationError(f"参数 {key} 被重复传入且值不同")
            merged[key] = value

        explicit = {
            "model": model,
            "prompt": prompt,
            "aspect_ratio": aspect_ratio,
            "quality": quality,
            "resolution": resolution,
            "official_channel": official_channel,
            "webhook": webhook,
            "urls": urls,
            "mask": mask,
        }
        for key, value in explicit.items():
            if value is None:
                continue
            if key in merged and merged[key] != value:
                raise ValidationError(f"参数 {key} 被重复传入且值不同")
            merged[key] = value

        if "officialChannel" in merged:
            legacy_channel = merged.pop("officialChannel")
            if "official_channel" in merged and merged["official_channel"] != legacy_channel:
                raise ValidationError("official_channel 与 officialChannel 不能同时传入不同值")
            merged["official_channel"] = legacy_channel

        if "aspectRatio" in merged:
            legacy_ratio = merged.pop("aspectRatio")
            if "aspect_ratio" in merged and merged["aspect_ratio"] != legacy_ratio:
                raise ValidationError("aspect_ratio 与 aspectRatio 不能同时传入不同值")
            merged["aspect_ratio"] = legacy_ratio

        allowed = {
            "model",
            "prompt",
            "aspect_ratio",
            "quality",
            "resolution",
            "official_channel",
            "webhook",
            "urls",
            "mask",
        }
        unknown = sorted(set(merged) - allowed)
        if unknown:
            raise ValidationError(f"存在不支持的参数：{'、'.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", "2K" if merged.get("model") in {"seedream-5.0-pro", "nano-banana-2.1"} else "1K"),
            "official_channel": merged.get("official_channel", "false"),
            "webhook": merged.get("webhook"),
            "urls": merged.get("urls"),
            "mask": merged.get("mask"),
        }

    @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"任务状态回调执行失败：{exc}。请保留任务 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 与 aspectRatio 不能同时传入不同值")
        return legacy_value

    @staticmethod
    def _reject_unknown_options(options: Mapping[str, Any]) -> None:
        if options:
            names = "、".join(sorted(options))
            raise ValidationError(f"存在不支持的参数：{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 已关闭")


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:
    """
    兼容原脚本的单函数用法，返回第一张图片 URL。

    正式网站建议自行持有并复用 ImgApiClient 实例，不要每张图创建一个连接池。
    """
    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:
    """本地调用示例。正式项目应将 on_task_created 替换为数据库写入。"""

    def save_task_id(task_id: str) -> None:
        # 生产环境：这里应先写数据库，再开始轮询。
        print(f"任务 ID：{task_id}")

    def show_status(snapshot: TaskSnapshot) -> None:
        print(f"任务状态：{snapshot.status}")

    try:
        with ImgApiClient() as client:
            result = client.generate_image(
                model="gpt-image-2",
                prompt="一只戴着墨镜的柴犬坐在沙滩上喝可乐，赛博朋克风格",
                aspect_ratio="1:1",
                quality="auto",
                resolution="1K",
                urls=[],
                on_task_created=save_task_id,
                on_status=show_status,
            )

        print("生成成功：")
        for image_url in result.image_urls:
            print(image_url)
    except SubmissionUncertainError as exc:
        print(f"提交结果不确定：{exc}")
        raise SystemExit(2) from exc
    except ImgApiError as exc:
        print(f"生成失败：{exc}")
        if exc.task_id:
            print(f"请保留任务 ID：{exc.task_id}")
        raise SystemExit(1) from exc


if __name__ == "__main__":
    main()