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

"""音频降噪引擎 — 基于 FFmpeg afftdn 滤镜.

支持对音频进行背景噪音消除、人声增强，适用于语音录制、采访等场景。

领域模型已抽离至 packages/domain/noise_reduction_config.py，本模块保留薄包装以维持向后兼容。
"""

from __future__ import annotations

import logging

from packages.domain.noise_reduction_config import (  # noqa: F401 — 向后兼容
    NoiseReductionLevel,
    NoiseReductionConfig,
    build_afftdn_filter as _build_afftdn_filter_base,
    build_arnndn_filter as _build_arnndn_filter_base,
    apply_noise_reduction_if_needed as _apply_noise_reduction_if_needed_base,
)

logger = logging.getLogger(__name__)


class NoiseReductionEngine:
    """音频降噪引擎 — 薄包装，实际逻辑在 domain.noise_reduction_config.

    基于 FFmpeg afftdn（Audio FFt Denoiser）滤镜实现：
    - 使用短时傅里叶变换分析音频频谱
    - 识别并消除稳态背景噪音
    - 保留人声等非稳态信号
    """

    def __init__(self, config: NoiseReductionConfig):
        self.config = config

    def build_filter(self, input_label: str, output_label: str) -> str:
        """构建音频降噪滤镜字符串.

        Args:
            input_label: 输入标签，如 "[0:a]" 或 "[a0]"
            output_label: 输出标签，如 "[nr0]"

        Returns:
            FFmpeg 滤镜字符串
        """
        return _build_afftdn_filter_base(self.config, input_label, output_label)

    def build_filter_arnndn(self, input_label: str, output_label: str, model_file: str) -> str:
        """使用 RNN 降噪滤镜（arnndn，效果更好但需要模型文件）.

        Args:
            input_label: 输入标签
            output_label: 输出标签
            model_file: RNNNoise 模型文件路径

        Returns:
            FFmpeg 滤镜字符串
        """
        return _build_arnndn_filter_base(self.config, input_label, output_label, model_file)


def apply_noise_reduction_if_needed(config_data, input_label: str, output_label: str):
    """便捷函数：根据配置判断是否需要应用音频降噪.

    Args:
        config_data: 降噪配置字典
        input_label: 输入标签
        output_label: 输出标签

    Returns:
        滤镜字符串，不需要降噪时返回 None
    """
    return _apply_noise_reduction_if_needed_base(config_data, input_label, output_label)