fix(documentary): harden fast-path fallback and cache key prefix

app/services/documentary/frame_analysis_service.py

@@ -63,6 +63,8 @@ JSON 必须包含以下键：
         except OSError:
             video_mtime = 0
 
+        legacy_prefix = utils.md5(f"{video_path}{video_mtime}")
+
         payload = "|".join(
             [
                 str(video_path),
@@ -75,4 +77,4 @@ JSON 必须包含以下键：
                 "documentary-frame-analysis-v2",
             ]
         )
-        return utils.md5(payload)
+        return f"{legacy_prefix}_{utils.md5(payload)}"

app/utils/utils.py

@@ -586,7 +586,11 @@ def clear_keyframes_cache(video_path: str = None, cache_scope: str = "keyframes"
 
         if video_path:
             # 清理指定视频的缓存（兼容前缀扩展键）
-            video_hash = md5(video_path + str(os.path.getmtime(video_path)))
+            try:
+                video_mtime = os.path.getmtime(video_path)
+            except OSError:
+                video_mtime = 0
+            video_hash = md5(video_path + str(video_mtime))
             for entry in os.listdir(cache_dir):
                 if not entry.startswith(video_hash):
                     continue

app/utils/video_processor.py

@@ -189,12 +189,16 @@ class VideoProcessor:
         """
         先尝试单次 ffmpeg 快路径抽帧，失败时回退到高兼容方案。
         """
+        if interval_seconds <= 0:
+            raise ValueError("interval_seconds must be > 0")
+
         os.makedirs(output_dir, exist_ok=True)
 
         try:
             return self._extract_frames_fast_path(output_dir, interval_seconds=interval_seconds)
         except Exception as exc:
             logger.warning(f"快路径抽帧失败，回退到兼容模式: {exc}")
+            self._cleanup_fast_path_artifacts(output_dir)
             self.extract_frames_by_interval_ultra_compatible(output_dir, interval_seconds=interval_seconds)
             return self._collect_extracted_frame_paths(output_dir)
 
@@ -258,9 +262,18 @@ class VideoProcessor:
         return sorted(
             os.path.join(output_dir, name)
             for name in os.listdir(output_dir)
-            if name.endswith(".jpg")
+            if re.fullmatch(r"keyframe_\d{6}_\d{9}\.jpg", name)
         )
 
+    @staticmethod
+    def _cleanup_fast_path_artifacts(output_dir: str) -> None:
+        for name in os.listdir(output_dir):
+            if not re.fullmatch(r"fastframe_\d{6}\.jpg", name):
+                continue
+            artifact_path = os.path.join(output_dir, name)
+            if os.path.isfile(artifact_path):
+                os.remove(artifact_path)
+
     def _extract_single_frame_optimized(self, timestamp: float, output_path: str,
                                        use_hw_accel: bool, hwaccel_type: str) -> bool:
         """

tests/test_documentary_frame_analysis_service.py

@@ -84,17 +84,31 @@ class DocumentaryFrameAnalysisServiceTests(unittest.TestCase):
 
         self.assertNotEqual(key_a, key_b)
 
+    def test_cache_key_starts_with_legacy_video_hash_prefix(self):
+        service = DocumentaryFrameAnalysisService()
+
+        with patch("app.services.documentary.frame_analysis_service.os.path.getmtime", return_value=123.0):
+            key = service._build_cache_key("video.mp4", 3.0, "prompt-v1", "model-a", 10, 2)
+
+        expected_prefix = utils.md5("video.mp4" + "123.0")
+        self.assertTrue(key.startswith(expected_prefix))
+
     def test_clear_keyframes_cache_respects_scope_and_prefix_match(self):
         with TemporaryDirectory() as temp_root:
+            service = DocumentaryFrameAnalysisService()
             analysis_dir = os.path.join(temp_root, "analysis")
             os.makedirs(analysis_dir, exist_ok=True)
 
-            with patch("app.utils.utils.os.path.getmtime", return_value=123.0):
-                prefix = utils.md5("video.mp4" + "123.0")
+            with patch("app.services.documentary.frame_analysis_service.os.path.getmtime", return_value=123.0):
+                target_key_a = service._build_cache_key("video.mp4", 3.0, "prompt-v1", "model-a", 10, 2)
+                target_key_b = service._build_cache_key("video.mp4", 5.0, "prompt-v1", "model-a", 10, 2)
+                keep_key = service._build_cache_key("other.mp4", 3.0, "prompt-v1", "model-a", 10, 2)
 
-            target_dir = os.path.join(analysis_dir, f"{prefix}_interval3")
-            keep_dir = os.path.join(analysis_dir, "other_video")
-            os.makedirs(target_dir, exist_ok=True)
+            target_dir_a = os.path.join(analysis_dir, target_key_a)
+            target_dir_b = os.path.join(analysis_dir, target_key_b)
+            keep_dir = os.path.join(analysis_dir, keep_key)
+            os.makedirs(target_dir_a, exist_ok=True)
+            os.makedirs(target_dir_b, exist_ok=True)
             os.makedirs(keep_dir, exist_ok=True)
 
             with patch("app.utils.utils.temp_dir", return_value=temp_root), patch(
@@ -102,7 +116,8 @@ class DocumentaryFrameAnalysisServiceTests(unittest.TestCase):
             ):
                 utils.clear_keyframes_cache(video_path="video.mp4", cache_scope="analysis")
 
-            self.assertFalse(os.path.exists(target_dir))
+            self.assertFalse(os.path.exists(target_dir_a))
+            self.assertFalse(os.path.exists(target_dir_b))
             self.assertTrue(os.path.exists(keep_dir))
 
 

tests/test_video_processor_documentary_unittest.py

@@ -44,3 +44,48 @@ class VideoProcessorDocumentaryTests(unittest.TestCase):
         self.assertEqual([expected_frame_path], result)
         fast_path.assert_called_once_with(output_dir, interval_seconds=3.0)
         fallback.assert_called_once_with(processor, output_dir, interval_seconds=3.0)
+
+    def test_extract_frames_by_interval_rejects_non_positive_interval(self):
+        processor = VideoProcessor.__new__(VideoProcessor)
+        processor.video_path = "demo.mp4"
+        processor.duration = 6.0
+        processor.fps = 25.0
+
+        with patch.object(VideoProcessor, "extract_frames_by_interval_ultra_compatible", autospec=True) as fallback:
+            with self.assertRaises(ValueError):
+                processor.extract_frames_by_interval_with_fallback("/tmp/out", interval_seconds=0)
+
+        fallback.assert_not_called()
+
+    def test_extract_frames_by_interval_fallback_cleans_partial_fast_path_artifacts(self):
+        processor = VideoProcessor.__new__(VideoProcessor)
+        processor.video_path = "demo.mp4"
+        processor.duration = 6.0
+        processor.fps = 25.0
+
+        with TemporaryDirectory() as output_dir:
+            stale_fastframe = os.path.join(output_dir, "fastframe_000000.jpg")
+            expected_keyframe = os.path.join(output_dir, "keyframe_000000_000000000.jpg")
+
+            def fast_path_with_partial_output(_output_dir, interval_seconds=5.0):
+                with open(stale_fastframe, "wb") as frame_file:
+                    frame_file.write(b"stale")
+                raise RuntimeError("simulated fast-path failure")
+
+            def ultra_compatible_fallback(self, output_dir_arg, interval_seconds=5.0):
+                with open(expected_keyframe, "wb") as frame_file:
+                    frame_file.write(b"frame")
+                return [0]
+
+            with patch.object(VideoProcessor, "_extract_frames_fast_path", side_effect=fast_path_with_partial_output) as fast_path, patch.object(
+                VideoProcessor,
+                "extract_frames_by_interval_ultra_compatible",
+                side_effect=ultra_compatible_fallback,
+                autospec=True,
+            ) as fallback:
+                result = processor.extract_frames_by_interval_with_fallback(output_dir, interval_seconds=3.0)
+
+            self.assertEqual([expected_keyframe], result)
+            self.assertFalse(os.path.exists(stale_fastframe))
+            fast_path.assert_called_once_with(output_dir, interval_seconds=3.0)
+            fallback.assert_called_once_with(processor, output_dir, interval_seconds=3.0)