Skip to content

Commit 5fbe84c

Browse files
shenjintishenjinti
authored
Merge pull request #8 from yeoleobun/aliyun-tts
fix: aliyun tts
2 parents 2260a8e + 6515179 commit 5fbe84c

2 files changed

Lines changed: 187 additions & 96 deletions

File tree

‎src/synthesis/aliyun.rs‎

Lines changed: 120 additions & 93 deletions
Original file line numberDiff line numberDiff line change
@@ -1,18 +1,19 @@
11
use super::{SynthesisClient, SynthesisOption, SynthesisType};
22
use anyhow::{anyhow, Result};
33
use async_trait::async_trait;
4-
use base64::{engine::general_purpose::STANDARD, Engine};
54
use futures::{stream, SinkExt, Stream, StreamExt};
6-
use http::{Request, StatusCode, Uri};
7-
use rand::random;
5+
use http::StatusCode;
86
use serde::{Deserialize, Serialize};
97
use std::pin::Pin;
10-
use tokio_tungstenite::{connect_async, tungstenite::Message};
8+
use tokio_tungstenite::{
9+
connect_async,
10+
tungstenite::{client::IntoClientRequest, Message},
11+
};
1112
use tracing::{debug, warn};
1213
use uuid::Uuid;
1314

1415
/// Aliyun CosyVoice WebSocket API Client
15-
/// https://help.aliyun.com/zh/model-studio/cosyvoice-websocket-api?spm=a2c4g.11186623.help-menu-2400256.d_2_5_0_2.470054e08UnCMU
16+
/// https://help.aliyun.com/zh/model-studio/cosyvoice-websocket-api
1617
#[derive(Debug)]
1718
pub struct AliyunTtsClient {
1819
option: SynthesisOption,
@@ -21,43 +22,78 @@ pub struct AliyunTtsClient {
2122
/// run-task command structure
2223
#[derive(Debug, Serialize)]
2324
struct RunTaskCommand {
25+
header: CommandHeader,
26+
payload: RunTaskPayload,
27+
}
28+
29+
#[derive(Debug, Serialize)]
30+
struct CommandHeader {
31+
action: String,
2432
task_id: String,
25-
command: String,
26-
model: String,
33+
stream: String,
34+
}
35+
36+
#[derive(Debug, Serialize)]
37+
struct RunTaskPayload {
38+
task_group: String,
39+
task: String,
2740
function: String,
41+
model: String,
2842
parameters: RunTaskParameters,
43+
input: Option<PayloadInput>,
2944
}
3045

3146
#[derive(Debug, Serialize)]
3247
struct RunTaskParameters {
33-
format: String,
34-
sample_rate: i32,
48+
text_type: String,
3549
voice: String,
36-
volume: f32,
37-
speed: f32,
38-
enable_ssml: bool,
50+
#[serde(skip_serializing_if = "Option::is_none")]
51+
format: Option<String>,
52+
#[serde(skip_serializing_if = "Option::is_none")]
53+
sample_rate: Option<u32>,
54+
#[serde(skip_serializing_if = "Option::is_none")]
55+
volume: Option<u32>,
56+
#[serde(skip_serializing_if = "Option::is_none")]
57+
rate: Option<f32>,
58+
#[serde(skip_serializing_if = "Option::is_none")]
59+
pitch: Option<f32>,
60+
#[serde(skip_serializing_if = "Option::is_none")]
61+
enable_ssml: Option<bool>,
3962
}
4063

41-
/// continue-task command structure
42-
#[derive(Debug, Serialize)]
43-
struct ContinueTaskCommand {
44-
task_id: String,
45-
command: String,
64+
#[derive(Debug, Serialize, Deserialize)]
65+
struct PayloadInput {
4666
text: String,
4767
}
4868

4969
/// finish-task command structure
5070
#[derive(Debug, Serialize)]
5171
struct FinishTaskCommand {
52-
task_id: String,
53-
command: String,
72+
header: CommandHeader,
73+
payload: FinishTaskPayload,
5474
}
5575

76+
#[derive(Debug, Serialize)]
77+
struct FinishTaskPayload {
78+
input: EmptyInput,
79+
}
80+
81+
#[derive(Debug, Serialize)]
82+
struct EmptyInput {}
83+
5684
/// WebSocket event response structure
5785
#[derive(Debug, Deserialize)]
5886
struct WebSocketEvent {
87+
header: WebSocketEventHeader,
88+
}
89+
90+
#[allow(dead_code)]
91+
#[derive(Debug, Deserialize)]
92+
struct WebSocketEventHeader {
93+
task_id: String,
5994
event: String,
60-
message: Option<String>,
95+
error_code: Option<String>,
96+
error_message: Option<String>,
6197
}
6298

6399
impl AliyunTtsClient {
@@ -70,17 +106,6 @@ impl AliyunTtsClient {
70106
Self { option }
71107
}
72108

73-
/// Build WebSocket connection URL
74-
fn build_websocket_url(&self, option: &SynthesisOption) -> String {
75-
let endpoint = option
76-
.endpoint
77-
.as_ref()
78-
.map(|e| e.trim_end_matches('/').to_string())
79-
.unwrap_or_else(|| "wss://dashscope.aliyuncs.com".to_string());
80-
81-
format!("{}/api/v1/services/aigc/text2speech/synthesis", endpoint)
82-
}
83-
84109
/// Get API Key from configuration or environment
85110
fn get_api_key(&self, option: &SynthesisOption) -> Result<String> {
86111
option
@@ -93,59 +118,73 @@ impl AliyunTtsClient {
93118
}
94119

95120
/// Create run-task command
96-
fn create_run_task_command(&self, option: &SynthesisOption, task_id: &str) -> RunTaskCommand {
121+
fn create_run_task_command(
122+
&self,
123+
option: &SynthesisOption,
124+
task_id: &str,
125+
text: &str,
126+
) -> RunTaskCommand {
97127
let model = option
98128
.extra
99129
.as_ref()
100130
.and_then(|e| e.get("model"))
101-
.map(|s| s.clone())
102-
.unwrap_or_else(|| "cosyvoice-v1".to_string());
131+
.cloned()
132+
.unwrap_or_else(|| "cosyvoice-v2".to_string());
103133

104134
let voice = option
105135
.speaker
106136
.clone()
107-
.unwrap_or_else(|| "zhichu_emo".to_string());
137+
.unwrap_or_else(|| "longyumi_v2".to_string());
108138

109139
let format = match option.codec.as_deref() {
110140
Some("mp3") => "mp3",
111141
Some("wav") => "wav",
112142
_ => "pcm",
113143
};
114144

115-
let sample_rate = option.samplerate.unwrap_or(16000);
116-
let volume = option.volume.unwrap_or(5) as f32 / 10.0; // Convert to 0.0-1.0 range
117-
let speed = option.speed.unwrap_or(1.0);
145+
let sample_rate = option.samplerate.unwrap_or(16000) as u32;
146+
let volume = (option.volume.unwrap_or(5) * 10) as u32; // Convert to 0 - 100 range
147+
let rate = option.speed.unwrap_or(1.0);
118148

119149
RunTaskCommand {
120-
task_id: task_id.to_string(),
121-
command: "run-task".to_string(),
122-
model,
123-
function: "text2speech".to_string(),
124-
parameters: RunTaskParameters {
125-
format: format.to_string(),
126-
sample_rate,
127-
voice,
128-
volume,
129-
speed,
130-
enable_ssml: false,
150+
header: CommandHeader {
151+
action: "run-task".to_string(),
152+
task_id: task_id.to_string(),
153+
stream: "duplex".to_string(),
154+
},
155+
payload: RunTaskPayload {
156+
task_group: "audio".to_string(),
157+
task: "tts".to_string(),
158+
function: "SpeechSynthesizer".to_string(),
159+
model,
160+
parameters: RunTaskParameters {
161+
text_type: "PlainText".to_string(),
162+
voice,
163+
format: Some(format.to_string()),
164+
sample_rate: Some(sample_rate),
165+
volume: Some(volume),
166+
rate: Some(rate),
167+
pitch: None,
168+
enable_ssml: None,
169+
},
170+
input: Some(PayloadInput {
171+
text: text.to_string(),
172+
}),
131173
},
132-
}
133-
}
134-
135-
/// Create continue-task command
136-
fn create_continue_task_command(&self, task_id: &str, text: &str) -> ContinueTaskCommand {
137-
ContinueTaskCommand {
138-
task_id: task_id.to_string(),
139-
command: "continue-task".to_string(),
140-
text: text.to_string(),
141174
}
142175
}
143176

144177
/// Create finish-task command
145178
fn create_finish_task_command(&self, task_id: &str) -> FinishTaskCommand {
146179
FinishTaskCommand {
147-
task_id: task_id.to_string(),
148-
command: "finish-task".to_string(),
180+
header: CommandHeader {
181+
action: "finish-task".to_string(),
182+
task_id: task_id.to_string(),
183+
stream: "duplex".to_string(),
184+
},
185+
payload: FinishTaskPayload {
186+
input: EmptyInput {},
187+
},
149188
}
150189
}
151190
}
@@ -163,24 +202,17 @@ impl SynthesisClient for AliyunTtsClient {
163202
) -> Result<Pin<Box<dyn Stream<Item = Result<Vec<u8>>> + Send + 'a>>> {
164203
let option = self.option.merge_with(option);
165204
let api_key = self.get_api_key(&option)?;
166-
let ws_url = self.build_websocket_url(&option);
167205
let task_id = Uuid::new_v4().to_string();
168-
206+
let ws_url = option
207+
.endpoint
208+
.as_deref()
209+
.unwrap_or("wss://dashscope.aliyuncs.com/api-ws/v1/inference");
169210
debug!("Connecting to Aliyun WebSocket URL: {}", ws_url);
170211

171-
// Parse WebSocket URL
172-
let ws_url = ws_url.parse::<Uri>()?;
173-
174-
// Create WebSocket request
175-
let request = Request::builder()
176-
.uri(&ws_url)
177-
.header("Host", ws_url.host().unwrap_or("dashscope.aliyuncs.com"))
178-
.header("Connection", "Upgrade")
179-
.header("Upgrade", "websocket")
180-
.header("Sec-WebSocket-Version", "13")
181-
.header("Sec-WebSocket-Key", STANDARD.encode(random::<[u8; 16]>()))
182-
.header("Authorization", format!("Bearer {}", api_key))
183-
.body(())?;
212+
let mut request = ws_url.into_client_request()?;
213+
let headers = request.headers_mut();
214+
headers.insert("Authorization", format!("Bearer {}", api_key).parse()?);
215+
headers.insert("X-DashScope-DataInspection", "enable".parse()?);
184216

185217
// Establish WebSocket connection
186218
let (ws_stream, response) = connect_async(request).await?;
@@ -199,30 +231,18 @@ impl SynthesisClient for AliyunTtsClient {
199231
let (mut ws_sink, ws_stream) = ws_stream.split();
200232

201233
// Send run-task command
202-
let run_task_cmd = self.create_run_task_command(&option, &task_id);
234+
let run_task_cmd = self.create_run_task_command(&option, &task_id, text);
203235
let run_task_json = serde_json::to_string(&run_task_cmd)?;
204236
debug!("Sending run-task command: {}", run_task_json);
205-
206237
ws_sink
207238
.send(Message::text(run_task_json))
208239
.await
209240
.map_err(|e| anyhow!("Failed to send run-task command: {}", e))?;
210241

211-
// Send continue-task command
212-
let continue_task_cmd = self.create_continue_task_command(&task_id, text);
213-
let continue_task_json = serde_json::to_string(&continue_task_cmd)?;
214-
debug!("Sending continue-task command: {}", continue_task_json);
215-
216-
ws_sink
217-
.send(Message::text(continue_task_json))
218-
.await
219-
.map_err(|e| anyhow!("Failed to send continue-task command: {}", e))?;
220-
221242
// Send finish-task command
222243
let finish_task_cmd = self.create_finish_task_command(&task_id);
223244
let finish_task_json = serde_json::to_string(&finish_task_cmd)?;
224245
debug!("Sending finish-task command: {}", finish_task_json);
225-
226246
ws_sink
227247
.send(Message::text(finish_task_json))
228248
.await
@@ -241,8 +261,8 @@ impl SynthesisClient for AliyunTtsClient {
241261
// Handle JSON event messages
242262
match serde_json::from_str::<WebSocketEvent>(&text) {
243263
Ok(event) => {
244-
debug!("Received event: {}", event.event);
245-
match event.event.as_str() {
264+
debug!("Received event: {:?}", event);
265+
match event.header.event.as_str() {
246266
"task-started" => {
247267
debug!("Task started");
248268
Some((Ok(Vec::new()), (ws_stream, false)))
@@ -252,10 +272,17 @@ impl SynthesisClient for AliyunTtsClient {
252272
Some((Ok(Vec::new()), (ws_stream, true)))
253273
}
254274
"task-failed" => {
275+
let error_code = event
276+
.header
277+
.error_code
278+
.unwrap_or_else(|| "Unknown error".to_string());
255279
let error_msg = event
256-
.message
280+
.header
281+
.error_message
257282
.unwrap_or_else(|| "Unknown error".to_string());
258-
warn!("Task failed: {}", error_msg);
283+
let error_msg =
284+
format!("Task failed: {} {}", error_code, error_msg);
285+
warn!("Task failed: {}:{}", error_code, error_msg);
259286
Some((
260287
Err(anyhow!("Task failed: {}", error_msg)),
261288
(ws_stream, true),
@@ -266,7 +293,7 @@ impl SynthesisClient for AliyunTtsClient {
266293
Some((Ok(Vec::new()), (ws_stream, false)))
267294
}
268295
_ => {
269-
debug!("Ignoring unknown event: {}", event.event);
296+
debug!("Ignoring unknown event: {}", event.header.event);
270297
Some((Ok(Vec::new()), (ws_stream, false)))
271298
}
272299
}

0 commit comments

Comments
 (0)