会议纪要实时音频转写交互协议（WebSocket）

package nls.alibaba;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.commons.codec.binary.Hex;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.handshake.ServerHandshake;

import java.io.File;
import java.io.IOException;
import java.net.URI;
import java.nio.file.Files;
import java.security.SecureRandom;
import java.util.HashMap;
import java.util.Map;

import static java.lang.Thread.sleep;

public class TingWuMeetingApiUsage extends WebSocketClient {

  private final byte[] audioData;

  private final String taskId;

  private final ObjectMapper objectMapper = new ObjectMapper();

  private boolean audioSent = false;

  private static final int CHUNK_SIZE = 3200;

  private static final long SEND_INTERVAL_MS = 100; // 模拟流式发送间隔

  private static final String apiKey = "替换为您的api-key";

  public TingWuMeetingApiUsage(URI serverUri, String audioFilePath) throws IOException {
    super(serverUri, createHeaders());
    this.audioData = Files.readAllBytes(new File(audioFilePath).toPath());

    // 生成随机16位字符串
    SecureRandom random = new SecureRandom();
    byte[] taskIdBytes = new byte[8];
    random.nextBytes(taskIdBytes);
    this.taskId = Hex.encodeHexString(taskIdBytes);
  }

  private static Map<String, String> createHeaders() {
    Map<String, String> headers = new HashMap<>();
    headers.put("Authorization", "Bearer " + apiKey);
    headers.put("Upgrade", "websocket");
    headers.put("Connection", "Upgrade");
    return headers;
  }

  @Override
  public void onOpen(ServerHandshake serverHandshake) {
    System.out.println("WebSocket connected.");
    System.out.println("Task ID: " + taskId);

    // 构造 run-task 消息
    String runTaskMessage = "{\n" +
      "  \"header\": {\n" +
      "    \"action\": \"run-task\",\n" +
      "    \"task_id\": \"" + taskId + "\",\n" +
      "    \"streaming\": \"duplex\"\n" +
      "  },\n" +
      "  \"payload\": {\n" +
      "    \"model\": \"tingwu-meeting-realtime\",\n" +
      "    \"task_group\": \"aigc\",\n" +
      "    \"task\": \"multimodal-generation\",\n" +
      "    \"function\": \"generation\",\n" +
      "    \"input\": {\n" +
      "      \"appId\": \"替换为您的应用Id\",\n" +
      "      \"dataId\": \"替换为CreateTask中获取的dataId\",\n" +
      "      \"directive\": \"start\"\n" +
      "    }\n" +
      "  }\n" +
      "}";

    System.out.println("Sending run-task: " + runTaskMessage);
    send(runTaskMessage);
  }

  @Override
  public void onMessage(String message) {
    System.out.println("Received text message: " + message);

    try {
      JsonNode root = objectMapper.readTree(message);

            JsonNode header = root.get("header");
            JsonNode payload = root.get("payload");

            if (header != null) {
                String event = header.path("event").asText("");
                // 可以忽略此消息
                if ("task-started".equals(event)) {
                    return;
                }

                String action = payload.path("output").path("action").asText("");
                switch (action) {
                    case "speech-listen":
                        System.out.println("Received speech-listen event, sending full audio...");
                        sendAudioFile();
                        break;
                    case "task-failed":
                        String errorCode = payload.path("output").path("errorCode").asText("N/A");
                        String errorMsg = payload.path("output").path("errorMessage").asText("N/A");
                        System.out.println("Task failed: " + errorCode + " - " + errorMsg);
                        close();
                        break;
                    case "recognize-result":
                        JsonNode output = payload.get("output");
                        System.out.println("Recognition result: " + (output != null ? output.toString() : "N/A"));
                        break;
                    case "ping":
                        // 保活，可忽略
                        System.out.println("Received Ping action, do nothing...");
                        break;
                    case "speech-end":
                        System.out.println("Speech ended. Closing...");
                        close();
                        break;
                    default:
                        System.out.println("Unknown message received: " + message);
                        break;
                }
            } else {
                System.out.println("Unknown message format (no header): " + message);
            }

        } catch (JsonProcessingException e) {
            System.err.println("Invalid JSON received: " + e.getMessage());
            System.out.println("Raw message: " + message);
        }
    }

    @Override
    public void onClose(int code, String reason, boolean b) {
        System.out.println("WebSocket closed. Code: " + code + ", Reason: " + reason);
    }

    @Override
    public void onError(Exception ex) {
        System.err.println("WebSocket error: " + ex.getMessage());
        ex.printStackTrace();
    }

    private void sendAudioFile() {
        if (audioSent) {
            System.out.println("Audio already sent, skipping...");
            return;
        }
        audioSent = true;

        int offset = 0;
        while (offset < audioData.length) {
            int len = Math.min(CHUNK_SIZE, audioData.length - offset);
            byte[] chunk = new byte[len];
            System.arraycopy(audioData, offset, chunk, 0, len);
            try {
                send(chunk);
                offset += len;
                System.out.println("Sent chunk: " + offset + "/" + audioData.length);
                sleep(SEND_INTERVAL_MS);
            } catch (Exception e) {
                System.err.println("Send failed at offset " + offset);
                e.printStackTrace();
                close();
                return;
            }
        }
        System.out.println("Audio sent completely.");
        sendFinishTask();

    }

