CosyVoice WebSocket API

using System.Net.WebSockets;
using System.Text;
using System.Text.Json;

class Program {
  // 如果未设置环境变量，替换下一行为: private static readonly string ApiKey = "YOUR_API_KEY"
  private static readonly string ApiKey = Environment.GetEnvironmentVariable("DASHSCOPE_API_KEY") ?? throw new InvalidOperationException("DASHSCOPE_API_KEY environment variable is not set.");

  private const string WebSocketUrl = "wss://dashscope.aliyuncs.com/api-ws/v1/inference/";
  // 输出文件路径
  private const string OutputFilePath = "output.mp3";

  // WebSocket 客户端
  private static ClientWebSocket _webSocket = new ClientWebSocket();
  // 取消令牌源
  private static CancellationTokenSource _cancellationTokenSource = new CancellationTokenSource();
  // 任务 ID
  private static string? _taskId;
  // 任务已启动标志
  private static TaskCompletionSource<bool> _taskStartedTcs = new TaskCompletionSource<bool>();

  static async Task Main(string[] args) {
    try {
      // 清空输出文件
      ClearOutputFile(OutputFilePath);

      // 连接 WebSocket 服务
      await ConnectToWebSocketAsync(WebSocketUrl);

      // 开始接收消息
      Task receiveTask = ReceiveMessagesAsync();

      // 发送启动指令
      _taskId = GenerateTaskId();
      await SendRunTaskCommandAsync(_taskId);

      // 等待任务已启动事件
      await _taskStartedTcs.Task;

      // 发送续传指令
      string[] texts = {
        "床前明月光",
        "疑是地上霜",
        "举头望明月",
        "低头思故乡"
      };
      foreach (string text in texts) {
        await SendContinueTaskCommandAsync(text);
      }

      // 发送结束指令
      await SendFinishTaskCommandAsync(_taskId);

      // 等待接收任务完成
      await receiveTask;

      Console.WriteLine("Task completed. Connection closed.");
    } catch (OperationCanceledException) {
      Console.WriteLine("Task canceled.");
    } catch (Exception ex) {
      Console.WriteLine($"Error: {ex.Message}");
    } finally {
      _cancellationTokenSource.Cancel();
      _webSocket.Dispose();
    }
  }

  private static void ClearOutputFile(string filePath) {
    if (File.Exists(filePath)) {
      File.WriteAllText(filePath, string.Empty);
      Console.WriteLine("Output file cleared.");
    } else {
      Console.WriteLine("Output file does not exist. No action needed.");
    }
  }

  private static async Task ConnectToWebSocketAsync(string url) {
    var uri = new Uri(url);
    if (_webSocket.State == WebSocketState.Connecting || _webSocket.State == WebSocketState.Open) {
      return;
    }

    // 设置 WebSocket 请求头
    _webSocket.Options.SetRequestHeader("Authorization", $"bearer {ApiKey}");
    _webSocket.Options.SetRequestHeader("X-DashScope-DataInspection", "enable");

    try {
      await _webSocket.ConnectAsync(uri, _cancellationTokenSource.Token);
      Console.WriteLine("Successfully connected to WebSocket service.");
    } catch (OperationCanceledException) {
      Console.WriteLine("WebSocket connection canceled.");
    } catch (Exception ex) {
      Console.WriteLine($"WebSocket connection failed: {ex.Message}");
      throw;
    }
  }

  private static async Task SendRunTaskCommandAsync(string taskId) {
    var command = CreateCommand("run-task", taskId, "duplex", new {
      task_group = "audio",
      task = "tts",
      function = "SpeechSynthesizer",
      model = "cosyvoice-v3-flash",
      parameters = new
      {
        text_type = "PlainText",
        voice = "longanyang",
        format = "mp3",
        sample_rate = 22050,
        volume = 50,
        rate = 1,
        pitch = 1,
        // 如果 enable_ssml 为 true，仅允许发送一条续传指令。否则会返回"Text request limit violated, expected 1."
        enable_ssml = false
      },
      input = new { }
    });

    await SendJsonMessageAsync(command);
    Console.WriteLine("Sent run-task instruction.");
  }

