"""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