# -*- coding: utf-8 -*- import asyncio import base64 import json import os import sys import aiohttp from loguru import logger class VolcEngineTTSClient: """ VolcEngine TTS客户端 参数范围说明: - speed_rate: -50~100 (100代表2倍速, -50代表0.5倍速, 0为正常语速) - loudness_rate: -50~100 (100代表2倍音量, -50代表0.5倍音量, 0为正常音量) - emotion_scale: 1-5 """ def __init__(self, voices_list_file=None): self.url = "https://openspeech.bytedance.com/api/v3/tts/unidirectional" self.appid = os.getenv("VOLCENGINE_APPID") self.access_token = os.getenv("VOLCENGINE_ACCESS_TOKEN") self.proxy = os.getenv("HTTPS_PROXY", None) if self.proxy: logger.info(f"volcengine tts use proxy: {self.proxy}") if voices_list_file is not None: with open(voices_list_file, "r", encoding="utf-8") as f: self.voices_list = json.load(f) else: self.voices_list = None def get_voice_list(self): return self.voices_list async def tts_http_stream(self, headers, params, audio_save_path): """执行TTS流式请求""" try: logger.info(f"volcengine tts params: {params}") audio_data = bytearray() total_audio_size = 0 async with aiohttp.ClientSession() as session: async with session.post(self.url, json=params, headers=headers, proxy=self.proxy) as response: response.raise_for_status() async for chunk in response.content: if not chunk: continue try: data = json.loads(chunk.decode("utf-8").strip()) if data.get("code", 0) == 0 and "data" in data and data["data"]: chunk_audio = base64.b64decode(data["data"]) audio_size = len(chunk_audio) total_audio_size += audio_size audio_data.extend(chunk_audio) continue if data.get("code", 0) == 0 and "sentence" in data and data["sentence"]: continue if data.get("code", 0) == 20000000: break if data.get("code", 0) > 0: logger.warning(f"volcengine tts error response: {data}") break except Exception as e: logger.warning(f"Failed to parse volcengine tts chunk: {e}") # save audio file if audio_data: with open(audio_save_path, "wb") as f: f.write(audio_data) logger.info(f"audio saved to {audio_save_path}, audio size: {len(audio_data) / 1024:.2f} KB") # set correct permissions os.chmod(audio_save_path, 0o644) return True else: logger.warning("No tts audio data received") return False except Exception as e: logger.warning(f"VolcEngineTTSClient tts request failed: {e}") return False async def tts_request( self, text, voice_type, context_texts="", emotion="", emotion_scale=4, speed_rate=0, loudness_rate=0, output="tts_output.mp3", resource_id="seed-tts-2.0", app_key="aGjiRDfUWi", uid="123123", format="mp3", sample_rate=24000, enable_timestamp=True, ): """ 执行TTS请求 Args: text: 要转换的文本 voice_type: 声音类型 emotion: 情感类型 emotion_scale: 情感强度 (1-5) speed_rate: 语速调节 (-50~100, 100代表2倍速, -50代表0.5倍速, 0为正常语速) loudness_rate: 音量调节 (-50~100, 100代表2倍音量, -50代表0.5倍音量, 0为正常音量) output: 输出文件路径 resource_id: 资源ID app_key: 应用密钥 uid: 用户ID format: 音频格式 sample_rate: 采样率 enable_timestamp: 是否启用时间戳 """ # 验证参数范围 if not (-50 <= speed_rate <= 100): logger.warning(f"speed_rate {speed_rate} 超出有效范围 [-50, 100]，将使用默认值 0") speed_rate = 0 if not (-50 <= loudness_rate <= 100): logger.warning(f"loudness_rate {loudness_rate} 超出有效范围 [-50, 100]，将使用默认值 0") loudness_rate = 0 if not (1 <= emotion_scale <= 5): logger.warning(f"emotion_scale {emotion_scale} 超出有效范围 [1, 5]，将使用默认值 3") emotion_scale = 3 headers = { "X-Api-App-Id": self.appid, "X-Api-Access-Key": self.access_token, "X-Api-Resource-Id": resource_id, "X-Api-App-Key": app_key, "Content-Type": "application/json", "Connection": "keep-alive", } additions = json.dumps({"explicit_language": "zh", "disable_markdown_filter": True, "enable_timestamp": True, "context_texts": [context_texts] if context_texts else None}) payload = { "user": {"uid": uid}, "req_params": { "text": text, "speaker": voice_type, "audio_params": { "format": format, "sample_rate": sample_rate, "enable_timestamp": enable_timestamp, "emotion": emotion, "emotion_scale": emotion_scale, "speed_rate": speed_rate, "loudness_rate": loudness_rate, }, "additions": additions, }, } success = await self.tts_http_stream(headers=headers, params=payload, audio_save_path=output) if success: logger.info(f"VolcEngineTTSClient tts request for '{text}': success") else: logger.warning(f"VolcEngineTTSClient tts request for '{text}': failed") return success async def test(args): """ TTS测试函数 Args: args: list, e.g. [text, voice_type, emotion, emotion_scale, speed_rate, loudness_rate, output, resource_id, app_key, uid, format, sample_rate, enable_timestamp] Provide as many as needed, from left to right. Parameter ranges: - speed_rate: -50~100 (100代表2倍速, -50代表0.5倍速, 0为正常语速) - loudness_rate: -50~100 (100代表2倍音量, -50代表0.5倍音量, 0为正常音量) - emotion_scale: 1-5 """ client = VolcEngineTTSClient() # 设置默认参数 params = { "text": "", "voice_type": "", "context_texts": "", "emotion": "", "emotion_scale": 4, "speed_rate": 0, "loudness_rate": 0, "output": "tts_output.mp3", "resource_id": "seed-tts-2.0", "app_key": "aGjiRDfUWi", "uid": "123123", "format": "mp3", "sample_rate": 24000, "enable_timestamp": True, } keys = list(params.keys()) # 覆盖默认参数 for i, arg in enumerate(args): # 类型转换 if keys[i] == "sample_rate": params[keys[i]] = int(arg) elif keys[i] == "enable_timestamp": # 支持多种布尔输入 params[keys[i]] = str(arg).lower() in ("1", "true", "yes", "on") else: params[keys[i]] = arg await client.tts_request( params["text"], params["voice_type"], params["context_texts"], params["emotion"], params["emotion_scale"], params["speed_rate"], params["loudness_rate"], params["output"], params["resource_id"], params["app_key"], params["uid"], params["format"], params["sample_rate"], params["enable_timestamp"], ) if __name__ == "__main__": asyncio.run(test(sys.argv[1:]))