feat: index video files as their extracted still frame

photomap/backend/embeddings.py

@@ -41,12 +41,14 @@
     capture_download_progress,
     get_cached_encoder,
 )
-from .media_types import IMAGE_EXTENSIONS, is_video
+from .media_types import IMAGE_EXTENSIONS, INDEXABLE_EXTENSIONS, is_video
 from .metadata_extraction import MetadataExtractor
 from .metadata_formatting import format_metadata
 from .metadata_modules import SlideSummary
 from .progress import IndexingCancelled, progress_tracker
 from .util import atomic_savez
+from .video import VIDEO_METADATA_KEY, extract_video_frame
+from .video_cache import VideoFrameCache
 
 logger = logging.getLogger(__name__)
 
@@ -105,6 +107,11 @@
 # on every single update.
 SCAN_REJECTS_FILENAME = "scan_rejects.npz"
 
+# Bump to discard every existing scan-reject cache once. Version 1 retires
+# caches written while videos still went through the pixel gate, which could
+# hold permanent rejections for perfectly good video files.
+SCAN_REJECTS_VERSION = 1
+
 # Process-wide gate around the GPU-using portion of indexing. Two concurrent
 # albums each spinning up a CLIP/SigLIP encoder will OOM a typical 8-12 GiB
 # card; this serializes them so the second album waits its turn. Created
@@ -461,6 +468,11 @@ class Embeddings(BaseModel):
     # heavy upscaling. Default mirrors the Album field default in config.py.
     min_image_dimension: int = 256
     min_image_bytes: int = DIMENSION_REJECT_MIN_BYTES
+    # Album this index belongs to, used to address the per-album video-frame
+    # cache. Optional because the CLI entry points build an Embeddings from a
+    # bare .npz path with no album behind it; videos still index there, their
+    # stills just aren't persisted and get re-extracted on demand.
+    album_key: str | None = None
 
     def __init__(self, **data):
         """Ensure embeddings_path is always resolved to prevent cache key mismatches."""
@@ -536,6 +548,21 @@ def _passes_dimension_gate(self, path: Path, st: os.stat_result | None = None) -
         surfaced side by side in the album editor; either can be disabled.
         Callers that already stat'ed the file pass the result as ``st``.
         """
+        # Videos bypass the gate entirely. Its bands are image heuristics: the
+        # middle band opens the file with ``Image.open(path).size``, which
+        # raises on a video, and the caller memoizes that ``False`` into
+        # scan_rejects.npz keyed by (size, mtime) — so a video rejected once
+        # would stay invisible until the file itself changed, with no UI to
+        # clear it. An explicit early return is the point, not an accident of
+        # the byte bands: most real videos exceed the probe ceiling and would
+        # pass on size alone, so the bug would never reproduce in manual
+        # testing with real footage, only with small clips.
+        #
+        # Nor is the pixel gate applied after extraction: a 240p home video is
+        # legitimately worth indexing.
+        if is_video(path):
+            return True
+
         min_dim = self.min_image_dimension
         byte_floor = self.min_image_bytes
         pixel_gate_active = min_dim > 1
@@ -592,6 +619,18 @@ def _load_scan_rejects(self) -> dict[str, tuple[int, float]]:
                     or int(data["min_bytes"]) != self.min_image_bytes
                 ):
                     return {}
+                # A cache written before videos bypassed the gate can hold
+                # permanent rejections for them: the gate opened the file with
+                # PIL, that raised, and the verdict was memoized against
+                # (size, mtime). Since those only change if the file itself
+                # does, such a video would stay invisible forever with no UI
+                # to clear it. Discarding the whole cache once on version
+                # change is cheap — it only costs one re-probe per file.
+                version = (
+                    int(data["cache_version"]) if "cache_version" in data.files else 0
+                )
+                if version != SCAN_REJECTS_VERSION:
+                    return {}
                 return {
                     str(key): (int(size), float(mtime))
                     for key, size, mtime in zip(
@@ -615,14 +654,15 @@ def _save_scan_rejects(self, rejects: dict[str, tuple[int, float]]) -> None:
                 mtimes=np.array([v[1] for v in rejects.values()], dtype=np.float64),
                 min_dim=np.int64(self.min_image_dimension),
                 min_bytes=np.int64(self.min_image_bytes),
+                cache_version=np.int64(SCAN_REJECTS_VERSION),
             )
         except OSError as e:
             logger.warning(f"Could not save scan-reject cache {path}: {e}")
 
     def get_image_files_from_directory(
         self,
         directory: Path,
-        exts: AbstractSet[str] = SUPPORTED_EXTENSIONS,
+        exts: AbstractSet[str] = INDEXABLE_EXTENSIONS,
         progress_callback: Callable | None = None,
         update_interval: int = 100,
         apply_dimension_gate: bool = True,
@@ -713,7 +753,7 @@ def get_image_files_from_directory(
     def get_image_files(
         self,
         image_paths_or_dir: list[Path] | Path,
-        exts: AbstractSet[str] = SUPPORTED_EXTENSIONS,
+        exts: AbstractSet[str] = INDEXABLE_EXTENSIONS,
         progress_callback: Callable | None = None,
         apply_dimension_gate: bool = True,
         reject_sink: dict[str, tuple[int, float]] | None = None,
@@ -827,14 +867,53 @@ def _clip_root(self) -> str | None:
         """Root directory for CLIP model caching (None = use the default cache)."""
         return None
 
-    def _load_image(
-        self, image_path: Path
-    ) -> tuple[Image.Image, float, dict] | None:
+    def _load_video(self, video_path: Path) -> tuple[Image.Image, float, dict] | None:
+        """Extract a video's still frame, cache it, and describe the video.
+
+        Returns the same shape as :meth:`_load_image` so the batch encoder
+        treats a video exactly like a photo from here on. ``None`` on any
+        failure — the caller already records that as a ``bad_file`` and moves
+        on, so "skip with a warning, never abort" is the pre-existing
+        contract rather than something new.
+
+        Thread-safe: each call spawns and owns its own ffmpeg process.
+        """
+        try:
+            extracted = extract_video_frame(video_path)
+            if extracted is None:
+                return None
+            frame, info = extracted
+
+            # Persist the still so the grid, slideshow poster and UMAP
+            # thumbnails have something to show without re-running ffmpeg.
+            # Extraction happens here, at index time, and never in a request
+            # handler — a lazy path would spawn dozens of concurrent ffmpeg
+            # processes on the first grid paint, with no timeout, no cancel
+            # and nowhere to surface a warning.
+            if self.album_key:
+                VideoFrameCache(self.album_key).store(video_path, frame)
+
+            # Video facts ride inside the existing per-image metadata dict, so
+            # every .npz rewrite path carries them for free and indexes
+            # predating video support need no migration.
+            metadata = {VIDEO_METADATA_KEY: info.model_dump()}
+            # Videos carry no EXIF for _get_modification_time to read.
+            return frame, video_path.stat().st_mtime, metadata
+        except Exception as e:
+            logger.error(f"Error processing video {video_path}: {e}")
+            return None
+
+    def _load_image(self, image_path: Path) -> tuple[Image.Image, float, dict] | None:
         """Open an image and extract modtime + metadata. Returns None on failure.
 