    // 发送 finish-task 消息
    private void sendFinishTask() {
        String finishTaskMessage = "{\n" +
                "  \"header\": {\n" +
                "    \"action\": \"finish-task\",\n" +
                "    \"task_id\": \"" + taskId + "\",\n" +
                "    \"streaming\": \"duplex\"\n" +
                "  },\n" +
                "  \"payload\": {\n" +
                "    \"model\": \"tingwu-meeting-realtime\",\n" +
                "    \"task_group\": \"aigc\",\n" +
                "    \"task\": \"multimodal-generation\",\n" +
                "    \"function\": \"generation\",\n" +
                "    \"input\": {\n" +
                "      \"directive\": \"stop\"\n" +
                "    }\n" +
                "  }\n" +
                "}";

        System.out.println("Sending finish-task: " + finishTaskMessage);
        send(finishTaskMessage);
    }

    public static void main(String[] args) {
        try {
            URI wsUri = new URI("wss://dashscope.aliyuncs.com/api-ws/v1/inference");
            String audioPath = "替换为您的音频路径";
            TingWuMeetingApiUsage client = new TingWuMeetingApiUsage(wsUri, audioPath);
            client.connect();
        } catch (Exception e) {
            System.err.println("Initialization error: " + e.getMessage());
            e.printStackTrace();
        }
    }

}

const WebSocket = require('ws');
const fs = require('fs');
const path = require('path');
const crypto = require('crypto');

const wsUrl = "wss://dashscope.aliyuncs.com/api-ws/v1/inference";
const audioFilePath = path.resolve('请替换为你自己的音频文件路径');

// 创建WebSocket连接
console.log('开始连接WebSocket...');
const ws = new WebSocket(wsUrl, {
  headers: {
    'Authorization': 'Bearer ' + '请替换为您的api-key',
    'Upgrade': 'websocket',
    'Connection': 'Upgrade'
  }
});

// 生成随机16位taskId
ws.taskId = crypto.randomBytes(8).toString('hex');

ws.on('open', () => {
  console.log('WebSocket connected.' );
  console.log(ws.taskId);

  // 构造run-task消息
  const runTaskMessage = {
    header: {
      action: 'run-task',
      task_id: ws.taskId,
      streaming: 'duplex'
    },
    payload: {
      model: 'tingwu-meeting-realtime',
      task_group: 'aigc',
      task: "multimodal-generation",
      function: "generation",
      input: {
        appId: '请替换为您的应用Id',
        dataId: '替换为CreateTask中获取的dataId',
        directive: 'start'
      }
    }
  };

  console.log('发送run-task指令:', runTaskMessage);
  ws.send(JSON.stringify(runTaskMessage), { binary: false }, (err) => {
    if (err) {
      console.error('发送 run-task 失败:', err);
      return;
    }
    console.log(`任务ID ${ws.taskId}：已发送 run-task 指令`);
  });
});

// 处理服务端消息
ws.on('message', (data) => {
  let message;
  try {
    message = JSON.parse(data.toString());
  } catch (e) {
    console.warn('无法解析非JSON消息:', data);
    return;
  }
  console.log('解析后的 message:', message);

  // 处理不同事件
  let action;

  if (message.payload && message.payload.output && message.payload.output.action) {
    action = message.payload.output.action;
  }
  switch(action) {
    case 'speech-listen':
      console.log('收到 speech-listen 事件，开始发送音频数据, id:', message.payload.output.dataId);
      sendAudioFile(ws, audioFilePath);
      break;

    case 'recognize-result':
      console.log('识别结果:', message.payload.output);
      break;

    case 'ping':
      console.log('收到 ping 事件，无需处理');
      break;

    case 'speech-end':
      console.log('收到 speech-end 事件，关闭连接');
      ws.close();
      break;

    case 'task-failed':
      console.log('收到 task-failed 事件，错误为' + message.payload.output.errorCode + ' ' +
                  message.payload.output.errorMessage + '关闭连接');
      ws.close();
      break;

    default:
      console.log('收到未知事件:', message);
  }
});

let once = false;

// 发送音频文件
function sendAudioFile(socket, filePath) {
  if (once) {
    console.log('已经发送过音频文件，跳过');
    return;
  }
  once = true;

  const CHUNK_SIZE = 3200;
  let buffer;

  try {
    buffer = fs.readFileSync(filePath);
  } catch (err) {
    console.error(`读取音频文件失败: ${filePath}`, err);
    return;
  }

  let offset = 0;

  function sendChunk() {
    if (offset >= buffer.length) {
      console.log('音频文件发送完成，发送 finish-task');
      sendFinishTask(socket);
      return;
    }

    const end = Math.min(offset + CHUNK_SIZE, buffer.length);
        const chunk = buffer.slice(offset, end);

        socket.send(chunk, { binary: true }, (err) => {
            if (err) {
                console.error('发送音频数据失败:', err);
                return;
            }

            offset = end;
            console.log("sending chunk: ", offset)
            // 控制发送间隔（模拟流式传输）
            setTimeout(sendChunk, 100);
        });
    }

    sendChunk();
}

// 发送finish-task指令
function sendFinishTask(socket) {
    const finishTaskMessage = JSON.stringify({
        header: {
            action: 'finish-task',
            task_id: ws.taskId,
            streaming: 'duplex'
        },
        payload: {
            model: 'tingwu-meeting-realtime',
            task_group: 'aigc',
            task: "multimodal-generation",
            function: "generation",
            input: {
                directive: 'stop'
            },
        }
    });

    socket.send(finishTaskMessage, { binary: false }, (err) => {
        if (err) {
            console.error('发送 finish-task 失败:', err);
            return;
        }
        console.log('已发送 finish-task 指令');
    });
}

// 错误处理
ws.on('error', (err) => {
    console.error('WebSocket 错误:', err);
});

// 关闭连接处理
ws.on('close', () => {
    console.log('WebSocket 连接已关闭');
});