diff --git a/apps/worker/video_processing/cover_generator.py b/apps/worker/video_processing/cover_generator.py index b2cc5bad3..8e656e6d2 100755 --- a/apps/worker/video_processing/cover_generator.py +++ b/apps/worker/video_processing/cover_generator.py @@ -93,20 +93,23 @@ class CoverGenerator: time_sec = 0 # scale + crop 实现 cover 裁剪(铺满输出尺寸) - vf = ( - f"scale={width}:{height}:force_original_aspect_ratio=increase," - f"crop={width}:{height}" - ) + vf = f"scale={width}:{height}:force_original_aspect_ratio=increase," f"crop={width}:{height}" command = [ FFMPEG_BIN, "-y", - "-ss", f"{time_sec:.3f}", - "-i", str(video_path), - "-vframes", "1", - "-vf", vf, - "-q:v", str(quality), - "-f", "mjpeg", + "-ss", + f"{time_sec:.3f}", + "-i", + str(video_path), + "-vframes", + "1", + "-vf", + vf, + "-q:v", + str(quality), + "-f", + "mjpeg", str(output_path), ] @@ -159,9 +162,12 @@ class CoverGenerator: if duration <= 0 or frame_count <= 1: # 无法获取时长或只有1帧,退化为普通抽帧 return CoverGenerator.extract_frame( - video_path, output_path, + video_path, + output_path, time_sec=min(DEFAULT_COVER_TIME, max(0, duration / 2)), - width=width, height=height, quality=quality, + width=width, + height=height, + quality=quality, ) # 临时目录 @@ -185,9 +191,12 @@ class CoverGenerator: frame_path = work_dir / f"cover_candidate_{i}.jpg" try: CoverGenerator.extract_frame( - video_path, frame_path, + video_path, + frame_path, time_sec=t, - width=width, height=height, quality=quality, + width=width, + height=height, + quality=quality, ) candidate_frames.append((t, frame_path)) except Exception as e: @@ -198,14 +207,18 @@ class CoverGenerator: # 全部失败,退化到普通抽帧 logger.warning("智能封面所有候选帧抽取失败,退化为普通抽帧") return CoverGenerator.extract_frame( - video_path, output_path, + video_path, + output_path, time_sec=min(DEFAULT_COVER_TIME, duration / 2), - width=width, height=height, quality=quality, + width=width, + height=height, + quality=quality, ) if len(candidate_frames) == 1: # 只有一帧,直接用 import shutil + shutil.copy2(candidate_frames[0][1], output_path) return output_path @@ -217,11 +230,14 @@ class CoverGenerator: # 复制最佳帧到输出路径 import shutil + shutil.copy2(best_frame[1], output_path) logger.info( "智能封面生成完成: 候选%d帧, 最佳t=%.2fs, 大小=%d字节", - len(candidate_frames), best_frame[0], output_path.stat().st_size, + len(candidate_frames), + best_frame[0], + output_path.stat().st_size, ) # 清理临时文件 @@ -265,18 +281,19 @@ class CoverGenerator: output_path.parent.mkdir(parents=True, exist_ok=True) # scale + crop 实现 cover 裁剪 - vf = ( - f"scale={width}:{height}:force_original_aspect_ratio=increase," - f"crop={width}:{height}" - ) + vf = f"scale={width}:{height}:force_original_aspect_ratio=increase," f"crop={width}:{height}" command = [ FFMPEG_BIN, "-y", - "-i", str(image_path), - "-vf", vf, - "-q:v", str(quality), - "-f", "mjpeg", + "-i", + str(image_path), + "-vf", + vf, + "-q:v", + str(quality), + "-f", + "mjpeg", str(output_path), ] @@ -288,6 +305,7 @@ class CoverGenerator: # 处理失败,直接复制原图 logger.warning("自定义封面处理失败,使用原图") import shutil + shutil.copy2(image_path, output_path) return output_path @@ -321,20 +339,29 @@ class CoverGenerator: """ if mode == "custom" and custom_image: return CoverGenerator.process_custom_cover( - custom_image, output_path, - width=width, height=height, quality=quality, + custom_image, + output_path, + width=width, + height=height, + quality=quality, ) elif mode == "time": return CoverGenerator.extract_frame( - video_path, output_path, + video_path, + output_path, time_sec=time_sec, - width=width, height=height, quality=quality, + width=width, + height=height, + quality=quality, ) else: # 默认智能封面 return CoverGenerator.extract_smart_cover( - video_path, output_path, - width=width, height=height, quality=quality, + video_path, + output_path, + width=width, + height=height, + quality=quality, ) @@ -379,7 +406,8 @@ def generate_cover_from_plan( custom_path = cover_config.get("custom_image_path") if custom_path and Path(custom_path).exists(): return CoverGenerator.process_custom_cover( - custom_path, output_path, + custom_path, + output_path, ) else: logger.warning("自定义封面图片路径无效,退化为智能封面") @@ -388,12 +416,15 @@ def generate_cover_from_plan( if mode == "time": time_sec = float(cover_config.get("time_sec", DEFAULT_COVER_TIME)) return CoverGenerator.extract_frame( - video_path, output_path, time_sec=time_sec, + video_path, + output_path, + time_sec=time_sec, ) else: # smart return CoverGenerator.extract_smart_cover( - video_path, output_path, + video_path, + output_path, ) except Exception as e: logger.warning("封面生成失败: %s", e) diff --git a/apps/worker/video_processing/render_audio.py b/apps/worker/video_processing/render_audio.py index 20ec7a698..59246be1f 100755 --- a/apps/worker/video_processing/render_audio.py +++ b/apps/worker/video_processing/render_audio.py @@ -156,9 +156,7 @@ def concat_main_audio( reverse_config = ReverseConfig.from_dict(clip.config.get("reverse")) af_filters = [] if reverse_config.enabled and reverse_config.reverse_audio: - reverse_filter = ReverseEngine.build_audio_filter( - reverse_config, duration=effective_duration - ) + reverse_filter = ReverseEngine.build_audio_filter(reverse_config, duration=effective_duration) if reverse_filter: af_filters.append(reverse_filter) @@ -198,9 +196,7 @@ def concat_main_audio( # 音频倒放 reverse_config = ReverseConfig.from_dict(clip.config.get("reverse")) if reverse_config.enabled and reverse_config.reverse_audio: - reverse_filter = ReverseEngine.build_audio_filter( - reverse_config, duration=effective_duration - ) + reverse_filter = ReverseEngine.build_audio_filter(reverse_config, duration=effective_duration) if reverse_filter: audio_filters.append(reverse_filter) diff --git a/apps/worker/video_processing/reverse_engine.py b/apps/worker/video_processing/reverse_engine.py index e21128bf8..d72322534 100755 --- a/apps/worker/video_processing/reverse_engine.py +++ b/apps/worker/video_processing/reverse_engine.py @@ -27,8 +27,8 @@ class ReverseConfig: """ enabled: bool = False - reverse_video: bool = True # 是否倒放视频 - reverse_audio: bool = True # 是否倒放音频 + reverse_video: bool = True # 是否倒放视频 + reverse_audio: bool = True # 是否倒放音频 @classmethod def from_dict(cls, data: dict[str, Any] | None) -> "ReverseConfig": diff --git a/apps/worker/video_processing/sticker_engine.py b/apps/worker/video_processing/sticker_engine.py index 86fc8e8c1..0cb4068c9 100755 --- a/apps/worker/video_processing/sticker_engine.py +++ b/apps/worker/video_processing/sticker_engine.py @@ -238,17 +238,13 @@ class StickerEngine: # 贴纸预处理 if pre_filters: - filter_parts.append( - f"[{sticker_idx + 1}:v]{','.join(pre_filters)}[{sticker_label}]" - ) + filter_parts.append(f"[{sticker_idx + 1}:v]{','.join(pre_filters)}[{sticker_label}]") sticker_source = f"[{sticker_label}]" else: sticker_source = f"[{sticker_idx + 1}:v]" # overlay 合成 - filter_parts.append( - f"{input_label}{sticker_source}overlay={pos_x:.0f}:{pos_y:.0f}{enable_expr}{output_label}" - ) + filter_parts.append(f"{input_label}{sticker_source}overlay={pos_x:.0f}:{pos_y:.0f}{enable_expr}{output_label}") return ";".join(filter_parts) @@ -308,9 +304,7 @@ class StickerEngine: # 时间范围 if sticker.duration > 0: - drawtext_params.append( - f"enable='between(t,{sticker.start_time},{sticker.start_time + sticker.duration})'" - ) + drawtext_params.append(f"enable='between(t,{sticker.start_time},{sticker.start_time + sticker.duration})'") # 淡入淡出(drawtext 没有直接的淡入淡出,用 alpha 表达式模拟) if sticker.fade_in > 0 or sticker.fade_out > 0: @@ -318,14 +312,12 @@ class StickerEngine: parts: list[str] = [] if sticker.fade_in > 0: parts.append( - f"if(lt(t,{sticker.start_time + sticker.fade_in})," - f"(t-{sticker.start_time})/{sticker.fade_in},1)" + f"if(lt(t,{sticker.start_time + sticker.fade_in})," f"(t-{sticker.start_time})/{sticker.fade_in},1)" ) if sticker.fade_out > 0 and sticker.duration > 0: fade_out_start = sticker.start_time + sticker.duration - sticker.fade_out parts.append( - f"if(gt(t,{fade_out_start})," - f"({sticker.start_time + sticker.duration}-t)/{sticker.fade_out},1)" + f"if(gt(t,{fade_out_start})," f"({sticker.start_time + sticker.duration}-t)/{sticker.fade_out},1)" ) if parts: alpha_expr = "*".join(parts) @@ -522,15 +514,11 @@ class StickerEngine: # 淡入淡出(使用 fade 的 alpha 模式) fade_filters: list[str] = [] if sticker.fade_in > 0: - fade_filters.append( - f"fade=in:st={sticker.start_time}:d={sticker.fade_in}:alpha=1" - ) + fade_filters.append(f"fade=in:st={sticker.start_time}:d={sticker.fade_in}:alpha=1") if sticker.fade_out > 0 and sticker.duration > 0: fade_out_start = sticker.start_time + sticker.duration - sticker.fade_out if fade_out_start > 0: - fade_filters.append( - f"fade=out:st={fade_out_start}:d={sticker.fade_out}:alpha=1" - ) + fade_filters.append(f"fade=out:st={fade_out_start}:d={sticker.fade_out}:alpha=1") # 估算贴纸尺寸用于位置计算 est_w = int(canvas_w * 0.3 * sticker.scale) if not sticker.width else sticker.width @@ -540,26 +528,20 @@ class StickerEngine: # enable 表达式 enable_expr = "" if sticker.duration > 0: - enable_expr = ( - f":enable='between(t,{sticker.start_time},{sticker.start_time + sticker.duration})'" - ) + enable_expr = f":enable='between(t,{sticker.start_time},{sticker.start_time + sticker.duration})'" parts: list[str] = [] # 贴纸预处理 all_pre = pre_filters + fade_filters if all_pre: - parts.append( - f"[{sticker_input_idx}:v]{','.join(all_pre)}[{scaled_label}]" - ) + parts.append(f"[{sticker_input_idx}:v]{','.join(all_pre)}[{scaled_label}]") sticker_source = f"[{scaled_label}]" else: sticker_source = f"[{sticker_input_idx}:v]" # overlay 合成 - parts.append( - f"{input_label}{sticker_source}overlay={pos_x:.0f}:{pos_y:.0f}{enable_expr}{output_label}" - ) + parts.append(f"{input_label}{sticker_source}overlay={pos_x:.0f}:{pos_y:.0f}{enable_expr}{output_label}") return ";".join(parts) diff --git a/apps/worker/video_processing/unified_render_service.py b/apps/worker/video_processing/unified_render_service.py index 00d7070d5..128e4165f 100755 --- a/apps/worker/video_processing/unified_render_service.py +++ b/apps/worker/video_processing/unified_render_service.py @@ -838,9 +838,7 @@ class UnifiedRenderService: # 倒放滤镜 reverse_config = ReverseConfig.from_dict(clip.config.get("reverse")) if reverse_config.enabled and reverse_config.reverse_video: - reverse_filter = ReverseEngine.build_video_filter( - reverse_config, duration=effective_duration - ) + reverse_filter = ReverseEngine.build_video_filter(reverse_config, duration=effective_duration) if reverse_filter: filters.append(reverse_filter) @@ -905,9 +903,7 @@ class UnifiedRenderService: # 音频倒放 reverse_config = ReverseConfig.from_dict(clip.config.get("reverse")) if reverse_config.enabled and reverse_config.reverse_audio: - af_filter = ReverseEngine.build_audio_filter( - reverse_config, duration=effective_duration - ) + af_filter = ReverseEngine.build_audio_filter(reverse_config, duration=effective_duration) if af_filter: command.extend(["-af", af_filter]) @@ -1068,9 +1064,7 @@ class UnifiedRenderService: # 倒放滤镜(在 trim 之后、scale 之前应用) reverse_config = ReverseConfig.from_dict(clip.config.get("reverse")) if reverse_config.enabled and reverse_config.reverse_video: - reverse_filter = ReverseEngine.build_video_filter( - reverse_config, duration=effective_duration - ) + reverse_filter = ReverseEngine.build_video_filter(reverse_config, duration=effective_duration) if reverse_filter: filters.append(reverse_filter) @@ -1240,9 +1234,7 @@ class UnifiedRenderService: ) raise - def _build_sticker_filters( - self, input_label: str, output_label: str - ) -> tuple[str, list[str]]: + def _build_sticker_filters(self, input_label: str, output_label: str) -> tuple[str, list[str]]: """构建贴纸叠加滤镜链. Args: diff --git a/tests/unit/test_cover_reverse_sticker.py b/tests/unit/test_cover_reverse_sticker.py index b788d1a60..b2460494b 100755 --- a/tests/unit/test_cover_reverse_sticker.py +++ b/tests/unit/test_cover_reverse_sticker.py @@ -11,18 +11,17 @@ from typing import Any from unittest.mock import MagicMock, patch import pytest - from video_processing.cover_generator import ( - CoverGenerator, DEFAULT_COVER_HEIGHT, DEFAULT_COVER_WIDTH, + CoverGenerator, generate_cover_from_plan, ) from video_processing.reverse_engine import ReverseConfig, ReverseEngine from video_processing.sticker_engine import ( - ImageStickerConfig, POSITION_PRESETS, STICKER_CATEGORIES, + ImageStickerConfig, StickerEngine, TextStickerConfig, get_sticker_categories, @@ -33,13 +32,13 @@ from video_processing.unified_render_service import ( UnifiedRenderService, ) - # ── Fixtures ────────────────────────────────────────────────────────────────── @dataclass class FakePlan: """模拟 EditPlan.""" + id: str = "plan_001" name: str = "测试计划" config: dict[str, Any] = field(default_factory=dict) @@ -90,22 +89,26 @@ class TestReverseConfig: def test_video_only(self): """只倒放视频.""" - config = ReverseConfig.from_dict({ - "enabled": True, - "reverse_video": True, - "reverse_audio": False, - }) + config = ReverseConfig.from_dict( + { + "enabled": True, + "reverse_video": True, + "reverse_audio": False, + } + ) assert config.enabled is True assert config.reverse_video is True assert config.reverse_audio is False def test_audio_only(self): """只倒放音频.""" - config = ReverseConfig.from_dict({ - "enabled": True, - "reverse_video": False, - "reverse_audio": True, - }) + config = ReverseConfig.from_dict( + { + "enabled": True, + "reverse_video": False, + "reverse_audio": True, + } + ) assert config.reverse_video is False assert config.reverse_audio is True @@ -253,9 +256,7 @@ class TestTextSticker: font_color="#FFFFFF", position="center", ) - f = StickerEngine._build_drawtext_filter( - sticker, "[in]", "[out]", 1080, 1920 - ) + f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920) assert "drawtext" in f assert "Hello World" in f assert "fontsize=36" in f @@ -270,9 +271,7 @@ class TestTextSticker: stroke_width=3, stroke_color="#FF0000", ) - f = StickerEngine._build_drawtext_filter( - sticker, "[in]", "[out]", 1080, 1920 - ) + f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920) assert "borderw=3" in f assert "bordercolor=#FF0000" in f @@ -285,9 +284,7 @@ class TestTextSticker: shadow_y=4, shadow_alpha=0.5, ) - f = StickerEngine._build_drawtext_filter( - sticker, "[in]", "[out]", 1080, 1920 - ) + f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920) assert "shadowx=4" in f assert "shadowy=4" in f @@ -299,17 +296,13 @@ class TestTextSticker: start_time=2.0, duration=3.0, ) - f = StickerEngine._build_drawtext_filter( - sticker, "[in]", "[out]", 1080, 1920 - ) + f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920) assert "enable='between(t,2.0,5.0)'" in f def test_drawtext_empty_text(self): """空文字直通.""" sticker = TextStickerConfig(enabled=True, text="") - f = StickerEngine._build_drawtext_filter( - sticker, "[in]", "[out]", 1080, 1920 - ) + f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920) assert "[in]copy[out]" in f def test_drawtext_with_fade(self): @@ -322,9 +315,7 @@ class TestTextSticker: fade_in=0.5, fade_out=0.5, ) - f = StickerEngine._build_drawtext_filter( - sticker, "[in]", "[out]", 1080, 1920 - ) + f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920) assert "alpha=" in f @@ -334,14 +325,16 @@ class TestImageSticker: def test_image_sticker_overlay(self, sample_image): """图片贴纸 overlay 滤镜生成.""" result = StickerEngine.build_sticker_chain( - stickers=[{ - "type": "image", - "image_path": str(sample_image), - "position": "top_right", - "scale": 0.5, - "opacity": 0.8, - "z_index": 10, - }], + stickers=[ + { + "type": "image", + "image_path": str(sample_image), + "position": "top_right", + "scale": 0.5, + "opacity": 0.8, + "z_index": 10, + } + ], input_label="[base]", output_label="[final]", canvas_w=1080, @@ -355,11 +348,13 @@ class TestImageSticker: def test_image_sticker_missing_file(self): """图片贴纸素材不存在时跳过.""" result = StickerEngine.build_sticker_chain( - stickers=[{ - "type": "image", - "image_path": "/nonexistent/image.png", - "position": "center", - }], + stickers=[ + { + "type": "image", + "image_path": "/nonexistent/image.png", + "position": "center", + } + ], input_label="[in]", output_label="[out]", canvas_w=1080, @@ -486,11 +481,14 @@ class TestCoverGenerator: output_path = Path(cmd[-1]) output_path.parent.mkdir(parents=True, exist_ok=True) output_path.write_bytes(b"fake jpeg data") + mock_run.side_effect = fake_run_ffmpeg output = tmp_path / "cover.jpg" result = CoverGenerator.extract_frame( - sample_video, output, time_sec=2.0, + sample_video, + output, + time_sec=2.0, ) assert result == output @@ -512,11 +510,14 @@ class TestCoverGenerator: output_path = Path(cmd[-1]) output_path.parent.mkdir(parents=True, exist_ok=True) output_path.write_bytes(b"fake jpeg data") + mock_run.side_effect = fake_run_ffmpeg output = tmp_path / "cover.jpg" CoverGenerator.extract_frame( - sample_video, output, time_sec=100.0, # 超过视频时长 + sample_video, + output, + time_sec=100.0, # 超过视频时长 ) cmd = mock_run.call_args[0][0] @@ -535,11 +536,14 @@ class TestCoverGenerator: output_path = Path(cmd[-1]) output_path.parent.mkdir(parents=True, exist_ok=True) output_path.write_bytes(b"fake jpeg data") + mock_run.side_effect = fake_run_ffmpeg output = tmp_path / "cover.jpg" CoverGenerator.extract_frame( - sample_video, output, time_sec=-5.0, + sample_video, + output, + time_sec=-5.0, ) cmd = mock_run.call_args[0][0] @@ -578,7 +582,9 @@ class TestCoverGenerator: output = tmp_path / "smart_cover.jpg" result = CoverGenerator.extract_smart_cover( - sample_video, output, frame_count=3, + sample_video, + output, + frame_count=3, ) assert result == output @@ -605,7 +611,8 @@ class TestCoverGenerator: output = tmp_path / "custom_cover.jpg" result = CoverGenerator.process_custom_cover( - sample_image, output, + sample_image, + output, ) assert result == output @@ -628,7 +635,10 @@ class TestCoverGenerator: mock_extract.return_value = output result = CoverGenerator.generate_cover( - sample_video, output, mode="time", time_sec=3.0, + sample_video, + output, + mode="time", + time_sec=3.0, ) assert result == output @@ -641,7 +651,9 @@ class TestCoverGenerator: mock_smart.return_value = output result = CoverGenerator.generate_cover( - sample_video, output, mode="smart", + sample_video, + output, + mode="smart", ) assert result == output @@ -654,7 +666,10 @@ class TestCoverGenerator: mock_custom.return_value = output result = CoverGenerator.generate_cover( - sample_video, output, mode="custom", custom_image=sample_image, + sample_video, + output, + mode="custom", + custom_image=sample_image, ) assert result == output @@ -692,8 +707,7 @@ class TestGenerateCoverFromPlan: # ═══════════════════════════════════════════════════════════════════════════════ -def _make_clip(clip_id="c1", asset_id="a1", path=Path("/fake/video.mp4"), - clip_type="main", config=None): +def _make_clip(clip_id="c1", asset_id="a1", path=Path("/fake/video.mp4"), clip_type="main", config=None): """创建测试用 ResolvedClip.""" return ResolvedClip( clip_id=clip_id, @@ -712,6 +726,7 @@ def _make_clip(clip_id="c1", asset_id="a1", path=Path("/fake/video.mp4"), def _make_service(plan, clips, asset_path_map=None, work_dir=None, tmp_path=None): """创建测试用 UnifiedRenderService.""" from pathlib import Path as P + work_dir = work_dir or (tmp_path or P("/tmp")) / "render_test" work_dir.mkdir(exist_ok=True, parents=True) return UnifiedRenderService( @@ -748,6 +763,7 @@ class TestReverseIntegration: # 直接测 _build_filter_complex from video_processing.unified_render_service import RenderLayer + layer = RenderLayer(role="main", clips=[clip, clip2]) filter_str, inputs = service._build_filter_complex([layer]) @@ -761,6 +777,7 @@ class TestReverseIntegration: service = _make_service(plan, [clip], tmp_path=tmp_path) from video_processing.unified_render_service import RenderLayer + layer = RenderLayer(role="main", clips=[clip]) layers = [layer] @@ -772,14 +789,13 @@ class TestStickerIntegration: def test_can_use_pass_through_with_stickers(self, tmp_path): """有贴纸时禁用直通模式.""" - plan = FakePlan(id="p1", config={ - "stickers": [{"type": "text", "text": "Hello", "position": "center"}] - }) + plan = FakePlan(id="p1", config={"stickers": [{"type": "text", "text": "Hello", "position": "center"}]}) clip = _make_clip() clip.actual_duration = 5.0 service = _make_service(plan, [clip], tmp_path=tmp_path) from video_processing.unified_render_service import RenderLayer + layer = RenderLayer(role="main", clips=[clip]) layers = [layer] @@ -793,6 +809,7 @@ class TestStickerIntegration: service = _make_service(plan, [clip], tmp_path=tmp_path) from video_processing.unified_render_service import RenderLayer + layer = RenderLayer(role="main", clips=[clip]) layers = [layer] @@ -800,11 +817,9 @@ class TestStickerIntegration: def test_build_sticker_filters_text(self, tmp_path): """文字贴纸滤镜构建.""" - plan = FakePlan(id="p1", config={ - "stickers": [ - {"type": "text", "text": "Hello", "position": "top_center", "z_index": 10} - ] - }) + plan = FakePlan( + id="p1", config={"stickers": [{"type": "text", "text": "Hello", "position": "top_center", "z_index": 10}]} + ) service = _make_service(plan, [], tmp_path=tmp_path) filter_str, extra_inputs = service._build_sticker_filters("in", "out") @@ -824,16 +839,19 @@ class TestStickerIntegration: def test_build_sticker_filters_image(self, sample_image, tmp_path): """图片贴纸滤镜构建 + 额外输入.""" - plan = FakePlan(id="p1", config={ - "stickers": [ - { - "type": "image", - "image_path": str(sample_image), - "position": "bottom_right", - "z_index": 5, - } - ] - }) + plan = FakePlan( + id="p1", + config={ + "stickers": [ + { + "type": "image", + "image_path": str(sample_image), + "position": "bottom_right", + "z_index": 5, + } + ] + }, + ) service = _make_service(plan, [], tmp_path=tmp_path) filter_str, extra_inputs = service._build_sticker_filters("in", "out")