+        Videos are dispatched to :meth:`_load_video`, which returns the same
+        shape, so everything downstream of here is media-agnostic.
+
         Thread-safe: PIL decoders release the GIL during native I/O and the
         helpers used here don't share mutable state.
         """
+        if is_video(image_path):
+            return self._load_video(image_path)
         try:
             pil = Image.open(image_path)
             pil = ImageOps.exif_transpose(pil)
@@ -895,20 +974,38 @@ def _process_images_batch(
         buf_modtimes: list[float] = []
         buf_metadatas: list[dict] = []
 
+        def keep(j: int, path: Path, embedding) -> None:
+            embeddings.append(embedding)
+            filenames.append(path.resolve().as_posix())
+            modification_times.append(buf_modtimes[j])
+            metadatas.append(buf_metadatas[j])
+
         def flush() -> None:
             if not buf_images:
                 return
             try:
                 batch_emb = encoder.encode_images(buf_images)
             except Exception as e:
-                logger.error(f"Error encoding batch of {len(buf_images)} images: {e}")
-                bad_files.extend(buf_paths)
+                # Retry the batch one item at a time so a single hostile
+                # image costs only itself. Extracted video frames are the
+                # first realistic source of such an image, and losing a whole
+                # batch of unrelated photos to one of them would be a
+                # confusing, hard-to-attribute data loss.
+                logger.warning(
+                    f"Error encoding batch of {len(buf_images)} images ({e}); "
+                    "retrying them individually."
+                )
+                for j, path in enumerate(buf_paths):
+                    try:
+                        single = encoder.encode_images([buf_images[j]])
+                    except Exception as inner:
+                        logger.error(f"Error encoding {path}: {inner}")
+                        bad_files.append(path)
+                    else:
+                        keep(j, path, single[0])
             else:
                 for j, path in enumerate(buf_paths):
-                    embeddings.append(batch_emb[j])
-                    filenames.append(path.resolve().as_posix())
-                    modification_times.append(buf_modtimes[j])
-                    metadatas.append(buf_metadatas[j])
+                    keep(j, path, batch_emb[j])
             buf_paths.clear()
             buf_images.clear()
             buf_modtimes.clear()
@@ -1041,6 +1138,69 @@ def _save_embeddings(self, index_result: IndexResult) -> None:
         # Clear cache after saving
         _open_npz_file.cache_clear()
 
+        self._prune_video_frame_cache(
+            index_result.filenames, index_result.modification_times
+        )
+
+    @staticmethod
+    def _register_unreadable_files_warning(
+        album_key: str, result: "IndexResult | None"
+    ) -> None:
+        """Queue the "N files were skipped" notice for this run's completion.
+
+        Files that could not be read at all were collected in ``bad_files`` and
+        reported nowhere — the user just saw a smaller count than expected.
+        Videos make it far more likely (a truncated download, a codec ffmpeg
+        cannot handle), so it needs surfacing.
+
+        This has to run *before* ``complete_operation``, which is what folds
+        pending notices into the ProgressInfo the poller reads and clears the
+        queue. Registering it afterwards — from the router, once the index call
+        has returned — left the notice stranded in the queue: never shown for
+        this run, and silently attached to whichever run completed next.
+        """
+        if result is None or not result.bad_files:
+            return
+        count = len(result.bad_files)
+        noun, verb = ("file", "was") if count == 1 else ("files", "were")
+        progress_tracker.add_completion_warning(
+            album_key,
+            f"{count} {noun} could not be read and {verb} skipped.",
+        )
+        logger.warning(
+            f"Skipped {count} unreadable {noun} in album '{album_key}': "
+            + ", ".join(p.name for p in result.bad_files[:5])
+            + ("…" if count > 5 else "")
+        )
+
+    def _prune_video_frame_cache(self, filenames, modification_times) -> None:
+        """Drop cached stills that the just-written index no longer refers to.
+
+        One sweep here replaces what would otherwise be seven separate
+        cleanups — mtime changes, moves, copies, single and batch deletes, and
+        files removed outside the app all leave orphans behind, and each would
+        need its own hook. Running at save time means it runs exactly when the
+        index is authoritative.
+
+        Failures only waste disk, so they are logged and swallowed.
+        """
+        if not self.album_key:
+            return
+        try:
+            cache = VideoFrameCache(self.album_key)
+            if not cache.directory.is_dir():
+                return
+            keep = {
+                cache.key_for(Path(str(name)), float(mtime))
+                for name, mtime in zip(filenames, modification_times, strict=False)
+                if is_video(Path(str(name)))
+            }
+            removed = cache.prune(keep)
+            if removed:
+                logger.info(f"Removed {removed} stale video frame(s) from the cache")
+        except Exception as e:
+            logger.warning(f"Could not prune the video frame cache: {e}")
+
     @staticmethod
     def _path_compare_key(p: Path) -> str:
         """Canonical key for the new-vs-missing diff in
