video-workflow/cloud-functions/06-voiceManager.js

/**
 * 云函数：voiceManager（音色档案 + speaker_id 池）
 * 替代：
 *   POST /api/voice/auto-speaker-id   → action=autoSpeakerId
 *   POST /api/voice-profiles/sync     → action=syncProfile
 * 新增：
 *   action=listProfiles  列出所有音色档案
 *   action=initPool      初始化 speaker_id 池（首次部署用）
 *
 * 注意：speaker_id 池原本在 docs/音色创建/speaker_id.md 文件里。
 * 迁移到云函数后改为存储在 "VoiceSpeakerPool" 表。
 * 首次部署后请用 action=initPool 把现有池一次性灌入。
 */
const VOICE_TOKEN = readEnv('VOICE_TOKEN') || 'Bearer r:f0333969e312a40e4703e8fe4ed1c600';
const VOICE_TTS_BASE_URL = readEnv('VOICE_TTS_BASE_URL') || 'https://server.fmode.cn/api/volcengine/tts';
const VOICE_MAX_ATTEMPTS = Math.max(1, Number(readEnv('VOICE_MAX_ATTEMPTS') || 3));
const {
  requireSession,
  requireAdmin,
  assertRequestedUserMatchesSession,
} = require('./_session');

async function handler(request, response) {
  try {
    await Psql.query(`
      CREATE TABLE IF NOT EXISTS "VoiceProfile" (
        "objectId"   VARCHAR(50) PRIMARY KEY,
        "timbreId"   VARCHAR(255),
        "speakerId"  VARCHAR(255),
        "name"       VARCHAR(255) DEFAULT '',
        "data"       JSONB NOT NULL DEFAULT '{}',
        "userId"     VARCHAR(255) DEFAULT '',
        "occupied"   BOOLEAN DEFAULT TRUE,
        "createdAt"  TIMESTAMPTZ DEFAULT NOW(),
        "updatedAt"  TIMESTAMPTZ DEFAULT NOW()
      )
    `);
    await Psql.query(`DROP INDEX IF EXISTS idx_voice_timbre`);
    await Psql.query(`CREATE UNIQUE INDEX IF NOT EXISTS idx_voice_user_timbre ON "VoiceProfile" ("userId", "timbreId") WHERE "timbreId" IS NOT NULL AND "timbreId" <> ''`);
    await Psql.query(`CREATE INDEX IF NOT EXISTS idx_voice_speaker ON "VoiceProfile" ("speakerId")`);

    await Psql.query(`
      CREATE TABLE IF NOT EXISTS "VoiceSpeakerPool" (
        "speakerId"  VARCHAR(255) PRIMARY KEY,
        "trained"    BOOLEAN DEFAULT FALSE,
        "createdAt"  TIMESTAMPTZ DEFAULT NOW()
      )
    `);

    const action = pickParam(request, 'action') || 'listProfiles';
    const session = action === 'initPool'
      ? await requireAdmin(request, Psql)
      : await requireSession(request, Psql);
    assertRequestedUserMatchesSession(request, session);
    const userId = session.userId;

    if (action === 'autoSpeakerId') {
      const used = await Psql.query(`
        SELECT s."speakerId" FROM "VoiceSpeakerPool" s
        LEFT JOIN "VoiceProfile" v ON v."speakerId" = s."speakerId"
        WHERE v."speakerId" IS NULL AND s."trained" = FALSE
        ORDER BY s."createdAt" ASC
        LIMIT 1
      `);
      if (!used.length) {
        return response.json({ code: 500, success: false, error: 'speaker_id 池已耗尽，请先 initPool 添加' });
      }
      response.json({ code: 200, success: true, data: { speakerId: used[0].speakerId } });
      return;
    }

    if (action === 'syncProfile') {
      const body = pickParam(request, 'profile', 'data') || request.body || {};
      const timbreId = body.timbre_id || body.timbreId || '';
      const speakerId = body.speaker_id || body.speakerId || '';
      if (!timbreId && !speakerId) {
        return response.json({ code: 400, success: false, error: '缺少 timbre_id 或 speaker_id' });
      }
      const now = new Date().toISOString();
      const merged = { ...body, timbre_id: timbreId, speaker_id: speakerId, updated_at: now };

      let existing = [];
      if (timbreId) {
        existing = await Psql.query(`SELECT * FROM "VoiceProfile" WHERE "timbreId" = $1 AND "userId" = $2 LIMIT 1`, [timbreId, userId]);
      }
      if (!existing.length && speakerId) {
        existing = await Psql.query(`SELECT * FROM "VoiceProfile" WHERE "speakerId" = $1 AND "userId" = $2 LIMIT 1`, [speakerId, userId]);
      }

      if (existing.length) {
        const old = typeof existing[0].data === 'string' ? JSON.parse(existing[0].data) : (existing[0].data || {});
        const merged2 = { ...old, ...merged };
        await Psql.query(
          `UPDATE "VoiceProfile" SET "timbreId"=$1, "speakerId"=$2, "name"=$3, "data"=$4, "occupied"=$5, "updatedAt"=NOW() WHERE "objectId"=$6`,
          [timbreId, speakerId, body.name || merged2.name || '', JSON.stringify(merged2), body.occupied !== false, existing[0].objectId]
        );
        response.json({ code: 200, success: true, data: { ...merged2, objectId: existing[0].objectId } });
      } else {
        merged.created_at = now;
        const objectId = generateId();
        await Psql.query(
          `INSERT INTO "VoiceProfile" ("objectId","timbreId","speakerId","name","data","userId","occupied")
           VALUES ($1,$2,$3,$4,$5,$6,$7)`,
          [objectId, timbreId, speakerId, body.name || '', JSON.stringify(merged), userId, body.occupied !== false]
        );
        response.json({ code: 200, success: true, data: { ...merged, objectId } });
      }
      return;
    }

    if (action === 'listProfiles') {
      const params = [userId];
      const rows = await Psql.query(
        `SELECT * FROM "VoiceProfile" WHERE "userId" = $1 ORDER BY "createdAt" DESC LIMIT 1000`,
        params
      );
      response.json({ code: 200, success: true, data: rows.map(rowToObj) });
      return;
    }

    if (action === 'initPool') {
      const ids = pickParam(request, 'speakerIds') || [];
      const list = Array.isArray(ids) ? ids : String(ids || '').split(/[\s,]+/).filter(Boolean);
      if (!list.length) {
        return response.json({ code: 400, success: false, error: '缺少 speakerIds（数组或逗号分隔字符串）' });
      }
      let inserted = 0;
      for (const sid of list) {
        const r = await Psql.query(
          `INSERT INTO "VoiceSpeakerPool" ("speakerId") VALUES ($1) ON CONFLICT DO NOTHING RETURNING "speakerId"`,
          [String(sid).trim()]
        );
        if (r.length) inserted++;
      }
      response.json({ code: 200, success: true, data: { total: list.length, inserted } });
      return;
    }

    if (action === 'synthesize') {
      const payload = pickParam(request, 'payload', 'data') || {};
      if (!payload.text && !payload.ssml) {
        return response.json({ code: 400, success: false, error: '缺少合成文本' });
      }
      const upstreamPayload = normalizeSynthesisPayload(payload);
      const res = await postJson(`${VOICE_TTS_BASE_URL}/unidirectional`, upstreamPayload);
      response.json(res);
      return;
    }

    if (action === 'cloneVoice') {
      const payload = pickParam(request, 'payload', 'data') || {};
      if (!payload.name || !payload.speaker_id || !payload.audioData) {
        return response.json({ code: 400, success: false, error: '缺少声音训练参数' });
      }
      const upstreamPayload = stripEmpty({
        ...payload,
        token: VOICE_TOKEN,
      });
      const res = await postJson(`${VOICE_TTS_BASE_URL}/voice_clone`, upstreamPayload);
      response.json(res);
      return;
    }

    if (action === 'deleteProfile') {
      const objectId = pickParam(request, 'objectId');
      const timbreId = pickParam(request, 'timbreId');
      if (!objectId && !timbreId) {
        return response.json({ code: 400, success: false, error: '缺少 objectId 或 timbreId' });
      }
      const r = objectId
        ? await Psql.query(`DELETE FROM "VoiceProfile" WHERE "objectId" = $1 AND "userId" = $2 RETURNING "objectId"`, [objectId, userId])
        : await Psql.query(`DELETE FROM "VoiceProfile" WHERE "timbreId" = $1 AND "userId" = $2 RETURNING "objectId"`, [timbreId, userId]);
      response.json({ code: 200, success: true, data: { deleted: r.length > 0 } });
      return;
    }

    response.json({ code: 400, success: false, error: `未知 action: ${action}` });
  } catch (error) {
    console.error('❌ voiceManager 失败:', error.message);
    response.json({ code: error.status || 500, success: false, error: error.message });
  }
}

