import type { IncomingMessage, ServerResponse } from "node:http"; import { EdgeTTS } from "node-edge-tts"; import { speechVoiceIdSchema, type SpeechVoice } from "../src/shared/tts/voices"; import { createEdgeTtsVoiceCatalog } from "./edge-tts-voices"; const EDGE_TTS_PATH = "/api/tts/edge"; const EDGE_TTS_VOICES_PATH = `${EDGE_TTS_PATH}/voices`; const DEFAULT_VOICE = "zh-CN-XiaoxiaoNeural"; const MAX_REQUEST_BYTES = 8 * 1024; const MAX_TEXT_LENGTH = 520; const MAX_AUDIO_BYTES = 5 * 1024 * 1024; const DEFAULT_MAX_CONCURRENT = 8; const DEFAULT_RATE_LIMIT = 60; const DEFAULT_RATE_WINDOW_MS = 60_000; const DEFAULT_TIMEOUT_MS = 30_000; interface EdgeTtsPayload { text?: unknown; voice?: unknown; } interface RateBucket { count: number; startedAt: number; } export type EdgeTtsSynthesizer = (text: string, voice: string) => Promise; export type EdgeTtsMiddlewareOptions = { defaultVoice?: string; maxConcurrent?: number; maxRequestsPerWindow?: number; now?: () => number; rateWindowMs?: number; synthesize?: EdgeTtsSynthesizer; timeoutMs?: number; trustProxy?: boolean; listVoices?: () => Promise; }; function sendJson(response: ServerResponse, status: number, message: string) { response.statusCode = status; response.setHeader("Content-Type", "application/json; charset=utf-8"); response.setHeader("Cache-Control", "no-store"); response.end(JSON.stringify({ error: message })); } async function readJson(request: IncomingMessage): Promise { const chunks: Buffer[] = []; let byteLength = 0; for await (const chunk of request) { const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk); byteLength += buffer.byteLength; if (byteLength > MAX_REQUEST_BYTES) { throw new RangeError("请求内容过大"); } chunks.push(buffer); } const payload = JSON.parse(Buffer.concat(chunks).toString("utf8")) as unknown; if (!payload || typeof payload !== "object" || Array.isArray(payload)) { throw new SyntaxError("请求内容必须是 JSON 对象"); } return payload as EdgeTtsPayload; } function isLoopback(address: string | undefined) { return address === "127.0.0.1" || address === "::1" || address === "::ffff:127.0.0.1"; } function getClientKey(request: IncomingMessage, trustProxy: boolean) { const remoteAddress = request.socket.remoteAddress; if (trustProxy && isLoopback(remoteAddress)) { const forwardedFor = request.headers["x-forwarded-for"]; const firstAddress = Array.isArray(forwardedFor) ? forwardedFor[0] : forwardedFor?.split(",", 1)[0]; if (firstAddress?.trim()) return firstAddress.trim(); } return remoteAddress || "unknown"; } export async function synthesizeEdgeSpeech(text: string, voice: string): Promise { const communicate = new EdgeTTS({ voice, timeout: DEFAULT_TIMEOUT_MS }); const chunks: Uint8Array[] = []; let byteLength = 0; for await (const chunk of communicate.streamAudio(text)) { if (!Buffer.isBuffer(chunk)) continue; chunks.push(chunk); byteLength += chunk.byteLength; if (byteLength > MAX_AUDIO_BYTES) throw new Error("语音服务返回的音频过大"); } if (!byteLength) throw new Error("语音服务返回了空音频"); const audio = new Uint8Array(byteLength); let offset = 0; for (const chunk of chunks) { audio.set(chunk, offset); offset += chunk.byteLength; } return audio; } export function createEdgeTtsMiddleware(options: EdgeTtsMiddlewareOptions = {}) { const synthesize = options.synthesize ?? synthesizeEdgeSpeech; const configuredVoice = options.defaultVoice ?? process.env.EDGE_TTS_VOICE ?? DEFAULT_VOICE; const defaultVoice = speechVoiceIdSchema.safeParse(configuredVoice).success ? configuredVoice : DEFAULT_VOICE; const maxConcurrent = options.maxConcurrent ?? DEFAULT_MAX_CONCURRENT; const maxRequestsPerWindow = options.maxRequestsPerWindow ?? DEFAULT_RATE_LIMIT; const rateWindowMs = options.rateWindowMs ?? DEFAULT_RATE_WINDOW_MS; const timeoutMs = options.timeoutMs ?? DEFAULT_TIMEOUT_MS; const trustProxy = options.trustProxy ?? false; const now = options.now ?? Date.now; const listVoices = createEdgeTtsVoiceCatalog(options.listVoices, now); const rateBuckets = new Map(); let activeRequests = 0; return async (request: IncomingMessage, response: ServerResponse, next?: () => void) => { const path = request.url?.split("?", 1)[0]; const voicesRequest = path === EDGE_TTS_VOICES_PATH; if (path !== EDGE_TTS_PATH && !voicesRequest) { next?.(); return; } const method = voicesRequest ? "GET" : "POST"; if (request.method !== method) { response.setHeader("Allow", method); sendJson(response, 405, `仅支持 ${method} 请求`); return; } if (!voicesRequest && !request.headers["content-type"]?.toLowerCase().startsWith("application/json")) { sendJson(response, 415, "请求必须使用 application/json"); return; } const currentTime = now(); const clientKey = getClientKey(request, trustProxy); const existingBucket = rateBuckets.get(clientKey); const bucket = !existingBucket || currentTime - existingBucket.startedAt >= rateWindowMs ? { count: 0, startedAt: currentTime } : existingBucket; bucket.count += 1; rateBuckets.set(clientKey, bucket); if (rateBuckets.size > 1_024) { for (const [key, value] of rateBuckets) { if (currentTime - value.startedAt >= rateWindowMs) rateBuckets.delete(key); } } if (bucket.count > maxRequestsPerWindow) { response.setHeader("Retry-After", String(Math.max(1, Math.ceil(rateWindowMs / 1_000)))); sendJson(response, 429, "语音请求过于频繁,请稍后重试"); return; } if (voicesRequest) { try { const voices = await listVoices(); response.statusCode = 200; response.setHeader("Content-Type", "application/json; charset=utf-8"); response.setHeader("Cache-Control", "no-store"); response.end(JSON.stringify({ voices, defaultVoice })); } catch { sendJson(response, 503, "音色列表暂时不可用,请稍后重试"); } return; } let payload: EdgeTtsPayload; try { payload = await readJson(request); } catch (error) { sendJson(response, error instanceof RangeError ? 413 : 400, "请求内容不是有效 JSON"); return; } const text = typeof payload.text === "string" ? payload.text.trim() : ""; if (!text) { sendJson(response, 400, "缺少待朗读文本"); return; } if (text.length > MAX_TEXT_LENGTH) { sendJson(response, 413, `单次文本不能超过 ${MAX_TEXT_LENGTH} 字`); return; } let voice = defaultVoice; if (Object.prototype.hasOwnProperty.call(payload, "voice")) { const selectedVoice = speechVoiceIdSchema.safeParse(payload.voice); if (!selectedVoice.success) { sendJson(response, 400, "请选择有效音色"); return; } try { if (!(await listVoices()).some((item) => item.id === selectedVoice.data)) { sendJson(response, 400, "所选音色不可用,请重新选择"); return; } voice = selectedVoice.data; } catch { sendJson(response, 503, "无法确认音色是否可用,请稍后重试"); return; } } if (activeRequests >= maxConcurrent) { response.setHeader("Retry-After", "1"); sendJson(response, 429, "语音服务繁忙,请稍后重试"); return; } activeRequests += 1; const synthesisTask = Promise.resolve().then(() => synthesize(text, voice)); void synthesisTask.finally(() => { activeRequests -= 1; }).catch(() => undefined); let timeout: ReturnType | undefined; try { const audio = await Promise.race([ synthesisTask, new Promise((_, reject) => { timeout = setTimeout(() => reject(new DOMException("语音服务响应超时", "TimeoutError")), timeoutMs); }) ]); if (!audio.byteLength) { sendJson(response, 502, "语音服务返回了空音频"); return; } if (audio.byteLength > MAX_AUDIO_BYTES) { sendJson(response, 502, "语音服务返回的音频过大"); return; } response.statusCode = 200; response.setHeader("Content-Type", "audio/mpeg"); response.setHeader("Cache-Control", "no-store"); response.end(Buffer.from(audio)); } catch (error) { if (error instanceof DOMException && error.name === "TimeoutError") { sendJson(response, 504, "语音服务响应超时"); return; } console.error("[Agent TTS] 语音生成失败", error); sendJson(response, 502, "语音生成失败"); } finally { if (timeout) clearTimeout(timeout); } }; }