1 files changed, 166 insertions, 0 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/indexer/enrich.py b/packages/meshbay-node/src/meshbay_node/indexer/enrich.py
new file mode 100644
index 0000000..784b2a3
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/indexer/enrich.py
@@ -0,0 +1,166 @@
+"""
+Index-time enrichment for the Videos group app: technical probe (ffprobe),
+filename parsing (title_parse), and thumbnail generation (ffmpeg) for a
+newly-added video IndexEntry.
+
+Runs through its own small bounded worker pool — separate from the streaming
+transcode pool (docs/mediacenter.md §5.2, mirroring webrtc_server.py's
+`_transcode_semaphore`) — so indexing a large library never blocks on this,
+and enrichment never competes with an active viewer for CPU. The scan itself
+already put the entry in the index with hash/size/type only; this fills in
+the rest asynchronously and hands the result back via a callback.
+"""
+
+import asyncio
+import logging
+from pathlib import Path
+from typing import Awaitable, Callable
+
+import blake3
+
+from meshbay_common.protocol import IndexEntry
+from meshbay_node.indexer import title_parse
+from meshbay_node.indexer.indexer import MEDIA_EXTENSIONS
+from meshbay_node.media_cache import MediaCache
+from meshbay_node.media_probe import probe_video
+
+log = logging.getLogger(__name__)
+
+DEFAULT_MAX_CONCURRENT = 2
+PROBE_TIMEOUT_SECS = 30
+THUMB_TIMEOUT_SECS = 30
+THUMB_WIDTH = 320
+# "Show/SeasonFolder/episode.mkv" is the expected shape, with a little slack
+# for an extra wrapper folder — not an attempt to find the exact group root.
+MAX_ANCESTOR_DEPTH = 4
+# Bounds the "borrow a title from a sibling episode filename" scan (§3.4) so
+# a folder with thousands of files costs a fixed, small amount of work.
+MAX_SIBLINGS_CHECKED = 20
+
+
+def _season_from_ancestors(file_path: Path) -> int | None:
+    folder = file_path.parent
+    for _ in range(MAX_ANCESTOR_DEPTH):
+        if folder is None or folder == folder.parent:
+            break
+        season = title_parse.season_from_folder_name(folder.name)
+        if season is not None:
+            return season
+        folder = folder.parent
+    return None
+
+
+def _title_from_siblings(file_path: Path) -> str | None:
+    """
+    §3.4: an episode filename with no show name in it borrows the title from
+    a representative sibling in the same folder, never from the folder name
+    alone (an acronym-named show folder is a real, observed case).
+    """
+    try:
+        names = sorted(p.name for p in file_path.parent.iterdir() if p.is_file())
+    except OSError:
+        return None
+    checked = 0
+    for name in names:
+        if name == file_path.name:
+            continue
+        if Path(name).suffix.lower() not in MEDIA_EXTENSIONS["video"]:
+            continue
+        checked += 1
+        if checked > MAX_SIBLINGS_CHECKED:
+            break
+        parsed = title_parse.parse_episode_filename(name)
+        if parsed.display_title:
+            return parsed.display_title
+    return None
+
+
+async def _make_thumbnail(file_path: Path, duration: float | None) -> bytes | None:
+    """One ffmpeg frame grab at ~10% of duration (or 5s if unknown), scaled down."""
+    seek = max(0.0, (duration or 50.0) * 0.1)
+    proc = await asyncio.create_subprocess_exec(
+        "ffmpeg", "-v", "error", "-ss", str(seek), "-i", str(file_path),
+        "-frames:v", "1", "-vf", f"scale={THUMB_WIDTH}:-1",
+        "-f", "image2", "-c:v", "mjpeg", "pipe:1",
+        stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
+    )
+    try:
+        stdout, _ = await asyncio.wait_for(proc.communicate(), THUMB_TIMEOUT_SECS)
+    except asyncio.TimeoutError:
+        proc.kill()
+        await proc.wait()
+        return None
+    return stdout or None
+
+
+class Enricher:
+    """Owns the node's bounded index-time enrichment pool."""
+
+    def __init__(self, media_cache: MediaCache, max_concurrent: int = DEFAULT_MAX_CONCURRENT):
+        self._media_cache = media_cache
+        self._sem = asyncio.Semaphore(max_concurrent)
+        self._tasks: set[asyncio.Task] = set()
+
+    def spawn(
+        self, entry: IndexEntry, file_path: Path,
+        on_done: Callable[[str, dict], Awaitable[None]],
+    ) -> asyncio.Task:
+        """
+        Fire-and-forget one file's enrichment. `on_done(file_id, fields)` is
+        awaited with the index fields to merge in once ready — never blocks
+        the caller (a scan or watchdog event). The reference this method
+        returns is what keeps the task alive; callers should hold it the
+        same way `WebRTCPeerSession._spawn` holds streaming tasks.
+        """
+        task = asyncio.ensure_future(self._run(entry, file_path, on_done))
+        self._tasks.add(task)
+
+        def _cleanup(t: asyncio.Task) -> None:
+            self._tasks.discard(t)
+            if not t.cancelled() and t.exception():
+                log.error("Enrichment failed for %s: %s", entry.id[:12], t.exception(),
+                          exc_info=t.exception())
+        task.add_done_callback(_cleanup)
+        return task
+
+    async def _run(
+        self, entry: IndexEntry, file_path: Path,
+        on_done: Callable[[str, dict], Awaitable[None]],
+    ) -> None:
+        async with self._sem:
+            fields: dict = {}
+            duration: float | None = None
+            try:
+                _codec, duration, _has_audio, width, height = await asyncio.wait_for(
+                    probe_video(str(file_path)), timeout=PROBE_TIMEOUT_SECS)
+                fields["duration"] = int(duration) if duration else None
+                fields["width"] = width
+                fields["height"] = height
+            except Exception as e:
+                log.warning("Probe failed for %s: %s", file_path, e)
+
+            ep = title_parse.parse_episode_filename(entry.name)
+            if ep.episode is not None:
+                title = ep.display_title or await asyncio.to_thread(
+                    _title_from_siblings, file_path)
+                season = ep.season
+                if season is None:
+                    season = await asyncio.to_thread(_season_from_ancestors, file_path)
+                fields["display_title"] = title or title_parse.naive_title(entry.name)
+                fields["season"] = season
+                fields["episode"] = ep.episode
+            else:
+                mv = title_parse.parse_movie_filename(entry.name)
+                fields["display_title"] = mv.display_title or mv.naive_title
+
+            try:
+                thumb = await _make_thumbnail(file_path, duration)
+            except Exception as e:
+                log.warning("Thumbnail generation failed for %s: %s", file_path, e)
+                thumb = None
+            if thumb:
+                thumb_hash = blake3.blake3(thumb).hexdigest()
+                await self._media_cache.put_thumb(thumb_hash, entry.id, thumb)
+                fields["thumb_hash"] = thumb_hash
+
+            await on_done(entry.id, fields)