From 17e6abd035b2727d11147abeb04639bf2e812be9 Mon Sep 17 00:00:00 2001 From: Zhengshou Lai Date: Sun, 2 Aug 2026 16:08:13 +0800 Subject: [PATCH] feat(tts): support Volcengine X-Api-Key auth (new api-key mode) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Newer Volcengine speech synthesis auths with a plain API key (secrets. api_keys.volc_tts / VOLC_TTS_API_KEY) via the HTTP endpoint and X-Api-Key header — no AppID/Token, and a quota pool that is separate from the legacy app. synthesize_tts now prefers api_key (HTTP) and falls back to the legacy WebSocket appid/access_token path. STT still uses appid/token (unchanged). --- mytoolkit/api/voice.py | 123 ++++++++++++++++++++++++++++++++---- mytoolkit/commands/voice.py | 99 +++++++++++++++++++++++------ 2 files changed, 188 insertions(+), 34 deletions(-) diff --git a/mytoolkit/api/voice.py b/mytoolkit/api/voice.py index 64c8883..56e5460 100644 --- a/mytoolkit/api/voice.py +++ b/mytoolkit/api/voice.py @@ -9,18 +9,14 @@ from pathlib import Path _RESOURCE_TTS_1 = "volc.service_type.10029" _RESOURCE_TTS_2 = "seed-tts-2.0" _ENDPOINT = "wss://openspeech.bytedance.com/api/v3/tts/unidirectional/stream" +# Newer HTTP endpoint: X-Api-Key auth (no AppID/Token), streamed base64. +# Note: the legacy appid/token path is WebSocket (_ENDPOINT); the api-key path +# is this HTTP endpoint (the /plan/ variant is Agent-Plan-only and rejects it). +_ENDPOINT_HTTP = "https://openspeech.bytedance.com/api/v3/tts/unidirectional" # Product labels: 经典男声 / 经典女声 / 阳光甜妹 -_VOICE_CHOICES = [ - "zh_male_m191_uranus_bigtts", - "zh_female_xiaohe_uranus_bigtts", - "ICL_uranus_zh_female_yuanqitianmei_tob", -] - _DEFAULT_VOICE = "zh_female_xiaohe_uranus_bigtts" -_FORMAT_CHOICES = ["mp3", "wav", "pcm"] - def resource_id_for_speaker(speaker: str) -> str: """Pick Volcengine resource id for a speaker.""" @@ -38,18 +34,38 @@ def synthesize_tts( fmt: str = "mp3", speech_rate: int = 0, output_path: str = "output.mp3", + api_key: str | None = None, ) -> str: - """Synchronous TTS synthesis via Volcengine WebSocket API. + """Synchronous TTS synthesis via Volcengine API. - Returns absolute path to the output audio file. + Prefers the newer plan HTTP endpoint (``api_key``, X-Api-Key auth); falls + back to the legacy WebSocket endpoint (appid + access_token). Returns the + absolute path to the output audio file. """ + if api_key: + return synthesize_tts_http( + api_key, + text=text, + speaker=speaker, + fmt=fmt, + speech_rate=speech_rate, + output_path=output_path, + ) + # Delayed import to avoid circular dependency when bin.commands is not yet loaded - from mytoolkit.protocols import EventType, MsgType, full_client_request, receive_message + from mytoolkit.protocols import ( + EventType, + MsgType, + full_client_request, + receive_message, + ) try: import websockets except ImportError as e: - raise RuntimeError(f"Missing dependency: {e}. Run: pip install websockets") from e + raise RuntimeError( + f"Missing dependency: {e}. Run: pip install websockets" + ) from e headers = { "X-Api-App-Id": appid, @@ -84,7 +100,10 @@ def synthesize_tts( audio_data = bytearray() while True: msg = await receive_message(ws) - if msg.type == MsgType.FullServerResponse and msg.event == EventType.SessionFinished: + if ( + msg.type == MsgType.FullServerResponse + and msg.event == EventType.SessionFinished + ): break elif msg.type == MsgType.AudioOnlyServer: audio_data.extend(msg.payload) @@ -104,3 +123,81 @@ def synthesize_tts( out = Path(output_path) out.write_bytes(audio_data) return str(out.resolve()) + + +def synthesize_tts_http( + api_key: str, + text: str, + *, + speaker: str = _DEFAULT_VOICE, + fmt: str = "mp3", + speech_rate: int = 0, + output_path: str = "output.mp3", + timeout_sec: float = 60.0, +) -> str: + """TTS via Volcengine v3 plan HTTP endpoint (X-Api-Key auth). + + One POST, streamed newline-delimited JSON; each line carries a base64 + audio chunk plus a terminal ``code == 20000000`` line. No AppID/Token. + """ + import base64 + import json + import urllib.error + import urllib.request + + payload = { + "req_params": { + "text": text, + "speaker": speaker, + "audio_params": { + "format": fmt, + "sample_rate": 24000, + "speech_rate": speech_rate, + }, + } + } + req = urllib.request.Request( + _ENDPOINT_HTTP, + data=json.dumps(payload).encode("utf-8"), + method="POST", + ) + req.add_header("X-Api-Key", api_key) + req.add_header("X-Api-Resource-Id", _RESOURCE_TTS_2) + req.add_header("Content-Type", "application/json") + req.add_header("Connection", "keep-alive") + req.add_header("X-Control-Require-Usage-Tokens-Return", "*") + + try: + resp = urllib.request.urlopen(req, timeout=timeout_sec) + except urllib.error.HTTPError as exc: + detail = exc.read()[:300].decode("utf-8", errors="replace") + raise RuntimeError( + f"TTS service error: HTTP {exc.code} {detail}" + ) from exc + except OSError as exc: + raise RuntimeError(f"TTS service error: {exc}") from exc + + audio_data = bytearray() + for raw in resp: + line = raw.decode("utf-8", errors="replace").strip() + if not line: + continue + try: + obj = json.loads(line) + except json.JSONDecodeError: + continue + code = obj.get("code", 0) + if obj.get("data"): + audio_data.extend(base64.b64decode(obj["data"])) + if code == 20000000: # terminal success + break + if code > 0: + raise RuntimeError( + f"TTS service error: {json.dumps(obj, ensure_ascii=False)[:300]}" + ) + if not audio_data: + raise RuntimeError("No audio data received from TTS service") + + out = Path(output_path) + out.write_bytes(audio_data) + return str(out.resolve()) diff --git a/mytoolkit/commands/voice.py b/mytoolkit/commands/voice.py index 6cd8495..c355aac 100644 --- a/mytoolkit/commands/voice.py +++ b/mytoolkit/commands/voice.py @@ -22,7 +22,11 @@ _FORMAT_CHOICES = ["mp3", "wav", "pcm"] def _get_credentials(): - """Get Volcengine credentials from config.""" + """Get Volcengine appid + access_token (STT, legacy TTS). + + STT still uses the legacy appid/token app auth; the newer X-Api-Key path is + TTS-only and lives in ``_get_tts_credentials``. + """ appid = config.resolve_key("secrets.volcengine.app_id", "VOLC_APPID") access_token = config.resolve_key( "secrets.volcengine.access_token", "VOLC_ACCESS_TOKEN" @@ -32,6 +36,23 @@ def _get_credentials(): return appid, access_token +def _get_tts_credentials() -> dict: + """TTS credentials from config (new api_key preferred). + + Newer X-Api-Key auth (``secrets.api_keys.volc_tts`` / ``VOLC_TTS_API_KEY``) + takes precedence; legacy appid + access_token is the fallback. + """ + api_key = config.resolve_key( + "secrets.api_keys.volc_tts", "VOLC_TTS_API_KEY" + ) + if api_key: + return {"api_key": api_key} + appid, access_token = _get_credentials() + if not appid or not access_token: + return {} + return {"appid": appid, "token": access_token} + + def _synthesize_sync( text: str, speaker: str, @@ -39,18 +60,20 @@ def _synthesize_sync( speech_rate: int, output_path: str, ) -> str: - """Synchronous wrapper around async WebSocket TTS.""" - appid, access_token = _get_credentials() - if not appid or not access_token: + """Synchronous wrapper around async TTS.""" + creds = _get_tts_credentials() + if not creds: raise click.UsageError( - "Volcengine credentials not set. Run:\n" - " mytoolkit env set secrets.volcengine.app_id \n" + "Volcengine TTS credentials not set. Run:\n" + " mytoolkit env set secrets.api_keys.volc_tts \n" + " (or legacy) mytoolkit env set secrets.volcengine.app_id \n" " mytoolkit env set secrets.volcengine.access_token " ) return synthesize_tts( - appid=appid, - token=access_token, + appid=creds.get("appid", ""), + token=creds.get("token", ""), + api_key=creds.get("api_key"), text=text, speaker=speaker, fmt=fmt, @@ -67,7 +90,9 @@ def voice(): @voice.command("tts") @click.argument("text", required=False) -@click.option("-f", "--file", type=click.Path(exists=True), help="从文件读取文本") +@click.option( + "-f", "--file", type=click.Path(exists=True), help="从文件读取文本" +) @click.option( "-v", "--voice", @@ -76,7 +101,13 @@ def voice(): default=_DEFAULT_VOICE, help="音色", ) -@click.option("--format", "fmt", type=click.Choice(_FORMAT_CHOICES), default="mp3", help="输出格式") +@click.option( + "--format", + "fmt", + type=click.Choice(_FORMAT_CHOICES), + default="mp3", + help="输出格式", +) @click.option( "--speed", type=click.IntRange(-50, 100), @@ -144,7 +175,9 @@ def stt_cmd(input_path, output, history_file, hotword, correct): if history_file: history = [ ln.strip() - for ln in Path(history_file).read_text(encoding="utf-8").splitlines() + for ln in Path(history_file) + .read_text(encoding="utf-8") + .splitlines() if ln.strip() ] hotwords = list(hotword) if hotword else None @@ -168,7 +201,9 @@ def stt_cmd(input_path, output, history_file, hotword, correct): ark_api_key=ark_key, deepseek_api_key=deepseek_key, ark_model=ark_model, - on_skip=lambda msg: click.echo(f"ASR correct skipped: {msg}", err=True), + on_skip=lambda msg: click.echo( + f"ASR correct skipped: {msg}", err=True + ), ) if output: Path(output).write_text(text + "\n", encoding="utf-8") @@ -179,15 +214,26 @@ def stt_cmd(input_path, output, history_file, hotword, correct): def _natural_sort_key(path: Path): name = path.stem - return [int(text) if text.isdigit() else text.lower() - for text in re.split(r"(\d+)", name)] + return [ + int(text) if text.isdigit() else text.lower() + for text in re.split(r"(\d+)", name) + ] @voice.command("tts-batch") @click.argument("text_dir", type=click.Path(exists=True, file_okay=False)) -@click.option("--voice", "speaker", type=click.Choice(_VOICE_CHOICES), default=_DEFAULT_VOICE) -@click.option("--output-dir", "-o", type=click.Path(), help="输出目录（默认与输入相同）") -@click.option("--format", "fmt", type=click.Choice(_FORMAT_CHOICES), default="mp3") +@click.option( + "--voice", + "speaker", + type=click.Choice(_VOICE_CHOICES), + default=_DEFAULT_VOICE, +) +@click.option( + "--output-dir", "-o", type=click.Path(), help="输出目录（默认与输入相同）" +) +@click.option( + "--format", "fmt", type=click.Choice(_FORMAT_CHOICES), default="mp3" +) @handle_errors def tts_batch_cmd(text_dir, speaker, output_dir, fmt): """批量将目录下的 .txt 文件转为语音。""" @@ -213,10 +259,17 @@ def tts_batch_cmd(text_dir, speaker, output_dir, fmt): @voice.command("tts-script") @click.argument("script", type=click.Path(exists=True)) @click.option("--output-dir", "-o", type=click.Path(), default="audio") -@click.option("--voice", "speaker", type=click.Choice(_VOICE_CHOICES), default=_DEFAULT_VOICE) +@click.option( + "--voice", + "speaker", + type=click.Choice(_VOICE_CHOICES), + default=_DEFAULT_VOICE, +) @click.option("--prefix", "-p", default="page") @click.option("--zero-pad", "-z", type=int, default=2) -@click.option("--format", "fmt", type=click.Choice(_FORMAT_CHOICES), default="mp3") +@click.option( + "--format", "fmt", type=click.Choice(_FORMAT_CHOICES), default="mp3" +) @handle_errors def tts_script_cmd(script, output_dir, speaker, prefix, zero_pad, fmt): """从 PPT 配音脚本（Markdown，每页一个引用块）批量生成 TTS。""" @@ -247,8 +300,12 @@ def tts_script_cmd(script, output_dir, speaker, prefix, zero_pad, fmt): continue filename = f"{prefix}{str(i).zfill(zero_pad)}.{fmt}" out_path = output_dir / filename - click.echo(f"[{i}] {filename}: {script_text[:40]}{'...' if len(script_text) > 40 else ''}") + click.echo( + f"[{i}] {filename}: {script_text[:40]}{'...' if len(script_text) > 40 else ''}" + ) _synthesize_sync(script_text, speaker, fmt, 0, str(out_path)) # type: ignore[reportCallIssue] generated += 1 - click.echo(f"\nDone: {generated} generated, {skipped} skipped (empty scripts).") + click.echo( + f"\nDone: {generated} generated, {skipped} skipped (empty scripts)." + )