  private static async Task SendContinueTaskCommandAsync(string text) {
    if (_taskId == null) {
      throw new InvalidOperationException("Task ID not initialized.");
    }

    var command = CreateCommand("continue-task", _taskId, "duplex", new {
      input = new {
        text
      }
    });

    await SendJsonMessageAsync(command);
    Console.WriteLine("Sent continue-task instruction.");
  }

  private static async Task SendFinishTaskCommandAsync(string taskId) {
    var command = CreateCommand("finish-task", taskId, "duplex", new {
      input = new { }
    });

    await SendJsonMessageAsync(command);
    Console.WriteLine("Sent finish-task instruction.");
  }

  private static async Task SendJsonMessageAsync(string message) {
    var buffer = Encoding.UTF8.GetBytes(message);
    try {
      await _webSocket.SendAsync(new ArraySegment<byte>(buffer), WebSocketMessageType.Text, true, _cancellationTokenSource.Token);
    } catch (OperationCanceledException) {
      Console.WriteLine("Message send canceled.");
    }
  }

  private static async Task ReceiveMessagesAsync() {
    while (_webSocket.State == WebSocketState.Open) {
      var response = await ReceiveMessageAsync();
      if (response != null) {
        var eventStr = response.RootElement.GetProperty("header").GetProperty("event").GetString();
        switch (eventStr) {
          case "task-started":
            Console.WriteLine("Task started.");
            _taskStartedTcs.TrySetResult(true);
            break;
          case "task-finished":
            Console.WriteLine("Task completed.");
            _cancellationTokenSource.Cancel();
            break;
          case "task-failed":
            Console.WriteLine("Task failed: " + response.RootElement.GetProperty("header").GetProperty("error_message").GetString());
            _cancellationTokenSource.Cancel();
            break;
          default:
            // 在此处理 result-generated
            break;
        }
      }
    }
  }

  private static async Task<JsonDocument?> ReceiveMessageAsync() {
    var buffer = new byte[1024 * 4];
    var segment = new ArraySegment<byte>(buffer);

    try {
      WebSocketReceiveResult result = await _webSocket.ReceiveAsync(segment, _cancellationTokenSource.Token);

      if (result.MessageType == WebSocketMessageType.Close) {
        await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, "Closing", _cancellationTokenSource.Token);
        return null;
      }

      if (result.MessageType == WebSocketMessageType.Binary) {
        // 处理二进制数据
        Console.WriteLine("Received binary data...");

        // 保存二进制数据到文件
        using (var fileStream = new FileStream(OutputFilePath, FileMode.Append)) {
          fileStream.Write(buffer, 0, result.Count);
        }

        return null;
      }

      string message = Encoding.UTF8.GetString(buffer, 0, result.Count);
      return JsonDocument.Parse(message);
    } catch (OperationCanceledException) {
      Console.WriteLine("Message receive canceled.");
      return null;
    }
  }

  private static string GenerateTaskId() {
    return Guid.NewGuid().ToString("N").Substring(0, 32);
  }

  private static string CreateCommand(string action, string taskId, string streaming, object payload) {
    var command = new {
      header = new {
        action,
        task_id = taskId,
        streaming
      },
      payload
    };

    return JsonSerializer.Serialize(command);
  }
}

{
  "require": {
    "react/event-loop": "^1.3",
    "react/socket": "^1.11",
    "react/stream": "^1.2",
    "react/http": "^1.1",
    "ratchet/pawl": "^0.4"
  },
  "autoload": {
    "psr-4": {
      "App\\": "src/"
    }
  }
}

<?php

require __DIR__ . '/vendor/autoload.php';

use Ratchet\Client\Connector;
use React\EventLoop\Loop;
use React\Socket\Connector as SocketConnector;

// 如果未设置环境变量，替换下一行为: $api_key = "YOUR_API_KEY"
$api_key = getenv("DASHSCOPE_API_KEY");
$websocket_url = 'wss://dashscope.aliyuncs.com/api-ws/v1/inference/'; // WebSocket 服务器地址
$output_file = 'output.mp3'; // 输出文件路径

$loop = Loop::get();

if (file_exists($output_file)) {
    // 清空文件内容
    file_put_contents($output_file, '');
}

