Compare commits

...

2 Commits

Author SHA1 Message Date
CI Bot c6bc8170d1 chore: rebase到develop + 格式化代码
CI/CD Pipeline / Validate Code Quality And Tests (pull_request) Successful in 40s
CI/CD Pipeline / Unit Tests (pull_request) Successful in 1m17s
CI/CD Pipeline / Integration Tests (pull_request) Failing after 1m26s
CI/CD Pipeline / Frontend Lint (pull_request) Successful in 4m48s
CI/CD Pipeline / Build & Push Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Build Production Runtime Images (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
2026-07-14 10:08:03 +08:00
CI Bot a1285be576 feat: TTS文字转语音配音引擎
- TTS服务抽象层 (TtsService Port),支持多provider切换
- Mock TTS实现:基于FFmpeg sine波合成,8种预设音色
- 配音核心引擎 (TtsEngine):整段配音+字幕联动配音
- 语速/语调调整:speed 0.5x~2.0x,pitch -12~12半音
- 8种预设音色:温暖女声/沉稳男声/活泼女声/可爱童声/客服女声/新闻男声/轻柔女声/磁性男声
- 字幕联动:按字幕片段分段合成,自动对齐时间轴
- 降级策略:TTS失败自动跳过,不阻断渲染
- 集成到UnifiedRenderService:配音作为独立音轨混入
- 音色列表API:GET /api/v1/tts/presets 支持按性别/风格/关键词筛选
- 40个新增单测全绿 + 134个现有测试全绿,无回归
2026-07-14 10:08:03 +08:00
9 changed files with 1500 additions and 0 deletions
+29
View File
@@ -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)
+66
View File
@@ -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())
+275
View File
@@ -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:
"""判断是否可以走直通优化路径。
+231
View File
@@ -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("输出文件为空或不存在")
+102
View File
@@ -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
+222
View File
@@ -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]
+78
View File
@@ -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
+413
View File
@@ -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