zhaoshang_client_ui/harness/tools/probe-nls-realtime.mjs

/**
 * 探针：**直接连阿里云 NLS 实时语音识别网关**，把上下行的原始帧打出来看。
 *
 * 为什么要有它：NLS 的实时协议（`ws/v1`）细节以**实测**为准 —— 官方文档在
 * `help.aliyun.com`（本机 WebFetch 被策略挡掉），而自述的帧结构一旦有一处不对，
 * 网关返回的是 `TaskFailed` / 直接断连，光看封装层代码排查不出来
 * （本项目在别处踩过同样的坑：协议问题必须读原始 SSE/WS 载荷）。
 *
 * 怎么跑（凭据从 .env.development.local 读，见 harness/tools/nls-token.mjs）：
 *   node harness/tools/probe-nls-realtime.mjs
 *   node harness/tools/probe-nls-realtime.mjs --audio=<16k单声道16bit的pcm/raw文件>
 *
 * 默认不发音频（只发 start → 等待 → stop），用于确认**握手与事件名**；
 * 带 `--audio=` 时按 ~100ms 一帧推流，用于确认**识别结果**。
 */

import { existsSync } from 'node:fs';
import { randomUUID } from 'node:crypto';
import {
  NLS_LOCAL_ENV_FILE,
  resolveNlsToken,
  maskCredential,
  readDotEnvFile,
  readNlsCredentials,
} from './nls-token.mjs';
import { readAudioAsPcm16kMono, synthesizeSpeechToWav } from './nls-test-audio.mjs';


const credentials = readNlsCredentials({ ...readDotEnvFile(NLS_LOCAL_ENV_FILE), ...process.env });
if (!credentials.ak || !credentials.sk || !credentials.appkey) {
  console.error('缺凭据：ALIYUN_NLS_AK / ALIYUN_NLS_SK / ALIYUN_NLS_APPKEY');
  process.exit(2);
}

const GATEWAY = process.env.NLS_GATEWAY || 'wss://nls-gateway-cn-shanghai.aliyuncs.com/ws/v1';
const audioArg = process.argv.find((a) => a.startsWith('--audio='));
const sayArg = process.argv.find((a) => a.startsWith('--say='));
const waitMs = Number((process.argv.find((a) => a.startsWith('--wait=')) || '').slice(7)) || 4000;
/** 端点静音断句阈值（0 = 用协议层默认 800）。延迟高先怀疑它 */
const silenceArg = process.argv.find((a) => a.startsWith('--silence='));
const maxSentenceSilence = silenceArg ? Number(silenceArg.slice('--silence='.length)) : 0;
/** 语义断句（试它是否能提高中间结果频率） */
const semanticDetection = process.argv.includes('--semantic');
/** 模拟握手期间已攒下的音频（毫秒） */
const prebufferMs = Number((process.argv.find((a) => a.startsWith('--prebuffer=')) || '').slice(12)) || 0;
/** 每个 tick 最多发几帧（100ms 一帧）：1 = 1× 实时，4 = 4× 实时 */
const drainFrames = Number((process.argv.find((a) => a.startsWith('--drain=')) || '').slice(8)) || 1;

const stamp = () => new Date().toISOString().slice(11, 23);
const log = (...args) => console.log(`[${stamp()}]`, ...args);
const raw = (value) => (typeof value === 'string' ? value : `(binary ${value?.byteLength ?? value?.length ?? '?'} bytes)`);

let audioFile = audioArg ? audioArg.slice('--audio='.length) : '';
if (sayArg) {
  const text = sayArg.slice('--say='.length);
  log(`用系统 TTS 合成语音：「${text}」`);
  audioFile = synthesizeSpeechToWav(text);
}
if (audioFile && !existsSync(audioFile)) {
  console.error(`音频文件不存在：${audioFile}`);
  process.exit(2);
}
let audioPcm = null;
if (audioFile) {
  const { pcm, note } = readAudioAsPcm16kMono(audioFile);
  audioPcm = pcm;
  log(`音频：${audioFile}（${audioPcm.length} 字节 = ${(audioPcm.length / 32000).toFixed(2)} 秒）${note}`);
}

const { token, expireAt, source } = await resolveNlsToken();
log(`token 到手（来源=${source}，按 ${new Date(expireAt).toLocaleString('zh-CN')} 过期）appkey=${maskCredential(credentials.appkey)}`);

const url = `${GATEWAY}?token=${encodeURIComponent(token)}`;
log('连接', url.replace(/token=[^&]+/, 'token=****'));

const ws = new WebSocket(url);
const newId = () => randomUUID().replace(/-/g, ''); // 文档要求 32 位十六进制
const startMessageId = newId();
const taskId = newId();
let stopSent = false;
let audioStartAt = 0;      // 第一帧音频发出的时刻
let sentBytes = 0;         // 已推给网关的音频字节数