// 创建自定义连接器
$socketConnector = new SocketConnector($loop, [
    'tcp' => [
      'bindto' => '0.0.0.0:0',
    ],
    'tls' => [
      'verify_peer' => false,
      'verify_peer_name' => false,
    ],
]);

$connector = new Connector($loop, $socketConnector);

$headers = [
    'Authorization' => 'bearer ' . $api_key,
    'X-DashScope-DataInspection' => 'enable'
];

$connector($websocket_url, [], $headers)->then(function ($conn) use ($loop, $output_file) {
    echo "Connected to WebSocket server\n";

    // 生成任务 ID
    $taskId = generateTaskId();

    // 发送启动指令
    sendRunTaskMessage($conn, $taskId);

    // 定义发送续传指令的函数
    $sendContinueTask = function() use ($conn, $loop, $taskId) {
      // 待发送的文本
      $texts = ["床前明月光", "疑是地上霜", "举头望明月", "低头思故乡"];
      $continueTaskCount = 0;
      foreach ($texts as $text) {
        $continueTaskMessage = json_encode([
          "header" => [
            "action" => "continue-task",
            "task_id" => $taskId,
            "streaming" => "duplex"
          ],
          "payload" => [
            "input" => [
              "text" => $text
            ]
          ]
        ]);
        echo "Preparing to send continue-task instruction: " . $continueTaskMessage . "\n";
        $conn->send($continueTaskMessage);
        $continueTaskCount++;
      }
      echo "Number of continue-task instructions sent: " . $continueTaskCount . "\n";

      // 发送结束指令
      sendFinishTaskMessage($conn, $taskId);
    };

    // 任务已启动事件标志
    $taskStarted = false;

    // 监听消息
    $conn->on('message', function($msg) use ($conn, $sendContinueTask, $loop, &$taskStarted, $taskId, $output_file) {
      if ($msg->isBinary()) {
        // 将二进制数据写入本地文件
        file_put_contents($output_file, $msg->getPayload(), FILE_APPEND);
      } else {
        // 处理非二进制消息
        $response = json_decode($msg, true);

        if (isset($response['header']['event'])) {
          handleEvent($conn, $response, $sendContinueTask, $loop, $taskId, $taskStarted);
        } else {
          echo "Unknown message format\n";
        }
      }
    });

    // 监听连接关闭
    $conn->on('close', function($code = null, $reason = null) {
      echo "Connection closed\n";
      if ($code !== null) {
        echo "Close code: " . $code . "\n";
      }
      if ($reason !== null) {
        echo "Close reason: " . $reason . "\n";
      }
    });
}, function ($e) {
    echo "Cannot connect: {$e->getMessage()}\n";
});

$loop->run();

/**
  * 生成任务 ID
  * @return string
  */
function generateTaskId(): string {
    return bin2hex(random_bytes(16));
}

/**
  * 发送启动指令
  * @param $conn
  * @param $taskId
  */
function sendRunTaskMessage($conn, $taskId) {
    $runTaskMessage = json_encode([
      "header" => [
        "action" => "run-task",
        "task_id" => $taskId,
        "streaming" => "duplex"
      ],
      "payload" => [
        "task_group" => "audio",
        "task" => "tts",
        "function" => "SpeechSynthesizer",
        "model" => "cosyvoice-v3-flash",
        "parameters" => [
          "text_type" => "PlainText",
          "voice" => "longanyang",
          "format" => "mp3",
          "sample_rate" => 22050,
          "volume" => 50,
          "rate" => 1,
          "pitch" => 1,
          // 如果 enable_ssml 为 true，仅允许发送一条续传指令。否则会返回"Text request limit violated, expected 1."
          "enable_ssml" => false
        ],
        "input" => (object) []
      ]
    ]);
    echo "Preparing to send run-task instruction: " . $runTaskMessage . "\n";
    $conn->send($runTaskMessage);
    echo "run-task instruction sent\n";
}

/**
  * 读取音频文件
  * @param string $filePath
  * @return bool|string
  */
function readAudioFile(string $filePath) {
    $voiceData = file_get_contents($filePath);
    if ($voiceData === false) {
      echo "Cannot read audio file\n";
    }
    return $voiceData;
}

