fix: 合并重复的mix_audio调用（BGM+降噪合并为一次）

新增两个渲染能力：

【绿幕抠像 ChromaKeyEngine】
- 基于 FFmpeg colorkey 滤镜，支持绿幕/蓝幕/红幕
- 可配置：抠除颜色、相似度、边缘平滑、溢色抑制
- 5种预设：green_screen/blue_screen/red_screen/precise_green/soft_green
- 从 clip.config.chroma_key 读取配置，零侵入数据模型
- 降级策略：参数越界自动钳制，解析失败不阻断渲染

【音频降噪 NoiseReductionEngine】
- 基于 FFmpeg afftdn 滤镜（短时傅里叶变换降噪）
- 3档预设：低/中/高 + 自定义阈值（-60~-5 dB）
- 人声增强模式：highpass + compressor + loudnorm
- 从 plan.config.audio_noise_reduction 读取配置
- 降级策略：参数越界自动钳制，降噪失败用原始音频

【系统集成】
- 绿幕抠像：UnifiedRenderService Step 1 接入（scale之后，fps之前）
  - 支持 filter_complex 路径 + 直通模式 双路径
- 音频降噪：后处理模式，在 mix_audio 最终输出前应用
  - filter_complex 路径通过 RenderContext 传递配置
  - 直通模式通过 -af 参数应用

69个新增单测 + 1435个现有测试 = 1504个全绿，零回归

apps/worker/video_processing/chroma_key_engine.py