function normalizeSynthesisPayload(input) {
  const payload = { ...(input || {}) };
  const officialVoiceType = String(payload.volcengine_voice_type || payload.voiceType || payload.voice_type || '').trim();
  if (officialVoiceType && !payload.speaker_id && !payload.speakerId && !payload.timbreId && !payload.timbre_id) {
    payload.speaker_id = officialVoiceType;
  }
  if (payload.timbre_id && !payload.timbreId) {
    payload.timbreId = payload.timbre_id;
  }

  return stripEmpty({
    ...payload,
    token: VOICE_TOKEN,
    isStream: payload.isStream === true,
  });
}

function rowToObj(row) {
  const data = typeof row.data === 'string' ? JSON.parse(row.data) : (row.data || {});
  return { ...data, objectId: row.objectId, occupied: row.occupied, createdAt: row.createdAt, updatedAt: row.updatedAt };
}

function pickParam(request, ...names) {
  const sources = [request.params, request.body, request];
  for (const src of sources) {
    if (!src || typeof src !== 'object') continue;
    for (const n of names) {
      const v = src[n];
      if (v !== undefined && v !== null && v !== '') return v;
    }
  }
  return null;
}

function generateId() {
  const chars = 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789';
  let s = '';
  for (let i = 0; i < 10; i++) s += chars.charAt(Math.floor(Math.random() * chars.length));
  return s;
}