/**
  * 拆分音频数据
  * @param string $data
  * @param int $chunkSize
  * @return array
  */
function splitAudioData(string $data, int $chunkSize): array {
    return str_split($data, $chunkSize);
}

/**
  * 发送结束指令
  * @param $conn
  * @param $taskId
  */
function sendFinishTaskMessage($conn, $taskId) {
    $finishTaskMessage = json_encode([
      "header" => [
        "action" => "finish-task",
        "task_id" => $taskId,
        "streaming" => "duplex"
      ],
      "payload" => [
        "input" => (object) []
      ]
    ]);
    echo "Preparing to send finish-task instruction: " . $finishTaskMessage . "\n";
    $conn->send($finishTaskMessage);
    echo "finish-task instruction sent\n";
}

/**
  * 处理事件
  * @param $conn
  * @param $response
  * @param $sendContinueTask
  * @param $loop
  * @param $taskId
  * @param $taskStarted
  */
function handleEvent($conn, $response, $sendContinueTask, $loop, $taskId, &$taskStarted) {
    switch ($response['header']['event']) {
      case 'task-started':
        echo "Task started. Sending continue-task instructions...\n";
        $taskStarted = true;
        // 发送续传指令
        $sendContinueTask();
        break;
      case 'result-generated':
        // 收到 result-generated 事件
        break;
      case 'task-finished':
        echo "Task completed\n";
        $conn->close();
        break;
      case 'task-failed':
        echo "Task failed\n";
        echo "Error code: " . $response['header']['error_code'] . "\n";
        echo "Error message: " . $response['header']['error_message'] . "\n";
        $conn->close();
        break;
      case 'error':
        echo "Error: " . $response['payload']['message'] . "\n";
        break;
      default:
        echo "Unknown event: " . $response['header']['event'] . "\n";
        break;
    }

    // 任务完成后关闭连接
    if ($response['header']['event'] == 'task-finished') {
      // 等待 1 秒确保所有数据传输完成
      $loop->addTimer(1, function() use ($conn) {
        $conn->close();
        echo "Client closed connection\n";
      });
    }

    // 未收到任务已启动事件时关闭连接
    if (!$taskStarted && in_array($response['header']['event'], ['task-failed', 'error'])) {
      $conn->close();
    }
}

npm install ws
npm install uuid

const WebSocket = require('ws');
const fs = require('fs');
const uuid = require('uuid').v4;

// 如果未设置环境变量，替换下一行为: const apiKey = "YOUR_API_KEY"
const apiKey = process.env.DASHSCOPE_API_KEY;
const url = 'wss://dashscope.aliyuncs.com/api-ws/v1/inference/';
// 输出文件路径
const outputFilePath = 'output.mp3';

// 清空输出文件
fs.writeFileSync(outputFilePath, '');

// 创建 WebSocket 客户端
const ws = new WebSocket(url, {
  headers: {
  Authorization: `bearer ${apiKey}`,
  'X-DashScope-DataInspection': 'enable'
  }
});

let taskStarted = false;
let taskId = uuid();

ws.on('open', () => {
  console.log('Connected to WebSocket server');

  // 发送启动指令
  const runTaskMessage = JSON.stringify({
  header: {
      action: 'run-task',
      task_id: taskId,
      streaming: 'duplex'
  },
  payload: {
      task_group: 'audio',
      task: 'tts',
      function: 'SpeechSynthesizer',
      model: 'cosyvoice-v3-flash',
      parameters: {
    text_type: 'PlainText',
    voice: 'longanyang', // 音色
    format: 'mp3', // 音频格式
    sample_rate: 22050, // 采样率
    volume: 50, // 音量
    rate: 1, // 语速
    pitch: 1, // 音调
    enable_ssml: false // 启用 SSML。如果为 true，仅允许发送一条续传指令。否则会返回"Text request limit violated, expected 1."
      },
      input: {}
  }
  });
  ws.send(runTaskMessage);
  console.log('Sent run-task message');
});

