xiaoxia-saas/packages/adapters/tts/mock_tts_service.py

"""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% 深度

        # 构建滤镜链

        # 生成基频 + 泛音（让声音更丰富）
        # 用多个 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("输出文件为空或不存在")