function stripEmpty(value) {
  if (!value || typeof value !== 'object') return value;
  const out = Array.isArray(value) ? [] : {};
  for (const [key, val] of Object.entries(value)) {
    if (val === undefined || val === null || val === '') continue;
    if (val && typeof val === 'object' && !Array.isArray(val)) {
      const nested = stripEmpty(val);
      if (Object.keys(nested).length) out[key] = nested;
    } else {
      out[key] = val;
    }
  }
  return out;
}

async function postJson(url, body) {
  let lastError = null;

  for (let attempt = 1; attempt <= VOICE_MAX_ATTEMPTS; attempt += 1) {
    try {
      const { status, ok, rawText, data, transport } = await requestJson('POST', url, {
        headers: { 'Content-Type': 'application/json', 'Accept': 'application/json' },
        body: JSON.stringify(body),
      });
      const code = Number(data && data.code ? data.code : status || 0);
      const retryable = !ok || code === 408 || code === 429 || code >= 500;

      if (!retryable || attempt >= VOICE_MAX_ATTEMPTS) {
        if (!ok) {
          return {
            code: status || code || 500,
            success: false,
            error: readError(data, rawText) || `HTTP ${status}`,
            upstream: { transport, attempt, maxAttempts: VOICE_MAX_ATTEMPTS },
          };
        }
        return data || { code: 500, success: false, error: '服务返回异常' };
      }
    } catch (error) {
      lastError = error;
      if (attempt >= VOICE_MAX_ATTEMPTS) break;
    }

    await sleep(Math.min(12000, 1200 * attempt * attempt));
  }

  return {
    code: 500,
    success: false,
    error: `TTS upstream request failed: ${formatFetchError(lastError)}`,
  };
}