const fileStream = fs.createWriteStream(outputFilePath, { flags: 'a' });
ws.on('message', (data, isBinary) => {
  if (isBinary) {
  // 将二进制数据写入文件
  fileStream.write(data);
  } else {
  const message = JSON.parse(data);

  switch (message.header.event) {
      case 'task-started':
    taskStarted = true;
    console.log('Task started');
    // 发送续传指令
    sendContinueTasks(ws);
    break;
      case 'task-finished':
    console.log('Task completed');
    ws.close();
    fileStream.end(() => {
      console.log('File stream closed');
    });
    break;
      case 'task-failed':
    console.error('Task failed: ', message.header.error_message);
    ws.close();
    fileStream.end(() => {
      console.log('File stream closed');
    });
    break;
      default:
    // 在此处理 result-generated
    break;
  }
  }
});

function sendContinueTasks(ws) {
  const texts = [
  '床前明月光，',
  '疑是地上霜。',
  '举头望明月，',
  '低头思故乡。'
  ];

  texts.forEach((text, index) => {
  setTimeout(() => {
      if (taskStarted) {
    const continueTaskMessage = JSON.stringify({
      header: {
      action: 'continue-task',
      task_id: taskId,
      streaming: 'duplex'
      },
      payload: {
      input: {
      text: text
      }
      }
    });
    ws.send(continueTaskMessage);
    console.log(`Sent continue-task, text: ${text}`);
      }
  }, index * 1000); // 每秒发送一次
  });

  // 发送结束指令
  setTimeout(() => {
  if (taskStarted) {
      const finishTaskMessage = JSON.stringify({
    header: {
      action: 'finish-task',
      task_id: taskId,
      streaming: 'duplex'
    },
    payload: {
      input: {}
    }
      });
      ws.send(finishTaskMessage);
      console.log('Sent finish-task');
  }
  }, texts.length * 1000 + 1000); // 最后一条续传指令后 1 秒发送
}

ws.on('close', () => {
  console.log('Disconnected from WebSocket server');
});

<dependencies>
  <!-- WebSocket 客户端 -->
  <dependency>
    <groupId>org.java-websocket</groupId>
    <artifactId>Java-WebSocket</artifactId>
    <version>1.5.3</version>
  </dependency>

  <!-- JSON 处理 -->
  <dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
    <version>2.13.0</version>
  </dependency>
</dependencies>

// 省略其他代码
dependencies {
  // WebSocket 客户端
  implementation 'org.java-websocket:Java-WebSocket:1.5.3'
  // JSON 处理
  implementation 'com.fasterxml.jackson.core:jackson-databind:2.13.0'
}
// 省略其他代码

import com.fasterxml.jackson.databind.ObjectMapper;

import org.java_websocket.client.WebSocketClient;
import org.java_websocket.handshake.ServerHandshake;

import java.io.FileOutputStream;
import java.io.IOException;
import java.net.URI;
import java.nio.ByteBuffer;
import java.util.*;

public class TTSWebSocketClient extends WebSocketClient {
  private final String taskId = UUID.randomUUID().toString();
  private final String outputFile = "output_" + System.currentTimeMillis() + ".mp3";
  private boolean taskFinished = false;

  public TTSWebSocketClient(URI serverUri, Map<String, String> headers) {
    super(serverUri, headers);
  }

  @Override
  public void onOpen(ServerHandshake serverHandshake) {
    System.out.println("Connection successful");

    // 发送启动指令
    // 如果 enable_ssml 为 true，仅允许发送一条续传指令。否则会返回"Text request limit violated, expected 1."
    String runTaskCommand = "{ \"header\": { \"action\": \"run-task\", \"task_id\": \"" + taskId + "\", \"streaming\": \"duplex\" }, \"payload\": { \"task_group\": \"audio\", \"task\": \"tts\", \"function\": \"SpeechSynthesizer\", \"model\": \"cosyvoice-v3-flash\", \"parameters\": { \"text_type\": \"PlainText\", \"voice\": \"longanyang\", \"format\": \"mp3\", \"sample_rate\": 22050, \"volume\": 50, \"rate\": 1, \"pitch\": 1, \"enable_ssml\": false }, \"input\": {} }}";
    send(runTaskCommand);
  }

