diff --git a/apps/worker/video_processing/pip_engine.py b/apps/worker/video_processing/pip_engine.py index 09c768fae..050a25f42 100755 --- a/apps/worker/video_processing/pip_engine.py +++ b/apps/worker/video_processing/pip_engine.py @@ -47,10 +47,10 @@ _VALID_POSITIONS = { } # 动画类型 -ANIMATION_FADE = "fade" # 淡入淡出 -ANIMATION_SLIDE_LEFT = "slide_left" # 从左滑入 +ANIMATION_FADE = "fade" # 淡入淡出 +ANIMATION_SLIDE_LEFT = "slide_left" # 从左滑入 ANIMATION_SLIDE_RIGHT = "slide_right" # 从右滑入 -ANIMATION_SLIDE_TOP = "slide_top" # 从上滑入 +ANIMATION_SLIDE_TOP = "slide_top" # 从上滑入 ANIMATION_SLIDE_BOTTOM = "slide_bottom" # 从下滑入 _VALID_ANIMATIONS = { @@ -81,7 +81,7 @@ class PiPLayerConfig: # 大小配置 width: int | str = "25%" # 宽度(像素或百分比) - height: int | str = "" # 高度(空则按比例自适应) + height: int | str = "" # 高度(空则按比例自适应) # 样式 opacity: float = 1.0 # 透明度 0.0-1.0 @@ -452,9 +452,7 @@ class PiPEngine: # overlay 滤镜 overlay_filter = ( - f"[{current_label}][{pre_label}]" - f"overlay={x_expr}:{y_expr}{enable_expr}" - f"[{combined_label}]" + f"[{current_label}][{pre_label}]" f"overlay={x_expr}:{y_expr}{enable_expr}" f"[{combined_label}]" ) filter_parts.append(overlay_filter) diff --git a/apps/worker/video_processing/unified_render_service.py b/apps/worker/video_processing/unified_render_service.py index 263fbd3b1..ec25de543 100755 --- a/apps/worker/video_processing/unified_render_service.py +++ b/apps/worker/video_processing/unified_render_service.py @@ -256,9 +256,7 @@ class UnifiedRenderService: # 追加画中画滤镜 if has_pip: - filter_complex, input_args = self._append_pip_filters( - filter_complex, input_args, pip_sources - ) + filter_complex, input_args = self._append_pip_filters(filter_complex, input_args, pip_sources) self._execute_ffmpeg(filter_complex, input_args, video_only_path) @@ -1002,9 +1000,7 @@ class UnifiedRenderService: # ── 画中画(PiP)相关方法 ────────────────────────────────────────────────── - def _resolve_pip_sources( - self, pip_config: PiPConfig - ) -> list[tuple[str, "PiPLayerConfig", Path]]: + def _resolve_pip_sources(self, pip_config: PiPConfig) -> list[tuple[str, "PiPLayerConfig", Path]]: """解析画中画图层的素材源,返回可用的图层列表. 降级策略:素材不存在或无效的图层自动跳过,不阻断渲染。 @@ -1026,9 +1022,7 @@ class UnifiedRenderService: for i, layer in enumerate(pip_config.layers): path = engine.validate_layer_source(layer, self.asset_path_map) if path is None: - logger.warning( - "PiP图层素材不可用,跳过: layer_index=%d source=%s", i, layer.source - ) + logger.warning("PiP图层素材不可用,跳过: layer_index=%d source=%s", i, layer.source) continue # 标签占位,实际输入索引由 build_pip_filters 内部管理 result.append((f"pip_src_{i}", layer, path)) diff --git a/tests/unit/test_pip_engine.py b/tests/unit/test_pip_engine.py index 7b14a00ae..7a5e71cff 100755 --- a/tests/unit/test_pip_engine.py +++ b/tests/unit/test_pip_engine.py @@ -6,13 +6,12 @@ from pathlib import Path from unittest.mock import patch import pytest - from video_processing.pip_engine import ( ANIMATION_FADE, + ANIMATION_SLIDE_BOTTOM, ANIMATION_SLIDE_LEFT, ANIMATION_SLIDE_RIGHT, ANIMATION_SLIDE_TOP, - ANIMATION_SLIDE_BOTTOM, POSITION_BOTTOM_LEFT, POSITION_BOTTOM_RIGHT, POSITION_CENTER, @@ -23,7 +22,6 @@ from video_processing.pip_engine import ( PiPLayerConfig, ) - # ── PiPLayerConfig.validate 测试 ────────────────────────────────────────────── @@ -383,9 +381,7 @@ class TestPiPEngineBuildFilters: ) sources = [("pip_src_0", layer, fake_video)] - filters, inputs, final_label = engine.build_pip_filters( - "base_video", sources, base_input_idx=3 - ) + filters, inputs, final_label = engine.build_pip_filters("base_video", sources, base_input_idx=3) # 应该有2个滤镜: 预处理 + overlay assert len(filters) == 2 @@ -422,9 +418,7 @@ class TestPiPEngineBuildFilters: ("s1", layer2, fake_video), ] - filters, inputs, final_label = engine.build_pip_filters( - "base", sources, base_input_idx=0 - ) + filters, inputs, final_label = engine.build_pip_filters("base", sources, base_input_idx=0) # 2层 × 2个滤镜(预处理+overlay)= 4个滤镜 assert len(filters) == 4 @@ -440,9 +434,7 @@ class TestPiPEngineBuildFilters: def test_with_opacity(self, engine, fake_video): """透明度应该在滤镜中体现.""" - layer = PiPLayerConfig( - source="a", position=POSITION_CENTER, opacity=0.5 - ) + layer = PiPLayerConfig(source="a", position=POSITION_CENTER, opacity=0.5) sources = [("s0", layer, fake_video)] filters, _, _ = engine.build_pip_filters("base", sources) @@ -452,9 +444,7 @@ class TestPiPEngineBuildFilters: def test_with_corner_radius(self, engine, fake_video): """圆角裁剪应该在滤镜中体现.""" - layer = PiPLayerConfig( - source="a", position=POSITION_CENTER, corner_radius=20 - ) + layer = PiPLayerConfig(source="a", position=POSITION_CENTER, corner_radius=20) sources = [("s0", layer, fake_video)] filters, _, _ = engine.build_pip_filters("base", sources) @@ -463,10 +453,7 @@ class TestPiPEngineBuildFilters: def test_with_border(self, engine, fake_video): """边框应该在滤镜中体现.""" - layer = PiPLayerConfig( - source="a", position=POSITION_CENTER, - border_width=3, border_color="red" - ) + layer = PiPLayerConfig(source="a", position=POSITION_CENTER, border_width=3, border_color="red") sources = [("s0", layer, fake_video)] filters, _, _ = engine.build_pip_filters("base", sources) @@ -476,10 +463,7 @@ class TestPiPEngineBuildFilters: def test_timing_start_time_and_duration(self, engine, fake_video): """时间控制应该生成enable表达式.""" - layer = PiPLayerConfig( - source="a", position=POSITION_CENTER, - start_time=5.0, duration=10.0 - ) + layer = PiPLayerConfig(source="a", position=POSITION_CENTER, start_time=5.0, duration=10.0) sources = [("s0", layer, fake_video)] filters, _, _ = engine.build_pip_filters("base", sources) @@ -489,10 +473,7 @@ class TestPiPEngineBuildFilters: def test_timing_start_time_only(self, engine, fake_video): """只有开始时间(全程显示到结束).""" - layer = PiPLayerConfig( - source="a", position=POSITION_CENTER, - start_time=3.0, duration=0.0 - ) + layer = PiPLayerConfig(source="a", position=POSITION_CENTER, start_time=3.0, duration=0.0) sources = [("s0", layer, fake_video)] filters, _, _ = engine.build_pip_filters("base", sources) @@ -502,10 +483,7 @@ class TestPiPEngineBuildFilters: def test_no_timing_no_enable(self, engine, fake_video): """无时间限制时不应该有enable表达式.""" - layer = PiPLayerConfig( - source="a", position=POSITION_CENTER, - start_time=0.0, duration=0.0 - ) + layer = PiPLayerConfig(source="a", position=POSITION_CENTER, start_time=0.0, duration=0.0) sources = [("s0", layer, fake_video)] filters, _, _ = engine.build_pip_filters("base", sources) @@ -515,7 +493,8 @@ class TestPiPEngineBuildFilters: def test_fade_animation(self, engine, fake_video): """淡入淡出动画.""" layer = PiPLayerConfig( - source="a", position=POSITION_CENTER, + source="a", + position=POSITION_CENTER, animation_in=ANIMATION_FADE, animation_out=ANIMATION_FADE, duration=10.0, @@ -532,7 +511,8 @@ class TestPiPEngineBuildFilters: def test_slide_animation_in(self, engine, fake_video): """滑入动画应该在overlay表达式中.""" layer = PiPLayerConfig( - source="a", position=POSITION_CENTER, + source="a", + position=POSITION_CENTER, animation_in=ANIMATION_SLIDE_LEFT, animation_duration=0.5, ) @@ -545,9 +525,7 @@ class TestPiPEngineBuildFilters: def test_full_opacity_no_alpha(self, engine, fake_video): """opacity=1时不应该有colorchannelmixer.""" - layer = PiPLayerConfig( - source="a", position=POSITION_CENTER, opacity=1.0 - ) + layer = PiPLayerConfig(source="a", position=POSITION_CENTER, opacity=1.0) sources = [("s0", layer, fake_video)] filters, _, _ = engine.build_pip_filters("base", sources) @@ -556,9 +534,7 @@ class TestPiPEngineBuildFilters: def test_zero_corner_radius_no_geq(self, engine, fake_video): """corner_radius=0时不应该有geq滤镜.""" - layer = PiPLayerConfig( - source="a", position=POSITION_CENTER, corner_radius=0 - ) + layer = PiPLayerConfig(source="a", position=POSITION_CENTER, corner_radius=0) sources = [("s0", layer, fake_video)] filters, _, _ = engine.build_pip_filters("base", sources)