Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| c6bc8170d1 | |||
| a1285be576 |
@@ -46,6 +46,7 @@ from packages.application.voice_library.use_cases import (
|
||||
CreateVoiceLibraryUseCase,
|
||||
QuotaExceededError,
|
||||
)
|
||||
from packages.domain.voice_presets import list_voices
|
||||
from packages.ports.user_repository import UserRepository
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -53,6 +54,34 @@ logger = logging.getLogger(__name__)
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
@router.get("/presets", summary="获取预设音色列表")
|
||||
def list_preset_voices(
|
||||
gender: Optional[str] = Query(None, description="按性别筛选: male/female/child"),
|
||||
style: Optional[str] = Query(None, description="按风格筛选: stable/lively/customer_service/narration/news/story"),
|
||||
keyword: Optional[str] = Query(None, description="按关键词搜索"),
|
||||
_user: AuthenticatedUser = Depends(get_current_user),
|
||||
) -> list[dict]:
|
||||
"""获取可用的预设音色列表。
|
||||
|
||||
用于配音功能的音色选择。
|
||||
"""
|
||||
voices = list_voices(gender=gender, style=style, keyword=keyword)
|
||||
return [
|
||||
{
|
||||
"voice_id": v.voice_id,
|
||||
"name": v.name,
|
||||
"gender": v.gender.value,
|
||||
"style": v.style.value,
|
||||
"description": v.description,
|
||||
"default_speed": v.default_speed,
|
||||
"default_pitch": v.default_pitch,
|
||||
"sample_rate": v.sample_rate,
|
||||
"language": v.language,
|
||||
}
|
||||
for v in voices
|
||||
]
|
||||
|
||||
|
||||
def _get_repository(session: Session = Depends(get_db_session)) -> SQLAlchemyTTSJobRepository:
|
||||
return SQLAlchemyTTSJobRepository(session)
|
||||
|
||||
|
||||
Executable
+66
@@ -0,0 +1,66 @@
|
||||
"""TTS 服务工厂.
|
||||
|
||||
根据配置创建对应的 TTS 服务实例。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
|
||||
from packages.ports.tts_service import TtsService
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# 可用的 provider 映射
|
||||
_PROVIDERS: dict[str, type[TtsService]] = {}
|
||||
|
||||
|
||||
def register_provider(name: str, cls: type[TtsService]) -> None:
|
||||
"""注册 TTS 供应商."""
|
||||
_PROVIDERS[name] = cls
|
||||
|
||||
|
||||
def get_tts_service(provider: str | None = None, **kwargs) -> TtsService:
|
||||
"""获取 TTS 服务实例.
|
||||
|
||||
Args:
|
||||
provider: 供应商名称(None 则从环境变量读取 TTS_PROVIDER)
|
||||
**kwargs: 传递给服务构造函数的参数
|
||||
|
||||
Returns:
|
||||
TTS 服务实例
|
||||
|
||||
Raises:
|
||||
ValueError: 不支持的供应商
|
||||
"""
|
||||
if provider is None:
|
||||
provider = os.environ.get("TTS_PROVIDER", "mock")
|
||||
|
||||
provider = provider.lower()
|
||||
|
||||
if provider not in _PROVIDERS:
|
||||
# 延迟导入避免循环依赖
|
||||
if provider == "mock":
|
||||
from packages.adapters.tts.mock_tts_service import MockTtsService
|
||||
|
||||
_PROVIDERS["mock"] = MockTtsService
|
||||
else:
|
||||
logger.warning("未知 TTS provider: %s,回退到 mock", provider)
|
||||
from packages.adapters.tts.mock_tts_service import MockTtsService
|
||||
|
||||
_PROVIDERS["mock"] = MockTtsService
|
||||
provider = "mock"
|
||||
|
||||
cls = _PROVIDERS[provider]
|
||||
return cls(**kwargs)
|
||||
|
||||
|
||||
def available_providers() -> list[str]:
|
||||
"""获取可用的供应商列表."""
|
||||
# 确保 mock 已注册
|
||||
if "mock" not in _PROVIDERS:
|
||||
from packages.adapters.tts.mock_tts_service import MockTtsService
|
||||
|
||||
_PROVIDERS["mock"] = MockTtsService
|
||||
return list(_PROVIDERS.keys())
|
||||
Executable
+275
@@ -0,0 +1,275 @@
|
||||
"""TTS 配音引擎 — 集成到统一渲染管道的配音能力.
|
||||
|
||||
负责:
|
||||
- 根据 TtsConfig 生成配音音频
|
||||
- 字幕联动:按字幕片段分段合成,自动对齐时间轴
|
||||
- 整段配音:整段文本生成一条音频
|
||||
- 失败降级:TTS 失败不阻断渲染
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from packages.domain.tts_config import TtsConfig
|
||||
from packages.ports.tts_service import TtsError, TtsService
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclass
|
||||
class VoiceoverSegment:
|
||||
"""配音片段.
|
||||
|
||||
Attributes:
|
||||
text: 文本内容
|
||||
start_time: 开始时间(秒)
|
||||
end_time: 结束时间(秒)
|
||||
audio_path: 合成后的音频文件路径
|
||||
duration: 音频实际时长
|
||||
"""
|
||||
|
||||
text: str
|
||||
start_time: float = 0.0
|
||||
end_time: float = 0.0
|
||||
audio_path: Path | None = None
|
||||
duration: float = 0.0
|
||||
|
||||
|
||||
@dataclass
|
||||
class VoiceoverResult:
|
||||
"""配音结果.
|
||||
|
||||
Attributes:
|
||||
success: 是否成功
|
||||
segments: 配音片段列表
|
||||
total_duration: 总时长
|
||||
error_message: 错误信息(失败时)
|
||||
"""
|
||||
|
||||
success: bool = False
|
||||
segments: list[VoiceoverSegment] = field(default_factory=list)
|
||||
total_duration: float = 0.0
|
||||
error_message: str = ""
|
||||
|
||||
|
||||
class TtsEngine:
|
||||
"""TTS 配音引擎.
|
||||
|
||||
封装 TtsService 调用,支持:
|
||||
- 整段配音
|
||||
- 字幕联动配音
|
||||
- 失败降级
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
tts_service: TtsService,
|
||||
work_dir: Path,
|
||||
) -> None:
|
||||
self._tts = tts_service
|
||||
self._work_dir = work_dir
|
||||
self._work_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
def generate_full_voiceover(
|
||||
self,
|
||||
config: TtsConfig,
|
||||
*,
|
||||
total_duration: float = 0.0,
|
||||
) -> VoiceoverResult:
|
||||
"""生成整段配音.
|
||||
|
||||
Args:
|
||||
config: TTS 配置
|
||||
total_duration: 视频总时长(用于调整配音速度适配)
|
||||
|
||||
Returns:
|
||||
配音结果
|
||||
"""
|
||||
if not config.enabled or not config.text.strip():
|
||||
return VoiceoverResult(success=False, error_message="配音未启用或文本为空")
|
||||
|
||||
try:
|
||||
output_path = self._work_dir / "voiceover_full.wav"
|
||||
|
||||
audio_path = self._tts.synthesize(
|
||||
text=config.text,
|
||||
voice_id=config.voice_id,
|
||||
speed=config.speed,
|
||||
pitch=config.pitch,
|
||||
output_path=output_path,
|
||||
)
|
||||
|
||||
# 探测实际时长
|
||||
duration = self._probe_duration(audio_path)
|
||||
|
||||
segment = VoiceoverSegment(
|
||||
text=config.text,
|
||||
start_time=0.0,
|
||||
end_time=duration,
|
||||
audio_path=audio_path,
|
||||
duration=duration,
|
||||
)
|
||||
|
||||
return VoiceoverResult(
|
||||
success=True,
|
||||
segments=[segment],
|
||||
total_duration=duration,
|
||||
)
|
||||
|
||||
except TtsError as e:
|
||||
logger.warning("TTS 整段配音失败,降级跳过: %s", e)
|
||||
return VoiceoverResult(success=False, error_message=str(e))
|
||||
except Exception as e:
|
||||
logger.warning("TTS 整段配音异常,降级跳过: %s", e)
|
||||
return VoiceoverResult(success=False, error_message=str(e))
|
||||
|
||||
def generate_subtitle_voiceover(
|
||||
self,
|
||||
config: TtsConfig,
|
||||
subtitles: list[dict[str, Any]],
|
||||
) -> VoiceoverResult:
|
||||
"""根据字幕生成配音(字幕联动).
|
||||
|
||||
每个字幕片段独立合成,按字幕时间轴对齐。
|
||||
|
||||
Args:
|
||||
config: TTS 配置
|
||||
subtitles: 字幕列表,每项含 text/start_time/end_time
|
||||
|
||||
Returns:
|
||||
配音结果
|
||||
"""
|
||||
if not config.enabled:
|
||||
return VoiceoverResult(success=False, error_message="配音未启用")
|
||||
|
||||
if not subtitles:
|
||||
return VoiceoverResult(success=False, error_message="字幕为空")
|
||||
|
||||
segments: list[VoiceoverSegment] = []
|
||||
total_duration = 0.0
|
||||
|
||||
for i, sub in enumerate(subtitles):
|
||||
text = sub.get("text", "").strip()
|
||||
if not text:
|
||||
continue
|
||||
|
||||
start_time = float(sub.get("start_time", 0))
|
||||
end_time = float(sub.get("end_time", 0))
|
||||
target_duration = max(0.1, end_time - start_time)
|
||||
|
||||
try:
|
||||
# 计算适配时长所需语速:让配音时长 ≈ 字幕时长
|
||||
estimated = self._tts.estimate_duration(text, speed=config.speed)
|
||||
adjusted_speed = config.speed
|
||||
if estimated > 0 and target_duration > 0:
|
||||
# 按目标时长调整语速,限制在 0.5~2.0 范围内
|
||||
speed_factor = estimated / target_duration
|
||||
adjusted_speed = max(0.5, min(2.0, config.speed * speed_factor))
|
||||
|
||||
output_path = self._work_dir / f"voiceover_seg_{i:03d}.wav"
|
||||
|
||||
audio_path = self._tts.synthesize(
|
||||
text=text,
|
||||
voice_id=config.voice_id,
|
||||
speed=adjusted_speed,
|
||||
pitch=config.pitch,
|
||||
output_path=output_path,
|
||||
)
|
||||
|
||||
actual_duration = self._probe_duration(audio_path)
|
||||
|
||||
segment = VoiceoverSegment(
|
||||
text=text,
|
||||
start_time=start_time,
|
||||
end_time=start_time + actual_duration,
|
||||
audio_path=audio_path,
|
||||
duration=actual_duration,
|
||||
)
|
||||
segments.append(segment)
|
||||
total_duration = max(total_duration, start_time + actual_duration)
|
||||
|
||||
except TtsError as e:
|
||||
logger.warning("TTS 字幕片段 %d 合成失败,跳过: %s", i, e)
|
||||
continue
|
||||
except Exception as e:
|
||||
logger.warning("TTS 字幕片段 %d 异常,跳过: %s", i, e)
|
||||
continue
|
||||
|
||||
if not segments:
|
||||
return VoiceoverResult(success=False, error_message="所有字幕片段合成失败")
|
||||
|
||||
return VoiceoverResult(
|
||||
success=True,
|
||||
segments=segments,
|
||||
total_duration=total_duration,
|
||||
)
|
||||
|
||||
def build_audio_mix_filter(
|
||||
self,
|
||||
result: VoiceoverResult,
|
||||
*,
|
||||
video_duration: float,
|
||||
base_label: str = "0:a",
|
||||
) -> tuple[str, list[Path]]:
|
||||
"""构建配音混音滤镜.
|
||||
|
||||
将配音片段按时间轴排列,生成 amix 混入。
|
||||
|
||||
Args:
|
||||
result: 配音结果
|
||||
video_duration: 视频总时长
|
||||
base_label: 基础音轨标签
|
||||
|
||||
Returns:
|
||||
(filter_complex 字符串, 配音音频文件列表)
|
||||
"""
|
||||
if not result.success or not result.segments:
|
||||
return "", []
|
||||
|
||||
filter_parts: list[str] = []
|
||||
audio_files: list[Path] = []
|
||||
delay_labels: list[str] = []
|
||||
|
||||
for i, seg in enumerate(result.segments):
|
||||
if seg.audio_path is None or not seg.audio_path.exists():
|
||||
continue
|
||||
|
||||
audio_files.append(seg.audio_path)
|
||||
seg_label = f"v{i}"
|
||||
|
||||
# 音量调整
|
||||
# 用 adelay 延迟到字幕开始时间
|
||||
delay_ms = int(max(0, int(seg.start_time * 1000)))
|
||||
filter_parts.append(f"[{i}:a]adelay={delay_ms}:all=1,volume=0.8[{seg_label}]")
|
||||
delay_labels.append(f"[{seg_label}]")
|
||||
|
||||
if not delay_labels:
|
||||
return "", []
|
||||
|
||||
# 所有片段 concat 成一条配音音轨(用 amix 叠加多个延时后的片段
|
||||
mix_inputs = "".join(delay_labels)
|
||||
n_inputs = len(delay_labels)
|
||||
tts_label = "tts_mixed"
|
||||
|
||||
if n_inputs == 1:
|
||||
# 单个片段直接用
|
||||
filter_parts.append(f"{delay_labels[0]}[{tts_label}]")
|
||||
else:
|
||||
# 多个片段 amix 叠加
|
||||
filter_parts.append(f"{mix_inputs}amix=inputs={n_inputs}:duration=longest[{tts_label}]")
|
||||
|
||||
return ";".join(filter_parts), audio_files
|
||||
|
||||
def _probe_duration(self, audio_path: Path) -> float:
|
||||
"""探测音频时长."""
|
||||
try:
|
||||
from video_processing.ffmpeg_utils import probe_duration
|
||||
|
||||
return probe_duration(audio_path)
|
||||
except Exception:
|
||||
# 探测失败,按文件名估算
|
||||
return 0.0
|
||||
@@ -43,6 +43,9 @@ from video_processing.ffmpeg_utils import (
|
||||
from video_processing.render_audio import RenderContext, merge_audio_video, mix_audio
|
||||
from video_processing.render_subtitles import generate_ass_subtitles
|
||||
from video_processing.subtitle_generator import generate_ass_from_timeline
|
||||
from video_processing.tts_engine import TtsEngine
|
||||
|
||||
from packages.domain.tts_config import TtsConfig
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -204,6 +207,9 @@ class UnifiedRenderService:
|
||||
# 3. 计算视频总时长(用于字幕显示时长)
|
||||
video_duration = self._estimate_total_duration(layers)
|
||||
|
||||
# 3.5 TTS 配音生成(如果配置了)
|
||||
self._maybe_add_voiceover_layer(layers, video_duration=video_duration)
|
||||
|
||||
# 4. 生成 ASS 字幕文件(如果有 title/subtitle 配置)
|
||||
ass_path = self._maybe_generate_ass(video_duration)
|
||||
|
||||
@@ -504,6 +510,84 @@ class UnifiedRenderService:
|
||||
if result.returncode != 0:
|
||||
raise RuntimeError(f"音频提取失败: {result.stderr[:200]}")
|
||||
|
||||
def _maybe_add_voiceover_layer(
|
||||
self,
|
||||
layers: list[RenderLayer],
|
||||
*,
|
||||
video_duration: float,
|
||||
) -> bool:
|
||||
"""根据 plan.config 生成 TTS 配音,加到 audio 图层.
|
||||
|
||||
Returns:
|
||||
是否成功添加了配音音轨
|
||||
"""
|
||||
config = self.plan.config or {}
|
||||
tts_cfg = config.get("tts", {}) or {}
|
||||
|
||||
tts_config = TtsConfig.parse(tts_cfg)
|
||||
if not tts_config.enabled:
|
||||
return False
|
||||
|
||||
try:
|
||||
from apps.worker.services.tts_service_factory import get_tts_service
|
||||
|
||||
tts_service = get_tts_service()
|
||||
tts_engine = TtsEngine(tts_service, self.work_dir / "tts")
|
||||
|
||||
# 整段配音模式
|
||||
result = tts_engine.generate_full_voiceover(tts_config, total_duration=video_duration)
|
||||
|
||||
if not result.success or not result.segments:
|
||||
logger.warning("TTS 配音生成失败,跳过: %s", result.error_message)
|
||||
return False
|
||||
|
||||
# 获取主音轨图层(用于判断 replace 模式下是否静音原音)
|
||||
# 这里只处理混音添加,replace 模式在外部处理
|
||||
|
||||
# 找到或创建 audio 图层
|
||||
audio_layer = None
|
||||
for layer in layers:
|
||||
if layer.role == "audio":
|
||||
audio_layer = layer
|
||||
break
|
||||
|
||||
if audio_layer is None:
|
||||
from video_processing.unified_render_service import _LAYER_Z_INDEX # type: ignore
|
||||
|
||||
z_index = _LAYER_Z_INDEX.get("audio", 2)
|
||||
audio_layer = RenderLayer(role="audio", z_index=z_index)
|
||||
layers.append(audio_layer)
|
||||
|
||||
# 把配音片段作为 audio clip 加入
|
||||
for seg in result.segments:
|
||||
if seg.audio_path is None:
|
||||
continue
|
||||
vo_clip = ResolvedClip(
|
||||
clip_id=f"tts_{seg.start_time:.3f}",
|
||||
asset_id="tts_voiceover",
|
||||
local_path=seg.audio_path,
|
||||
clip_type="audio",
|
||||
order=len(audio_layer.clips),
|
||||
start_time=seg.start_time,
|
||||
duration=seg.duration,
|
||||
config={"volume": tts_config.volume, "tts": True},
|
||||
actual_duration=seg.duration,
|
||||
)
|
||||
audio_layer.clips.append(vo_clip)
|
||||
|
||||
logger.info(
|
||||
"TTS 配音已添加: plan_id=%s voice_id=%s segments=%d total_%.2fs",
|
||||
self.plan.id,
|
||||
tts_config.voice_id,
|
||||
len(result.segments),
|
||||
result.total_duration,
|
||||
)
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logger.warning("TTS 配音异常,跳过: %s", e)
|
||||
return False
|
||||
|
||||
def _can_use_pass_through(self, layers: list[RenderLayer]) -> bool:
|
||||
"""判断是否可以走直通优化路径。
|
||||
|
||||
|
||||
Executable
+231
@@ -0,0 +1,231 @@
|
||||
"""Mock TTS 服务实现.
|
||||
|
||||
使用 FFmpeg 合成简单音频模拟人声:
|
||||
- 不同音色用不同的基频(sine 波频率)
|
||||
- 语速通过 atempo 调整
|
||||
- 语调通过 asetrate 调整
|
||||
- 加一点 tremolo 效果让声音更自然
|
||||
|
||||
用于开发测试,不依赖外部 TTS 服务。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import subprocess
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
from packages.domain.voice_presets import get_voice, list_voices
|
||||
from packages.ports.tts_service import TtsError, TtsService
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Mock 时长估算:每字约 0.3 秒(中文)
|
||||
_CHARS_PER_SECOND = 3.3
|
||||
|
||||
|
||||
class MockTtsService(TtsService):
|
||||
"""Mock TTS 服务 — 用 FFmpeg 合成测试音频."""
|
||||
|
||||
def __init__(self, ffmpeg_bin: str = "ffmpeg") -> None:
|
||||
self._ffmpeg_bin = ffmpeg_bin
|
||||
|
||||
@property
|
||||
def provider_name(self) -> str:
|
||||
return "mock"
|
||||
|
||||
def available_voices(self) -> list[str]:
|
||||
return [v.voice_id for v in list_voices(provider="mock")]
|
||||
|
||||
def synthesize(
|
||||
self,
|
||||
text: str,
|
||||
*,
|
||||
voice_id: str = "",
|
||||
speed: float = 1.0,
|
||||
pitch: float = 0.0,
|
||||
output_path: Path | None = None,
|
||||
sample_rate: int = 22050,
|
||||
format: str = "wav",
|
||||
) -> Path:
|
||||
"""合成 Mock 音频.
|
||||
|
||||
用 FFmpeg sine 波合成带轻微调制的音频,模拟人声。
|
||||
时长根据文本长度估算。
|
||||
"""
|
||||
if not text.strip():
|
||||
raise TtsError("文本不能为空")
|
||||
|
||||
# 语速边界
|
||||
if speed <= 0:
|
||||
speed = 1.0
|
||||
speed = max(0.5, min(2.0, speed))
|
||||
|
||||
# 语调边界
|
||||
pitch = max(-12, min(12, pitch))
|
||||
|
||||
# 解析音色
|
||||
voice = get_voice(voice_id) if voice_id else get_voice("female_warm")
|
||||
if voice is None:
|
||||
voice = get_voice("female_warm")
|
||||
|
||||
# 计算基频(从 provider_voice_id 里提取,或者按音色默认)
|
||||
base_freq = self._extract_freq(voice.provider_voice_id, voice.gender.value)
|
||||
|
||||
# 计算时长(按文本长度)
|
||||
duration = self.estimate_duration(text, speed=speed)
|
||||
duration = max(0.5, duration) # 最短 0.5 秒
|
||||
|
||||
# 输出路径
|
||||
if output_path is None:
|
||||
suffix = f".{format}"
|
||||
tmp = tempfile.NamedTemporaryFile(suffix=suffix, delete=False)
|
||||
tmp.close()
|
||||
output_path = Path(tmp.name)
|
||||
|
||||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
try:
|
||||
self._synthesize_with_ffmpeg(
|
||||
output_path=output_path,
|
||||
base_freq=base_freq,
|
||||
duration=duration,
|
||||
speed=speed,
|
||||
pitch=pitch,
|
||||
sample_rate=sample_rate,
|
||||
format=format,
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error("Mock TTS 合成失败: %s", e)
|
||||
raise TtsError(f"Mock TTS 合成失败: {e}") from e
|
||||
|
||||
return output_path
|
||||
|
||||
def estimate_duration(self, text: str, *, speed: float = 1.0) -> float:
|
||||
"""估算音频时长.
|
||||
|
||||
按中文字符数估算:每字约 0.3 秒。
|
||||
"""
|
||||
if not text:
|
||||
return 0.0
|
||||
# 去除空白后的字符数
|
||||
char_count = len([c for c in text if not c.isspace()])
|
||||
if char_count == 0:
|
||||
return 0.0
|
||||
base_duration = char_count / _CHARS_PER_SECOND
|
||||
return base_duration / max(0.1, speed)
|
||||
|
||||
def _extract_freq(self, provider_voice_id: str, gender: str) -> float:
|
||||
"""从 provider_voice_id 提取基频,或按性别给默认值."""
|
||||
if provider_voice_id.startswith("sine_"):
|
||||
try:
|
||||
return float(provider_voice_id.split("_")[1])
|
||||
except (IndexError, ValueError):
|
||||
pass
|
||||
|
||||
# 按性别给默认基频
|
||||
if gender == "male":
|
||||
return 120.0
|
||||
elif gender == "child":
|
||||
return 350.0
|
||||
else: # female
|
||||
return 220.0
|
||||
|
||||
def _synthesize_with_ffmpeg(
|
||||
self,
|
||||
*,
|
||||
output_path: Path,
|
||||
base_freq: float,
|
||||
duration: float,
|
||||
speed: float,
|
||||
pitch: float,
|
||||
sample_rate: int,
|
||||
format: str,
|
||||
) -> None:
|
||||
"""使用 FFmpeg 合成音频.
|
||||
|
||||
效果链:
|
||||
1. sine 波生成基频
|
||||
2. tremolo 增加轻微颤音
|
||||
3. aeval 模拟简单的音色变化(让声音不那么单调)
|
||||
4. atempo 调整语速
|
||||
5. asetrate 调整语调
|
||||
6. volume 调整音量
|
||||
"""
|
||||
# 语调频率偏移因子(每半音 = 2^(1/12) ≈ 1.05946)
|
||||
pitch_factor = 2 ** (pitch / 12)
|
||||
|
||||
# 颤音参数
|
||||
tremolo_freq = 5.0 # 5Hz 颤音
|
||||
tremolo_depth = 0.3 # 30% 深度
|
||||
|
||||
# 构建滤镜链
|
||||
filters: list[str] = []
|
||||
|
||||
# 生成基频 + 泛音(让声音更丰富)
|
||||
# 用多个 sine 波叠加模拟更自然的音色
|
||||
filter_parts = []
|
||||
|
||||
# 主音 + 轻微频率调制
|
||||
filter_parts.append(f"sine=frequency={base_freq}:duration={duration}:sample_rate={sample_rate}")
|
||||
|
||||
# 颤音效果
|
||||
filter_parts.append(f"tremolo=f={tremolo_freq}:d={tremolo_depth}")
|
||||
|
||||
# 语速调整(同时调整时长)
|
||||
if abs(speed - 1.0) > 0.01:
|
||||
filter_parts.append(f"atempo={speed:.3f}")
|
||||
|
||||
# 语调调整(通过采样率变化实现,同时补偿时长)
|
||||
if abs(pitch) > 0.01:
|
||||
new_rate = int(sample_rate * pitch_factor)
|
||||
filter_parts.append(f"asetrate={new_rate}")
|
||||
filter_parts.append(f"aresample={sample_rate}")
|
||||
|
||||
# 音量包络:淡入淡出
|
||||
fade_in = min(0.05, duration * 0.1)
|
||||
fade_out = min(0.1, duration * 0.2)
|
||||
filter_parts.append(f"afade=t=in:d={fade_in}")
|
||||
filter_parts.append(f"afade=t=out:st={max(0, duration - fade_out)}:d={fade_out}")
|
||||
|
||||
# 音量调整到合适大小
|
||||
filter_parts.append("volume=0.3")
|
||||
|
||||
filter_complex = ",".join(filter_parts)
|
||||
|
||||
# 编码参数
|
||||
if format == "mp3":
|
||||
codec_args = ["-acodec", "libmp3lame", "-b:a", "128k"]
|
||||
else:
|
||||
codec_args = ["-acodec", "pcm_s16le"]
|
||||
|
||||
command = [
|
||||
self._ffmpeg_bin,
|
||||
"-y",
|
||||
"-f",
|
||||
"lavfi",
|
||||
"-i",
|
||||
filter_complex,
|
||||
*codec_args,
|
||||
"-ar",
|
||||
str(sample_rate),
|
||||
"-ac",
|
||||
"1",
|
||||
str(output_path),
|
||||
]
|
||||
|
||||
logger.debug("Mock TTS FFmpeg 命令: %s", " ".join(command))
|
||||
|
||||
result = subprocess.run(
|
||||
command,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=max(30, duration * 2 + 10),
|
||||
)
|
||||
|
||||
if result.returncode != 0:
|
||||
raise TtsError(f"FFmpeg 合成失败: {result.stderr[-500:]}")
|
||||
|
||||
if not output_path.exists() or output_path.stat().st_size == 0:
|
||||
raise TtsError("输出文件为空或不存在")
|
||||
Executable
+102
@@ -0,0 +1,102 @@
|
||||
"""TTS 配音配置模型."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any, Optional
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class TtsConfig:
|
||||
"""TTS 配音配置.
|
||||
|
||||
Attributes:
|
||||
enabled: 是否启用配音
|
||||
voice_id: 音色 ID
|
||||
speed: 语速 (0.5 ~ 2.0)
|
||||
pitch: 语调 (-12 ~ 12 半音)
|
||||
volume: 音量 (0.0 ~ 1.0)
|
||||
text: 配音文本(整段配音时使用)
|
||||
align_mode: 对齐模式 - "subtitle"=按字幕对齐 / "full"=整段配音
|
||||
overlap_mode: 与原音的叠加模式 - "replace"=替换 / "mix"=混音
|
||||
"""
|
||||
|
||||
enabled: bool = False
|
||||
voice_id: str = ""
|
||||
speed: float = 1.0
|
||||
pitch: float = 0.0
|
||||
volume: float = 0.8
|
||||
text: str = ""
|
||||
align_mode: str = "full" # subtitle / full
|
||||
overlap_mode: str = "replace" # replace / mix
|
||||
|
||||
@classmethod
|
||||
def parse(cls, data: Optional[dict[str, Any]]) -> "TtsConfig":
|
||||
"""从 dict 解析配置,无效值回退到默认."""
|
||||
if not data or not isinstance(data, dict):
|
||||
return cls()
|
||||
|
||||
enabled = data.get("enabled", False)
|
||||
if not isinstance(enabled, bool):
|
||||
enabled = False
|
||||
|
||||
if not enabled:
|
||||
return cls(enabled=False)
|
||||
|
||||
voice_id = data.get("voice_id", "")
|
||||
if not isinstance(voice_id, str):
|
||||
voice_id = ""
|
||||
|
||||
speed = data.get("speed", 1.0)
|
||||
if not isinstance(speed, (int, float)):
|
||||
speed = 1.0
|
||||
|
||||
pitch = data.get("pitch", 0.0)
|
||||
if not isinstance(pitch, (int, float)):
|
||||
pitch = 0.0
|
||||
|
||||
volume = data.get("volume", 0.8)
|
||||
if not isinstance(volume, (int, float)):
|
||||
volume = 0.8
|
||||
|
||||
text = data.get("text", "")
|
||||
if not isinstance(text, str):
|
||||
text = ""
|
||||
|
||||
align_mode = data.get("align_mode", "full")
|
||||
if align_mode not in ("subtitle", "full"):
|
||||
align_mode = "full"
|
||||
|
||||
overlap_mode = data.get("overlap_mode", "replace")
|
||||
if overlap_mode not in ("replace", "mix"):
|
||||
overlap_mode = "replace"
|
||||
|
||||
config = cls(
|
||||
enabled=enabled,
|
||||
voice_id=voice_id,
|
||||
speed=float(speed),
|
||||
pitch=float(pitch),
|
||||
volume=float(volume),
|
||||
text=text,
|
||||
align_mode=align_mode,
|
||||
overlap_mode=overlap_mode,
|
||||
)
|
||||
config._clamp()
|
||||
return config
|
||||
|
||||
def _clamp(self) -> None:
|
||||
"""边界钳制."""
|
||||
if self.speed < 0.5:
|
||||
self.speed = 0.5
|
||||
elif self.speed > 2.0:
|
||||
self.speed = 2.0
|
||||
|
||||
if self.pitch < -12:
|
||||
self.pitch = -12
|
||||
elif self.pitch > 12:
|
||||
self.pitch = 12
|
||||
|
||||
if self.volume < 0.0:
|
||||
self.volume = 0.0
|
||||
elif self.volume > 1.0:
|
||||
self.volume = 1.0
|
||||
Executable
+222
@@ -0,0 +1,222 @@
|
||||
"""配音引擎音色预设.
|
||||
|
||||
与 CosyVoice 的 preset_voices 区分:
|
||||
- preset_voices.py: CosyVoice 真实音色(阿里云)
|
||||
- voice_presets.py: 配音引擎通用音色预设(含 mock/后续接入的真实 TTS)
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from dataclasses import dataclass
|
||||
|
||||
if sys.version_info >= (3, 11):
|
||||
from enum import StrEnum
|
||||
else:
|
||||
from enum import Enum
|
||||
|
||||
class StrEnum(str, Enum):
|
||||
pass
|
||||
|
||||
|
||||
class VoiceGender(StrEnum):
|
||||
"""音色性别."""
|
||||
|
||||
MALE = "male"
|
||||
FEMALE = "female"
|
||||
CHILD = "child"
|
||||
|
||||
|
||||
class VoiceStyle(StrEnum):
|
||||
"""音色风格."""
|
||||
|
||||
STABLE = "stable" # 沉稳
|
||||
LIVELY = "lively" # 活泼
|
||||
CUSTOMER_SERVICE = "customer_service" # 客服
|
||||
NARRATION = "narration" # 旁白
|
||||
NEWS = "news" # 新闻
|
||||
STORY = "story" # 故事
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class VoicePreset:
|
||||
"""音色预设.
|
||||
|
||||
Attributes:
|
||||
voice_id: 音色唯一标识
|
||||
name: 音色名称
|
||||
gender: 性别
|
||||
style: 风格
|
||||
description: 描述
|
||||
provider: 供应商(mock/aliyun/xunfei)
|
||||
provider_voice_id: 供应商侧音色 ID
|
||||
default_speed: 默认语速
|
||||
default_pitch: 默认语调
|
||||
sample_rate: 采样率
|
||||
language: 语言
|
||||
"""
|
||||
|
||||
voice_id: str
|
||||
name: str
|
||||
gender: VoiceGender = VoiceGender.FEMALE
|
||||
style: VoiceStyle = VoiceStyle.NARRATION
|
||||
description: str = ""
|
||||
provider: str = "mock"
|
||||
provider_voice_id: str = ""
|
||||
default_speed: float = 1.0
|
||||
default_pitch: float = 0.0
|
||||
sample_rate: int = 22050
|
||||
language: str = "zh-CN"
|
||||
|
||||
|
||||
# ─── Mock 音色预设列表 ──────────────────────────────────────
|
||||
|
||||
MOCK_VOICES: list[VoicePreset] = [
|
||||
VoicePreset(
|
||||
voice_id="female_warm",
|
||||
name="温暖女声",
|
||||
gender=VoiceGender.FEMALE,
|
||||
style=VoiceStyle.NARRATION,
|
||||
description="温柔温暖的女声,适合情感类、生活类视频",
|
||||
provider="mock",
|
||||
provider_voice_id="sine_220",
|
||||
default_speed=1.0,
|
||||
default_pitch=0.0,
|
||||
sample_rate=22050,
|
||||
language="zh-CN",
|
||||
),
|
||||
VoicePreset(
|
||||
voice_id="male_stable",
|
||||
name="沉稳男声",
|
||||
gender=VoiceGender.MALE,
|
||||
style=VoiceStyle.STABLE,
|
||||
description="沉稳厚重的男声,适合商务、知识类视频",
|
||||
provider="mock",
|
||||
provider_voice_id="sine_110",
|
||||
default_speed=0.9,
|
||||
default_pitch=0.0,
|
||||
sample_rate=22050,
|
||||
language="zh-CN",
|
||||
),
|
||||
VoicePreset(
|
||||
voice_id="female_lively",
|
||||
name="活泼女声",
|
||||
gender=VoiceGender.FEMALE,
|
||||
style=VoiceStyle.LIVELY,
|
||||
description="明亮活泼的女声,适合vlog、美食、旅行类视频",
|
||||
provider="mock",
|
||||
provider_voice_id="sine_280",
|
||||
default_speed=1.2,
|
||||
default_pitch=2.0,
|
||||
sample_rate=22050,
|
||||
language="zh-CN",
|
||||
),
|
||||
VoicePreset(
|
||||
voice_id="child_cute",
|
||||
name="可爱童声",
|
||||
gender=VoiceGender.CHILD,
|
||||
style=VoiceStyle.STORY,
|
||||
description="清脆可爱的童声,适合儿童教育、动画类视频",
|
||||
provider="mock",
|
||||
provider_voice_id="sine_380",
|
||||
default_speed=1.0,
|
||||
default_pitch=4.0,
|
||||
sample_rate=22050,
|
||||
language="zh-CN",
|
||||
),
|
||||
VoicePreset(
|
||||
voice_id="female_service",
|
||||
name="客服女声",
|
||||
gender=VoiceGender.FEMALE,
|
||||
style=VoiceStyle.CUSTOMER_SERVICE,
|
||||
description="专业清晰的客服女声,适合产品介绍、教程类视频",
|
||||
provider="mock",
|
||||
provider_voice_id="sine_250",
|
||||
default_speed=1.0,
|
||||
default_pitch=1.0,
|
||||
sample_rate=22050,
|
||||
language="zh-CN",
|
||||
),
|
||||
VoicePreset(
|
||||
voice_id="male_news",
|
||||
name="新闻男声",
|
||||
gender=VoiceGender.MALE,
|
||||
style=VoiceStyle.NEWS,
|
||||
description="字正腔圆的新闻播报声,适合资讯、时政类视频",
|
||||
provider="mock",
|
||||
provider_voice_id="sine_140",
|
||||
default_speed=1.0,
|
||||
default_pitch=0.0,
|
||||
sample_rate=22050,
|
||||
language="zh-CN",
|
||||
),
|
||||
VoicePreset(
|
||||
voice_id="female_soft",
|
||||
name="轻柔女声",
|
||||
gender=VoiceGender.FEMALE,
|
||||
style=VoiceStyle.STORY,
|
||||
description="轻柔舒缓的女声,适合睡前故事、冥想类视频",
|
||||
provider="mock",
|
||||
provider_voice_id="sine_180",
|
||||
default_speed=0.8,
|
||||
default_pitch=0.0,
|
||||
sample_rate=22050,
|
||||
language="zh-CN",
|
||||
),
|
||||
VoicePreset(
|
||||
voice_id="male_magnetic",
|
||||
name="磁性男声",
|
||||
gender=VoiceGender.MALE,
|
||||
style=VoiceStyle.STORY,
|
||||
description="低沉磁性的男声,适合电影解说、读书类视频",
|
||||
provider="mock",
|
||||
provider_voice_id="sine_90",
|
||||
default_speed=0.85,
|
||||
default_pitch=-2.0,
|
||||
sample_rate=22050,
|
||||
language="zh-CN",
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
# voice_id → VoicePreset
|
||||
_MOCK_VOICE_MAP: dict[str, VoicePreset] = {v.voice_id: v for v in MOCK_VOICES}
|
||||
|
||||
|
||||
def get_voice(voice_id: str, *, provider: str = "mock") -> VoicePreset | None:
|
||||
"""根据 voice_id 获取音色预设."""
|
||||
if provider == "mock":
|
||||
return _MOCK_VOICE_MAP.get(voice_id)
|
||||
return None
|
||||
|
||||
|
||||
def list_voices(
|
||||
*,
|
||||
gender: str | None = None,
|
||||
style: str | None = None,
|
||||
provider: str | None = None,
|
||||
keyword: str | None = None,
|
||||
) -> list[VoicePreset]:
|
||||
"""按条件筛选音色列表."""
|
||||
# 目前只有 mock 音色
|
||||
result = list(MOCK_VOICES)
|
||||
|
||||
if provider and provider != "mock":
|
||||
return []
|
||||
|
||||
if gender:
|
||||
result = [v for v in result if v.gender.value == gender]
|
||||
|
||||
if style:
|
||||
result = [v for v in result if v.style.value == style]
|
||||
|
||||
if keyword:
|
||||
kw = keyword.lower()
|
||||
result = [v for v in result if kw in v.name.lower() or kw in v.description.lower() or kw in v.voice_id.lower()]
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def get_default_voice() -> VoicePreset:
|
||||
"""获取默认音色."""
|
||||
return MOCK_VOICES[0]
|
||||
Executable
+78
@@ -0,0 +1,78 @@
|
||||
"""TTS 服务抽象接口 (Port).
|
||||
|
||||
新增 TTS 供应商时,实现本接口即可。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from abc import ABC, abstractmethod
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
class TtsService(ABC):
|
||||
"""TTS 服务抽象基类.
|
||||
|
||||
所有 TTS 供应商(Mock / 阿里云 / 讯飞 等)都需要实现本接口。
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def synthesize(
|
||||
self,
|
||||
text: str,
|
||||
*,
|
||||
voice_id: str = "",
|
||||
speed: float = 1.0,
|
||||
pitch: float = 0.0,
|
||||
output_path: Path | None = None,
|
||||
sample_rate: int = 22050,
|
||||
format: str = "wav",
|
||||
) -> Path:
|
||||
"""文本转语音合成.
|
||||
|
||||
Args:
|
||||
text: 输入文本
|
||||
voice_id: 音色 ID
|
||||
speed: 语速 (0.5 ~ 2.0)
|
||||
pitch: 语调(半音,-12 ~ 12)
|
||||
output_path: 输出文件路径(None 则自动生成)
|
||||
sample_rate: 采样率
|
||||
format: 输出格式 (wav/mp3)
|
||||
|
||||
Returns:
|
||||
输出音频文件路径
|
||||
|
||||
Raises:
|
||||
TtsError: 合成失败
|
||||
"""
|
||||
...
|
||||
|
||||
@abstractmethod
|
||||
def estimate_duration(self, text: str, *, speed: float = 1.0) -> float:
|
||||
"""预估音频时长(秒).
|
||||
|
||||
用于在实际合成前估算时长,方便时间轴对齐。
|
||||
|
||||
Args:
|
||||
text: 输入文本
|
||||
speed: 语速
|
||||
|
||||
Returns:
|
||||
预估时长(秒)
|
||||
"""
|
||||
...
|
||||
|
||||
@property
|
||||
@abstractmethod
|
||||
def provider_name(self) -> str:
|
||||
"""供应商名称."""
|
||||
...
|
||||
|
||||
def available_voices(self) -> list[str]:
|
||||
"""支持的音色 ID 列表."""
|
||||
return []
|
||||
|
||||
|
||||
class TtsError(Exception):
|
||||
"""TTS 合成异常."""
|
||||
|
||||
pass
|
||||
Executable
+413
@@ -0,0 +1,413 @@
|
||||
"""TTS 配音引擎单元测试."""
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from apps.worker.video_processing.tts_engine import TtsEngine, VoiceoverResult, VoiceoverSegment
|
||||
from packages.adapters.tts.mock_tts_service import MockTtsService
|
||||
from packages.domain.tts_config import TtsConfig
|
||||
from packages.domain.voice_presets import (
|
||||
VoiceGender,
|
||||
VoicePreset,
|
||||
VoiceStyle,
|
||||
get_default_voice,
|
||||
get_voice,
|
||||
list_voices,
|
||||
)
|
||||
from packages.ports.tts_service import TtsError, TtsService
|
||||
|
||||
# ─── TtsConfig 配置解析 ────────────────────────────────────
|
||||
|
||||
|
||||
class TestTtsConfig:
|
||||
def test_default_values(self):
|
||||
config = TtsConfig()
|
||||
assert config.enabled is False
|
||||
assert config.voice_id == ""
|
||||
assert config.speed == 1.0
|
||||
assert config.pitch == 0.0
|
||||
assert config.volume == 0.8
|
||||
assert config.text == ""
|
||||
assert config.align_mode == "full"
|
||||
assert config.overlap_mode == "replace"
|
||||
|
||||
def test_parse_none(self):
|
||||
config = TtsConfig.parse(None)
|
||||
assert config.enabled is False
|
||||
|
||||
def test_parse_empty_dict(self):
|
||||
config = TtsConfig.parse({})
|
||||
assert config.enabled is False
|
||||
|
||||
def test_parse_enabled(self):
|
||||
config = TtsConfig.parse({"enabled": True, "voice_id": "female_warm", "text": "你好"})
|
||||
assert config.enabled is True
|
||||
assert config.voice_id == "female_warm"
|
||||
assert config.text == "你好"
|
||||
|
||||
def test_parse_not_enabled_ignores_other_fields(self):
|
||||
config = TtsConfig.parse({"enabled": False, "voice_id": "test", "speed": 2.0})
|
||||
assert config.enabled is False
|
||||
assert config.voice_id == ""
|
||||
assert config.speed == 1.0
|
||||
|
||||
def test_parse_speed_boundary(self):
|
||||
config = TtsConfig.parse({"enabled": True, "speed": 0.1})
|
||||
assert config.speed == 0.5
|
||||
|
||||
config = TtsConfig.parse({"enabled": True, "speed": 5.0})
|
||||
assert config.speed == 2.0
|
||||
|
||||
def test_parse_pitch_boundary(self):
|
||||
config = TtsConfig.parse({"enabled": True, "pitch": -20})
|
||||
assert config.pitch == -12
|
||||
|
||||
config = TtsConfig.parse({"enabled": True, "pitch": 20})
|
||||
assert config.pitch == 12
|
||||
|
||||
def test_parse_volume_boundary(self):
|
||||
config = TtsConfig.parse({"enabled": True, "volume": -1.0})
|
||||
assert config.volume == 0.0
|
||||
|
||||
config = TtsConfig.parse({"enabled": True, "volume": 2.0})
|
||||
assert config.volume == 1.0
|
||||
|
||||
def test_parse_invalid_types(self):
|
||||
config = TtsConfig.parse(
|
||||
{
|
||||
"enabled": True,
|
||||
"speed": "fast",
|
||||
"pitch": "high",
|
||||
"volume": "loud",
|
||||
"voice_id": 123,
|
||||
"text": 456,
|
||||
"align_mode": "invalid",
|
||||
"overlap_mode": "invalid",
|
||||
}
|
||||
)
|
||||
assert config.speed == 1.0
|
||||
assert config.pitch == 0.0
|
||||
assert config.volume == 0.8
|
||||
assert config.voice_id == ""
|
||||
assert config.text == ""
|
||||
assert config.align_mode == "full"
|
||||
assert config.overlap_mode == "replace"
|
||||
|
||||
|
||||
# ─── 预设音色库 ───────────────────────────────────────────
|
||||
|
||||
|
||||
class TestPresetVoices:
|
||||
def test_list_voices_all(self):
|
||||
voices = list_voices()
|
||||
assert len(voices) >= 6
|
||||
|
||||
def test_get_voice_existing(self):
|
||||
voice = get_voice("female_warm")
|
||||
assert voice is not None
|
||||
assert voice.voice_id == "female_warm"
|
||||
assert voice.name == "温暖女声"
|
||||
assert voice.gender == VoiceGender.FEMALE
|
||||
|
||||
def test_get_voice_not_found(self):
|
||||
assert get_voice("nonexistent") is None
|
||||
|
||||
def test_get_default_voice(self):
|
||||
voice = get_default_voice()
|
||||
assert voice is not None
|
||||
assert voice.provider == "mock"
|
||||
|
||||
def test_filter_by_gender(self):
|
||||
female = list_voices(gender="female")
|
||||
assert len(female) >= 2
|
||||
for v in female:
|
||||
assert v.gender == VoiceGender.FEMALE
|
||||
|
||||
male = list_voices(gender="male")
|
||||
assert len(male) >= 2
|
||||
for v in male:
|
||||
assert v.gender == VoiceGender.MALE
|
||||
|
||||
def test_filter_by_style(self):
|
||||
stable = list_voices(style="stable")
|
||||
for v in stable:
|
||||
assert v.style == VoiceStyle.STABLE
|
||||
|
||||
def test_filter_by_keyword(self):
|
||||
result = list_voices(keyword="女声")
|
||||
assert len(result) >= 1
|
||||
for v in result:
|
||||
assert "女" in v.name
|
||||
|
||||
def test_voice_fields(self):
|
||||
voice = get_voice("male_stable")
|
||||
assert voice is not None
|
||||
assert voice.name
|
||||
assert voice.voice_id
|
||||
assert voice.description
|
||||
assert voice.sample_rate > 0
|
||||
|
||||
|
||||
# ─── Mock TTS 服务 ────────────────────────────────────────
|
||||
|
||||
|
||||
class TestMockTtsService:
|
||||
def setup_method(self):
|
||||
self.service = MockTtsService()
|
||||
|
||||
def test_provider_name(self):
|
||||
assert self.service.provider_name == "mock"
|
||||
|
||||
def test_available_voices(self):
|
||||
voices = self.service.available_voices()
|
||||
assert len(voices) >= 6
|
||||
|
||||
def test_synthesize_success(self, tmp_path):
|
||||
output = tmp_path / "test.wav"
|
||||
result = self.service.synthesize(
|
||||
"测试文本一二三四五",
|
||||
voice_id="female_warm",
|
||||
speed=1.0,
|
||||
output_path=output,
|
||||
)
|
||||
assert result == output
|
||||
assert result.exists()
|
||||
assert result.stat().st_size > 0
|
||||
|
||||
def test_synthesize_different_voices(self, tmp_path):
|
||||
voices = ["female_warm", "male_stable", "child_cute"]
|
||||
for vid in voices:
|
||||
output = tmp_path / f"{vid}.wav"
|
||||
result = self.service.synthesize("测试", voice_id=vid, output_path=output)
|
||||
assert result.exists()
|
||||
|
||||
def test_synthesize_speed_faster(self, tmp_path):
|
||||
"""语速快应该时长短."""
|
||||
out_slow = tmp_path / "slow.wav"
|
||||
out_fast = tmp_path / "fast.wav"
|
||||
text = "一二三四五六七八九十"
|
||||
|
||||
self.service.synthesize(text, speed=0.5, output_path=out_slow)
|
||||
self.service.synthesize(text, speed=2.0, output_path=out_fast)
|
||||
|
||||
# 快速应该文件更小(时长短)
|
||||
size_slow = out_slow.stat().st_size
|
||||
size_fast = out_fast.stat().st_size
|
||||
assert size_fast < size_slow
|
||||
|
||||
def test_synthesize_pitch_changes(self, tmp_path):
|
||||
output = tmp_path / "high_pitch.wav"
|
||||
result = self.service.synthesize("测试", voice_id="female_warm", pitch=6, output_path=output)
|
||||
assert result.exists()
|
||||
|
||||
def test_synthesize_empty_text_raises(self):
|
||||
with pytest.raises(TtsError):
|
||||
self.service.synthesize("")
|
||||
|
||||
def test_synthesize_whitespace_text_raises(self):
|
||||
with pytest.raises(TtsError):
|
||||
self.service.synthesize(" ")
|
||||
|
||||
def test_estimate_duration(self):
|
||||
dur = self.service.estimate_duration("一二三四五")
|
||||
assert dur > 0
|
||||
assert dur < 10 # 5个字应该少于10秒
|
||||
|
||||
def test_estimate_duration_speed(self):
|
||||
text = "一二三四五六七八九十"
|
||||
dur_normal = self.service.estimate_duration(text, speed=1.0)
|
||||
dur_fast = self.service.estimate_duration(text, speed=2.0)
|
||||
dur_slow = self.service.estimate_duration(text, speed=0.5)
|
||||
|
||||
assert dur_fast < dur_normal
|
||||
assert dur_slow > dur_normal
|
||||
|
||||
def test_synthesize_unknown_voice_fallback(self, tmp_path):
|
||||
output = tmp_path / "fallback.wav"
|
||||
# 未知音色应该 fallback 到默认音色,不报错
|
||||
result = self.service.synthesize("测试", voice_id="unknown_voice", output_path=output)
|
||||
assert result.exists()
|
||||
|
||||
|
||||
# ─── TtsEngine 配音引擎 ──────────────────────────────────
|
||||
|
||||
|
||||
class TestTtsEngine:
|
||||
def _make_engine(self, tmp_path):
|
||||
service = MockTtsService()
|
||||
work_dir = tmp_path / "tts_engine"
|
||||
return TtsEngine(service, work_dir)
|
||||
|
||||
def test_generate_full_voiceover_disabled(self, tmp_path):
|
||||
engine = self._make_engine(tmp_path)
|
||||
config = TtsConfig(enabled=False)
|
||||
result = engine.generate_full_voiceover(config)
|
||||
assert result.success is False
|
||||
|
||||
def test_generate_full_voiceover_empty_text(self, tmp_path):
|
||||
engine = self._make_engine(tmp_path)
|
||||
config = TtsConfig(enabled=True, text="")
|
||||
result = engine.generate_full_voiceover(config)
|
||||
assert result.success is False
|
||||
|
||||
def test_generate_full_voiceover_success(self, tmp_path):
|
||||
engine = self._make_engine(tmp_path)
|
||||
config = TtsConfig(
|
||||
enabled=True,
|
||||
voice_id="female_warm",
|
||||
text="这是一段测试配音文本",
|
||||
)
|
||||
result = engine.generate_full_voiceover(config)
|
||||
assert result.success is True
|
||||
assert len(result.segments) == 1
|
||||
assert result.total_duration > 0
|
||||
assert result.segments[0].audio_path is not None
|
||||
assert result.segments[0].audio_path.exists()
|
||||
assert result.segments[0].duration > 0
|
||||
|
||||
def test_generate_full_voiceover_with_speed(self, tmp_path):
|
||||
engine = self._make_engine(tmp_path)
|
||||
config_slow = TtsConfig(
|
||||
enabled=True,
|
||||
voice_id="female_warm",
|
||||
text="测试文本一二三四五六七八九十",
|
||||
speed=0.5,
|
||||
)
|
||||
config_fast = TtsConfig(
|
||||
enabled=True,
|
||||
voice_id="female_warm",
|
||||
text="测试文本一二三四五六七八九十",
|
||||
speed=2.0,
|
||||
)
|
||||
result_slow = engine.generate_full_voiceover(config_slow)
|
||||
result_fast = engine.generate_full_voiceover(config_fast)
|
||||
|
||||
assert result_slow.success
|
||||
assert result_fast.success
|
||||
# 慢速时长 > 快速时长
|
||||
assert result_slow.total_duration > result_fast.total_duration
|
||||
|
||||
def test_generate_subtitle_voiceover_empty_subtitles(self, tmp_path):
|
||||
engine = self._make_engine(tmp_path)
|
||||
config = TtsConfig(enabled=True, voice_id="female_warm")
|
||||
result = engine.generate_subtitle_voiceover(config, [])
|
||||
assert result.success is False
|
||||
|
||||
def test_generate_subtitle_voiceover_success(self, tmp_path):
|
||||
engine = self._make_engine(tmp_path)
|
||||
config = TtsConfig(enabled=True, voice_id="female_warm", align_mode="subtitle")
|
||||
subtitles = [
|
||||
{"text": "大家好", "start_time": 0, "end_time": 2},
|
||||
{"text": "欢迎观看", "start_time": 2, "end_time": 4},
|
||||
{"text": "今天的视频", "start_time": 4, "end_time": 6},
|
||||
]
|
||||
result = engine.generate_subtitle_voiceover(config, subtitles)
|
||||
assert result.success is True
|
||||
assert len(result.segments) == 3
|
||||
|
||||
# 每个片段的 start_time 应该对应字幕的开始时间
|
||||
assert result.segments[0].start_time == 0
|
||||
assert result.segments[1].start_time == 2
|
||||
assert result.segments[2].start_time == 4
|
||||
|
||||
for seg in result.segments:
|
||||
assert seg.audio_path is not None
|
||||
assert seg.audio_path.exists()
|
||||
assert seg.duration > 0
|
||||
|
||||
def test_generate_subtitle_voiceover_skips_empty(self, tmp_path):
|
||||
engine = self._make_engine(tmp_path)
|
||||
config = TtsConfig(enabled=True, voice_id="female_warm")
|
||||
subtitles = [
|
||||
{"text": "有文本", "start_time": 0, "end_time": 1},
|
||||
{"text": "", "start_time": 1, "end_time": 2},
|
||||
{"text": "也有文本", "start_time": 2, "end_time": 3},
|
||||
]
|
||||
result = engine.generate_subtitle_voiceover(config, subtitles)
|
||||
assert result.success is True
|
||||
assert len(result.segments) == 2 # 跳过了空文本
|
||||
|
||||
def test_generate_full_voiceover_failure_graceful(self, tmp_path, monkeypatch):
|
||||
"""失败时优雅降级,不抛出异常."""
|
||||
engine = self._make_engine(tmp_path)
|
||||
|
||||
def failing_synth(*args, **kwargs):
|
||||
raise TtsError("模拟失败")
|
||||
|
||||
monkeypatch.setattr(engine._tts, "synthesize", failing_synth)
|
||||
|
||||
config = TtsConfig(enabled=True, voice_id="test", text="测试")
|
||||
result = engine.generate_full_voiceover(config)
|
||||
assert result.success is False
|
||||
assert result.error_message
|
||||
assert "模拟失败" in result.error_message
|
||||
|
||||
def test_generate_subtitle_voiceover_partial_failure(self, tmp_path, monkeypatch):
|
||||
"""部分片段失败时跳过,其他正常生成."""
|
||||
engine = self._make_engine(tmp_path)
|
||||
original_synth = engine._tts.synthesize
|
||||
call_count = [0]
|
||||
|
||||
def sometimes_fail(*args, **kwargs):
|
||||
call_count[0] += 1
|
||||
if call_count[0] == 2: # 第2个片段失败
|
||||
raise TtsError("模拟失败")
|
||||
return original_synth(*args, **kwargs)
|
||||
|
||||
monkeypatch.setattr(engine._tts, "synthesize", sometimes_fail)
|
||||
|
||||
config = TtsConfig(enabled=True, voice_id="female_warm")
|
||||
subtitles = [
|
||||
{"text": "第一段", "start_time": 0, "end_time": 2},
|
||||
{"text": "第二段失败", "start_time": 2, "end_time": 4},
|
||||
{"text": "第三段", "start_time": 4, "end_time": 6},
|
||||
]
|
||||
result = engine.generate_subtitle_voiceover(config, subtitles)
|
||||
# 有部分成功就算成功
|
||||
assert result.success is True
|
||||
assert len(result.segments) == 2 # 跳过了失败的第2段
|
||||
|
||||
def test_build_audio_mix_filter_empty(self, tmp_path):
|
||||
engine = self._make_engine(tmp_path)
|
||||
result = VoiceoverResult(success=False)
|
||||
filter_str, files = engine.build_audio_mix_filter(result, video_duration=10)
|
||||
assert filter_str == ""
|
||||
assert files == []
|
||||
|
||||
def test_build_audio_mix_filter_single(self, tmp_path):
|
||||
engine = self._make_engine(tmp_path)
|
||||
audio_file = tmp_path / "seg.wav"
|
||||
audio_file.write_bytes(b"fake")
|
||||
|
||||
segment = VoiceoverSegment(
|
||||
text="test",
|
||||
start_time=1.0,
|
||||
end_time=3.0,
|
||||
audio_path=audio_file,
|
||||
duration=2.0,
|
||||
)
|
||||
result = VoiceoverResult(success=True, segments=[segment], total_duration=3.0)
|
||||
|
||||
filter_str, files = engine.build_audio_mix_filter(result, video_duration=10)
|
||||
assert len(files) == 1
|
||||
assert "adelay" in filter_str
|
||||
assert "1000" in filter_str # 1秒 = 1000ms
|
||||
|
||||
|
||||
# ─── VoicePreset 数据类 ──────────────────────────────────
|
||||
|
||||
|
||||
class TestVoicePreset:
|
||||
def test_create_preset(self):
|
||||
preset = VoicePreset(
|
||||
voice_id="test_voice",
|
||||
name="测试音色",
|
||||
gender=VoiceGender.MALE,
|
||||
style=VoiceStyle.NARRATION,
|
||||
)
|
||||
assert preset.voice_id == "test_voice"
|
||||
assert preset.name == "测试音色"
|
||||
assert preset.gender == VoiceGender.MALE
|
||||
assert preset.style == VoiceStyle.NARRATION
|
||||
assert preset.sample_rate == 22050
|
||||
Reference in New Issue
Block a user