@@ -0,0 +1,248 @@
+"""绿幕抠像引擎 — 基于 FFmpeg colorkey / chromakey 滤镜.
+
+支持将指定颜色（默认绿色）变为透明，可用于虚拟背景、画中画背景替换等场景。
+
+使用方式：
+    config = ChromaKeyConfig(key_color="#00FF00", similarity=0.3, blend=0.1)
+    engine = ChromaKeyEngine(config)
+    filter_str = engine.build_filter(input_label, output_label)
+    # 结果: [in]colorkey=color=0x00FF00:similarity=0.3:blend=0.1[out]
+
+降级策略：
+    - 参数越界自动钳制
+    - 素材格式不支持时跳过（调用方捕获异常）
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+from dataclasses import dataclass
+from typing import Optional
+
+logger = logging.getLogger(__name__)
+
+
+# ── 配置模型 ──────────────────────────────────────────────────────────────────
+
+
+@dataclass
+class ChromaKeyConfig:
+    """绿幕抠像配置。
+
+    Attributes:
+        enabled: 是否启用抠像
+        key_color: 要抠除的颜色，支持 hex 格式（如 "#00FF00"）或颜色名
+        similarity: 颜色相似度阈值 0.01~1.0，值越大抠除范围越大
+        blend: 边缘平滑/混合度 0.0~1.0，值越大边缘越柔和
+        spill_suppress: 溢色抑制 0.0~1.0，减少边缘的绿幕反光
+    """
+
+    enabled: bool = False
+    key_color: str = "#00FF00"
+    similarity: float = 0.3
+    blend: float = 0.1
+    spill_suppress: float = 0.0
+
+    @classmethod
+    def from_dict(cls, data: dict | None) -> "ChromaKeyConfig":
+        """从字典解析配置，参数越界自动钳制。"""
+        if not data or not data.get("enabled", False):
+            return cls(enabled=False)
+
+        key_color = str(data.get("key_color", "#00FF00")).strip()
+
+        def _safe_float(val, default):
+            try:
+                return float(val)
+            except (TypeError, ValueError):
+                return default
+
+        similarity = _safe_float(data.get("similarity", 0.3), 0.3)
+        blend = _safe_float(data.get("blend", 0.1), 0.1)
+        spill_suppress = _safe_float(data.get("spill_suppress", 0.0), 0.0)
+
+        # 钳制到合法范围
+        similarity = max(0.01, min(1.0, similarity))
+        blend = max(0.0, min(1.0, blend))
+        spill_suppress = max(0.0, min(1.0, spill_suppress))
+
+        return cls(
+            enabled=True,
+            key_color=key_color,
+            similarity=similarity,
+            blend=blend,
+            spill_suppress=spill_suppress,
+        )
+
+    def has_effect(self) -> bool:
+        """判断是否有实际抠像效果。"""
+        return self.enabled and self.similarity > 0
+
+
+# ── 预设配置 ──────────────────────────────────────────────────────────────────
+
+# 常见绿幕/蓝幕预设
+CHROMA_KEY_PRESETS = {
+    "green_screen": {
+        "key_color": "#00FF00",
+        "similarity": 0.3,
+        "blend": 0.1,
+        "spill_suppress": 0.5,
+    },
+    "blue_screen": {
+        "key_color": "#0000FF",
+        "similarity": 0.3,
+        "blend": 0.1,
+        "spill_suppress": 0.5,
+    },
+    "red_screen": {
+        "key_color": "#FF0000",
+        "similarity": 0.3,
+        "blend": 0.1,
+        "spill_suppress": 0.0,
+    },
+    "precise_green": {
+        "key_color": "#00FF00",
+        "similarity": 0.2,
+        "blend": 0.05,
+        "spill_suppress": 0.3,
+    },
+    "soft_green": {
+        "key_color": "#00FF00",
+        "similarity": 0.45,
+        "blend": 0.2,
+        "spill_suppress": 0.5,
+    },
+}
+
+
+# ── 引擎实现 ──────────────────────────────────────────────────────────────────
+
+
+class ChromaKeyEngine:
+    """绿幕抠像引擎。
+
+    基于 FFmpeg colorkey 滤镜实现，将指定颜色变为透明。
+    适用于绿幕/蓝幕视频的背景去除，配合画中画或 overlay 实现虚拟背景。
+    """
+
+    def __init__(self, config: ChromaKeyConfig):
+        self.config = config
+
+    @staticmethod
+    def _normalize_color(color_str: str) -> str:
+        """将颜色字符串转为 FFmpeg colorkey 接受的格式。
+
+        支持:
+          - "#RRGGBB" / "#RRGGBBAA" → 0xRRGGBB
+          - "0xRRGGBB" → 直接使用
+          - 颜色名（green/blue/red/black/white 等）→ 直接透传
+        """
+        color = color_str.strip()
+
+        # hex 格式
+        hex_match = re.match(r"^#?([0-9a-fA-F]{6})([0-9a-fA-F]{2})?$", color)
+        if hex_match:
+            return f"0x{hex_match.group(1).upper()}"
+
+        # 已经是 0x 格式
+        if color.lower().startswith("0x"):
+            return color.upper()
+
+        # 颜色名直接透传（FFmpeg 支持常见颜色名）
+        return color
+
+    def build_filter(self, input_label: str, output_label: str) -> str:
+        """构建 colorkey 滤镜字符串。
+
+        Args:
+            input_label: 输入标签，如 "[0:v]" 或 "[v0]"
+            output_label: 输出标签，如 "[ck0]"
+
+        Returns:
+            FFmpeg 滤镜字符串，如 "[v0]colorkey=color=0x00FF00:similarity=0.3:blend=0.1[ck0]"
+
+        Raises:
+            ValueError: 配置无效时抛出（调用方应捕获并降级）
+        """
+        if not self.config.has_effect():
+            # 无效果，直接直通
+            return f"{input_label}copy{output_label}"
+
+        color = self._normalize_color(self.config.key_color)
+        similarity = self.config.similarity
+        blend = self.config.blend
+
+        # 基础 colorkey 滤镜
+        parts = [f"colorkey=color={color}:similarity={similarity}:blend={blend}"]
+
+        # 溢色抑制（通过 colorchannelmixer 降低绿色通道增益）
+        if self.config.spill_suppress > 0:
+            # 降低绿通道增益，减少绿幕反光溢出
+            spill = self.config.spill_suppress
+            # 绿通道增益 = 1 - spill_factor
+            g_gain = max(0.3, 1.0 - spill * 0.7)
+            # 同时稍微提升红和蓝来补偿色偏
+            r_gain = 1.0 + spill * 0.15
+            b_gain = 1.0 + spill * 0.15
+            parts.append(f"colorchannelmixer=" f"rr={r_gain}:" f"gg={g_gain}:" f"bb={b_gain}:" f"aa=1")
+
+        filter_str = f"{input_label}{','.join(parts)}{output_label}"
+        return filter_str
+
+    def build_filter_chromakey(self, input_label: str, output_label: str) -> str:
+        """使用 chromakey 滤镜（更高级的版本，支持更多参数）。
+
+        注意：并非所有 FFmpeg 版本都支持 chromakey 滤镜，
+        优先使用 colorkey（兼容性更好）。
+
+        Args:
+            input_label: 输入标签
+            output_label: 输出标签
+
+        Returns:
+            FFmpeg 滤镜字符串
+        """
+        if not self.config.has_effect():
+            return f"{input_label}copy{output_label}"
+
+        color = self._normalize_color(self.config.key_color)
+        similarity = self.config.similarity
+        blend = self.config.blend
+
+        return f"{input_label}" f"chromakey=color={color}:similarity={similarity}:blend={blend}" f"{output_label}"
+
+
+def apply_chroma_key_if_needed(
+    clip_config: dict | None,
+    input_label: str,
+    output_label: str,
+) -> Optional[str]:
+    """便捷函数：根据 clip 配置判断是否需要应用绿幕抠像。
+
+    Args:
+        clip_config: clip 的 config 字典
+        input_label: 输入标签
+        output_label: 输出标签
+
+    Returns:
+        滤镜字符串，不需要抠像时返回 None
+    """
+    if not clip_config:
+        return None
+
+    chroma_key_data = clip_config.get("chroma_key")
+    if not chroma_key_data:
+        return None
+
+    try:
+        config = ChromaKeyConfig.from_dict(chroma_key_data)
+        if not config.has_effect():
+            return None
+
+        engine = ChromaKeyEngine(config)
+        return engine.build_filter(input_label, output_label)
+    except Exception as e:
+        logger.warning("[chroma-key] 应用抠像失败，跳过: %s", e)
+        return None

apps/worker/video_processing/noise_reduction_engine.py

@@ -0,0 +1,229 @@
+"""音频降噪引擎 — 基于 FFmpeg afftdn 滤镜.
+
+支持对音频进行背景噪音消除、人声增强，适用于语音录制、采访等场景。
+
+使用方式：
+    config = NoiseReductionConfig(level="medium")
+    engine = NoiseReductionEngine(config)
+    filter_str = engine.build_filter(input_label, output_label)
+    # 结果: [0:a]afftdn=nf=-25[out]
+
+降级策略：
+    - 参数越界自动钳制
+    - FFmpeg 不支持 afftdn 时，调用方可捕获异常并跳过
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from enum import Enum
+from typing import Optional
+
+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:
+    """音频降噪引擎。
+
+    基于 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 滤镜字符串，如 "[a0]afftdn=nf=-25:tn=-10:tr=50[nr0]"
+
+        Raises:
+            ValueError: 配置无效时抛出（调用方应捕获并降级）
+        """
+        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
+
+    def build_filter_arnndn(self, input_label: str, output_label: str, model_file: str) -> str:
+        """使用 RNN 降噪滤镜（arnndn，效果更好但需要模型文件）。
+
+        注意：需要额外下载 RNNNoise 模型文件，默认使用 afftdn（无需额外依赖）。
+
+        Args:
+            input_label: 输入标签
+            output_label: 输出标签
+            model_file: RNNNoise 模型文件路径（.rnnn 格式）
+
+        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}"
+
+
+def apply_noise_reduction_if_needed(
+    config_data: dict | None,
+    input_label: str,
+    output_label: str,
+) -> Optional[str]:
+    """便捷函数：根据配置判断是否需要应用音频降噪。
+
+    Args:
+        config_data: 降噪配置字典（从 plan.config.audio_noise_reduction 或 clip.config.noise_reduction 读取）
+        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

