375 lines
14 KiB
Python
375 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
简单的 Gemini TTS 测试脚本
|
|
通过 Vertex AI 鉴权调用 Gemini TTS API，将文本转为语音并保存为 WAV 文件。
|
|
|
|
使用方法:
|
|
pip install google-genai toml
|
|
python gemini_tts_test.py --text "你好，世界" --voice Kore --output output.wav
|
|
|
|
依赖:
|
|
pip install google-genai toml
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import struct
|
|
import sys
|
|
import wave
|
|
from pathlib import Path
|
|
|
|
import toml
|
|
from google import genai
|
|
from google.genai import types
|
|
from google.oauth2 import service_account
|
|
from niceutils import *
|
|
from upload_cos import upload_func
|
|
|
|
# 默认配置
|
|
DEFAULT_MODEL = "gemini-2.5-pro-tts"
|
|
DEFAULT_VOICE = "Kore"
|
|
DEFAULT_SAMPLE_RATE = 24000
|
|
DEFAULT_CHANNELS = 1
|
|
DEFAULT_BITS_PER_SAMPLE = 16
|
|
DEFAULT_PROJECT_ID = "dfvertex-tt6"
|
|
DEFAULT_LOCATION = "us-central1"
|
|
|
|
|
|
def load_vertex_ai_credentials(auth_toml_path: str) -> dict:
|
|
"""从 auth.toml 加载 Vertex AI 服务账号凭证"""
|
|
config = toml.load(auth_toml_path)
|
|
vertex_ai = config.get("vertex-ai", {})
|
|
if not vertex_ai.get("private_key"):
|
|
raise ValueError("auth.toml 中未找到 [vertex-ai] 凭证配置")
|
|
return vertex_ai
|
|
|
|
|
|
def create_credentials(vertex_ai_config: dict) -> service_account.Credentials:
|
|
"""根据 auth.toml 中的配置创建 GCP 服务账号凭证"""
|
|
# 构建与 GCP 服务账号密钥 JSON 格式一致的 dict
|
|
creds_info = {
|
|
"type": vertex_ai_config.get("type", "service_account"),
|
|
"project_id": vertex_ai_config.get("project_id", ""),
|
|
"private_key_id": vertex_ai_config.get("private_key_id", ""),
|
|
"private_key": vertex_ai_config.get("private_key", ""),
|
|
"client_email": vertex_ai_config.get("client_email", ""),
|
|
"client_id": vertex_ai_config.get("client_id", ""),
|
|
"auth_uri": vertex_ai_config.get("auth_uri", ""),
|
|
"token_uri": vertex_ai_config.get("token_uri", ""),
|
|
"auth_provider_x509_cert_url": vertex_ai_config.get("auth_provider_x509_cert_url", ""),
|
|
"client_x509_cert_url": vertex_ai_config.get("client_x509_cert_url", ""),
|
|
"universe_domain": vertex_ai_config.get("universe_domain", "googleapis.com"),
|
|
}
|
|
|
|
credentials = service_account.Credentials.from_service_account_info(
|
|
creds_info,
|
|
scopes=["https://www.googleapis.com/auth/cloud-platform"],
|
|
)
|
|
return credentials
|
|
|
|
|
|
def pcm_to_wav(pcm_data: bytes, sample_rate: int, channels: int = 1, bits_per_sample: int = 16) -> bytes:
|
|
"""将 PCM 原始数据转换为 WAV 格式"""
|
|
import io
|
|
buf = io.BytesIO()
|
|
with wave.open(buf, "wb") as wf:
|
|
wf.setnchannels(channels)
|
|
wf.setsampwidth(bits_per_sample // 8)
|
|
wf.setframerate(sample_rate)
|
|
wf.writeframes(pcm_data)
|
|
return buf.getvalue()
|
|
|
|
|
|
def call_gemini_tts(
|
|
text: str,
|
|
voice_id: str = DEFAULT_VOICE,
|
|
model: str = DEFAULT_MODEL,
|
|
project_id: str = DEFAULT_PROJECT_ID,
|
|
location: str = DEFAULT_LOCATION,
|
|
credentials=None,
|
|
) -> bytes:
|
|
"""调用 Gemini TTS API，返回 PCM 音频数据"""
|
|
|
|
# 创建 Gemini 客户端（Vertex AI 后端）
|
|
client = genai.Client(
|
|
vertexai=True,
|
|
project=project_id,
|
|
location=location,
|
|
credentials=credentials,
|
|
)
|
|
|
|
# 构建请求配置
|
|
config = types.GenerateContentConfig(
|
|
response_modalities=["AUDIO"],
|
|
speech_config=types.SpeechConfig(
|
|
voice_config=types.VoiceConfig(
|
|
prebuilt_voice_config=types.PrebuiltVoiceConfig(
|
|
voice_name=voice_id,
|
|
)
|
|
)
|
|
),
|
|
)
|
|
|
|
# 构建请求内容
|
|
contents = [
|
|
types.Content(
|
|
role="user",
|
|
parts=[types.Part(text=text)],
|
|
)
|
|
]
|
|
|
|
# print(f"[TTS] 调用 Gemini TTS, model={model}, voice={voice_id}, text_len={len(text)}")
|
|
|
|
# 调用 API（非流式）
|
|
response = client.models.generate_content(
|
|
model=model,
|
|
contents=contents,
|
|
config=config,
|
|
)
|
|
|
|
# 提取音频数据
|
|
if not response.candidates:
|
|
raise RuntimeError("Gemini TTS 返回空响应")
|
|
|
|
candidate = response.candidates[0]
|
|
finish_reason = candidate.finish_reason
|
|
# print(f"[TTS] finish_reason={finish_reason}")
|
|
|
|
if not candidate.content or not candidate.content.parts:
|
|
raise RuntimeError(f"Gemini TTS 无内容, finish_reason={finish_reason}")
|
|
|
|
pcm_data = None
|
|
for part in candidate.content.parts:
|
|
if part.inline_data and part.inline_data.mime_type and part.inline_data.mime_type.startswith("audio/"):
|
|
pcm_data = part.inline_data.data
|
|
# print(f"[TTS] 获取到音频数据, mime_type={part.inline_data.mime_type}, size={len(pcm_data)} bytes")
|
|
break
|
|
|
|
if pcm_data is None:
|
|
raise RuntimeError("Gemini TTS 响应中无音频数据")
|
|
|
|
return pcm_data
|
|
|
|
|
|
def call_gemini_tts_stream(
|
|
text: str,
|
|
voice_id: str = DEFAULT_VOICE,
|
|
model: str = DEFAULT_MODEL,
|
|
project_id: str = DEFAULT_PROJECT_ID,
|
|
location: str = DEFAULT_LOCATION,
|
|
credentials=None,
|
|
) -> bytes:
|
|
"""流式调用 Gemini TTS API，返回拼接后的 PCM 音频数据"""
|
|
|
|
# 创建 Gemini 客户端（Vertex AI 后端）
|
|
client = genai.Client(
|
|
vertexai=True,
|
|
project=project_id,
|
|
location=location,
|
|
credentials=credentials,
|
|
)
|
|
|
|
# 构建请求配置
|
|
config = types.GenerateContentConfig(
|
|
response_modalities=["AUDIO"],
|
|
speech_config=types.SpeechConfig(
|
|
voice_config=types.VoiceConfig(
|
|
prebuilt_voice_config=types.PrebuiltVoiceConfig(
|
|
voice_name=voice_id,
|
|
)
|
|
)
|
|
),
|
|
)
|
|
|
|
# 构建请求内容
|
|
contents = [
|
|
types.Content(
|
|
role="user",
|
|
parts=[types.Part(text=text)],
|
|
)
|
|
]
|
|
|
|
print(f"[TTS-Stream] 调用 Gemini TTS (流式), model={model}, voice={voice_id}, text_len={len(text)}")
|
|
|
|
# 流式调用 API
|
|
pcm_chunks = []
|
|
has_audio = False
|
|
finish_reason = None
|
|
finish_message = None
|
|
|
|
_i = 0
|
|
|
|
for response in client.models.generate_content_stream(
|
|
model=model,
|
|
contents=contents,
|
|
config=config,
|
|
):
|
|
# 调试：打印响应对象的基本信息
|
|
if not response.candidates:
|
|
print(f" [DEBUG] 响应中没有 candidates，跳过")
|
|
continue
|
|
|
|
candidate = response.candidates[0]
|
|
if candidate.finish_reason:
|
|
finish_reason = candidate.finish_reason
|
|
finish_message = candidate.finish_message
|
|
# 打印详细的 finish_reason 信息
|
|
if not candidate.content or not candidate.content.parts:
|
|
continue
|
|
|
|
for part in candidate.content.parts:
|
|
if part.inline_data and part.inline_data.mime_type and part.inline_data.mime_type.startswith("audio/"):
|
|
if part.inline_data.data:
|
|
has_audio = True
|
|
pcm_chunks.append(part.inline_data.data)
|
|
print(f" [chunk_{_i}] 收到音频帧, size={len(part.inline_data.data)} bytes")
|
|
_i += 1
|
|
|
|
print(f"[TTS-Stream] finish_reason={finish_reason}, finish_message={finish_message}, 共收到 {len(pcm_chunks)} 个音频帧")
|
|
|
|
if not has_audio:
|
|
raise RuntimeError(f"Gemini TTS 流式响应中无音频数据, finish_reason={finish_reason}")
|
|
|
|
return b"".join(pcm_chunks), finish_reason
|
|
|
|
|
|
def call_gemini_tts_func(voice, text):
|
|
# 加载凭证
|
|
vertex_ai_config = load_vertex_ai_credentials('./auth.toml')
|
|
credentials = create_credentials(vertex_ai_config)
|
|
print('开始生成音频')
|
|
pcm_data = call_gemini_tts(
|
|
text=text,
|
|
voice_id=voice,
|
|
model=DEFAULT_MODEL,
|
|
project_id=DEFAULT_PROJECT_ID,
|
|
location=DEFAULT_LOCATION,
|
|
credentials=credentials,
|
|
)
|
|
|
|
wav_data = pcm_to_wav(pcm_data, DEFAULT_SAMPLE_RATE)
|
|
url = upload_func('./audios/test.wav', wav_data)
|
|
return url
|
|
|
|
|
|
|
|
# 无marker
|
|
# text = '''
|
|
# 先帝创业未半而中道崩殂，今天下三分，益州疲弊，此诚危急存亡之秋也。然侍卫之臣不懈于内，忠志之士忘身于外者，盖追先帝之殊遇，欲报之于陛下也。诚宜开张圣听，以光先帝遗德，恢弘志士之气，不宜妄自菲薄，引喻失义，以塞忠谏之路也。宫中府中，俱为一体；陟罚臧否，不宜异同。若有作奸犯科及为忠善者，宜付有司论其刑赏，以昭陛下平明之理，不宜偏私，使内外异法也。
|
|
# '''
|
|
|
|
# 多marker
|
|
# text = '''
|
|
# [barely-audible whisper]先帝创业未半而中道崩殂，[soft inhale][teasing whisper]今天下三分，益州疲敝，[soft swallow]此诚危急存亡之秋也。[warm exhale]然侍卫之臣不懈于内，[lingering whisper]忠志之士忘身于外者，[vocal smile]盖追先帝之殊遇，[pause: 0.8s]欲报之于陛下也。[breathy whisper]诚宜开张圣听，[slow exhale]以光先帝遗德，[hushed pause]恢弘志士之气，[whispering]不宜妄自菲薄，引喻失义，[soft inhale][seductive whisper]以塞忠谏之路也。
|
|
# '''
|
|
|
|
# 单marker
|
|
text = '''
|
|
[barely-audible whisper]先帝创业未半而中道崩殂，[soft inhale]今天下三分，益州疲敝，[soft swallow]此诚危急存亡之秋也。[warm exhale]然侍卫之臣不懈于内，[lingering whisper]忠志之士忘身于外者，[vocal smile]盖追先帝之殊遇，[pause: 0.8s]欲报之于陛下也。[breathy whisper]诚宜开张圣听，[slow exhale]以光先帝遗德，[hushed pause]恢弘志士之气，[whispering]不宜妄自菲薄，引喻失义，[soft inhale]以塞忠谏之路也。
|
|
'''
|
|
|
|
text = '''
|
|
好吧... [soft inhale][barely-audible whisper]关于考试 - 或许你可以找到你的老师聊聊？ [teasing exhale]看看是否有机会获得更多的分数或者重新考试？ [soft swallow]关于你的父母。。。 [lingering whisper]或许我们可以提前演练一下说什么？ [warm inhale]有时一些话还是非常管用的。 [hushed giggle][intimate whisper]
|
|
'''
|
|
|
|
def main(__i):
|
|
parser = argparse.ArgumentParser(description="Gemini TTS 测试脚本")
|
|
parser.add_argument("--text", type=str, default=text, help="要转换的文本")
|
|
parser.add_argument("--voice", type=str, default=DEFAULT_VOICE, help=f"音色ID (默认: {DEFAULT_VOICE})")
|
|
parser.add_argument("--model", type=str, default=DEFAULT_MODEL, help=f"模型名称 (默认: {DEFAULT_MODEL})")
|
|
parser.add_argument("--output", type=str, default="output.wav", help="输出 WAV 文件路径 (默认: output.wav)")
|
|
parser.add_argument("--auth", type=str, default='auth.toml', help="auth.toml 文件路径 (默认: configs/auth.toml)")
|
|
parser.add_argument("--project", type=str, default=DEFAULT_PROJECT_ID, help=f"GCP 项目ID (默认: {DEFAULT_PROJECT_ID})")
|
|
parser.add_argument("--location", type=str, default=DEFAULT_LOCATION, help=f"GCP 区域 (默认: {DEFAULT_LOCATION})")
|
|
parser.add_argument("--stream", default=True, help="使用流式调用")
|
|
parser.add_argument("--sample-rate", type=int, default=DEFAULT_SAMPLE_RATE, help=f"采样率 (默认: {DEFAULT_SAMPLE_RATE})")
|
|
|
|
args = parser.parse_args()
|
|
|
|
# 确定 auth.toml 路径
|
|
if args.auth:
|
|
auth_path = args.auth
|
|
else:
|
|
# 尝试从项目根目录查找
|
|
script_dir = Path(__file__).resolve().parent
|
|
candidates = [
|
|
script_dir / "../../configs/auth.toml",
|
|
Path("configs/auth.toml"),
|
|
Path("auth.toml"),
|
|
]
|
|
auth_path = None
|
|
for p in candidates:
|
|
if p.exists():
|
|
auth_path = str(p.resolve())
|
|
break
|
|
if auth_path is None:
|
|
print("错误: 找不到 auth.toml，请通过 --auth 参数指定路径")
|
|
sys.exit(1)
|
|
|
|
print(f"[配置] auth.toml: {auth_path}")
|
|
print(f"[配置] project: {args.project}, location: {args.location}")
|
|
print(f"[配置] model: {args.model}, voice: {args.voice}")
|
|
print(f"[配置] sample_rate: {args.sample_rate}")
|
|
print(f"[配置] 流式模式: {'是' if args.stream else '否'}")
|
|
print()
|
|
|
|
# 加载凭证
|
|
vertex_ai_config = load_vertex_ai_credentials(auth_path)
|
|
credentials = create_credentials(vertex_ai_config)
|
|
|
|
# 调用 TTS
|
|
if args.stream:
|
|
pcm_data, finish_reason = call_gemini_tts_stream(
|
|
text=args.text,
|
|
voice_id=args.voice,
|
|
model=args.model,
|
|
project_id=args.project,
|
|
location=args.location,
|
|
credentials=credentials,
|
|
)
|
|
else:
|
|
finish_reason = None
|
|
pcm_data = call_gemini_tts(
|
|
text=args.text,
|
|
voice_id=args.voice,
|
|
model=args.model,
|
|
project_id=args.project,
|
|
location=args.location,
|
|
credentials=credentials,
|
|
)
|
|
|
|
# 转换为 WAV 并保存
|
|
wav_data = pcm_to_wav(pcm_data, args.sample_rate)
|
|
output_path = Path(f'./audios/{__i}.wav')
|
|
output_path.parent.mkdir(parents=True, exist_ok=True)
|
|
output_path.write_bytes(wav_data)
|
|
# print('开始上传cos')
|
|
# cos_url = upload_func('./test.mp3', wav_data)
|
|
|
|
return finish_reason
|
|
|
|
|
|
if __name__ == "__main__":
|
|
finish_reasons = list()
|
|
para_size = 100
|
|
|
|
# 串行执行任务
|
|
for i in range(para_size):
|
|
try:
|
|
result = main(i)
|
|
print(result)
|
|
finish_reasons.append([i, result])
|
|
print(f"[进度] 完成 {i+1}/{para_size}", flush=True)
|
|
# 每完成一个任务就立即保存
|
|
dump_json(finish_reasons, './finish_reasons.json')
|
|
except Exception as e:
|
|
print(f"[错误] 任务失败: {e}", flush=True)
|
|
finish_reasons.append(None)
|
|
# 失败也保存
|
|
dump_json(finish_reasons, './finish_reasons.json')
|
|
|
|
print(f"\n[汇总] 共完成 {len(finish_reasons)} 个任务")
|
|
print(finish_reasons)
|
|
|
|
# ps aux | grep call_gemini_tts_online.py | grep -v grep | awk '{print $2}' | xargs kill -9
|
|
# nohup python call_gemini_tts_online.py > call_gemini_tts_online_log.txt 2>&1 & |