201 lines
8.3 KiB
Python
201 lines
8.3 KiB
Python
import os
|
||
import asyncio
|
||
import aiohttp
|
||
import base64
|
||
from pathlib import Path
|
||
from pydub import AudioSegment
|
||
import io
|
||
import logging
|
||
from dotenv import load_dotenv
|
||
import shutil
|
||
import uuid
|
||
|
||
logging.basicConfig(
|
||
level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
|
||
)
|
||
logger = logging.getLogger("volcano_tts_test")
|
||
|
||
|
||
def check_dependencies():
|
||
dependencies = ["ffmpeg", "ffprobe"]
|
||
missing = []
|
||
for dep in dependencies:
|
||
if not shutil.which(dep):
|
||
missing.append(dep)
|
||
if missing:
|
||
logger.error(f"缺少必要依赖: {', '.join(missing)}")
|
||
logger.error(
|
||
"请安装缺失的依赖项。在Ubuntu上可以使用: sudo apt-get install ffmpeg"
|
||
)
|
||
return False
|
||
return True
|
||
|
||
|
||
env_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "volcano.env")
|
||
if (os.path.exists(env_path)):
|
||
load_dotenv(env_path)
|
||
logger.info(f"已加载环境变量文件: {env_path}")
|
||
else:
|
||
logger.warning(f"环境变量文件不存在: {env_path}")
|
||
|
||
|
||
async def test_volcano_tts():
|
||
|
||
if not check_dependencies():
|
||
return
|
||
api_access_token = os.getenv("VOLCANO_ACCESS_TOKEN", "")
|
||
appid = os.getenv("VOLCANO_APP_ID", "")
|
||
cluster = os.getenv("VOLCANO_CLUSTER", "volcano_tts")
|
||
base_url = os.getenv(
|
||
"VOLCANO_TTS_BASE_URL", "https://openspeech.bytedance.com/api/v1/tts"
|
||
)
|
||
role_name = os.getenv("ROLE_NAME", "Dundun Chicken")
|
||
voice_type = os.getenv("VOICE_TYPE", "S_NkHcFJam1")
|
||
speed_ratio = float(os.getenv("SPEED_RATIO", "1.0"))
|
||
phrases = [
|
||
"It's being upgraded. Please don't cut off the power.",
|
||
"Enter the network configuration mode.",
|
||
"Exit the network configuration mode.",
|
||
"I'll speak louder.",
|
||
"I'll speak softer.",
|
||
"There's no internet connection. It's time to get off work.",
|
||
f"Hello , I'm {role_name}",
|
||
"Connected to the network.",
|
||
"I'm out of power.",
|
||
"The upgrade was successful.",
|
||
"There seem to be some minor issues with the upgrade. Let's try it again.",
|
||
"It's already at the maximum volume.",
|
||
"You're so talkative.",
|
||
]
|
||
file_prefixes = [
|
||
"upgrading",
|
||
"enter_network_config",
|
||
"exit_network_config",
|
||
"volume_up",
|
||
"volume_down",
|
||
"network_lost",
|
||
"wakeup",
|
||
"network_connected",
|
||
"low_energy",
|
||
"upgrade_success",
|
||
"upgrade_failed",
|
||
"max_volume",
|
||
"too_talkative"
|
||
]
|
||
assets_dir = os.getenv("ASSETS_DIR", "assets")
|
||
output_dir = Path(assets_dir) / "tts_audio"
|
||
output_dir.mkdir(parents=True, exist_ok=True)
|
||
if not api_access_token or not appid:
|
||
logger.error("缺少必要的配置: VOLCANO_ACCESS_TOKEN 或 VOLCANO_APP_ID")
|
||
return
|
||
logger.info(f"开始测试火山引擎TTS服务，角色: {role_name}, 音色: {voice_type}")
|
||
for phrase, file_prefix in zip(phrases, file_prefixes):
|
||
output_file_prefix = str(output_dir / file_prefix)
|
||
unique_reqid = str(uuid.uuid4())
|
||
payload = {
|
||
"app": {
|
||
"appid": appid,
|
||
"token": "access_token",
|
||
"cluster": cluster,
|
||
},
|
||
"user": {"uid": "test_user"},
|
||
"audio": {
|
||
"voice_type": voice_type,
|
||
"encoding": "mp3",
|
||
"speed_ratio": float(speed_ratio),
|
||
},
|
||
"request": {"reqid": unique_reqid, "text": phrase, "operation": "query"},
|
||
}
|
||
headers = {
|
||
"Authorization": f"Bearer;{api_access_token}",
|
||
"Content-Type": "application/json",
|
||
}
|
||
logger.info(f"准备发送请求，文本内容: {phrase}, 请求ID: {unique_reqid}")
|
||
try:
|
||
async with aiohttp.ClientSession() as session:
|
||
async with session.post(
|
||
base_url, headers=headers, json=payload
|
||
) as response:
|
||
logid = response.headers.get('X-Tt-Logid', 'unknown')
|
||
logger.debug(f"服务端返回Logid: {logid}")
|
||
|
||
if response.status == 200:
|
||
resp_json = await response.json()
|
||
if "code" in resp_json:
|
||
code = resp_json["code"]
|
||
if code == 3000: # 成功状态码
|
||
audio_base64 = resp_json.get("data")
|
||
if audio_base64:
|
||
audio_data = base64.b64decode(audio_base64)
|
||
duration_ms = int(
|
||
resp_json.get("addition", {}).get("duration", "0")
|
||
)
|
||
duration = duration_ms / 1000 # 转换为秒
|
||
output_file = f"{output_file_prefix}.mp3"
|
||
audio_stream = io.BytesIO(audio_data)
|
||
sound = AudioSegment.from_file(
|
||
audio_stream, format="mp3"
|
||
)
|
||
sound = (
|
||
sound.set_frame_rate(16000)
|
||
.set_sample_width(2)
|
||
.set_channels(1)
|
||
)
|
||
sound.export(output_file, format="mp3", bitrate="16k")
|
||
logger.info(f"TTS合成成功! 音频时长: {duration}秒")
|
||
logger.info(f"保存音频到: {output_file}")
|
||
else:
|
||
error_msg = "音频数据不存在"
|
||
logger.error(f"TTS合成失败: {error_msg}")
|
||
else:
|
||
error_code = resp_json.get('code')
|
||
error_msg = resp_json.get('message', '未知错误')
|
||
error_info = get_error_description(error_code, error_msg)
|
||
logger.error(f"TTS合成失败: {error_info} (LogID: {logid})")
|
||
else:
|
||
logger.error(f"返回数据格式异常，缺少code字段: {resp_json}")
|
||
else:
|
||
error_text = await response.text()
|
||
logger.error(
|
||
f"TTS合成失败，状态码: {response.status}, 错误: {error_text} (LogID: {logid})"
|
||
)
|
||
except Exception as e:
|
||
logger.error(f"TTS请求出错: {str(e)}")
|
||
await asyncio.sleep(1)
|
||
|
||
|
||
def get_error_description(code, message):
|
||
"""根据错误码返回详细的错误描述"""
|
||
error_descriptions = {
|
||
3001: "无效的请求，请检查参数",
|
||
3003: "并发超限，请降低请求频率或增购并发",
|
||
3005: "后端服务忙，请稍后重试",
|
||
3006: "服务中断，请求已完成/失败之后，相同reqid再次请求",
|
||
3010: "文本长度超限，请减少文本长度",
|
||
3011: "无效文本，请检查文本内容",
|
||
3030: "处理超时，请重试或检查文本",
|
||
3031: "处理错误，后端出现异常",
|
||
3032: "等待获取音频超时，请重试",
|
||
3040: "后端链路连接错误，请重试",
|
||
3050: "音色不存在，请检查voice_type参数"
|
||
}
|
||
|
||
if "quota exceeded for types: xxxxxxxxx_lifetime" in message:
|
||
return "试用版用量用完，需开通正式版才能继续使用"
|
||
elif "quota exceeded for types: concurrency" in message:
|
||
return "并发超过限定值，需减少并发调用或增购并发"
|
||
elif "Init Engine Instance failed" in message:
|
||
return "voice_type或cluster参数错误"
|
||
elif "illegal input text" in message:
|
||
return "文本无效，无可合成的有效内容"
|
||
elif "requested grant not found" in message:
|
||
return "鉴权失败，请检查appid和token是否正确"
|
||
elif "access denied" in message:
|
||
return "未拥有当前音色授权，请在控制台购买该音色"
|
||
|
||
return f"错误码: {code}, 错误信息: {message} - {error_descriptions.get(code, '未知错误')}"
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(test_volcano_tts())
|