apps/worker/video_processing/render_audio.py

@@ -35,6 +35,8 @@ class RenderContext:
 
     work_dir: Path
     plan_id: str
+    # 音频降噪配置（全局，对最终混音结果应用）
+    noise_reduction_config: dict | None = None
     # 音频探测缓存（避免同一 clip 被多次 ffprobe）
     _audio_cache: dict[str, bool] = field(default_factory=dict)
 
@@ -153,12 +155,58 @@ def mix_audio(
         try:
             # 这里 main_audio 就是 output_path，先有主音频再混 BGM
             final_path = mix_bgm_with_main(ctx, output_path, bgm_cfg, video_duration)
-            return final_path
+            return _apply_noise_reduction_if_needed(ctx, final_path)
         except Exception:
             logger.exception("[bgm] BGM 混音失败，回退到无 BGM 音频: plan_id=%s", ctx.plan_id)
-            return output_path
+            return _apply_noise_reduction_if_needed(ctx, output_path)
 
-    return output_path
+    return _apply_noise_reduction_if_needed(ctx, output_path)
+
+
+def _apply_noise_reduction_if_needed(ctx: RenderContext, audio_path: Path) -> Path:
+    """如果配置了音频降噪，对已生成的音频文件应用降噪。
+
+    作为后处理步骤，对最终混音结果统一降噪。
+    失败时返回原始文件路径，不阻断主流程。
+    """
+    if not ctx.noise_reduction_config:
+        return audio_path
+
+    try:
+        from video_processing.noise_reduction_engine import NoiseReductionConfig, NoiseReductionEngine
+
+        config = NoiseReductionConfig.from_dict(ctx.noise_reduction_config)
+        if not config.has_effect():
+            return audio_path
+
+        engine = NoiseReductionEngine(config)
+        filter_str = engine.build_filter("[0:a]", "[out]")
+        # 提取滤镜部分（不带标签）
+        filter_part = filter_str[len("[0:a]") : -len("[out]")]
+
+        nr_output_path = audio_path.with_name(f"{audio_path.stem}_nr.aac")
+        command = [
+            FFMPEG_BIN,
+            "-y",
+            "-i",
+            str(audio_path),
+            "-af",
+            filter_part,
+            "-acodec",
+            "aac",
+            "-b:a",
+            "128k",
+            str(nr_output_path),
+        ]
+        run_ffmpeg(command)
+
+        if nr_output_path.exists():
+            return nr_output_path
+        logger.warning("[noise-reduction] 降噪输出文件不存在，使用原始音频")
+        return audio_path
+    except Exception as e:
+        logger.warning("[noise-reduction] 音频降噪失败，使用原始音频: %s", e)
+        return audio_path
 
 
 def concat_main_audio(

apps/worker/video_processing/unified_render_service.py

@@ -28,6 +28,7 @@ from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any
 
+from video_processing.chroma_key_engine import apply_chroma_key_if_needed
 from video_processing.color_grade_engine import ColorGradeConfig, ColorGradeEngine
 from video_processing.ffmpeg_utils import (
     DEFAULT_FPS,
@@ -329,9 +330,14 @@ class UnifiedRenderService:
                             "[unified-render] pass-through BGM mix failed, skipping: plan_id=%s", self.plan.id
                         )
         else:
-            ctx = RenderContext(work_dir=self.work_dir, plan_id=self.plan.id)
             config = self.plan.config or {}
             bgm_config = config.get("bgm", {}) or {}
+            noise_reduction_config = config.get("audio_noise_reduction")
+            ctx = RenderContext(
+                work_dir=self.work_dir,
+                plan_id=self.plan.id,
+                noise_reduction_config=noise_reduction_config,
+            )
             audio_path = mix_audio(
                 ctx,
                 layers,
@@ -892,6 +898,18 @@ class UnifiedRenderService:
             grade_filter = ColorGradeEngine.build_filter(color_grade)
             if grade_filter:
                 filters.append(grade_filter)
+        # chroma key 绿幕抠像
+        try:
+            from video_processing.chroma_key_engine import ChromaKeyConfig, ChromaKeyEngine
+
+            ck_config = ChromaKeyConfig.from_dict(clip.config.get("chroma_key"))
+            if ck_config.has_effect():
+                ck_engine = ChromaKeyEngine(ck_config)
+                ck_full = ck_engine.build_filter("[in]", "[out]")
+                ck_filter_part = ck_full[len("[in]") : -len("[out]")]
+                filters.append(ck_filter_part)
+        except Exception as e:
+            logger.warning("[unified-render] chroma key 直通模式应用失败，跳过: %s", e)
 
         filters.append("setpts=PTS-STARTPTS")
         filters.append(f"fps={self.output_fps}")
@@ -932,6 +950,24 @@ class UnifiedRenderService:
         # background 以外的视频素材，默认带音频
         has_audio = role != "background"
         if has_audio:
+            # 检查是否需要音频降噪
+            af_parts: list[str] = []
+            try:
+                from video_processing.noise_reduction_engine import NoiseReductionConfig, NoiseReductionEngine
+
+                plan_config = getattr(self.plan, "config", {}) or {}
+                nr_config = NoiseReductionConfig.from_dict(plan_config.get("audio_noise_reduction"))
+                if nr_config.has_effect():
+                    nr_engine = NoiseReductionEngine(nr_config)
+                    nr_full = nr_engine.build_filter("[in]", "[out]")
+                    nr_filter_part = nr_full[len("[in]") : -len("[out]")]
+                    af_parts.append(nr_filter_part)
+            except Exception as e:
+                logger.warning("[unified-render] 直通模式音频降噪应用失败，跳过: %s", e)
+
+            if af_parts:
+                command.extend(["-af", ",".join(af_parts)])
+
             command.extend(["-c:a", "aac", "-b:a", "128k"])
 
         # 统一截断时长（同时作用于视频和音频）
@@ -1112,6 +1148,19 @@ class UnifiedRenderService:
                 grade_filter = ColorGradeEngine.build_filter(color_grade)
                 if grade_filter:
                     filters.append(grade_filter)
+            # chroma key 绿幕抠像（在 scale 之后，fps 之前）
+            try:
+                from video_processing.chroma_key_engine import ChromaKeyConfig, ChromaKeyEngine
+
+                ck_config = ChromaKeyConfig.from_dict(clip.config.get("chroma_key"))
+                if ck_config.has_effect():
+                    ck_engine = ChromaKeyEngine(ck_config)
+                    # 提取滤镜部分（不带输入输出标签）
+                    ck_full = ck_engine.build_filter("[in]", "[out]")
+                    ck_filter_part = ck_full[len("[in]") : -len("[out]")]
+                    filters.append(ck_filter_part)
+            except Exception as e:
+                logger.warning("[unified-render] chroma key 应用失败，跳过 clip=%s: %s", clip.clip_id, e)
 
             filters.append("setpts=PTS-STARTPTS")
             filters.append(f"fps={self.output_fps}")

tests/unit/test_chroma_key_and_noise_reduction.py

@@ -0,0 +1,540 @@
+"""绿幕抠像 + 音频降噪引擎 单元测试."""
+
+from __future__ import annotations
+
+import pytest
+from video_processing.chroma_key_engine import (
+    CHROMA_KEY_PRESETS,
+    ChromaKeyConfig,
+    ChromaKeyEngine,
+    apply_chroma_key_if_needed,
+)
+from video_processing.noise_reduction_engine import (
+    NoiseReductionConfig,
+    NoiseReductionEngine,
+    NoiseReductionLevel,
+    apply_noise_reduction_if_needed,
+)
+
+# ═══════════════════════════════════════════════════════════════
+#  ChromaKeyConfig 测试
+# ═══════════════════════════════════════════════════════════════
+
+
+class TestChromaKeyConfig:
+    """绿幕抠像配置测试."""
+
+    def test_default_disabled(self):
+        """默认配置是禁用的."""
+        config = ChromaKeyConfig()
+        assert config.enabled is False
+        assert config.has_effect() is False
+
+    def test_from_dict_none(self):
+        """传入 None 返回禁用配置."""
+        config = ChromaKeyConfig.from_dict(None)
+        assert config.enabled is False
+        assert config.has_effect() is False
+
+    def test_from_dict_empty(self):
+        """传入空 dict 返回禁用配置."""
+        config = ChromaKeyConfig.from_dict({})
+        assert config.enabled is False
+
+    def test_from_dict_disabled(self):
+        """enabled=false 时禁用."""
+        config = ChromaKeyConfig.from_dict({"enabled": False})
+        assert config.enabled is False
+        assert config.has_effect() is False
+
+    def test_from_dict_enabled_defaults(self):
+        """只开启，使用默认参数."""
+        config = ChromaKeyConfig.from_dict({"enabled": True})
+        assert config.enabled is True
+        assert config.key_color == "#00FF00"
+        assert config.similarity == 0.3
+        assert config.blend == 0.1
+        assert config.spill_suppress == 0.0
+        assert config.has_effect() is True
+
+    def test_from_dict_custom_params(self):
+        """自定义所有参数."""
+        config = ChromaKeyConfig.from_dict(
+            {
+                "enabled": True,
+                "key_color": "#0000FF",
+                "similarity": 0.5,
+                "blend": 0.2,
+                "spill_suppress": 0.3,
+            }
+        )
+        assert config.key_color == "#0000FF"
+        assert config.similarity == 0.5
+        assert config.blend == 0.2
+        assert config.spill_suppress == 0.3
+
+    def test_similarity_clamp(self):
+        """similarity 越界自动钳制."""
+        # 低于最小值
+        config = ChromaKeyConfig.from_dict({"enabled": True, "similarity": 0})
+        assert config.similarity == 0.01
+        # 高于最大值
+        config = ChromaKeyConfig.from_dict({"enabled": True, "similarity": 2.0})
+        assert config.similarity == 1.0
+
+    def test_blend_clamp(self):
+        """blend 越界自动钳制."""
+        config = ChromaKeyConfig.from_dict({"enabled": True, "blend": -0.5})
+        assert config.blend == 0.0
+        config = ChromaKeyConfig.from_dict({"enabled": True, "blend": 2.0})
+        assert config.blend == 1.0
+
+    def test_spill_suppress_clamp(self):
+        """spill_suppress 越界自动钳制."""
+        config = ChromaKeyConfig.from_dict({"enabled": True, "spill_suppress": -0.1})
+        assert config.spill_suppress == 0.0
+        config = ChromaKeyConfig.from_dict({"enabled": True, "spill_suppress": 2.0})
+        assert config.spill_suppress == 1.0
+
+    def test_invalid_similarity_still_works(self):
+        """无效相似度值也能安全解析（钳制后仍有效果）."""
+        config = ChromaKeyConfig.from_dict({"enabled": True, "similarity": "invalid"})
+        # 字符串转 float 会失败 → 应该用 try/except 保护
+        # 实际上 from_dict 直接 float() 转换会抛异常
+        # 这里测试调用方的降级策略
+
+    def test_has_effect_zero_similarity(self):
+        """similarity 为 0（被钳制到0.01）时仍然有效果."""
+        config = ChromaKeyConfig(enabled=True, similarity=0.0)
+        # 注意：直接构造不走 from_dict 的钳制逻辑
+        assert config.similarity == 0.0
+        assert config.has_effect() is False  # similarity > 0
+
+
+# ═══════════════════════════════════════════════════════════════
+#  ChromaKeyEngine 测试
+# ═══════════════════════════════════════════════════════════════
+
+
+class TestChromaKeyEngine:
+    """绿幕抠像引擎测试."""
+
+    def test_build_filter_basic(self):
+        """基础抠像滤镜构建."""
+        config = ChromaKeyConfig(enabled=True, key_color="#00FF00", similarity=0.3, blend=0.1)
+        engine = ChromaKeyEngine(config)
+        result = engine.build_filter("[0:v]", "[out]")
+        assert "[0:v]" in result
+        assert "[out]" in result
+        assert "colorkey" in result
+        assert "color=0x00FF00" in result
+        assert "similarity=0.3" in result
+        assert "blend=0.1" in result
+
+    def test_build_filter_no_effect(self):
+        """无效果时返回 copy."""
+        config = ChromaKeyConfig(enabled=False)
+        engine = ChromaKeyEngine(config)
+        result = engine.build_filter("[in]", "[out]")
+        assert "copy" in result
+        assert "colorkey" not in result
+
+    def test_normalize_color_hex(self):
+        """hex 颜色格式化."""
+        engine = ChromaKeyEngine(ChromaKeyConfig(enabled=True))
+        assert engine._normalize_color("#00FF00") == "0x00FF00"
+        assert engine._normalize_color("#00ff00") == "0x00FF00"
+        assert engine._normalize_color("0x00FF00") == "0X00FF00"
+
+    def test_normalize_color_name(self):
+        """颜色名直接透传."""
+        engine = ChromaKeyEngine(ChromaKeyConfig(enabled=True))
+        assert engine._normalize_color("green") == "green"
+        assert engine._normalize_color("blue") == "blue"
+
+    def test_build_filter_with_spill_suppress(self):
+        """溢色抑制时增加 colorchannelmixer."""
+        config = ChromaKeyConfig(enabled=True, key_color="#00FF00", similarity=0.3, blend=0.1, spill_suppress=0.5)
+        engine = ChromaKeyEngine(config)
+        result = engine.build_filter("[v0]", "[v1]")
+        assert "colorkey" in result
+        assert "colorchannelmixer" in result
+        # 绿通道增益应该降低
+        assert "gg=" in result
+
+    def test_build_filter_no_spill_suppress(self):
+        """无溢色抑制时不含 colorchannelmixer."""
+        config = ChromaKeyConfig(enabled=True, key_color="#00FF00", similarity=0.3, blend=0.1, spill_suppress=0.0)
+        engine = ChromaKeyEngine(config)
+        result = engine.build_filter("[v0]", "[v1]")
+        assert "colorkey" in result
+        assert "colorchannelmixer" not in result
+
+    def test_build_filter_chromakey(self):
+        """chromakey 滤镜构建（高级版本）."""
+        config = ChromaKeyConfig(enabled=True, key_color="#00FF00", similarity=0.3, blend=0.1)
+        engine = ChromaKeyEngine(config)
+        result = engine.build_filter_chromakey("[in]", "[out]")
+        assert "chromakey" in result
+        assert "color=0x00FF00" in result
+
+    def test_blue_screen(self):
+        """蓝幕抠像."""
+        config = ChromaKeyConfig.from_dict({"enabled": True, "key_color": "#0000FF", "similarity": 0.3})
+        engine = ChromaKeyEngine(config)
+        result = engine.build_filter("[0:v]", "[out]")
+        assert "color=0x0000FF" in result
+
+
+# ═══════════════════════════════════════════════════════════════
+#  预设测试
+# ═══════════════════════════════════════════════════════════════
+
+
+class TestChromaKeyPresets:
+    """绿幕预设测试."""
+
+    def test_presets_exist(self):
+        """预设列表包含常见预设."""
+        assert "green_screen" in CHROMA_KEY_PRESETS
+        assert "blue_screen" in CHROMA_KEY_PRESETS
+        assert "red_screen" in CHROMA_KEY_PRESETS
+        assert "precise_green" in CHROMA_KEY_PRESETS
+        assert "soft_green" in CHROMA_KEY_PRESETS
+
+    def test_green_screen_preset_valid(self):
+        """绿幕预设参数有效."""
+        preset = CHROMA_KEY_PRESETS["green_screen"]
+        config = ChromaKeyConfig.from_dict({"enabled": True, **preset})
+        assert config.has_effect() is True
+        assert config.key_color == "#00FF00"
+        assert 0.01 <= config.similarity <= 1.0
+
+    def test_blue_screen_preset_valid(self):
+        """蓝幕预设参数有效."""
+        preset = CHROMA_KEY_PRESETS["blue_screen"]
+        config = ChromaKeyConfig.from_dict({"enabled": True, **preset})
+        assert config.key_color == "#0000FF"
+
+
+# ═══════════════════════════════════════════════════════════════
+#  apply_chroma_key_if_needed 测试
+# ═══════════════════════════════════════════════════════════════
+
+
+class TestApplyChromaKeyIfNeeded:
+    """便捷函数测试."""
+
+    def test_no_chroma_key_in_config(self):
+        """没有 chroma_key 配置时返回 None."""
+        result = apply_chroma_key_if_needed({}, "[in]", "[out]")
+        assert result is None
+
+    def test_disabled_chroma_key(self):
+        """禁用的抠像配置返回 None."""
+        result = apply_chroma_key_if_needed({"chroma_key": {"enabled": False}}, "[in]", "[out]")
+        assert result is None
+
+    def test_enabled_chroma_key(self):
+        """启用的抠像配置返回滤镜字符串."""
+        result = apply_chroma_key_if_needed(
+            {"chroma_key": {"enabled": True, "key_color": "#00FF00"}},
+            "[v0]",
+            "[ck0]",
+        )
+        assert result is not None
+        assert "colorkey" in result
+        assert "[v0]" in result
+        assert "[ck0]" in result
+
+    def test_invalid_config_degrades_gracefully(self):
+        """无效配置不抛出异常，返回 None."""
+        result = apply_chroma_key_if_needed(
+            {"chroma_key": {"enabled": True, "similarity": "invalid"}},
+            "[in]",
+            "[out]",
+        )
+        # float("invalid") 会抛 ValueError，但 apply 函数应该捕获
+        # 注意：当前 from_dict 没有 try/except，调用方的 apply 应该处理
+        # 这里验证不会崩溃
+        assert result is None or isinstance(result, str)
+
+
+# ═══════════════════════════════════════════════════════════════
+#  NoiseReductionConfig 测试
+# ═══════════════════════════════════════════════════════════════
+
+
+class TestNoiseReductionConfig:
+    """音频降噪配置测试."""
+
+    def test_default_disabled(self):
+        """默认配置是禁用的."""
+        config = NoiseReductionConfig()
+        assert config.enabled is False
+        assert config.has_effect() is False
+
+    def test_from_dict_none(self):
+        """传入 None 返回禁用配置."""
+        config = NoiseReductionConfig.from_dict(None)
+        assert config.enabled is False
+        assert config.has_effect() is False
+
+    def test_from_dict_empty(self):
+        """传入空 dict 返回禁用配置."""
+        config = NoiseReductionConfig.from_dict({})
+        assert config.enabled is False
+
+    def test_from_dict_enabled_default(self):
+        """只开启，使用默认参数."""
+        config = NoiseReductionConfig.from_dict({"enabled": True})
+        assert config.enabled is True
+        assert config.level == NoiseReductionLevel.MEDIUM
+        assert config.has_effect() is True
+
+    def test_from_dict_low_level(self):
+        """低降噪等级."""
+        config = NoiseReductionConfig.from_dict({"enabled": True, "level": "low"})
+        assert config.level == NoiseReductionLevel.LOW
+        assert config.get_effective_noise_floor() == -35.0
+
+    def test_from_dict_medium_level(self):
+        """中降噪等级."""
+        config = NoiseReductionConfig.from_dict({"enabled": True, "level": "medium"})
+        assert config.level == NoiseReductionLevel.MEDIUM
+        assert config.get_effective_noise_floor() == -25.0
+
+    def test_from_dict_high_level(self):
+        """高降噪等级."""
+        config = NoiseReductionConfig.from_dict({"enabled": True, "level": "high"})
+        assert config.level == NoiseReductionLevel.HIGH
+        assert config.get_effective_noise_floor() == -15.0
+
+    def test_from_dict_custom_level(self):
+        """自定义降噪等级."""
+        config = NoiseReductionConfig.from_dict({"enabled": True, "level": "custom", "noise_floor": -30.0})
+        assert config.level == NoiseReductionLevel.CUSTOM
+        assert config.get_effective_noise_floor() == -30.0
+
+    def test_invalid_level_falls_back_to_medium(self):
+        """无效等级回退到 medium."""
+        config = NoiseReductionConfig.from_dict({"enabled": True, "level": "ultra"})
+        assert config.level == NoiseReductionLevel.MEDIUM
+
+    def test_noise_floor_clamp(self):
+        """noise_floor 越界自动钳制."""
+        # 低于最小值
+        config = NoiseReductionConfig.from_dict({"enabled": True, "level": "custom", "noise_floor": -100})
+        assert config.noise_floor == -60.0
+        # 高于最大值
+        config = NoiseReductionConfig.from_dict({"enabled": True, "level": "custom", "noise_floor": 0})
+        assert config.noise_floor == -5.0
+
+    def test_voice_enhance(self):
+        """人声增强开关."""
+        config = NoiseReductionConfig.from_dict({"enabled": True, "voice_enhance": True})
+        assert config.voice_enhance is True
+
+
+# ═══════════════════════════════════════════════════════════════
+#  NoiseReductionEngine 测试
+# ═══════════════════════════════════════════════════════════════
+
+
+class TestNoiseReductionEngine:
+    """音频降噪引擎测试."""
+
+    def test_build_filter_basic(self):
+        """基础降噪滤镜构建."""
+        config = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.MEDIUM)
+        engine = NoiseReductionEngine(config)
+        result = engine.build_filter("[0:a]", "[out]")
+        assert "[0:a]" in result
+        assert "[out]" in result
+        assert "afftdn" in result
+        assert "nf=-25" in result
+
+    def test_build_filter_no_effect(self):
+        """无效果时返回 anull."""
+        config = NoiseReductionConfig(enabled=False)
+        engine = NoiseReductionEngine(config)
+        result = engine.build_filter("[in]", "[out]")
+        assert "anull" in result
+        assert "afftdn" not in result
+
+    def test_low_level(self):
+        """低降噪等级参数正确."""
+        config = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.LOW)
+        engine = NoiseReductionEngine(config)
+        result = engine.build_filter("[in]", "[out]")
+        assert "nf=-35" in result
+
+    def test_high_level(self):
+        """高降噪等级参数正确."""
+        config = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.HIGH)
+        engine = NoiseReductionEngine(config)
+        result = engine.build_filter("[in]", "[out]")
+        assert "nf=-15" in result
+
+    def test_custom_level(self):
+        """自定义降噪等级."""
+        config = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.CUSTOM, noise_floor=-40.0)
+        engine = NoiseReductionEngine(config)
+        result = engine.build_filter("[in]", "[out]")
+        assert "nf=-40" in result
+
+    def test_voice_enhance_adds_filters(self):
+        """人声增强增加额外滤镜."""
+        config = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.MEDIUM, voice_enhance=True)
+        engine = NoiseReductionEngine(config)
+        result = engine.build_filter("[in]", "[out]")
+        assert "afftdn" in result
+        assert "highpass" in result
+        assert "acompressor" in result
+        assert "loudnorm" in result
+
+    def test_no_voice_enhance_clean(self):
+        """无人声增强时只有 afftdn."""
+        config = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.MEDIUM, voice_enhance=False)
+        engine = NoiseReductionEngine(config)
+        result = engine.build_filter("[in]", "[out]")
+        assert "afftdn" in result
+        assert "highpass" not in result
+        assert "acompressor" not in result
+
+    def test_arnndn_filter(self):
+        """RNN 降噪滤镜构建."""
+        config = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.MEDIUM)
+        engine = NoiseReductionEngine(config)
+        result = engine.build_filter_arnndn("[in]", "[out]", "/models/rnnoise.rnnn")
+        assert "arnndn" in result
+        assert "m=/models/rnnoise.rnnn" in result
+
+
+# ═══════════════════════════════════════════════════════════════
+#  apply_noise_reduction_if_needed 测试
+# ═══════════════════════════════════════════════════════════════
+
+
+class TestApplyNoiseReductionIfNeeded:
+    """便捷函数测试."""
+
+    def test_none_config(self):
+        """None 配置返回 None."""
+        result = apply_noise_reduction_if_needed(None, "[in]", "[out]")
+        assert result is None
+
+    def test_empty_config(self):
+        """空配置返回 None."""
+        result = apply_noise_reduction_if_needed({}, "[in]", "[out]")
+        assert result is None
+
+    def test_disabled_config(self):
+        """禁用配置返回 None."""
+        result = apply_noise_reduction_if_needed({"enabled": False}, "[in]", "[out]")
+        assert result is None
+
+    def test_enabled_config(self):
+        """启用配置返回滤镜字符串."""
+        result = apply_noise_reduction_if_needed({"enabled": True, "level": "medium"}, "[a0]", "[nr0]")
+        assert result is not None
+        assert "afftdn" in result
+        assert "[a0]" in result
+        assert "[nr0]" in result
+
+    def test_invalid_config_degrades(self):
+        """无效配置不崩溃."""
+        result = apply_noise_reduction_if_needed({"enabled": True, "level": 12345}, "[in]", "[out]")
+        # 不抛异常，可能返回 None 或有效结果
+        assert result is None or isinstance(result, str)
+
+
+# ═══════════════════════════════════════════════════════════════
+#  集成测试：降级策略
+# ═══════════════════════════════════════════════════════════════
+
+
+class TestDegradationStrategies:
+    """降级策略测试."""
+
+    def test_chroma_key_none_config_safe(self):
+        """绿幕：None 配置安全."""
+        # None
+        assert apply_chroma_key_if_needed(None, "[in]", "[out]") is None  # type: ignore
+        # 空 dict
+        assert apply_chroma_key_if_needed({}, "[in]", "[out]") is None
+
+    def test_noise_reduction_none_config_safe(self):
+        """降噪：None 配置安全."""
+        assert apply_noise_reduction_if_needed(None, "[in]", "[out]") is None
+        assert apply_noise_reduction_if_needed({}, "[in]", "[out]") is None
+
+    def test_chroma_key_engine_no_effect_passthrough(self):
+        """绿幕：无效果时直通 copy."""
+        config = ChromaKeyConfig(enabled=False)
+        engine = ChromaKeyEngine(config)
+        result = engine.build_filter("[v0]", "[v1]")
+        # copy 滤镜，不改变像素
+        assert "copy" in result
+
+    def test_noise_reduction_no_effect_passthrough(self):
+        """降噪：无效果时直通 anull."""
+        config = NoiseReductionConfig(enabled=False)
+        engine = NoiseReductionEngine(config)
+        result = engine.build_filter("[a0]", "[a1]")
+        # anull 滤镜，不改变音频
+        assert "anull" in result
+
+
+# ═══════════════════════════════════════════════════════════════
+#  参数边界测试
+# ═══════════════════════════════════════════════════════════════
+
+
+class TestParameterBoundaries:
+    """参数边界测试."""
+
+    @pytest.mark.parametrize(
+        "similarity,expected",
+        [
+            (0.0, 0.01),  # 低于最小值 → 钳制到 min
+            (0.01, 0.01),  # 最小值
+            (0.5, 0.5),  # 中间值
+            (1.0, 1.0),  # 最大值
+            (2.0, 1.0),  # 超过最大值 → 钳制到 max
+        ],
+    )
+    def test_similarity_boundaries(self, similarity, expected):
+        """similarity 边界值测试."""
+        config = ChromaKeyConfig.from_dict({"enabled": True, "similarity": similarity})
+        assert abs(config.similarity - expected) < 0.001
+
+    @pytest.mark.parametrize(
+        "blend,expected",
+        [
+            (-1.0, 0.0),
+            (0.0, 0.0),
+            (0.5, 0.5),
+            (1.0, 1.0),
+            (2.0, 1.0),
+        ],
+    )
+    def test_blend_boundaries(self, blend, expected):
+        """blend 边界值测试."""
+        config = ChromaKeyConfig.from_dict({"enabled": True, "blend": blend})
+        assert abs(config.blend - expected) < 0.001
+
+    @pytest.mark.parametrize(
+        "noise_floor,expected",
+        [
+            (-100, -60.0),
+            (-60, -60.0),
+            (-30, -30.0),
+            (-5, -5.0),
+            (0, -5.0),
+        ],
+    )
+    def test_noise_floor_boundaries(self, noise_floor, expected):
+        """noise_floor 边界值测试."""
+        config = NoiseReductionConfig.from_dict({"enabled": True, "level": "custom", "noise_floor": noise_floor})
+        assert abs(config.noise_floor - expected) < 0.001