diff --git a/apps/api/app/api/routes/tts.py b/apps/api/app/api/routes/tts.py index 40c665685..2a01f06b2 100755 --- a/apps/api/app/api/routes/tts.py +++ b/apps/api/app/api/routes/tts.py @@ -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) diff --git a/apps/worker/services/tts_service_factory.py b/apps/worker/services/tts_service_factory.py new file mode 100755 index 000000000..d25aad513 --- /dev/null +++ b/apps/worker/services/tts_service_factory.py @@ -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()) diff --git a/apps/worker/video_processing/tts_engine.py b/apps/worker/video_processing/tts_engine.py new file mode 100755 index 000000000..be1e1f7d7 --- /dev/null +++ b/apps/worker/video_processing/tts_engine.py @@ -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 diff --git a/apps/worker/video_processing/unified_render_service.py b/apps/worker/video_processing/unified_render_service.py index adb344b89..379626b36 100755 --- a/apps/worker/video_processing/unified_render_service.py +++ b/apps/worker/video_processing/unified_render_service.py @@ -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: """判断是否可以走直通优化路径。 diff --git a/packages/adapters/tts/mock_tts_service.py b/packages/adapters/tts/mock_tts_service.py new file mode 100755 index 000000000..185c3344c --- /dev/null +++ b/packages/adapters/tts/mock_tts_service.py @@ -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("输出文件为空或不存在") diff --git a/packages/domain/tts_config.py b/packages/domain/tts_config.py new file mode 100755 index 000000000..76206152e --- /dev/null +++ b/packages/domain/tts_config.py @@ -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 diff --git a/packages/domain/voice_presets.py b/packages/domain/voice_presets.py new file mode 100755 index 000000000..857683625 --- /dev/null +++ b/packages/domain/voice_presets.py @@ -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] diff --git a/packages/ports/tts_service.py b/packages/ports/tts_service.py new file mode 100755 index 000000000..3211fe59e --- /dev/null +++ b/packages/ports/tts_service.py @@ -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 diff --git a/tests/unit/test_tts_voiceover.py b/tests/unit/test_tts_voiceover.py new file mode 100755 index 000000000..3c6d265b3 --- /dev/null +++ b/tests/unit/test_tts_voiceover.py @@ -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