diff --git a/apps/worker/video_processing/chroma_key_engine.py b/apps/worker/video_processing/chroma_key_engine.py new file mode 100755 index 000000000..5176f95f4 --- /dev/null +++ b/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 diff --git a/apps/worker/video_processing/noise_reduction_engine.py b/apps/worker/video_processing/noise_reduction_engine.py new file mode 100755 index 000000000..d1d5e053f --- /dev/null +++ b/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 diff --git a/apps/worker/video_processing/render_audio.py b/apps/worker/video_processing/render_audio.py index 5941cdc24..6a0e72d1a 100755 --- a/apps/worker/video_processing/render_audio.py +++ b/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( diff --git a/apps/worker/video_processing/unified_render_service.py b/apps/worker/video_processing/unified_render_service.py index a3c65ca70..396fb210c 100755 --- a/apps/worker/video_processing/unified_render_service.py +++ b/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, @@ -333,9 +334,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, @@ -975,6 +981,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}") @@ -1015,6 +1033,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"]) # 统一截断时长(同时作用于视频和音频) @@ -1258,6 +1294,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}") diff --git a/tests/unit/test_chroma_key_and_noise_reduction.py b/tests/unit/test_chroma_key_and_noise_reduction.py new file mode 100755 index 000000000..85454bd10 --- /dev/null +++ b/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