Accept near-complete frame extraction chunks

tests/test_ffmpeg_sampler.py

@@ -352,6 +352,48 @@ class FfmpegSamplerTests(unittest.TestCase):
             ]
             self.assertEqual(persisted, records)
 
+    def test_sample_video_frames_uses_nearly_complete_frames_when_ffmpeg_exits_nonzero(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            video_id = "video-abc"
+            frame_dir = root / "frames" / video_id
+
+            def run_with_nonzero_exit(command, *args, **kwargs):
+                frame_dir.mkdir(parents=True, exist_ok=True)
+                for index in range(1, 597):
+                    (frame_dir / f"{index:06d}.jpg").write_bytes(b"jpg")
+                raise subprocess.CalledProcessError(
+                    returncode=1,
+                    cmd=command,
+                    stderr="trailing decoder error after near-complete chunk",
+                )
+
+            with patch("subprocess.run", side_effect=run_with_nonzero_exit):
+                records = sample_video_frames(
+                    {
+                        "video_id": video_id,
+                        "path": str(root / "input.mp4"),
+                        "codec_name": "hevc",
+                        "actual_begin": 1000,
+                        "actual_end": 1600,
+                    },
+                    root,
+                    {
+                        "prefer_nvdec": True,
+                        "allow_cpu_fallback": False,
+                        "hwaccel": "cuda",
+                        "codec_decoders": {"hevc": "hevc_cuvid"},
+                        "frame_fps": 1,
+                        "frame_width": 640,
+                        "jpeg_quality": 4,
+                        "timeout_seconds_per_video": 30,
+                    },
+                )
+
+            self.assertEqual(len(records), 596)
+            self.assertEqual({record["status"] for record in records}, {"sampled"})
+            self.assertIn("near-complete chunk", records[0]["stderr_summary"])
+
 
 if __name__ == "__main__":
     unittest.main()

video_ai_analysis_poc/ffmpeg_sampler.py

@@ -126,7 +126,11 @@ def sample_video_frames(
             timeline_start_epoch=start_epoch,
             timezone_name=timezone_name,
         )
-        if records and (max_frames is None or len(records) >= max_frames):
+        if _has_usable_frames_after_nonzero_exit(
+            records,
+            max_frames=max_frames,
+            ffmpeg_config=ffmpeg_config,
+        ):
             _attach_success_evidence(
                 records,
                 command,
@@ -142,6 +146,24 @@ def sample_video_frames(
     return records
 
 
+def _has_usable_frames_after_nonzero_exit(
+    records: list[dict[str, Any]],
+    *,
+    max_frames: int | None,
+    ffmpeg_config: dict[str, Any],
+) -> bool:
+    if not records:
+        return False
+    if max_frames is None or len(records) >= max_frames:
+        return True
+    min_ratio = float(ffmpeg_config.get("min_success_frame_ratio", 0.98))
+    missing_tolerance = int(ffmpeg_config.get("max_missing_success_frames", 5))
+    return (
+        len(records) >= math.floor(max_frames * min_ratio)
+        or max_frames - len(records) <= missing_tolerance
+    )
+
+
 def _replace_video_records(
     manifest_path: Path,
     video_id: str,