"""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("输出文件为空或不存在")