  @Override
  public void onMessage(String message) {
    System.out.println("Received server message: " + message);
    try {
      // 解析 JSON 消息
      Map<String, Object> messageMap = new ObjectMapper().readValue(message, Map.class);

      if (messageMap.containsKey("header")) {
        Map<String, Object> header = (Map<String, Object>) messageMap.get("header");

        if (header.containsKey("event")) {
          String event = (String) header.get("event");

          if ("task-started".equals(event)) {
            System.out.println("Received task-started event");

            List<String> texts = Arrays.asList(
                "床前明月光，疑是地上霜",
                "举头望明月，低头思故乡"
            );

            for (String text : texts) {
              // 发送续传指令
              sendContinueTask(text);
            }

            // 发送结束指令
            sendFinishTask();
          } else if ("task-finished".equals(event)) {
            System.out.println("Received task-finished event");
            taskFinished = true;
            closeConnection();
          } else if ("task-failed".equals(event)) {
            System.out.println("Task failed: " + message);
            closeConnection();
          }
        }
      }
    } catch (Exception e) {
      System.err.println("Exception occurred: " + e.getMessage());
    }
  }

  @Override
  public void onMessage(ByteBuffer message) {
    System.out.println("Received binary audio data size: " + message.remaining());

    try (FileOutputStream fos = new FileOutputStream(outputFile, true)) {
      byte[] buffer = new byte[message.remaining()];
      message.get(buffer);
      fos.write(buffer);
      System.out.println("Audio data written to local file " + outputFile);
    } catch (IOException e) {
      System.err.println("Failed to write audio data to local file: " + e.getMessage());
    }
  }

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

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

  private void sendContinueTask(String text) {
    String command = "{ \"header\": { \"action\": \"continue-task\", \"task_id\": \"" + taskId + "\", \"streaming\": \"duplex\" }, \"payload\": { \"input\": { \"text\": \"" + text + "\" } }}";
    send(command);
  }

  private void sendFinishTask() {
    String command = "{ \"header\": { \"action\": \"finish-task\", \"task_id\": \"" + taskId + "\", \"streaming\": \"duplex\" }, \"payload\": { \"input\": {} }}";
    send(command);
  }

  private void closeConnection() {
    if (!isClosed()) {
      close();
    }
  }

  public static void main(String[] args) {
    try {
      // 如果未设置环境变量，替换下一行为: String apiKey = "YOUR_API_KEY"
      String apiKey = System.getenv("DASHSCOPE_API_KEY");
      if (apiKey == null || apiKey.isEmpty()) {
        System.err.println("Set DASHSCOPE_API_KEY environment variable");
        return;
      }

      Map<String, String> headers = new HashMap<>();
      headers.put("Authorization", "bearer " + apiKey);
      TTSWebSocketClient client = new TTSWebSocketClient(new URI("wss://dashscope.aliyuncs.com/api-ws/v1/inference/"), headers);

      client.connect();

      while (!client.isClosed() && !client.taskFinished) {
        Thread.sleep(1000);
      }
    } catch (Exception e) {
      System.err.println("Failed to connect to WebSocket service: " + e.getMessage());
      e.printStackTrace();
    }
  }
}

pip uninstall websocket-client
pip uninstall websocket
pip install websocket-client

import websocket
import json
import uuid
import os
import time