async function requestJson(method, url, init) {
  let lastError = null;

  if (typeof fetch === 'function') {
    try {
      const response = await fetch(url, { method, ...(init || {}) });
      return readResponse(response, 'fetch');
    } catch (error) {
      lastError = error;
    }
  }

  if (typeof require === 'function') {
    try {
      return await requestJsonWithNodeHttps(method, url, init);
    } catch (error) {
      lastError = error;
    }
  }

  if (typeof XMLHttpRequest !== 'undefined') {
    try {
      return await requestJsonWithXhr(method, url, init);
    } catch (error) {
      lastError = error;
    }
  }

  throw lastError || new Error('No HTTP client is available in this cloud runtime');
}

async function readResponse(response, transport) {
  const rawText = await response.text();
  let data = null;
  try { data = rawText ? JSON.parse(rawText) : null; } catch {}
  return {
    status: response.status,
    ok: !!response.ok,
    rawText,
    data,
    transport,
  };
}

function requestJsonWithNodeHttps(method, url, init) {
  return new Promise((resolve, reject) => {
    try {
      const parsed = new URL(url);
      const lib = parsed.protocol === 'http:' ? require('http') : require('https');
      const headers = init && init.headers ? init.headers : {};
      const body = init && init.body ? init.body : '';
      const req = lib.request({
        method,
        protocol: parsed.protocol,
        hostname: parsed.hostname,
        port: parsed.port,
        path: `${parsed.pathname}${parsed.search}`,
        headers: body ? { ...headers, 'Content-Length': Buffer.byteLength(body) } : headers,
      }, (res) => {
        const chunks = [];
        res.on('data', (chunk) => chunks.push(chunk));
        res.on('end', () => {
          const rawText = Buffer.concat(chunks).toString('utf8');
          let data = null;
          try { data = rawText ? JSON.parse(rawText) : null; } catch {}
          resolve({
            status: res.statusCode || 0,
            ok: res.statusCode >= 200 && res.statusCode < 300,
            rawText,
            data,
            transport: 'node-https',
          });
        });
      });
      req.on('error', reject);
      if (body) req.write(body);
      req.end();
    } catch (error) {
      reject(error);
    }
  });
}

function requestJsonWithXhr(method, url, init) {
  return new Promise((resolve, reject) => {
    const xhr = new XMLHttpRequest();
    xhr.open(method, url, true);
    const headers = init && init.headers ? init.headers : {};
    for (const [key, value] of Object.entries(headers)) {
      xhr.setRequestHeader(key, value);
    }
    xhr.onreadystatechange = function onReadyStateChange() {
      if (xhr.readyState !== 4) return;
      let data = null;
      try { data = xhr.responseText ? JSON.parse(xhr.responseText) : null; } catch {}
      resolve({
        status: xhr.status,
        ok: xhr.status >= 200 && xhr.status < 300,
        rawText: xhr.responseText || '',
        data,
        transport: 'xhr',
      });
    };
    xhr.onerror = function onXhrError() {
      reject(new Error('XMLHttpRequest failed'));
    };
    xhr.send(init && init.body ? init.body : null);
  });
}

function readError(data, fallback) {
  return data?.error?.message || data?.error || data?.message || data?.msg || fallback || '';
}

function formatFetchError(error) {
  if (!error) return 'unknown error';
  const message = error.message || String(error);
  const cause = error.cause ? `; cause=${error.cause.code || error.cause.message || error.cause}` : '';
  return `${message}${cause}`;
}

function sleep(ms) {
  return new Promise((resolve) => setTimeout(resolve, ms));
}

function readEnv(name) {
  if (typeof process !== 'undefined' && process.env && process.env[name]) {
    return process.env[name];
  }
  return '';
}