xiaoxia-saas/packages/application/tts_job/audio_merger.py

"""FFmpeg 音频合并器 — P1 长文本分段合成。

将多个分段音频文件合并为一个完整音频文件。
"""

from __future__ import annotations

import logging
import os
import shutil
import tempfile
from subprocess import CalledProcessError, TimeoutExpired

from packages.shared.ffmpeg_utils import FFMPEG_BIN, run_ffmpeg

logger = logging.getLogger(__name__)


class AudioMergeError(Exception):
    """音频合并异常。"""

    pass


class AudioMerger:
    """使用 FFmpeg 合并多个音频文件。"""

    def merge(self, audio_paths: list[str], output_format: str = "mp3") -> bytes:
        """合并多个音频文件，返回合并后的音频数据。

        使用 FFmpeg concat demuxer 按顺序拼接音频。
        所有输入文件必须为相同格式和采样率。

        Args:
            audio_paths: 音频文件路径列表（按合成顺序）
            output_format: 输出格式（mp3/wav/pcm）

        Returns:
            合并后的音频文件字节数据

        Raises:
            AudioMergeError: 合并失败
        """
        if not audio_paths:
            raise AudioMergeError("没有可合并的音频文件")

        if len(audio_paths) == 1:
            with open(audio_paths[0], "rb") as f:
                return f.read()

        temp_dir = tempfile.mkdtemp(prefix="tts_merge_")
        try:
            # 生成 concat demuxer 列表文件
            list_path = os.path.join(temp_dir, "concat_list.txt")
            with open(list_path, "w") as f:
                for path in audio_paths:
                    # FFmpeg concat 文件需要 file: 前缀，路径中的 ' 和 \n 需转义
                    escaped = path.replace("'", "'\\''").replace("\n", "\\n")
                    f.write(f"file '{escaped}'\n")

            output_path = os.path.join(temp_dir, f"merged.{output_format}")

            cmd = [
                FFMPEG_BIN,
                "-y",
                "-f",
                "concat",
                "-safe",
                "0",
                "-i",
                list_path,
                "-c",
                "copy",
                output_path,
            ]

            try:
                run_ffmpeg(cmd, timeout=120)
            except CalledProcessError as e:
                logger.error(f"FFmpeg 合并失败: stderr={e.stderr}")
                raise AudioMergeError(f"FFmpeg 合并失败: {str(e)[:500]}")

            with open(output_path, "rb") as f:
                return f.read()

        except TimeoutExpired:
            raise AudioMergeError("FFmpeg 合并超时（120 秒）")
        except AudioMergeError:
            raise
        except Exception as e:
            raise AudioMergeError(f"音频合并失败: {e}")
        finally:
            shutil.rmtree(temp_dir, ignore_errors=True)