aiartstudio/server/utils/openai.ts

// server/utils/openai.ts - OpenAI/NewAPI 调用工具：文本、视觉、Responses 流式和 Chat Completions 流式生图。
import OpenAI from "openai";
import type { BaseOptions, IImageGenerateData } from "#shared/types/openai";

/** 图像生成改走 Chat Completions 流式接口，通常能更快拿到中转平台返回的图片地址 */
const CHAT_COMPLETIONS_URL = "https://api.qflink.xyz/v1/chat/completions";
const IMAGE_GENERATION_MODEL = "gpt-image-2";
const DEFAULT_MAX_SSE_DATA_LINE_BYTES = 1024 * 1024;
const textEncoder = new TextEncoder();

/** Chat Completions SSE 每个 data chunk 的最小结构 */
interface IChatCompletionStreamChunk {
  id?: string;
  object?: string;
  created?: number;
  model?: string;
  metadata?: unknown;
  choices?: Array<{
    delta?: {
      content?: string;
      role?: string;
    };
    finish_reason?: string | null;
    index?: number;
  }>;
  usage?: unknown;
}

/** 聚合后的流式生图上游响应，供服务端入库排查使用，不包含 API Key */
interface IImageStreamUpstreamResponse {
  content: string;
  imageUrl?: string;
  contentLength: number;
  chunkCount: number;
  finishReason?: string | null;
  usage?: unknown;
}

export interface IAskImageResult extends IImageGenerateData {
  /** 完整上游生图接口返回结果，仅服务端内部保存 */
  upstreamResponse: unknown;
}

/** 通用 AI 调用函数（支持文本 / 图文 / 多模态） */
export const askAI = async ({
  apiKey,
  model,
  baseURL,
  input
}: BaseOptions & {
  input: OpenAI.Responses.ResponseCreateParams["input"];
}) => {
  const client = new OpenAI({ apiKey, baseURL });

  return await client.responses.create({
    model,
    input
  });
};

/** 调用文本模型 */
export const askText = async (
  options: BaseOptions & {
    text: string;
  }
) => {
  const res = await askAI({
    ...options,
    input: options.text
  });

  return res.output_text;
};

/**
 * 调用视觉模型：输入「文本 + 图片」，输出文本
 *
 * 分辨率参数说明：
 *
 * | 值   | 含义                     |
 * |------|--------------------------|
 * | low  | 低分辨率（更快、更省成本） |
 * | high | 高分辨率（更精准）         |
 * | auto | 自动选择                 |
 */
export const askVision = async (
  options: BaseOptions & {
    text: string;
    image: string;
    detail?: "low" | "high" | "auto";
  }
) => {
  // 如果图片不是 URL，也不是 data URL，就当成 base64 处理，自动补全 data URL 前缀
  const normalizeImage = (img: string) => {
    if (img.startsWith("http")) return img;
    if (!img.startsWith("data:")) {
      return `data:image/png;base64,${img}`;
    }
    return img;
  };

  const res = await askAI({
    ...options,
    input: [
      {
        role: "user",
        content: [
          {
            type: "input_text",
            text: options.text
          },
          {
            type: "input_image",
            image_url: normalizeImage(options.image),
            detail: options.detail || "auto"
          }
        ]
      }
    ]
  });

  return res.output_text;
};

/** 调用流式图片生成接口，完整 API Key 只在服务端使用 */
export const askImgStream = async ({
  apiKey,
  prompt
}: {
  apiKey: string;
  prompt: string;
}): Promise<IAskImageResult> => {
  const response = await fetch(CHAT_COMPLETIONS_URL, {
    method: "POST",
    headers: {
      Authorization: `Bearer ${apiKey}`,
      "Content-Type": "application/json"
    },
    body: JSON.stringify({
      model: IMAGE_GENERATION_MODEL,
      stream: true,
      messages: [
        {
          role: "user",
          content: prompt
        }
      ]
    })
  });

  if (!response.ok) {
    const message = await response.text().catch(() => "");
    throw new Error(message || `图片生成失败：${response.status}`);
  }

  if (!response.body) {
    throw new Error("图片生成失败：上游没有返回流");
  }

  const upstreamResponse = await readChatCompletionStream(response.body);
  const imageUrl = extractImageUrlFromStreamText(upstreamResponse.content);

  return {
    imageUrl,
    revisedPrompt: undefined,
    upstreamResponse: {
      ...upstreamResponse,
      imageUrl
    }
  };
};