@@ -1330,6 +1490,7 @@ def traversal_callback(count, message):
                 self.create_umap_index, result.embeddings
             )
             result.umap_embeddings = umap_embeddings
+            self._register_unreadable_files_warning(album_key, result)
             progress_tracker.complete_operation(
                 album_key, "Indexing completed successfully"
             )
@@ -1599,6 +1760,11 @@ def _on_save_start() -> None:
                 len(missing_image_paths),
                 on_save_start=_on_save_start,
             )
+            # Before either completion path below: a run that indexed nothing
+            # new can still have skipped files, and complete_operation is what
+            # consumes the queue.
+            self._register_unreadable_files_warning(album_key, result)
+
             if not did_rebuild:
                 logger.info(
                     "No new images needed to be indexed. Will not regenerate umap"

photomap/backend/routers/album.py

@@ -11,6 +11,7 @@
 from ..config import Album, create_album, default_board_index_path, get_config_manager
 from ..embeddings import Embeddings
 from ..encoders import default_encoder_spec
+from ..video_cache import VideoFrameCache
 
 
 class UmapEpsSetRequest(BaseModel):
@@ -105,6 +106,7 @@ def get_embeddings_for_album(album_key: str) -> Embeddings:
         encoder_spec=album_config.encoder_spec,
         min_image_dimension=album_config.min_image_dimension,
         min_image_bytes=album_config.min_image_bytes,
+        album_key=album_key,
     )
 
 
@@ -192,6 +194,19 @@ def _cleanup_derived_index(album: Album | None) -> None:
         logger.warning(f"Could not remove index directory {derived_dir}: {e}")
 
 
+def _cleanup_video_frames(album_key: str) -> None:
+    """Remove an album's extracted video stills when the album goes away.
+
+    The frame cache lives in the per-user cache directory, keyed by album, so
+    nothing else would ever reclaim it. Never raises: a failure here costs
+    disk space, not correctness.
+    """
+    try:
+        VideoFrameCache(album_key).clear()
+    except Exception as e:
+        logger.warning(f"Could not clear video frame cache for '{album_key}': {e}")
+
+
 def _album_public_dict(album: Album) -> dict[str, Any]:
     """Album fields as exposed to the frontend.
 
@@ -349,6 +364,7 @@ async def delete_album(album_key: str) -> JSONResponse:
         album = config_manager.get_album(album_key)
         if config_manager.delete_album(album_key):
             _cleanup_derived_index(album)
+            _cleanup_video_frames(album_key)
             return JSONResponse(
                 content={
                     "success": True,