488 lines
20 KiB
Python
488 lines
20 KiB
Python
import asyncio
|
||
import base64
|
||
import unittest
|
||
import os
|
||
import sys
|
||
import tempfile
|
||
import time
|
||
from datetime import timedelta
|
||
from pathlib import Path
|
||
from types import SimpleNamespace
|
||
from unittest.mock import patch
|
||
|
||
# add project root to python path
|
||
sys.path.insert(0, str(Path(__file__).parent.parent.parent))
|
||
|
||
from app.utils import utils
|
||
from app.services import voice as vs
|
||
from app.services import task as task_service
|
||
from pydub import AudioSegment
|
||
|
||
temp_dir = utils.storage_dir("temp")
|
||
|
||
text_en = """
|
||
What is the meaning of life?
|
||
This question has puzzled philosophers, scientists, and thinkers of all kinds for centuries.
|
||
Throughout history, various cultures and individuals have come up with their interpretations and beliefs around the purpose of life.
|
||
Some say it's to seek happiness and self-fulfillment, while others believe it's about contributing to the welfare of others and making a positive impact in the world.
|
||
Despite the myriad of perspectives, one thing remains clear: the meaning of life is a deeply personal concept that varies from one person to another.
|
||
It's an existential inquiry that encourages us to reflect on our values, desires, and the essence of our existence.
|
||
"""
|
||
|
||
text_zh = """
|
||
预计未来3天深圳冷空气活动频繁，未来两天持续阴天有小雨，出门带好雨具；
|
||
10-11日持续阴天有小雨，日温差小，气温在13-17℃之间，体感阴凉；
|
||
12日天气短暂好转，早晚清凉；
|
||
"""
|
||
|
||
voice_rate=1.0
|
||
voice_volume=1.0
|
||
|
||
class TestVoiceService(unittest.TestCase):
|
||
def setUp(self):
|
||
self.loop = asyncio.new_event_loop()
|
||
asyncio.set_event_loop(self.loop)
|
||
|
||
def tearDown(self):
|
||
self.loop.close()
|
||
|
||
def test_siliconflow(self):
|
||
# SiliconFlow 的 API Key 存在 [siliconflow].api_key 中，运行时代码也是从
|
||
# config.siliconflow 读取；这里必须使用同一配置源，避免正确配置凭据时
|
||
# 测试仍然被误跳过。
|
||
if not vs.config.siliconflow.get("api_key"):
|
||
self.skipTest("siliconflow_api_key is not configured")
|
||
|
||
voice_name = "siliconflow:FunAudioLLM/CosyVoice2-0.5B:alex-Male"
|
||
voice_name = vs.parse_voice_name(voice_name)
|
||
|
||
async def _do():
|
||
parts = voice_name.split(":")
|
||
if len(parts) >= 3:
|
||
model = parts[1]
|
||
# 移除性别后缀，例如 "alex-Male" -> "alex"
|
||
voice_with_gender = parts[2]
|
||
voice = voice_with_gender.split("-")[0]
|
||
# 构建完整的voice参数，格式为 "model:voice"
|
||
full_voice = f"{model}:{voice}"
|
||
voice_file = f"{temp_dir}/tts-siliconflow-{voice}.mp3"
|
||
subtitle_file = f"{temp_dir}/tts-siliconflow-{voice}.srt"
|
||
sub_maker = vs.siliconflow_tts(
|
||
text=text_zh, model=model, voice=full_voice, voice_file=voice_file, voice_rate=voice_rate, voice_volume=voice_volume
|
||
)
|
||
if not sub_maker:
|
||
self.fail("siliconflow tts failed")
|
||
vs.create_subtitle(sub_maker=sub_maker, text=text_zh, subtitle_file=subtitle_file)
|
||
audio_duration = vs.get_audio_duration(sub_maker)
|
||
print(f"voice: {voice_name}, audio duration: {audio_duration}s")
|
||
else:
|
||
self.fail("siliconflow invalid voice name")
|
||
|
||
self.loop.run_until_complete(_do())
|
||
|
||
def test_azure_tts_v1(self):
|
||
voice_name = "zh-CN-XiaoyiNeural-Female"
|
||
voice_name = vs.parse_voice_name(voice_name)
|
||
print(voice_name)
|
||
|
||
voice_file = f"{temp_dir}/tts-azure-v1-{voice_name}.mp3"
|
||
subtitle_file = f"{temp_dir}/tts-azure-v1-{voice_name}.srt"
|
||
sub_maker = vs.azure_tts_v1(
|
||
text=text_zh, voice_name=voice_name, voice_file=voice_file, voice_rate=voice_rate
|
||
)
|
||
if not sub_maker:
|
||
self.fail("azure tts v1 failed")
|
||
vs.create_subtitle(sub_maker=sub_maker, text=text_zh, subtitle_file=subtitle_file)
|
||
audio_duration = vs.get_audio_duration(sub_maker)
|
||
print(f"voice: {voice_name}, audio duration: {audio_duration}s")
|
||
|
||
def test_azure_tts_v1_supports_legacy_edge_tts_without_boundary(self):
|
||
"""
|
||
验证 Azure TTS V1 在旧版 edge_tts 依赖残留时仍可继续工作。
|
||
|
||
这个回归场景对应 Windows 便携包更新失败后，现场环境还停留在旧版
|
||
edge_tts 的情况：
|
||
1. `Communicate.__init__()` 不接受 `boundary`
|
||
2. 只有异步 `stream()`，没有 `stream_sync()`
|
||
"""
|
||
|
||
class _LegacyCommunicate:
|
||
def __init__(self, text, voice, rate="+0%"):
|
||
self.text = text
|
||
self.voice = voice
|
||
self.rate = rate
|
||
|
||
async def stream(self):
|
||
yield {"type": "audio", "data": b"legacy-audio"}
|
||
yield {
|
||
"type": "WordBoundary",
|
||
"offset": 0,
|
||
"duration": 10000000,
|
||
"text": "legacy",
|
||
}
|
||
|
||
class _FakeSubMaker:
|
||
def __init__(self):
|
||
self.events = []
|
||
|
||
def feed(self, chunk):
|
||
self.events.append(chunk)
|
||
|
||
def get_srt(self):
|
||
if not self.events:
|
||
return ""
|
||
return "1\n00:00:00,000 --> 00:00:01,000\nlegacy\n"
|
||
|
||
with tempfile.TemporaryDirectory() as tmp_dir, patch.object(
|
||
vs.edge_tts, "Communicate", _LegacyCommunicate
|
||
), patch.object(vs.edge_tts, "SubMaker", _FakeSubMaker):
|
||
voice_file = str(Path(tmp_dir) / "legacy-edge-tts.mp3")
|
||
sub_maker = vs.azure_tts_v1(
|
||
text="legacy edge tts compatibility",
|
||
voice_name="zh-CN-XiaoyiNeural-Female",
|
||
voice_file=voice_file,
|
||
voice_rate=1.0,
|
||
)
|
||
|
||
self.assertIsNotNone(sub_maker)
|
||
self.assertEqual(Path(voice_file).read_bytes(), b"legacy-audio")
|
||
self.assertEqual(len(sub_maker.events), 1)
|
||
self.assertEqual(sub_maker.events[0]["type"], "WordBoundary")
|
||
|
||
def test_azure_tts_v1_times_out_hanging_stream_sync(self):
|
||
"""
|
||
验证 Azure TTS V1 在 edge_tts 同步流卡住时能够快速失败。
|
||
|
||
真实现场里，网络异常、服务端限流、voice 语言与文本不匹配时，
|
||
`stream_sync()` 可能长时间不返回，导致 WebUI 任务只停在
|
||
`start, voice name...`。这里用阻塞的 fake stream 复现该场景，
|
||
确认超时保护会让函数结束并返回 None。
|
||
"""
|
||
|
||
class _HangingCommunicate:
|
||
def __init__(self, text, voice, rate="+0%", boundary=None):
|
||
self.text = text
|
||
self.voice = voice
|
||
self.rate = rate
|
||
self.boundary = boundary
|
||
|
||
def stream_sync(self):
|
||
time.sleep(10)
|
||
yield {"type": "audio", "data": b"unreachable"}
|
||
|
||
class _FakeSubMaker:
|
||
def feed(self, chunk):
|
||
return None
|
||
|
||
def get_srt(self):
|
||
return ""
|
||
|
||
with tempfile.TemporaryDirectory() as tmp_dir, patch.object(
|
||
vs.edge_tts, "Communicate", _HangingCommunicate
|
||
), patch.object(vs.edge_tts, "SubMaker", _FakeSubMaker), patch.object(
|
||
vs.config,
|
||
"app",
|
||
dict(vs.config.app, edge_tts_timeout=0.05),
|
||
):
|
||
voice_file = Path(tmp_dir) / "hanging-edge-tts.mp3"
|
||
started_at = time.monotonic()
|
||
sub_maker = vs.azure_tts_v1(
|
||
text="帮我生成一个花开花落的视频",
|
||
voice_name="en-AU-NatashaNeural-Female",
|
||
voice_file=str(voice_file),
|
||
voice_rate=1.0,
|
||
)
|
||
elapsed = time.monotonic() - started_at
|
||
self.assertFalse(voice_file.exists())
|
||
|
||
self.assertIsNone(sub_maker)
|
||
self.assertLess(elapsed, 2)
|
||
|
||
def test_azure_tts_v2(self):
|
||
if not vs.config.azure.get("speech_key") or not vs.config.azure.get("speech_region"):
|
||
self.skipTest("Azure speech key or region is not configured")
|
||
|
||
voice_name = "zh-CN-XiaoxiaoMultilingualNeural-V2-Female"
|
||
voice_name = vs.parse_voice_name(voice_name)
|
||
print(voice_name)
|
||
|
||
async def _do():
|
||
voice_file = f"{temp_dir}/tts-azure-v2-{voice_name}.mp3"
|
||
subtitle_file = f"{temp_dir}/tts-azure-v2-{voice_name}.srt"
|
||
sub_maker = vs.azure_tts_v2(
|
||
text=text_zh, voice_name=voice_name, voice_file=voice_file
|
||
)
|
||
if not sub_maker:
|
||
self.fail("azure tts v2 failed")
|
||
vs.create_subtitle(sub_maker=sub_maker, text=text_zh, subtitle_file=subtitle_file)
|
||
audio_duration = vs.get_audio_duration(sub_maker)
|
||
print(f"voice: {voice_name}, audio duration: {audio_duration}s")
|
||
|
||
self.loop.run_until_complete(_do())
|
||
|
||
def test_gemini_tts_uses_legacy_submaker_fields(self):
|
||
"""
|
||
验证 Gemini TTS 在 edge_tts 7.x 环境下仍会返回项目兼容的字幕结构，
|
||
并且可以被 `subtitle_provider=edge` 的字幕生成链路直接消费，
|
||
避免再次回退 Whisper。
|
||
"""
|
||
|
||
class _InlineData:
|
||
def __init__(self, data):
|
||
self.data = data
|
||
|
||
class _Part:
|
||
def __init__(self, data):
|
||
self.inline_data = _InlineData(data)
|
||
|
||
class _Content:
|
||
def __init__(self, data):
|
||
self.parts = [_Part(data)]
|
||
|
||
class _Candidate:
|
||
def __init__(self, data):
|
||
self.content = _Content(data)
|
||
|
||
class _Response:
|
||
def __init__(self, data):
|
||
self.candidates = [_Candidate(data)]
|
||
|
||
class _FakeModel:
|
||
def __init__(self, name):
|
||
self.name = name
|
||
|
||
def generate_content(self, contents, generation_config):
|
||
tone = (
|
||
AudioSegment.silent(duration=1800)
|
||
.set_frame_rate(24000)
|
||
.set_channels(1)
|
||
.set_sample_width(2)
|
||
)
|
||
return _Response(tone.raw_data)
|
||
|
||
voice_file = f"{temp_dir}/tts-gemini-Zephyr.mp3"
|
||
subtitle_file = f"{temp_dir}/tts-gemini-Zephyr.srt"
|
||
text = "Gemini subtitle generation should work now. Testing multiple lines."
|
||
|
||
with patch("google.generativeai.configure"), patch(
|
||
"google.generativeai.GenerativeModel", _FakeModel
|
||
), patch.object(vs.config, "app", dict(vs.config.app, gemini_api_key="test-key")):
|
||
sub_maker = vs.gemini_tts(
|
||
text=text,
|
||
voice_name="Zephyr",
|
||
voice_rate=1.0,
|
||
voice_file=voice_file,
|
||
)
|
||
|
||
self.assertIsNotNone(sub_maker)
|
||
self.assertEqual(
|
||
getattr(sub_maker, "subs", []),
|
||
["Gemini subtitle generation should work now", "Testing multiple lines"],
|
||
)
|
||
self.assertEqual(len(getattr(sub_maker, "offset", [])), 2)
|
||
self.assertEqual(sub_maker.offset[0][0], 0)
|
||
self.assertLess(sub_maker.offset[0][1], sub_maker.offset[1][1])
|
||
|
||
vs.create_subtitle(sub_maker=sub_maker, text=text, subtitle_file=subtitle_file)
|
||
subtitle_content = Path(subtitle_file).read_text(encoding="utf-8")
|
||
self.assertIn("Gemini subtitle generation should work now", subtitle_content)
|
||
self.assertIn("Testing multiple lines", subtitle_content)
|
||
|
||
def test_mimo_tts_uses_openai_compatible_audio_response(self):
|
||
"""
|
||
验证 Xiaomi MiMo TTS 可以消费 OpenAI-compatible 的音频响应结构。
|
||
|
||
这里用 fake OpenAI client 和 fake AudioSegment 覆盖真实网络与 ffmpeg，
|
||
确认运行时代码会把待合成文本放到 assistant message，并把返回的
|
||
base64 WAV 音频导出到项目后续流程使用的音频文件。
|
||
"""
|
||
|
||
class _FakeAudio:
|
||
def __init__(self):
|
||
self.data = base64.b64encode(b"RIFF-fake-wav").decode("utf-8")
|
||
|
||
class _FakeMessage:
|
||
def __init__(self):
|
||
self.audio = _FakeAudio()
|
||
|
||
class _FakeChoice:
|
||
def __init__(self):
|
||
self.message = _FakeMessage()
|
||
|
||
class _FakeCompletion:
|
||
def __init__(self):
|
||
self.choices = [_FakeChoice()]
|
||
|
||
class _FakeCompletions:
|
||
def create(self, **kwargs):
|
||
self.kwargs = kwargs
|
||
return _FakeCompletion()
|
||
|
||
class _FakeAudioSegment:
|
||
def __len__(self):
|
||
return 1800
|
||
|
||
def export(self, output_file, format):
|
||
Path(output_file).write_bytes(b"fake-mp3")
|
||
|
||
fake_completions = _FakeCompletions()
|
||
fake_client = SimpleNamespace(
|
||
chat=SimpleNamespace(completions=fake_completions)
|
||
)
|
||
|
||
with tempfile.TemporaryDirectory() as tmp_dir, patch.object(
|
||
vs,
|
||
"OpenAI",
|
||
return_value=fake_client,
|
||
) as openai_client, patch(
|
||
"pydub.AudioSegment.from_file",
|
||
return_value=_FakeAudioSegment(),
|
||
), patch.object(
|
||
vs.config,
|
||
"app",
|
||
dict(
|
||
vs.config.app,
|
||
mimo_api_key="mimo-key",
|
||
mimo_base_url="https://api.xiaomimimo.com/v1",
|
||
mimo_tts_model_name="mimo-v2.5-tts",
|
||
mimo_tts_style_prompt="用清晰的中文旁白朗读。",
|
||
),
|
||
):
|
||
voice_file = str(Path(tmp_dir) / "mimo-tts.mp3")
|
||
sub_maker = vs.mimo_tts(
|
||
text="小米语音合成测试。第二句话。",
|
||
voice_name="冰糖",
|
||
voice_rate=1.0,
|
||
voice_file=voice_file,
|
||
voice_volume=1.0,
|
||
)
|
||
generated_audio = Path(voice_file).read_bytes()
|
||
|
||
openai_client.assert_called_once_with(
|
||
api_key="mimo-key",
|
||
base_url="https://api.xiaomimimo.com/v1",
|
||
)
|
||
self.assertEqual(fake_completions.kwargs["model"], "mimo-v2.5-tts")
|
||
self.assertEqual(
|
||
fake_completions.kwargs["messages"],
|
||
[
|
||
{"role": "user", "content": "用清晰的中文旁白朗读。"},
|
||
{"role": "assistant", "content": "小米语音合成测试。第二句话。"},
|
||
],
|
||
)
|
||
self.assertEqual(
|
||
fake_completions.kwargs["audio"],
|
||
{"format": "wav", "voice": "冰糖"},
|
||
)
|
||
self.assertEqual(generated_audio, b"fake-mp3")
|
||
self.assertIsNotNone(sub_maker)
|
||
self.assertEqual(getattr(sub_maker, "subs", []), ["小米语音合成测试", "第二句话"])
|
||
self.assertEqual(len(getattr(sub_maker, "offset", [])), 2)
|
||
|
||
def test_generate_subtitle_keeps_edge_provider_for_gemini_legacy_submaker(self):
|
||
"""
|
||
验证 Gemini TTS 返回的 legacy 字幕结构在 edge provider 下可以直接产出
|
||
SRT，不会因为匹配失败而回退到 Whisper。
|
||
"""
|
||
script = "Gemini subtitle generation should work now. Testing multiple lines."
|
||
sub_maker = vs.populate_legacy_submaker_with_full_text(
|
||
vs.ensure_legacy_submaker_fields(vs.SubMaker()),
|
||
script,
|
||
2.4,
|
||
)
|
||
|
||
with tempfile.TemporaryDirectory() as tmp_dir, patch.object(
|
||
task_service.config,
|
||
"app",
|
||
dict(task_service.config.app, subtitle_provider="edge"),
|
||
), patch("app.services.subtitle.create") as whisper_create, patch(
|
||
"app.utils.utils.task_dir",
|
||
lambda tid="": str(Path(tmp_dir) / tid) if tid else str(Path(tmp_dir)),
|
||
):
|
||
task_id = "gemini-subtitle-edge-task"
|
||
Path(tmp_dir, task_id).mkdir(parents=True, exist_ok=True)
|
||
subtitle_path = task_service.generate_subtitle(
|
||
task_id=task_id,
|
||
params=type("Params", (), {"subtitle_enabled": True})(),
|
||
video_script=script,
|
||
sub_maker=sub_maker,
|
||
audio_file="",
|
||
)
|
||
|
||
self.assertTrue(subtitle_path.endswith("subtitle.srt"))
|
||
self.assertTrue(Path(subtitle_path).exists())
|
||
self.assertFalse(whisper_create.called)
|
||
subtitle_content = Path(subtitle_path).read_text(encoding="utf-8")
|
||
self.assertIn("Gemini subtitle generation should work now", subtitle_content)
|
||
self.assertIn("Testing multiple lines", subtitle_content)
|
||
|
||
def test_script_split_keeps_thousand_separator_comma(self):
|
||
"""
|
||
Edge TTS 会把 "1,000 years" 作为连续文本返回。脚本断句时不能把
|
||
数字中间的英文逗号当成句子边界，否则字幕聚合会出现 issue #894
|
||
里的 sub_items 数量少于 script_lines，并错误回退 Whisper。
|
||
"""
|
||
text = (
|
||
"It takes about 1,000 years for a single drop of water to finish "
|
||
"the whole trip!"
|
||
)
|
||
|
||
self.assertEqual(
|
||
utils.split_string_by_punctuations(text),
|
||
[
|
||
(
|
||
"It takes about 1,000 years for a single drop of water to finish "
|
||
"the whole trip"
|
||
)
|
||
],
|
||
)
|
||
|
||
def test_edge_cue_aggregation_handles_thousand_separator_comma(self):
|
||
"""
|
||
复现 issue #894 的关键形态：Edge cues 中最后一句作为连续文本返回，
|
||
包含 `1,000 years`。脚本断句必须与 cues 聚合结果一致，不能把它
|
||
拆成两条字幕。
|
||
"""
|
||
text = (
|
||
"The ocean isn't just sitting stil, it moves around the world like a massive "
|
||
"amusement park ride! Cold water at the North and South Poles sinks to the "
|
||
"bottom because it is heavy and salty. At the same time, warm water from the "
|
||
"sunny equator flows along the top to take its place. This creates a giant "
|
||
"underwater conveyor belt that travels all the way around the Earth. It takes "
|
||
"about 1,000 years for a single drop of water to finish the whole trip!"
|
||
)
|
||
script_lines = utils.split_string_by_punctuations(text)
|
||
cues = []
|
||
for index, line in enumerate(script_lines):
|
||
# Edge 的 cue content 经常没有脚本里的空格和标点布局，这里去掉空格
|
||
# 来模拟更严格的匹配场景。
|
||
cues.append(
|
||
SimpleNamespace(
|
||
content=line.replace(" ", ""),
|
||
start=timedelta(seconds=index),
|
||
end=timedelta(seconds=index + 0.8),
|
||
)
|
||
)
|
||
sub_maker = SimpleNamespace(cues=cues)
|
||
|
||
sub_items = vs._build_subtitle_items_from_edge_cues(sub_maker, script_lines)
|
||
|
||
self.assertEqual(len(sub_items), len(script_lines))
|
||
self.assertIn("1,000 years", sub_items[-1])
|
||
|
||
def test_convert_rate_to_percent_signs_zero_rate(self):
|
||
# Rates near but not exactly 1.0 round to 0 percent. edge-tts rejects
|
||
# an unsigned "0%" (ValueError: Invalid rate '0%'), so the helper must
|
||
# emit a sign-prefixed "+0%". Regression test for that crash.
|
||
self.assertEqual(vs.convert_rate_to_percent(1.0), "+0%")
|
||
self.assertEqual(vs.convert_rate_to_percent(1.004), "+0%")
|
||
self.assertEqual(vs.convert_rate_to_percent(0.997), "+0%")
|
||
self.assertEqual(vs.convert_rate_to_percent(1.5), "+50%")
|
||
self.assertEqual(vs.convert_rate_to_percent(0.8), "-20%")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# python -m unittest test.services.test_voice.TestVoiceService.test_azure_tts_v1
|
||
# python -m unittest test.services.test_voice.TestVoiceService.test_azure_tts_v2
|
||
unittest.main()
|