| 1 | """MiniMax T2A backend for narration audio generation.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import binascii |
| 6 | import json |
| 7 | import math |
| 8 | import os |
| 9 | from pathlib import Path |
| 10 | |
| 11 | from tts_backends.backend_common import ( |
| 12 | extension_from_format, |
| 13 | get_bytes, |
| 14 | post_json, |
| 15 | publish_audio_bytes, |
| 16 | publish_audio_subtitle, |
| 17 | read_api_key, |
| 18 | ) |
| 19 | from tts_backends.backend_edge import ( |
| 20 | DEFAULT_SUBTITLE_MAX_CHARS, |
| 21 | format_word_timed_srt, |
| 22 | ) |
| 23 | |
| 24 | |
| 25 | DEFAULT_ENDPOINT = "https://api.minimaxi.com/v1/t2a_v2" |
| 26 | DEFAULT_MODEL = "speech-2.8-hd" |
| 27 | _TICKS_PER_MILLISECOND = 10_000 |
| 28 | |
| 29 | # International fallback: set MINIMAX_TTS_BASE_URL=https://api.minimax.io if needed. |
| 30 | |
| 31 | |
| 32 | def output_extension(audio_format: str) -> str: |
| 33 | return extension_from_format(audio_format) |
| 34 | |
| 35 | |
| 36 | def read_minimax_api_key(env_name: str) -> str: |
| 37 | return read_api_key(env_name, label="MiniMax") |
| 38 | |
| 39 | |
| 40 | def resolve_url(base_url: str | None = None) -> str: |
| 41 | base = (base_url or os.environ.get("MINIMAX_TTS_BASE_URL") or DEFAULT_ENDPOINT).rstrip("/") |
| 42 | if base.endswith("/t2a_v2"): |
| 43 | return base |
| 44 | if base.endswith("/v1"): |
| 45 | return base + "/t2a_v2" |
| 46 | return base + "/v1/t2a_v2" |
| 47 | |
| 48 | |
| 49 | def _ticks(value: object, *, field: str) -> int: |
| 50 | if isinstance(value, bool): |
| 51 | raise RuntimeError(f"MiniMax subtitle {field} is not numeric") |
| 52 | try: |
| 53 | numeric = float(value) |
| 54 | except (TypeError, ValueError) as exc: |
| 55 | raise RuntimeError(f"MiniMax subtitle {field} is not numeric") from exc |
| 56 | if not math.isfinite(numeric): |
| 57 | raise RuntimeError(f"MiniMax subtitle {field} is not finite") |
| 58 | return int(round(numeric * _TICKS_PER_MILLISECOND)) |
| 59 | |
| 60 | |
| 61 | def _word_boundaries(raw: object) -> list[dict]: |
| 62 | if not isinstance(raw, list) or not raw: |
| 63 | raise RuntimeError("MiniMax subtitle response contains no sentence blocks") |
| 64 | |
| 65 | boundaries: list[dict] = [] |
| 66 | for sentence in raw: |
| 67 | if not isinstance(sentence, dict): |
| 68 | raise RuntimeError("MiniMax subtitle sentence block is not an object") |
| 69 | words = sentence.get("timestamped_words") |
| 70 | if not isinstance(words, list) or not words: |
| 71 | raise RuntimeError("MiniMax subtitle sentence block contains no word timings") |
| 72 | for item in words: |
| 73 | if not isinstance(item, dict): |
| 74 | raise RuntimeError("MiniMax subtitle word timing is not an object") |
| 75 | word = item.get("word") |
| 76 | if not isinstance(word, str) or not word: |
| 77 | raise RuntimeError("MiniMax subtitle word timing contains no text") |
| 78 | start = _ticks(item.get("time_begin"), field="start time") |
| 79 | end = _ticks(item.get("time_end"), field="end time") |
| 80 | if start < 0 or end <= start: |
| 81 | raise RuntimeError( |
| 82 | "MiniMax subtitle response contains an invalid word interval" |
| 83 | ) |
| 84 | if boundaries and start < boundaries[-1]["offset"]: |
| 85 | raise RuntimeError("MiniMax subtitle words are not in chronological order") |
| 86 | boundaries.append( |
| 87 | { |
| 88 | "text": word, |
| 89 | "offset": start, |
| 90 | "duration": end - start, |
| 91 | } |
| 92 | ) |
| 93 | return boundaries |
| 94 | |
| 95 | |
| 96 | def _download_subtitle( |
| 97 | url: str, |
| 98 | text: str, |
| 99 | max_chars: int, |
| 100 | ) -> str: |
| 101 | try: |
| 102 | raw = json.loads(get_bytes(url).decode("utf-8-sig")) |
| 103 | except (UnicodeError, json.JSONDecodeError) as exc: |
| 104 | raise RuntimeError("MiniMax subtitle response is not valid UTF-8 JSON") from exc |
| 105 | return format_word_timed_srt( |
| 106 | text, |
| 107 | _word_boundaries(raw), |
| 108 | max_chars, |
| 109 | provider_label="MiniMax", |
| 110 | ) |
| 111 | |
| 112 | |
| 113 | def generate( |
| 114 | text: str, |
| 115 | output_path: Path, |
| 116 | *, |
| 117 | api_key: str, |
| 118 | voice_id: str, |
| 119 | model: str, |
| 120 | audio_format: str, |
| 121 | sample_rate: int, |
| 122 | bitrate: int, |
| 123 | channel: int, |
| 124 | speed: float, |
| 125 | volume: float, |
| 126 | pitch: int, |
| 127 | language_boost: str, |
| 128 | base_url: str | None, |
| 129 | subtitle_path: Path | None = None, |
| 130 | subtitle_max_chars: int = DEFAULT_SUBTITLE_MAX_CHARS, |
| 131 | ) -> None: |
| 132 | payload = { |
| 133 | "model": model, |
| 134 | "text": text, |
| 135 | "stream": False, |
| 136 | "language_boost": language_boost, |
| 137 | "output_format": "hex", |
| 138 | "voice_setting": { |
| 139 | "voice_id": voice_id, |
| 140 | "speed": speed, |
| 141 | "vol": volume, |
| 142 | "pitch": pitch, |
| 143 | }, |
| 144 | "audio_setting": { |
| 145 | "sample_rate": sample_rate, |
| 146 | "bitrate": bitrate, |
| 147 | "format": audio_format, |
| 148 | "channel": channel, |
| 149 | }, |
| 150 | } |
| 151 | if subtitle_path is not None: |
| 152 | payload["subtitle_enable"] = True |
| 153 | payload["subtitle_type"] = "word" |
| 154 | |
| 155 | data = post_json( |
| 156 | resolve_url(base_url), |
| 157 | headers={"Authorization": f"Bearer {api_key}"}, |
| 158 | payload=payload, |
| 159 | timeout=180, |
| 160 | ) |
| 161 | base_resp = data.get("base_resp") or {} |
| 162 | if base_resp.get("status_code") not in (None, 0, "0"): |
| 163 | raise RuntimeError(f"MiniMax TTS failed: {data}") |
| 164 | |
| 165 | response_data = data.get("data") or {} |
| 166 | audio_hex = response_data.get("audio") |
| 167 | if not audio_hex: |
| 168 | raise RuntimeError(f"MiniMax response missing audio data: {data}") |
| 169 | try: |
| 170 | audio = binascii.unhexlify(audio_hex) |
| 171 | except (binascii.Error, ValueError) as exc: |
| 172 | raise RuntimeError("MiniMax response audio is not valid hex data") from exc |
| 173 | |
| 174 | if subtitle_path is None: |
| 175 | publish_audio_bytes(audio, output_path) |
| 176 | return |
| 177 | |
| 178 | subtitle_url = response_data.get("subtitle_file") |
| 179 | if not isinstance(subtitle_url, str) or not subtitle_url.strip(): |
| 180 | trace_id = data.get("trace_id") or "unknown" |
| 181 | raise RuntimeError( |
| 182 | f"MiniMax response missing subtitle file (trace_id={trace_id})" |
| 183 | ) |
| 184 | subtitle = _download_subtitle( |
| 185 | subtitle_url, |
| 186 | text, |
| 187 | subtitle_max_chars, |
| 188 | ) |
| 189 | publish_audio_subtitle(audio, output_path, subtitle, subtitle_path) |
| 190 | |
| 191 | |
| 192 | def print_voices() -> None: |
| 193 | print("MiniMax TTS voices are selected by voice_id.") |
| 194 | print("Use a system voice ID or a cloned voice_id from MiniMax Voice Clone.") |
| 195 | print("Example domestic system voice from MiniMax docs: male-qn-qingse") |
| 196 |