| 1 | """Alibaba CosyVoice backend for narration audio generation.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import json |
| 6 | import math |
| 7 | import os |
| 8 | from pathlib import Path |
| 9 | from urllib import error, request |
| 10 | |
| 11 | from tts_backends.backend_common import ( |
| 12 | download_audio, |
| 13 | extension_from_format, |
| 14 | get_bytes, |
| 15 | post_json, |
| 16 | publish_audio_subtitle, |
| 17 | read_api_key, |
| 18 | read_http_error, |
| 19 | ) |
| 20 | from tts_backends.backend_edge import ( |
| 21 | DEFAULT_SUBTITLE_MAX_CHARS, |
| 22 | format_word_timed_srt, |
| 23 | ) |
| 24 | |
| 25 | |
| 26 | DEFAULT_ENDPOINT = "https://dashscope.aliyuncs.com/api/v1/services/audio/tts/SpeechSynthesizer" |
| 27 | DEFAULT_MODEL = "cosyvoice-v3-flash" |
| 28 | _TICKS_PER_MILLISECOND = 10_000 |
| 29 | |
| 30 | |
| 31 | def output_extension(audio_format: str) -> str: |
| 32 | return extension_from_format(audio_format) |
| 33 | |
| 34 | |
| 35 | def read_cosyvoice_api_key(env_name: str) -> str: |
| 36 | env_names = tuple(dict.fromkeys([env_name, "COSYVOICE_API_KEY", "DASHSCOPE_API_KEY"])) |
| 37 | return read_api_key(*env_names, label="CosyVoice/DashScope") |
| 38 | |
| 39 | |
| 40 | def resolve_url(base_url: str | None = None) -> str: |
| 41 | base = (base_url or os.environ.get("COSYVOICE_TTS_BASE_URL") or DEFAULT_ENDPOINT).rstrip("/") |
| 42 | if base.endswith("/SpeechSynthesizer"): |
| 43 | return base |
| 44 | return base + "/api/v1/services/audio/tts/SpeechSynthesizer" |
| 45 | |
| 46 | |
| 47 | def _parse_sse_event(data_lines: list[str]) -> dict | None: |
| 48 | payload = "\n".join(data_lines).strip() |
| 49 | if not payload or payload == "[DONE]": |
| 50 | return None |
| 51 | try: |
| 52 | event = json.loads(payload) |
| 53 | except json.JSONDecodeError as exc: |
| 54 | raise RuntimeError("CosyVoice streaming response contains invalid JSON") from exc |
| 55 | if not isinstance(event, dict): |
| 56 | raise RuntimeError("CosyVoice streaming event is not an object") |
| 57 | return event |
| 58 | |
| 59 | |
| 60 | def _parse_sse(raw: bytes) -> list[dict]: |
| 61 | try: |
| 62 | body = raw.decode("utf-8-sig") |
| 63 | except UnicodeError as exc: |
| 64 | raise RuntimeError("CosyVoice streaming response is not valid UTF-8") from exc |
| 65 | |
| 66 | if not any(line.startswith("data:") for line in body.splitlines()): |
| 67 | event = _parse_sse_event([body]) |
| 68 | return [event] if event is not None else [] |
| 69 | |
| 70 | events: list[dict] = [] |
| 71 | data_lines: list[str] = [] |
| 72 | for line in body.splitlines(): |
| 73 | if not line: |
| 74 | event = _parse_sse_event(data_lines) |
| 75 | if event is not None: |
| 76 | events.append(event) |
| 77 | data_lines = [] |
| 78 | continue |
| 79 | if line.startswith("data:"): |
| 80 | data_lines.append(line[5:].lstrip()) |
| 81 | event = _parse_sse_event(data_lines) |
| 82 | if event is not None: |
| 83 | events.append(event) |
| 84 | if not events: |
| 85 | raise RuntimeError("CosyVoice streaming response contains no events") |
| 86 | return events |
| 87 | |
| 88 | |
| 89 | def _post_streaming( |
| 90 | url: str, |
| 91 | *, |
| 92 | api_key: str, |
| 93 | payload: dict, |
| 94 | ) -> list[dict]: |
| 95 | body = json.dumps(payload, ensure_ascii=False).encode("utf-8") |
| 96 | req = request.Request( |
| 97 | url, |
| 98 | data=body, |
| 99 | headers={ |
| 100 | "Authorization": f"Bearer {api_key}", |
| 101 | "Content-Type": "application/json", |
| 102 | "X-DashScope-SSE": "enable", |
| 103 | }, |
| 104 | method="POST", |
| 105 | ) |
| 106 | try: |
| 107 | with request.urlopen(req, timeout=180) as response: |
| 108 | return _parse_sse(response.read()) |
| 109 | except error.HTTPError as exc: |
| 110 | raise RuntimeError(read_http_error(exc)) from exc |
| 111 | except error.URLError as exc: |
| 112 | raise RuntimeError( |
| 113 | f"CosyVoice streaming request failed: {exc.reason}" |
| 114 | ) from exc |
| 115 | |
| 116 | |
| 117 | def _ticks(value: object, *, field: str) -> int: |
| 118 | if isinstance(value, bool): |
| 119 | raise RuntimeError(f"CosyVoice word {field} is not numeric") |
| 120 | try: |
| 121 | numeric = float(value) |
| 122 | except (TypeError, ValueError) as exc: |
| 123 | raise RuntimeError(f"CosyVoice word {field} is not numeric") from exc |
| 124 | if not math.isfinite(numeric): |
| 125 | raise RuntimeError(f"CosyVoice word {field} is not finite") |
| 126 | return int(round(numeric * _TICKS_PER_MILLISECOND)) |
| 127 | |
| 128 | |
| 129 | def _word_boundaries(events: list[dict]) -> tuple[list[dict], str]: |
| 130 | sentence_text: dict[int, str] = {} |
| 131 | sentence_words: dict[int, list] = {} |
| 132 | audio_url = "" |
| 133 | |
| 134 | for event in events: |
| 135 | code = event.get("code") |
| 136 | message = event.get("message") |
| 137 | if code or message: |
| 138 | details = ": ".join(str(value) for value in (code, message) if value) |
| 139 | raise RuntimeError(f"CosyVoice TTS failed: {details}") |
| 140 | output = event.get("output") or {} |
| 141 | if not isinstance(output, dict): |
| 142 | continue |
| 143 | audio = output.get("audio") or {} |
| 144 | if isinstance(audio, dict) and isinstance(audio.get("url"), str): |
| 145 | audio_url = audio["url"] |
| 146 | |
| 147 | sentence = output.get("sentence") or {} |
| 148 | if not isinstance(sentence, dict): |
| 149 | continue |
| 150 | index = sentence.get("index") |
| 151 | if isinstance(index, bool) or not isinstance(index, int) or index < 0: |
| 152 | continue |
| 153 | original_text = output.get("original_text") |
| 154 | if isinstance(original_text, str) and original_text: |
| 155 | sentence_text[index] = original_text |
| 156 | words = sentence.get("words") |
| 157 | if isinstance(words, list) and words: |
| 158 | sentence_words[index] = words |
| 159 | |
| 160 | if not audio_url: |
| 161 | raise RuntimeError("CosyVoice streaming response contains no audio URL") |
| 162 | if not sentence_words: |
| 163 | raise RuntimeError( |
| 164 | "CosyVoice returned no word timestamps. Use a cloned voice from " |
| 165 | "cosyvoice-v3.5-plus/flash, cosyvoice-v3-plus/flash, or cosyvoice-v2, " |
| 166 | "or a system voice marked as timestamp-supported; otherwise pass " |
| 167 | "--cosyvoice-audio-only." |
| 168 | ) |
| 169 | |
| 170 | boundaries: list[dict] = [] |
| 171 | for sentence_index in sorted(sentence_words): |
| 172 | original_text = sentence_text.get(sentence_index, "") |
| 173 | for item in sentence_words[sentence_index]: |
| 174 | if not isinstance(item, dict): |
| 175 | raise RuntimeError("CosyVoice word timing is not an object") |
| 176 | word = item.get("text") |
| 177 | begin_index = item.get("begin_index") |
| 178 | end_index = item.get("end_index") |
| 179 | if ( |
| 180 | original_text |
| 181 | and isinstance(begin_index, int) |
| 182 | and not isinstance(begin_index, bool) |
| 183 | and isinstance(end_index, int) |
| 184 | and not isinstance(end_index, bool) |
| 185 | and 0 <= begin_index < end_index <= len(original_text) |
| 186 | ): |
| 187 | word = original_text[begin_index:end_index] |
| 188 | if not isinstance(word, str) or not word: |
| 189 | raise RuntimeError("CosyVoice word timing contains no text") |
| 190 | start = _ticks(item.get("begin_time"), field="start time") |
| 191 | end = _ticks(item.get("end_time"), field="end time") |
| 192 | if start < 0 or end <= start: |
| 193 | raise RuntimeError( |
| 194 | "CosyVoice response contains an invalid word interval" |
| 195 | ) |
| 196 | if boundaries and start < boundaries[-1]["offset"]: |
| 197 | raise RuntimeError( |
| 198 | "CosyVoice word timestamps are not in chronological order" |
| 199 | ) |
| 200 | boundaries.append({ |
| 201 | "text": word, |
| 202 | "offset": start, |
| 203 | "duration": end - start, |
| 204 | }) |
| 205 | return boundaries, audio_url |
| 206 | |
| 207 | |
| 208 | def generate( |
| 209 | text: str, |
| 210 | output_path: Path, |
| 211 | *, |
| 212 | api_key: str, |
| 213 | voice_id: str, |
| 214 | model: str, |
| 215 | audio_format: str, |
| 216 | sample_rate: int, |
| 217 | volume: int | None, |
| 218 | rate: float | None, |
| 219 | pitch: float | None, |
| 220 | instruction: str | None, |
| 221 | language_hint: str | None, |
| 222 | base_url: str | None, |
| 223 | subtitle_path: Path | None = None, |
| 224 | subtitle_max_chars: int = DEFAULT_SUBTITLE_MAX_CHARS, |
| 225 | ) -> None: |
| 226 | input_payload: dict[str, object] = { |
| 227 | "text": text, |
| 228 | "voice": voice_id, |
| 229 | "format": audio_format, |
| 230 | "sample_rate": sample_rate, |
| 231 | } |
| 232 | if volume is not None: |
| 233 | input_payload["volume"] = volume |
| 234 | if rate is not None: |
| 235 | input_payload["rate"] = rate |
| 236 | if pitch is not None: |
| 237 | input_payload["pitch"] = pitch |
| 238 | if instruction: |
| 239 | input_payload["instruction"] = instruction |
| 240 | if language_hint: |
| 241 | input_payload["language_hints"] = [language_hint] |
| 242 | |
| 243 | payload = { |
| 244 | "model": model, |
| 245 | "input": input_payload, |
| 246 | } |
| 247 | url = resolve_url(base_url) |
| 248 | |
| 249 | if subtitle_path is not None: |
| 250 | input_payload["word_timestamp_enabled"] = True |
| 251 | events = _post_streaming( |
| 252 | url, |
| 253 | api_key=api_key, |
| 254 | payload=payload, |
| 255 | ) |
| 256 | boundaries, audio_url = _word_boundaries(events) |
| 257 | subtitle = format_word_timed_srt( |
| 258 | text, |
| 259 | boundaries, |
| 260 | subtitle_max_chars, |
| 261 | provider_label="CosyVoice", |
| 262 | ) |
| 263 | publish_audio_subtitle( |
| 264 | get_bytes(audio_url), |
| 265 | output_path, |
| 266 | subtitle, |
| 267 | subtitle_path, |
| 268 | ) |
| 269 | return |
| 270 | |
| 271 | data = post_json( |
| 272 | url, |
| 273 | headers={"Authorization": f"Bearer {api_key}"}, |
| 274 | payload=payload, |
| 275 | timeout=180, |
| 276 | ) |
| 277 | audio = (data.get("output") or {}).get("audio") or {} |
| 278 | audio_url = audio.get("url") |
| 279 | if not audio_url: |
| 280 | raise RuntimeError(f"CosyVoice response missing audio URL: {data}") |
| 281 | download_audio(audio_url, output_path) |
| 282 | |
| 283 | |
| 284 | def print_voices() -> None: |
| 285 | print("CosyVoice voices are selected by voice.") |
| 286 | print("Use a system voice name or a cloned/designed voice_id from CosyVoice.") |
| 287 | print("Timestamp-supported system voice example: longanyang") |
| 288 | print( |
| 289 | "For SRT, use a supported CosyVoice cloned voice or a system voice " |
| 290 | "marked timestamp-supported in the provider catalog." |
| 291 | ) |
| 292 |