From 201634853b39348812458cbe916ae8f05346ba81 Mon Sep 17 00:00:00 2001 From: "russell.zhang" Date: Wed, 21 Oct 2026 12:40:00 +0800 Subject: [PATCH] =?UTF-8?q?feat(backend):=20=E8=A7=86=E9=A2=91=E7=90=86?= =?UTF-8?q?=E8=A7=A3=E6=94=AF=E6=8C=81=E5=9C=BA=E6=99=AF=E6=A3=80=E6=B5=8B?= =?UTF-8?q?=E6=99=BA=E8=83=BD=E6=8A=BD=E5=B8=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 抽帧计划优先使用 ffmpeg 场景切换检测（select='gt(scene,t)'）按镜头 切点采样，替代固定间隔盲采： - 始终包含首帧；相邻采样点带最小间隔（frame_interval/2）防抖； 保留 max_frames 上限 - 检测失败或未检出切换（纯静止视频）时回退原固定间隔逻辑，行为 完全兼容；smart_sampling=False 可显式关闭 - 新增单测覆盖 pts_time 解析、采样计划边界、场景采样/无切点回退/ 关闭开关三条路径，并通过真实 ffmpeg 对 24s 合成视频验证切点检出 Closes #464 --- backend/app/utils/video_reader.py | 56 ++++- .../tests/test_video_reader_smart_sampling.py | 237 ++++++++++++++++++ 2 files changed, 292 insertions(+), 1 deletion(-) create mode 100644 backend/tests/test_video_reader_smart_sampling.py diff --git a/backend/app/utils/video_reader.py b/backend/app/utils/video_reader.py index 100dbff61..b3e824689 100644 --- a/backend/app/utils/video_reader.py +++ b/backend/app/utils/video_reader.py @@ -17,6 +17,8 @@ def __init__(self, grid_size=(3, 3), frame_interval=2, dedupe_enabled=True, + smart_sampling=True, + scene_threshold=0.3, unit_width=960, unit_height=540, save_quality=90, @@ -27,6 +29,8 @@ def __init__(self, self.grid_size = grid_size self.frame_interval = frame_interval self.dedupe_enabled = dedupe_enabled + self.smart_sampling = smart_sampling + self.scene_threshold = scene_threshold self.unit_width = unit_width self.unit_height = unit_height self.save_quality = save_quality @@ -67,12 +71,62 @@ def _extract_single_frame(self, ts: int) -> str | None: except subprocess.CalledProcessError: return None + def _detect_scene_cuts(self) -> list[int]: + """用 ffmpeg 场景检测返回画面突变的整秒时间戳（升序去重）。 + + ffmpeg 不可用、超时或未检出任何切换时返回空列表，由调用方回退固定间隔采样。 + """ + cmd = [ + "ffmpeg", "-hide_banner", "-nostats", + "-i", self.video_path, + "-vf", f"select='gt(scene,{self.scene_threshold})',metadata=print:file=-", + "-an", "-sn", "-dn", "-f", "null", "-", + ] + try: + proc = subprocess.run(cmd, capture_output=True, text=True, timeout=600) + except Exception as exc: + logger.warning(f"场景检测失败，回退固定间隔抽帧：{exc}") + return [] + stamps = set() + for line in (proc.stdout or "").splitlines(): + if "pts_time:" not in line: + continue + try: + stamps.add(int(float(line.split("pts_time:")[1].split()[0]))) + except (ValueError, IndexError): + continue + return sorted(stamps) + + @staticmethod + def _plan_scene_timestamps(cuts: list[int], duration: int, max_frames: int, min_gap: int) -> list[int]: + """把场景切点整理成抽帧计划：始终包含首帧，相邻采样点间隔不小于 min_gap，最多 max_frames 个。""" + picked: list[int] = [] + last: int | None = None + for ts in [0] + sorted(cuts): + if ts > duration: + break + if last is not None and ts - last < min_gap: + continue + picked.append(ts) + last = ts + if len(picked) >= max_frames: + break + return picked + def extract_frames(self, max_frames=1000) -> list[str]: try: os.makedirs(self.frame_dir, exist_ok=True) duration = float(ffmpeg.probe(self.video_path)["format"]["duration"]) - timestamps = [i for i in range(0, int(duration), self.frame_interval)][:max_frames] + timestamps = None + if self.smart_sampling: + cuts = self._detect_scene_cuts() + if cuts: + min_gap = max(1, self.frame_interval // 2) + timestamps = self._plan_scene_timestamps(cuts, int(duration), max_frames, min_gap) + logger.info(f"场景检测到 {len(cuts)} 个切点，计划抽帧 {len(timestamps)} 张") + if not timestamps: + timestamps = [i for i in range(0, int(duration), self.frame_interval)][:max_frames] # 并行提取帧 max_workers = min(os.cpu_count() or 4, 8, len(timestamps)) diff --git a/backend/tests/test_video_reader_smart_sampling.py b/backend/tests/test_video_reader_smart_sampling.py new file mode 100644 index 000000000..1831dc8ed --- /dev/null +++ b/backend/tests/test_video_reader_smart_sampling.py @@ -0,0 +1,237 @@ +import importlib.util +import pathlib +import re +import sys +import tempfile +import types +import unittest +from unittest.mock import patch + + +def _install_stubs(): + app_mod = types.ModuleType("app") + utils_pkg = types.ModuleType("app.utils") + + logger_mod = types.ModuleType("app.utils.logger") + + class _Logger: + @staticmethod + def info(*_args, **_kwargs): + return None + + @staticmethod + def warning(*_args, **_kwargs): + return None + + @staticmethod + def error(*_args, **_kwargs): + return None + + def _get_logger(_name): + return _Logger() + + logger_mod.get_logger = _get_logger + + path_helper_mod = types.ModuleType("app.utils.path_helper") + ffmpeg_mod = types.ModuleType("ffmpeg") + + pil_mod = types.ModuleType("PIL") + pil_image_mod = types.ModuleType("PIL.Image") + pil_draw_mod = types.ModuleType("PIL.ImageDraw") + pil_font_mod = types.ModuleType("PIL.ImageFont") + + class _FakeImage: + pass + + class _FakeImageDraw: + @staticmethod + def Draw(*_args, **_kwargs): + return None + + class _FakeImageFont: + @staticmethod + def truetype(*_args, **_kwargs): + return None + + @staticmethod + def load_default(): + return None + + pil_image_mod.Image = _FakeImage + pil_draw_mod.ImageDraw = _FakeImageDraw + pil_font_mod.ImageFont = _FakeImageFont + + def _get_app_dir(name): + return name + + path_helper_mod.get_app_dir = _get_app_dir + ffmpeg_mod.probe = lambda *_args, **_kwargs: {"format": {"duration": "0"}} + + sys.modules.setdefault("app", app_mod) + sys.modules.setdefault("app.utils", utils_pkg) + sys.modules["PIL"] = pil_mod + sys.modules["PIL.Image"] = pil_image_mod + sys.modules["PIL.ImageDraw"] = pil_draw_mod + sys.modules["PIL.ImageFont"] = pil_font_mod + sys.modules["ffmpeg"] = ffmpeg_mod + sys.modules["app.utils.logger"] = logger_mod + sys.modules["app.utils.path_helper"] = path_helper_mod + + +def _load_video_reader_module(): + _install_stubs() + root = pathlib.Path(__file__).resolve().parents[1] + module_path = root / "app" / "utils" / "video_reader.py" + spec = importlib.util.spec_from_file_location("video_reader", module_path) + if spec is None or spec.loader is None: + raise ImportError("video_reader module spec not found") + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +video_reader_module = _load_video_reader_module() +VideoReader = video_reader_module.VideoReader + + +def _make_fake_ffmpeg_runner(scene_stdout, colors_by_second): + """同时模拟场景检测与抽帧两类 ffmpeg 调用。""" + + def _run(cmd, check=True, capture_output=False, text=False, timeout=None): + args = [arg for arg in cmd if isinstance(arg, str)] + if any("select=" in arg for arg in args): + return types.SimpleNamespace(stdout=scene_stdout, stderr="") + output_path = next((arg for arg in args if arg.endswith(".jpg")), None) + if output_path is None: + raise AssertionError("Output path not found in ffmpeg cmd") + match = re.search(r"frame_(\d{2})_(\d{2})\.jpg$", output_path) + if match is None: + raise AssertionError("Unexpected output path") + sec = int(match.group(1)) * 60 + int(match.group(2)) + payload = colors_by_second[sec] + with open(output_path, "wb") as f: + f.write(payload) + return types.SimpleNamespace(stdout="", stderr="") + + return _run + + +class TestDetectSceneCuts(unittest.TestCase): + def test_parses_pts_time_lines_into_sorted_unique_seconds(self): + with tempfile.TemporaryDirectory() as tmp_dir: + reader = VideoReader(video_path="dummy.mp4", frame_dir=str(tmp_dir), grid_dir=str(tmp_dir)) + stdout = ( + "frame:12 pts:1234 pts_time:16.08\n" + "lavfi.scene_score=0.420000\n" + "frame:48 pts:4321 pts_time:8.40\n" + "lavfi.scene_score=0.510000\n" + "frame:96 pts:8640 pts_time:16.9\n" + ) + with patch.object(video_reader_module.subprocess, "run", + return_value=types.SimpleNamespace(stdout=stdout, stderr="")): + cuts = reader._detect_scene_cuts() + self.assertEqual(cuts, [8, 16]) + + def test_returns_empty_list_when_ffmpeg_fails(self): + with tempfile.TemporaryDirectory() as tmp_dir: + reader = VideoReader(video_path="dummy.mp4", frame_dir=str(tmp_dir), grid_dir=str(tmp_dir)) + + def _boom(*_args, **_kwargs): + raise RuntimeError("ffmpeg missing") + + with patch.object(video_reader_module.subprocess, "run", side_effect=_boom): + cuts = reader._detect_scene_cuts() + self.assertEqual(cuts, []) + + +class TestPlanSceneTimestamps(unittest.TestCase): + def test_always_includes_first_frame_and_respects_min_gap(self): + planned = VideoReader._plan_scene_timestamps([1, 2, 60], duration=120, max_frames=100, min_gap=10) + self.assertEqual(planned, [0, 60]) + + def test_caps_at_max_frames(self): + planned = VideoReader._plan_scene_timestamps([5, 10, 20, 30], duration=100, max_frames=2, min_gap=1) + self.assertEqual(planned, [0, 5]) + + def test_drops_cuts_beyond_duration(self): + planned = VideoReader._plan_scene_timestamps([5, 90], duration=60, max_frames=100, min_gap=1) + self.assertEqual(planned, [0, 5]) + + +class TestExtractFramesWithSmartSampling(unittest.TestCase): + def test_uses_scene_timestamps_when_available(self): + with tempfile.TemporaryDirectory() as tmp_dir: + frame_dir = pathlib.Path(tmp_dir) / "frames" + grid_dir = pathlib.Path(tmp_dir) / "grids" + reader = VideoReader( + video_path="dummy.mp4", + frame_interval=10, + frame_dir=str(frame_dir), + grid_dir=str(grid_dir), + smart_sampling=True, + ) + + scene_stdout = "frame:0 pts_time:8.4\nlavfi.scene_score=0.51\nframe:1 pts_time:16.2\n" + fake_colors = {0: b"frame-a", 8: b"frame-b", 10: b"frame-c", 16: b"frame-d", 20: b"frame-e"} + + with patch.object(video_reader_module.ffmpeg, "probe", return_value={"format": {"duration": "24"}}), \ + patch.object(video_reader_module.subprocess, "run", + side_effect=_make_fake_ffmpeg_runner(scene_stdout, fake_colors)): + paths = reader.extract_frames(max_frames=100) + + names = [pathlib.Path(p).name for p in paths] + self.assertEqual(names, ["frame_00_00.jpg", "frame_00_08.jpg", "frame_00_16.jpg"]) + + def test_falls_back_to_fixed_interval_when_no_scene_detected(self): + with tempfile.TemporaryDirectory() as tmp_dir: + frame_dir = pathlib.Path(tmp_dir) / "frames" + grid_dir = pathlib.Path(tmp_dir) / "grids" + reader = VideoReader( + video_path="dummy.mp4", + frame_interval=10, + frame_dir=str(frame_dir), + grid_dir=str(grid_dir), + smart_sampling=True, + ) + + fake_colors = {0: b"frame-a", 10: b"frame-b", 20: b"frame-c"} + + with patch.object(video_reader_module.ffmpeg, "probe", return_value={"format": {"duration": "24"}}), \ + patch.object(video_reader_module.subprocess, "run", + side_effect=_make_fake_ffmpeg_runner("", fake_colors)): + paths = reader.extract_frames(max_frames=100) + + names = [pathlib.Path(p).name for p in paths] + self.assertEqual(names, ["frame_00_00.jpg", "frame_00_10.jpg", "frame_00_20.jpg"]) + + def test_fixed_interval_used_when_smart_sampling_disabled(self): + with tempfile.TemporaryDirectory() as tmp_dir: + frame_dir = pathlib.Path(tmp_dir) / "frames" + grid_dir = pathlib.Path(tmp_dir) / "grids" + reader = VideoReader( + video_path="dummy.mp4", + frame_interval=10, + frame_dir=str(frame_dir), + grid_dir=str(grid_dir), + smart_sampling=False, + ) + + fake_colors = {0: b"frame-a", 10: b"frame-b", 20: b"frame-c"} + base_runner = _make_fake_ffmpeg_runner("", fake_colors) + + def _run(cmd, check=True, capture_output=False, text=False, timeout=None): + args = [arg for arg in cmd if isinstance(arg, str)] + if any("select=" in arg for arg in args): + raise AssertionError("scene detection should not run when smart_sampling is disabled") + return base_runner(cmd, check=check, capture_output=capture_output, text=text, timeout=timeout) + + with patch.object(video_reader_module.ffmpeg, "probe", return_value={"format": {"duration": "24"}}), \ + patch.object(video_reader_module.subprocess, "run", side_effect=_run): + paths = reader.extract_frames(max_frames=100) + + names = [pathlib.Path(p).name for p in paths] + self.assertEqual(names, ["frame_00_00.jpg", "frame_00_10.jpg", "frame_00_20.jpg"]) + + +if __name__ == "__main__": + unittest.main()