From 815998acaa8b46e84b6296ac81cbd835d7c9eda9 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Mon, 27 Jul 2026 08:30:37 +0800 Subject: [PATCH] =?UTF-8?q?test(wave119):=20=E6=8A=BD=E7=A6=BBnoise=5Fredu?= =?UTF-8?q?ction=5Fconfig=E9=A2=86=E5=9F=9F=E6=A8=A1=E5=9E=8B=20+=2039?= =?UTF-8?q?=E5=8D=95=E6=B5=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 从 noise_reduction_engine.py 抽离领域模型到 packages/domain/noise_reduction_config.py - noise_reduction_engine.py: 229→72行 (-69%) - 新增 39 个单测，覆盖 NoiseReductionConfig 全部纯逻辑 - 原有 27 个测试无回归 - 保持向后兼容：引擎模块保留全部公开接口作为薄包装 --- .../noise_reduction_engine.py | 197 ++----------- packages/domain/noise_reduction_config.py | 231 ++++++++++++++++ tests/unit/test_noise_reduction_config.py | 258 ++++++++++++++++++ 3 files changed, 509 insertions(+), 177 deletions(-) create mode 100755 packages/domain/noise_reduction_config.py create mode 100755 tests/unit/test_noise_reduction_config.py diff --git a/apps/worker/video_processing/noise_reduction_engine.py b/apps/worker/video_processing/noise_reduction_engine.py index d1d5e053f..3484d8af7 100755 --- a/apps/worker/video_processing/noise_reduction_engine.py +++ b/apps/worker/video_processing/noise_reduction_engine.py @@ -2,126 +2,26 @@ 支持对音频进行背景噪音消除、人声增强，适用于语音录制、采访等场景。 -使用方式： - config = NoiseReductionConfig(level="medium") - engine = NoiseReductionEngine(config) - filter_str = engine.build_filter(input_label, output_label) - # 结果: [0:a]afftdn=nf=-25[out] - -降级策略： - - 参数越界自动钳制 - - FFmpeg 不支持 afftdn 时，调用方可捕获异常并跳过 +领域模型已抽离至 packages/domain/noise_reduction_config.py，本模块保留薄包装以维持向后兼容。 """ from __future__ import annotations import logging -from dataclasses import dataclass -from enum import Enum -from typing import Optional + +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 NoiseReductionLevel(str, Enum): - """降噪等级预设。""" - - LOW = "low" # 轻度降噪，保留细节，适合轻微背景噪音 - MEDIUM = "medium" # 中度降噪，平衡效果和音质 - HIGH = "high" # 高度降噪，适合嘈杂环境，可能轻微影响音质 - CUSTOM = "custom" # 自定义参数 - - -# 各等级对应的降噪参数（afftdn 的 noise floor，单位 dB） -# 值越大（越接近 0），降噪越强；值越小（越负），降噪越弱 -_LEVEL_PARAMS = { - NoiseReductionLevel.LOW: { - "nf": -35, # 噪音阈值（dB），越负越保守 - "tn": -10, # 噪音频谱平滑度 - "tr": 50, # 时间分辨率（ms） - }, - NoiseReductionLevel.MEDIUM: { - "nf": -25, - "tn": -10, - "tr": 50, - }, - NoiseReductionLevel.HIGH: { - "nf": -15, - "tn": -5, - "tr": 30, - }, -} - - -# ── 配置模型 ────────────────────────────────────────────────────────────────── - - -@dataclass -class NoiseReductionConfig: - """音频降噪配置。 - - Attributes: - enabled: 是否启用降噪 - level: 降噪等级 low/medium/high/custom - noise_floor: 自定义噪音阈值（dB），仅 level=custom 时有效，范围 -60 ~ -5 - voice_enhance: 是否启用人声增强 - output_format: 输出格式描述（内部使用） - """ - - enabled: bool = False - level: NoiseReductionLevel = NoiseReductionLevel.MEDIUM - noise_floor: float = -25.0 # dB - voice_enhance: bool = False - - @classmethod - def from_dict(cls, data: dict | None) -> "NoiseReductionConfig": - """从字典解析配置，参数越界自动钳制。""" - if not data or not data.get("enabled", False): - return cls(enabled=False) - - level_str = str(data.get("level", "medium")).lower() - try: - level = NoiseReductionLevel(level_str) - except ValueError: - level = NoiseReductionLevel.MEDIUM - - try: - noise_floor = float(data.get("noise_floor", -25.0)) - except (TypeError, ValueError): - noise_floor = -25.0 - - voice_enhance = bool(data.get("voice_enhance", False)) - - # 钳制到合法范围 - noise_floor = max(-60.0, min(-5.0, noise_floor)) - - return cls( - enabled=True, - level=level, - noise_floor=noise_floor, - voice_enhance=voice_enhance, - ) - - def has_effect(self) -> bool: - """判断是否有实际降噪效果。""" - return self.enabled - - def get_effective_noise_floor(self) -> float: - """获取实际生效的噪音阈值（dB）。""" - if self.level == NoiseReductionLevel.CUSTOM: - return self.noise_floor - params = _LEVEL_PARAMS.get(self.level, _LEVEL_PARAMS[NoiseReductionLevel.MEDIUM]) - return float(params["nf"]) - - -# ── 引擎实现 ────────────────────────────────────────────────────────────────── - - class NoiseReductionEngine: - """音频降噪引擎。 + """音频降噪引擎 — 薄包装，实际逻辑在 domain.noise_reduction_config. 基于 FFmpeg afftdn（Audio FFt Denoiser）滤镜实现： - 使用短时傅里叶变换分析音频频谱 @@ -133,97 +33,40 @@ class NoiseReductionEngine: 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 滤镜字符串，如 "[a0]afftdn=nf=-25:tn=-10:tr=50[nr0]" - - Raises: - ValueError: 配置无效时抛出（调用方应捕获并降级） + FFmpeg 滤镜字符串 """ - if not self.config.has_effect(): - return f"{input_label}anull{output_label}" - - # 获取参数 - if self.config.level == NoiseReductionLevel.CUSTOM: - nf = self.config.noise_floor - tn = -10 # 默认频谱平滑度 - tr = 50 # 默认时间分辨率 - else: - params = _LEVEL_PARAMS.get( - self.config.level, - _LEVEL_PARAMS[NoiseReductionLevel.MEDIUM], - ) - nf = float(params["nf"]) - tn = float(params["tn"]) - tr = float(params["tr"]) - - # 构建 afftdn 滤镜 - # nf: noise floor (dB) - # tn: temporal noise floor smoothing (dB) - # tr: time resolution (ms) - filter_parts = [f"afftdn=nf={nf}:tn={tn}:tr={tr}"] - - # 人声增强：通过 highpass + 轻微压缩实现 - if self.config.voice_enhance: - # 1. 高通滤波，去除低频噪音 - filter_parts.append("highpass=f=80") - # 2. 轻微压缩，提升人声清晰度 - filter_parts.append("acompressor=threshold=-20:ratio=2:attack=5:release=50") - # 3. 响度归一化 - filter_parts.append("loudnorm=I=-16:TP=-1.5:LRA=11") - - filter_str = f"{input_label}{','.join(filter_parts)}{output_label}" - return filter_str + 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，效果更好但需要模型文件）。 - - 注意：需要额外下载 RNNNoise 模型文件，默认使用 afftdn（无需额外依赖）。 + """使用 RNN 降噪滤镜（arnndn，效果更好但需要模型文件）. Args: input_label: 输入标签 output_label: 输出标签 - model_file: RNNNoise 模型文件路径（.rnnn 格式） + model_file: RNNNoise 模型文件路径 Returns: FFmpeg 滤镜字符串 """ - if not self.config.has_effect(): - return f"{input_label}anull{output_label}" - - return f"{input_label}arnndn=m={model_file}{output_label}" + return _build_arnndn_filter_base(self.config, input_label, output_label, model_file) -def apply_noise_reduction_if_needed( - config_data: dict | None, - input_label: str, - output_label: str, -) -> Optional[str]: - """便捷函数：根据配置判断是否需要应用音频降噪。 +def apply_noise_reduction_if_needed(config_data, input_label: str, output_label: str): + """便捷函数：根据配置判断是否需要应用音频降噪. Args: - config_data: 降噪配置字典（从 plan.config.audio_noise_reduction 或 clip.config.noise_reduction 读取） + config_data: 降噪配置字典 input_label: 输入标签 output_label: 输出标签 Returns: 滤镜字符串，不需要降噪时返回 None """ - if not config_data: - return None - - try: - config = NoiseReductionConfig.from_dict(config_data) - if not config.has_effect(): - return None - - engine = NoiseReductionEngine(config) - return engine.build_filter(input_label, output_label) - except Exception as e: - logger.warning("[noise-reduction] 应用降噪失败，跳过: %s", e) - return None + return _apply_noise_reduction_if_needed_base(config_data, input_label, output_label) diff --git a/packages/domain/noise_reduction_config.py b/packages/domain/noise_reduction_config.py new file mode 100755 index 000000000..6ab44a452 --- /dev/null +++ b/packages/domain/noise_reduction_config.py @@ -0,0 +1,231 @@ +"""音频降噪配置领域模型 — 纯逻辑，无FFmpeg依赖. + +抽离自 noise_reduction_engine.py，包含： +- NoiseReductionLevel 枚举（low/medium/high/custom） +- NoiseReductionConfig 数据类（解析/钳制/效果判断） +- 等级预设参数 +- afftdn / arnndn 滤镜构建 +- 便捷函数（apply_noise_reduction_if_needed） +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass +from enum import Enum +from typing import Any + +logger = logging.getLogger(__name__) + + +# ── 降噪等级 ────────────────────────────────────────────────────────────────── + + +class NoiseReductionLevel(str, Enum): + """降噪等级预设.""" + + LOW = "low" # 轻度降噪，保留细节，适合轻微背景噪音 + MEDIUM = "medium" # 中度降噪，平衡效果和音质 + HIGH = "high" # 高度降噪，适合嘈杂环境，可能轻微影响音质 + CUSTOM = "custom" # 自定义参数 + + +# 各等级对应的降噪参数（afftdn 的 noise floor，单位 dB） +# 值越大（越接近 0），降噪越强；值越小（越负），降噪越弱 +_LEVEL_PARAMS: dict[NoiseReductionLevel, dict[str, float]] = { + NoiseReductionLevel.LOW: { + "nf": -35, # 噪音阈值（dB），越负越保守 + "tn": -10, # 噪音频谱平滑度 + "tr": 50, # 时间分辨率（ms） + }, + NoiseReductionLevel.MEDIUM: { + "nf": -25, + "tn": -10, + "tr": 50, + }, + NoiseReductionLevel.HIGH: { + "nf": -15, + "tn": -5, + "tr": 30, + }, +} + +# 参数范围 +MIN_NOISE_FLOOR = -60.0 +MAX_NOISE_FLOOR = -5.0 + +# 默认值 +DEFAULT_LEVEL = NoiseReductionLevel.MEDIUM +DEFAULT_NOISE_FLOOR = -25.0 + + +# ── 配置模型 ────────────────────────────────────────────────────────────────── + + +@dataclass +class NoiseReductionConfig: + """音频降噪配置. + + Attributes: + enabled: 是否启用降噪 + level: 降噪等级 low/medium/high/custom + noise_floor: 自定义噪音阈值（dB），仅 level=custom 时有效，范围 -60 ~ -5 + voice_enhance: 是否启用人声增强 + """ + + enabled: bool = False + level: NoiseReductionLevel = DEFAULT_LEVEL + noise_floor: float = DEFAULT_NOISE_FLOOR # dB + voice_enhance: bool = False + + @classmethod + def from_dict(cls, data: dict[str, Any] | None) -> NoiseReductionConfig: + """从字典解析配置，参数越界自动钳制.""" + if not data or not data.get("enabled", False): + return cls(enabled=False) + + level_str = str(data.get("level", "medium")).lower() + try: + level = NoiseReductionLevel(level_str) + except ValueError: + level = DEFAULT_LEVEL + + try: + noise_floor = float(data.get("noise_floor", DEFAULT_NOISE_FLOOR)) + except (TypeError, ValueError): + noise_floor = DEFAULT_NOISE_FLOOR + + voice_enhance = bool(data.get("voice_enhance", False)) + + # 钳制到合法范围 + noise_floor = max(MIN_NOISE_FLOOR, min(MAX_NOISE_FLOOR, noise_floor)) + + return cls( + enabled=True, + level=level, + noise_floor=noise_floor, + voice_enhance=voice_enhance, + ) + + def has_effect(self) -> bool: + """判断是否有实际降噪效果.""" + return self.enabled + + def get_effective_noise_floor(self) -> float: + """获取实际生效的噪音阈值（dB）.""" + if self.level == NoiseReductionLevel.CUSTOM: + return self.noise_floor + params = _LEVEL_PARAMS.get(self.level, _LEVEL_PARAMS[DEFAULT_LEVEL]) + return float(params["nf"]) + + def get_level_params(self) -> dict[str, float]: + """获取当前等级的完整参数字典.""" + if self.level == NoiseReductionLevel.CUSTOM: + return { + "nf": self.noise_floor, + "tn": -10.0, + "tr": 50.0, + } + params = _LEVEL_PARAMS.get(self.level, _LEVEL_PARAMS[DEFAULT_LEVEL]) + return {k: float(v) for k, v in params.items()} + + def validate(self) -> tuple[bool, str]: + """校验配置是否有效.""" + if not self.enabled: + return True, "" + + if not (MIN_NOISE_FLOOR <= self.noise_floor <= MAX_NOISE_FLOOR): + return False, f"noise_floor 必须在 {MIN_NOISE_FLOOR}~{MAX_NOISE_FLOOR} dB 之间" + + return True, "" + + +# ── 滤镜构建 ──────────────────────────────────────────────────────────────── + + +def build_afftdn_filter( + config: NoiseReductionConfig, + input_label: str, + output_label: str, +) -> str: + """构建 afftdn 音频降噪滤镜字符串. + + Args: + config: 降噪配置 + input_label: 输入标签，如 "[0:a]" 或 "[a0]" + output_label: 输出标签，如 "[nr0]" + + Returns: + FFmpeg 滤镜字符串 + """ + if not config.has_effect(): + return f"{input_label}anull{output_label}" + + params = config.get_level_params() + nf = params["nf"] + tn = params["tn"] + tr = params["tr"] + + # 构建 afftdn 滤镜 + filter_parts = [f"afftdn=nf={nf}:tn={tn}:tr={tr}"] + + # 人声增强：通过 highpass + 压缩 + 响度归一化实现 + if config.voice_enhance: + filter_parts.append("highpass=f=80") + filter_parts.append("acompressor=threshold=-20:ratio=2:attack=5:release=50") + filter_parts.append("loudnorm=I=-16:TP=-1.5:LRA=11") + + return f"{input_label}{','.join(filter_parts)}{output_label}" + + +def build_arnndn_filter( + config: NoiseReductionConfig, + input_label: str, + output_label: str, + model_file: str, +) -> str: + """使用 RNN 降噪滤镜（arnndn，效果更好但需要模型文件）. + + 注意：需要额外下载 RNNNoise 模型文件，默认使用 afftdn（无需额外依赖）。 + """ + if not config.has_effect(): + return f"{input_label}anull{output_label}" + + return f"{input_label}arnndn=m={model_file}{output_label}" + + +# ── 便捷函数 ──────────────────────────────────────────────────────────────── + + +def apply_noise_reduction_if_needed( + config_data: dict[str, Any] | None, + input_label: str, + output_label: str, +) -> str | None: + """便捷函数：根据配置判断是否需要应用音频降噪. + + Args: + config_data: 降噪配置字典 + input_label: 输入标签 + output_label: 输出标签 + + Returns: + 滤镜字符串，不需要降噪时返回 None + """ + if not config_data: + return None + + try: + config = NoiseReductionConfig.from_dict(config_data) + if not config.has_effect(): + return None + + return build_afftdn_filter(config, input_label, output_label) + except Exception as e: + logger.warning("[noise-reduction] 应用降噪失败，跳过: %s", e) + return None + + +def get_level_names() -> list[str]: + """获取所有降噪等级名称列表.""" + return [level.value for level in NoiseReductionLevel] diff --git a/tests/unit/test_noise_reduction_config.py b/tests/unit/test_noise_reduction_config.py new file mode 100755 index 000000000..0a41a231d --- /dev/null +++ b/tests/unit/test_noise_reduction_config.py @@ -0,0 +1,258 @@ +"""noise_reduction_config 领域模型单测.""" + +from __future__ import annotations + +import pytest + +from packages.domain.noise_reduction_config import ( + DEFAULT_LEVEL, + DEFAULT_NOISE_FLOOR, + MAX_NOISE_FLOOR, + MIN_NOISE_FLOOR, + NoiseReductionConfig, + NoiseReductionLevel, + apply_noise_reduction_if_needed, + build_afftdn_filter, + build_arnndn_filter, + get_level_names, +) + +# ── NoiseReductionLevel 枚举测试 ─────────────────────────────────────────── + + +class TestNoiseReductionLevel: + def test_four_levels(self): + assert len(NoiseReductionLevel) == 4 + + def test_level_values(self): + assert NoiseReductionLevel.LOW.value == "low" + assert NoiseReductionLevel.MEDIUM.value == "medium" + assert NoiseReductionLevel.HIGH.value == "high" + assert NoiseReductionLevel.CUSTOM.value == "custom" + + def test_from_string(self): + assert NoiseReductionLevel("low") == NoiseReductionLevel.LOW + assert NoiseReductionLevel("medium") == NoiseReductionLevel.MEDIUM + assert NoiseReductionLevel("high") == NoiseReductionLevel.HIGH + assert NoiseReductionLevel("custom") == NoiseReductionLevel.CUSTOM + + +# ── NoiseReductionConfig.from_dict 测试 ─────────────────────────────────── + + +class TestNoiseReductionConfigFromDict: + def test_none_returns_disabled(self): + cfg = NoiseReductionConfig.from_dict(None) + assert cfg.enabled is False + + def test_empty_dict_returns_disabled(self): + cfg = NoiseReductionConfig.from_dict({}) + assert cfg.enabled is False + + def test_disabled_returns_disabled(self): + cfg = NoiseReductionConfig.from_dict({"enabled": False}) + assert cfg.enabled is False + + def test_enabled_default_params(self): + cfg = NoiseReductionConfig.from_dict({"enabled": True}) + assert cfg.enabled is True + assert cfg.level == NoiseReductionLevel.MEDIUM + assert cfg.noise_floor == DEFAULT_NOISE_FLOOR + assert cfg.voice_enhance is False + + def test_custom_level(self): + cfg = NoiseReductionConfig.from_dict({"enabled": True, "level": "custom", "noise_floor": -30.0}) + assert cfg.level == NoiseReductionLevel.CUSTOM + assert cfg.noise_floor == -30.0 + + def test_invalid_level_defaults_medium(self): + cfg = NoiseReductionConfig.from_dict({"enabled": True, "level": "invalid"}) + assert cfg.level == NoiseReductionLevel.MEDIUM + + def test_noise_floor_clamped_low(self): + cfg = NoiseReductionConfig.from_dict({"enabled": True, "level": "custom", "noise_floor": -100.0}) + assert cfg.noise_floor == MIN_NOISE_FLOOR + + def test_noise_floor_clamped_high(self): + cfg = NoiseReductionConfig.from_dict({"enabled": True, "level": "custom", "noise_floor": 0.0}) + assert cfg.noise_floor == MAX_NOISE_FLOOR + + def test_invalid_noise_floor_type_uses_default(self): + cfg = NoiseReductionConfig.from_dict({"enabled": True, "level": "custom", "noise_floor": "not_a_number"}) + assert cfg.noise_floor == DEFAULT_NOISE_FLOOR + + def test_voice_enhance_true(self): + cfg = NoiseReductionConfig.from_dict({"enabled": True, "voice_enhance": True}) + assert cfg.voice_enhance is True + + def test_case_insensitive_level(self): + cfg = NoiseReductionConfig.from_dict({"enabled": True, "level": "HIGH"}) + assert cfg.level == NoiseReductionLevel.HIGH + + +# ── has_effect / get_effective_noise_floor 测试 ─────────────────────────── + + +class TestConfigProperties: + def test_disabled_no_effect(self): + cfg = NoiseReductionConfig(enabled=False) + assert cfg.has_effect() is False + + def test_enabled_has_effect(self): + cfg = NoiseReductionConfig(enabled=True) + assert cfg.has_effect() is True + + def test_effective_noise_floor_low(self): + cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.LOW) + assert cfg.get_effective_noise_floor() == -35.0 + + def test_effective_noise_floor_medium(self): + cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.MEDIUM) + assert cfg.get_effective_noise_floor() == -25.0 + + def test_effective_noise_floor_high(self): + cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.HIGH) + assert cfg.get_effective_noise_floor() == -15.0 + + def test_effective_noise_floor_custom(self): + cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.CUSTOM, noise_floor=-40.0) + assert cfg.get_effective_noise_floor() == -40.0 + + def test_get_level_params_medium(self): + cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.MEDIUM) + params = cfg.get_level_params() + assert params["nf"] == -25.0 + assert params["tn"] == -10.0 + assert params["tr"] == 50.0 + + def test_get_level_params_custom(self): + cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.CUSTOM, noise_floor=-30.0) + params = cfg.get_level_params() + assert params["nf"] == -30.0 + assert "tn" in params + assert "tr" in params + + +# ── validate 测试 ───────────────────────────────────────────────────────── + + +class TestValidate: + def test_disabled_valid(self): + cfg = NoiseReductionConfig(enabled=False) + ok, msg = cfg.validate() + assert ok is True + assert msg == "" + + def test_enabled_valid(self): + cfg = NoiseReductionConfig(enabled=True, noise_floor=-25.0) + ok, msg = cfg.validate() + assert ok is True + + def test_noise_floor_out_of_range(self): + cfg = NoiseReductionConfig(enabled=True, noise_floor=-100.0) + ok, msg = cfg.validate() + assert ok is False + assert "noise_floor" in msg + + +# ── build_afftdn_filter 测试 ─────────────────────────────────────────────── + + +class TestBuildAfftdnFilter: + def test_disabled_returns_anull(self): + cfg = NoiseReductionConfig(enabled=False) + result = build_afftdn_filter(cfg, "[in]", "[out]") + assert "anull" in result + assert "[in]" in result + assert "[out]" in result + + def test_medium_level(self): + cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.MEDIUM) + result = build_afftdn_filter(cfg, "[a]", "[nr]") + assert "afftdn=" in result + assert "nf=-25.0" in result or "nf=-25" in result + assert "[a]" in result + assert "[nr]" in result + + def test_high_level(self): + cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.HIGH) + result = build_afftdn_filter(cfg, "[in]", "[out]") + assert "afftdn=" in result + assert "nf=-15.0" in result or "nf=-15" in result + + def test_low_level(self): + cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.LOW) + result = build_afftdn_filter(cfg, "[in]", "[out]") + assert "afftdn=" in result + assert "nf=-35.0" in result or "nf=-35" in result + + def test_custom_level(self): + cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.CUSTOM, noise_floor=-40.0) + result = build_afftdn_filter(cfg, "[in]", "[out]") + assert "afftdn=" in result + assert "nf=-40.0" in result or "nf=-40" in result + + def test_voice_enhance_adds_filters(self): + cfg = NoiseReductionConfig(enabled=True, voice_enhance=True) + result = build_afftdn_filter(cfg, "[in]", "[out]") + assert "highpass" in result + assert "acompressor" in result + assert "loudnorm" in result + + def test_no_voice_enhance_no_extra_filters(self): + cfg = NoiseReductionConfig(enabled=True, voice_enhance=False) + result = build_afftdn_filter(cfg, "[in]", "[out]") + assert "highpass" not in result + assert "acompressor" not in result + + +# ── build_arnndn_filter 测试 ─────────────────────────────────────────────── + + +class TestBuildArnndnFilter: + def test_disabled_returns_anull(self): + cfg = NoiseReductionConfig(enabled=False) + result = build_arnndn_filter(cfg, "[in]", "[out]", "model.rnnn") + assert "anull" in result + + def test_enabled_returns_arnndn(self): + cfg = NoiseReductionConfig(enabled=True) + result = build_arnndn_filter(cfg, "[a]", "[nr]", "/path/to/model.rnnn") + assert "arnndn=" in result + assert "m=/path/to/model.rnnn" in result + assert "[a]" in result + assert "[nr]" in result + + +# ── apply_noise_reduction_if_needed 测试 ───────────────────────────────── + + +class TestApplyNoiseReductionIfNeeded: + def test_none_config_returns_none(self): + assert apply_noise_reduction_if_needed(None, "[in]", "[out]") is None + + def test_disabled_returns_none(self): + assert apply_noise_reduction_if_needed({"enabled": False}, "[in]", "[out]") is None + + def test_enabled_returns_filter(self): + result = apply_noise_reduction_if_needed({"enabled": True, "level": "medium"}, "[in]", "[out]") + assert result is not None + assert "afftdn" in result + + def test_invalid_config_handles_exception(self): + # 异常情况应该返回 None 而不是抛出 + result = apply_noise_reduction_if_needed("invalid", "[in]", "[out]") + assert result is None + + +# ── 工具函数测试 ─────────────────────────────────────────────────────────── + + +class TestUtils: + def test_get_level_names_returns_four(self): + names = get_level_names() + assert len(names) == 4 + assert "low" in names + assert "medium" in names + assert "high" in names + assert "custom" in names -- 2.54.0