/** 从流式累积文本中提取图片地址，兼容 Markdown 图片、常见图片 URL 和无扩展下载链接 */
export const extractImageUrlFromStreamText = (text: string): string => {
  const markdownImageMatch = text.match(/!\[[^\]]*]\((https?:\/\/[^)\s]+)\)/i);
  if (markdownImageMatch?.[1]) {
    return cleanupImageUrl(markdownImageMatch[1]);
  }

  const imageUrlMatch = text.match(
    /https?:\/\/[^\s)>'"]+(?:\.(?:png|jpe?g|webp|gif)(?:\?[^\s)>'"]*)?|\/file_download\/[^\s)>'"]+)/i
  );
  if (imageUrlMatch?.[0]) {
    return cleanupImageUrl(imageUrlMatch[0]);
  }

  const urlMatch = text.match(/https?:\/\/[^\s)>'"]+/i);
  if (urlMatch?.[0]) {
    return cleanupImageUrl(urlMatch[0]);
  }

  throw new Error("图片生成失败：未找到图片地址");
};

/** 调用 Responses API 流式接口，保留给其他文本/多模态场景复用 */
export const askStream = async (
  options: BaseOptions & {
    input: OpenAI.Responses.ResponseCreateParams["input"];
  }
) => {
  const client = new OpenAI({
    apiKey: options.apiKey,
    baseURL: options.baseURL
  });

  const stream = await client.responses.stream({
    model: options.model,
    input: options.input
  });

  return stream;
};

/** 读取 Chat Completions SSE 流，累积 delta.content 并提取 usage/chunk 元信息 */
const readChatCompletionStream = async (
  stream: ReadableStream<Uint8Array>
): Promise<IImageStreamUpstreamResponse> => {
  const reader = stream.getReader();
  const decoder = new TextDecoder();
  const maxDataLineBytes = getPositiveIntegerEnv(
    "IMAGE_STREAM_MAX_SSE_LINE_BYTES",
    DEFAULT_MAX_SSE_DATA_LINE_BYTES
  );
  let buffer = "";
  let content = "";
  let chunkCount = 0;
  let finishReason: string | null | undefined;
  let usage: unknown;

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;

    buffer += decoder.decode(value, { stream: true });
    assertSseBufferSize(buffer, maxDataLineBytes);
    const lines = buffer.split(/\r?\n/);
    buffer = lines.pop() ?? "";

    for (const line of lines) {
      const chunk = parseSseDataLine(line, maxDataLineBytes);
      if (!chunk) continue;
      chunkCount += 1;

      const deltaContent = collectDeltaContent(chunk);
      if (deltaContent) {
        content += deltaContent;
      }

      if (chunk.usage) {
        usage = chunk.usage;
      }

      finishReason =
        chunk.choices?.find((choice) => choice.finish_reason)
          ?.finish_reason ?? finishReason;
    }
  }

  const finalText = buffer + decoder.decode();
  for (const line of finalText.split(/\r?\n/)) {
    const chunk = parseSseDataLine(line, maxDataLineBytes);
    if (!chunk) continue;
    chunkCount += 1;

    const deltaContent = collectDeltaContent(chunk);
    if (deltaContent) {
      content += deltaContent;
    }

    if (chunk.usage) {
      usage = chunk.usage;
    }

    finishReason =
      chunk.choices?.find((choice) => choice.finish_reason)?.finish_reason ??
      finishReason;
  }

  return {
    content,
    contentLength: content.length,
    chunkCount,
    finishReason,
    usage
  };
};

/** 解析单行 SSE data，跳过空行和 [DONE] */
const parseSseDataLine = (
  line: string,
  maxDataLineBytes: number
): IChatCompletionStreamChunk | null => {
  const trimmed = line.trim();
  if (!trimmed.startsWith("data:")) return null;

  const data = trimmed.slice("data:".length).trim();
  if (!data || data === "[DONE]") return null;

  if (textEncoder.encode(data).byteLength > maxDataLineBytes) {
    throw new Error("图片生成失败：流式响应过大");
  }

  return JSON.parse(data) as IChatCompletionStreamChunk;
};

/** 收集一个 chunk 中所有 choice 的 delta.content */
const collectDeltaContent = (chunk: IChatCompletionStreamChunk) => {
  return (
    chunk.choices
      ?.map((choice) => choice.delta?.content || "")
      .filter(Boolean)
      .join("") || ""
  );
};

/** 清理模型文本里 URL 后面可能粘上的句末标点 */
const cleanupImageUrl = (url: string) => {
  return url.replace(/[,.!?，。！？]+$/u, "");
};

/** 限制尚未切分出换行的 SSE 缓冲区，避免异常长 data 行持续堆在内存里 */
const assertSseBufferSize = (buffer: string, maxDataLineBytes: number) => {
  if (
    !buffer.includes("\n") &&
    textEncoder.encode(buffer).byteLength > maxDataLineBytes
  ) {
    throw new Error("图片生成失败：流式响应过大");
  }
};

const getPositiveIntegerEnv = (name: string, fallback: number) => {
  const value = Number.parseInt(process.env[name] ?? "", 10);
  return Number.isInteger(value) && value > 0 ? value : fallback;
};