xiaoxia-saas/apps/worker/video_processing/tts_engine.py

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