const send = (payload) => {
  const text = JSON.stringify(payload);
  log('→ 发送', text.length > 400 ? `${text.slice(0, 400)}…` : text);
  ws.send(text);
};

/**
 * 发一条指令。⚠️ **每条指令都要带 appkey**（实测踩过：只有 start 带 appkey 时，
 * stop 会回 `TaskFailed 40000002 Gateway:MESSAGE_INVALID:Missing message appkey!`），
 * 所以这里统一注入 header，杜绝漏字段。
 */
const sendCommand = ({ name, task, payload }) => {
  send({
    header: {
      appkey: credentials.appkey,
      message_id: newId(),
      task_id: taskId,
      namespace: 'SpeechTranscriber',
      task,
      name,
    },
    ...(payload ? { payload } : {}),
  });
};

const finish = (code) => {
  try {
    ws.close();
  } catch {
    /* ignore */
  }
  setTimeout(() => process.exit(code), 200);
};

const timer = setTimeout(() => {
  log('⚠️ 超时退出（网关未在预期时间内给终态）');
  finish(1);
}, 25000);

ws.onopen = () => {
  log('✅ WebSocket 已连接');
  sendCommand({
    name: 'StartTranscription',
    task: 'start',
    payload: {
      format: 'pcm',
      sample_rate: 16000,
      enable_intermediate_result: true,
      enable_punctuation_prediction: true,
      enable_inverse_text_normalization: true,
      enable_voice_detection: true,
      ...(maxSentenceSilence ? { max_sentence_silence: maxSentenceSilence } : {}),
      ...(semanticDetection ? { enable_semantic_sentence_detection: true } : {}),
    },
  });
  if (audioPcm) audioStartAt = Date.now();

  if (audioPcm) {
    // 模拟真实采集：**可用音频 = 握手积压 + 已过时间**（1× 实时到达），
    // 每个 tick 最多排空 drainFrames 帧 —— 这正是前端 nls.ts 的排空策略。
    // `--drain=1` 复现"只按 1× 排空"的旧行为（积压永远排不掉）；
    // `--drain=4` 是现在前端的策略（积压时最多 4× 追）。
    const pcm = audioPcm;
    const frame = 3200; // 100ms
    const bytesPerMs = 32; // 16k/16bit/mono → 32 字节/ms
    const startedAt = Date.now();
    let offset = 0;
    const pump = setInterval(() => {
      const elapsedMs = Date.now() - startedAt;
      const available = Math.min(pcm.length, Math.round((prebufferMs + elapsedMs) * bytesPerMs));
      let sentThisTick = 0;
      while (offset < available && sentThisTick < drainFrames) {
        const end = Math.min(offset + frame, pcm.length, available);
        const slice = pcm.subarray(offset, end);
        offset = end;
        if (ws.readyState === WebSocket.OPEN) {
          ws.send(slice);
          sentBytes += slice.length;
        }
        sentThisTick += 1;
      }
      if (offset >= pcm.length) {
        clearInterval(pump);
        log(`音频推完（${pcm.length} 字节）`);
        if (!stopSent) {
          stopSent = true;
          sendCommand({ name: 'StopTranscription', task: 'stop' });
        }
      }
    }, 100);
  } else {
    setTimeout(() => {
      if (stopSent) return;
      stopSent = true;
      sendCommand({ name: 'StopTranscription', task: 'stop' });
    }, waitMs);
  }
};

ws.onmessage = (event) => {
  const text = typeof event.data === 'string' ? event.data : '';
  log('← 收到', raw(event.data));
  if (text) {
    try {
      const body = JSON.parse(text);
      const name = body?.header?.name || '(无 header.name)';
      const gwMs = Number(body?.payload?.time ?? NaN);
      const sentMs = Math.round((sentBytes / 32000) * 1000);
      // 感知延迟 = 「麦克风已经说了多少」−「网关处理到第几毫秒」。
      // 用采集量（积压 + 已过时间）而不是发送量来算 —— 用户感受到的是前者。
      const capturedMs = audioStartAt ? prebufferMs + (Date.now() - audioStartAt) : sentMs;
      const lag = Number.isFinite(gwMs) ? capturedMs - gwMs : null;
      log(
        `   事件=${name} 网关已处理=${Number.isFinite(gwMs) ? gwMs + 'ms' : '-'} 已推=${sentMs}ms ` +
          `${lag === null ? '' : `**感知延迟=${lag}ms** `}结果=${JSON.stringify(body?.payload?.result ?? '')}`
      );
      if (name === 'TranscriptionCompleted') finish(0);
      if (name === 'TaskFailed') finish(3);
    } catch {
      log('   （不是 JSON）');
    }
  }
};

ws.onerror = (event) => log('❌ WebSocket 错误', event?.message || event?.type || '');
ws.onclose = (event) => {
  clearTimeout(timer);
  log(`连接关闭 code=${event.code} reason=${event.reason || ''}`);
};