class TTSClient:
  def __init__(self, api_key, uri):
    """
  初始化 TTSClient 实例

  参数:
    api_key (str): 用于认证的 API Key
    uri (str): WebSocket 服务地址
  """
    self.api_key = api_key  # 替换为你的 API Key
    self.uri = uri  # 替换为你的 WebSocket 地址
    self.task_id = str(uuid.uuid4())  # 生成唯一的任务 ID
    self.output_file = f"output_{int(time.time())}.mp3"  # 输出音频文件路径
    self.ws = None  # WebSocketApp 实例
    self.task_started = False  # 是否已收到任务已启动事件
    self.task_finished = False  # 是否已收到任务已完成或任务失败事件

  def on_open(self, ws):
    """
  连接建立时的回调
  发送启动指令开始语音合成
  """
    print("WebSocket connected")

    # 构建启动指令
    run_task_cmd = {
      "header": {
        "action": "run-task",
        "task_id": self.task_id,
        "streaming": "duplex"
      },
      "payload": {
        "task_group": "audio",
        "task": "tts",
        "function": "SpeechSynthesizer",
        "model": "cosyvoice-v3-flash",
        "parameters": {
          "text_type": "PlainText",
          "voice": "longanyang",
          "format": "mp3",
          "sample_rate": 22050,
          "volume": 50,
          "rate": 1,
          "pitch": 1,
          # 如果 enable_ssml 为 True，仅允许发送一条续传指令。否则会返回"Text request limit violated, expected 1."
          "enable_ssml": False
        },
        "input": {}
      }
    }

    # 发送启动指令
    ws.send(json.dumps(run_task_cmd))
    print("Sent run-task instruction")

  def on_message(self, ws, message):
    """
  收到消息时的回调
  分别处理文本和二进制消息
  """
    if isinstance(message, str):
      # 处理 JSON 文本消息
      try:
        msg_json = json.loads(message)
        print(f"Received JSON message: {msg_json}")

        if "header" in msg_json:
          header = msg_json["header"]

          if "event" in header:
            event = header["event"]

            if event == "task-started":
              print("Task started")
              self.task_started = True

              # 发送续传指令
              texts = [
                "床前明月光，疑是地上霜",
                "举头望明月，低头思故乡"
              ]

              for text in texts:
                self.send_continue_task(text)

              # 所有续传指令后发送结束指令
              self.send_finish_task()

            elif event == "task-finished":
              print("Task completed")
              self.task_finished = True
              self.close(ws)

            elif event == "task-failed":
              error_msg = msg_json.get("error_message", "Unknown error")
              print(f"Task failed: {error_msg}")
              self.task_finished = True
              self.close(ws)

      except json.JSONDecodeError as e:
        print(f"JSON parsing failed: {e}")
    else:
      # 处理二进制消息（音频数据）
      print(f"Received binary message, size: {len(message)} bytes")
      with open(self.output_file, "ab") as f:
        f.write(message)
      print(f"Audio data written to local file {self.output_file}")

  def on_error(self, ws, error):
    """发生错误时的回调"""
    print(f"WebSocket error: {error}")

  def on_close(self, ws, close_status_code, close_msg):
    """连接关闭时的回调"""
    print(f"WebSocket closed: {close_msg} ({close_status_code})")

  def send_continue_task(self, text):
    """发送续传指令，包含待合成的文本"""
    cmd = {
      "header": {
        "action": "continue-task",
        "task_id": self.task_id,
        "streaming": "duplex"
      },
      "payload": {
        "input": {
          "text": text
        }
      }
    }

    self.ws.send(json.dumps(cmd))
    print(f"Sent continue-task instruction, text: {text}")

  def send_finish_task(self):
    """发送结束指令，结束语音合成"""
    cmd = {
      "header": {
        "action": "finish-task",
        "task_id": self.task_id,
        "streaming": "duplex"
      },
      "payload": {
        "input": {}
      }
    }

    self.ws.send(json.dumps(cmd))
    print("Sent finish-task instruction")

  def close(self, ws):
    """手动关闭连接"""
    if ws and ws.sock and ws.sock.connected:
      ws.close()
      print("Manually closed connection")

  def run(self):
    """启动 WebSocket 客户端"""
    # 设置请求头（认证）
    header = {
      "Authorization": f"bearer {self.api_key}",
      "X-DashScope-DataInspection": "enable"
    }

    # 创建 WebSocketApp 实例
    self.ws = websocket.WebSocketApp(
      self.uri,
      header=header,
      on_open=self.on_open,
      on_message=self.on_message,
      on_error=self.on_error,
      on_close=self.on_close
    )

    print("Listening for WebSocket messages...")
    self.ws.run_forever()  # 启动持久连接


# 使用示例
if __name__ == "__main__":
  # 如果未设置环境变量，替换下一行为: API_KEY = "YOUR_API_KEY"
  API_KEY = os.environ.get("DASHSCOPE_API_KEY")
  SERVER_URI = "wss://dashscope.aliyuncs.com/api-ws/v1/inference/"  # 替换为你的 WebSocket 地址

  client = TTSClient(API_KEY, SERVER_URI)
  client.run()