"""What the node does with a file once it is indexed, per app: the TMDB match and thumbnail for Videos, tags and cover for Music, thumbnail and EXIF for Photos. NodeDaemon runs it; the enrichers themselves are in indexer/.""" from dataclasses import replace from meshbay_node.indexer import DirectoryIndexer, GroupIndex from meshbay_node.roots import entry_abs_path def _under_any_directory(path: str, directories: list[str]) -> bool: """ Whether an entry's folder is one of an app's directories, or inside one. Mirrors `underAnyDirectory` in the SPA's app modules. One helper for every app since they all take a list: Videos and Music used to take a single folder and had a function each saying the same thing, which is how the two came to differ in what they did with a trailing slash. """ path = path or "" return any(path == d or path.startswith(d + "/") for d in directories) def _owning_directory(path: str, directories: list[str]) -> str | None: """ Which of an app's directories an entry belongs to — the deepest match. Deepest, because directories may nest: with both `Media` and `Media/Albums` configured, a file under the second belongs to the second. Taking the first match instead would measure it against a boundary one level too shallow, which for Music is the difference between reading a folder as an artist and reading it as a release. """ path = path or "" best: str | None = None for d in directories: if path == d or path.startswith(d + "/"): if best is None or len(d) > len(best): best = d return best class EnrichmentMixin: async def _enrich_new_video_entries(self, indexer: DirectoryIndexer, entries: list) -> None: """ Videos app: fire (never await further) enrichment for unattempted video entries under the group's configured video_root. A group with no video_root set yet does not enrich anything — TMDB lookups and ffmpeg thumbnailing are real, ongoing per-file cost (docs/MESHBAY_DESIGN.md §6.5), and running them over an operator's whole shared index before they have chosen which folder is actually their media library would burn both TMDB's rate limit and the node's CPU on files that were never meant to be in the Videos app at all. Once a root is set, `_enrich_video_root_now` (called when it changes) separately sweeps whatever it already contains — this path alone only ever sees entries new since the last broadcast. """ if not self._enricher or not self._roster: return video_dirs = await self._roster.app_directories(indexer.group_id, "video") if not video_dirs: return for entry in entries: if entry.type != "video" or (indexer.group_id, entry.id) in self._enriched_attempted: continue if not _under_any_directory(entry.path, video_dirs): continue file_path = entry_abs_path(indexer.roots, entry) if not file_path or not file_path.exists(): continue self._enriched_attempted.add((indexer.group_id, entry.id)) async def on_done(file_id: str, fields: dict, _indexer=indexer) -> None: await self._on_enriched(_indexer, file_id, fields) self._enricher.spawn(entry, file_path, on_done) async def _enrich_video_root_now(self, group_id: str) -> None: """ Videos app: sweep a group's existing index for enrichment right after its video root is set or changed. The ordinary path above only ever looks at entries new since the last broadcast, so a folder that already had files sitting in it before it became the video_root would otherwise never get enriched at all — nothing else re-visits already-indexed entries once they have been broadcast once. """ indexer = self._state.get("indexers", {}).get(group_id) if not indexer: return await self._enrich_new_video_entries(indexer, list(indexer.index.entries)) async def _reenrich_renamed_video_entries( self, indexer: DirectoryIndexer, updates: list, previous: GroupIndex, ) -> None: """ Videos app: found live — a French-named episode file, renamed by the operator to match its English-named siblings, kept showing as its own separate poster-grid card (and its own row in Flat list) indefinitely, because `_enriched_attempted` — there specifically to stop enrichment's own field-fill from re-triggering itself forever (see the caller) — also silently blocked the *new* filename from ever being title-parsed at all. `entry.id in self._enriched_attempted` is the same content, so simply discarding it here and re-running the ordinary enrichment path is enough: a fresh ffprobe/thumbnail for an unchanged file is redundant work, not a correctness issue, and renames are rare enough that the redundancy is not worth a separate "title-parse only" code path. """ for entry in updates: if entry.type != "video": continue old = previous.get_entry(entry.id) if old is None or (old.name == entry.name and old.path == entry.path): continue self._enriched_attempted.discard((indexer.group_id, entry.id)) # The rename re-derives the title (the whole point of this # method), which can change the correct TMDB match — but # file_tmdb is keyed by content hash, unchanged by a rename, so # nothing else would ever dislodge the old name's match. A # manual "Fix match" correction is kept (clear_file_tmdb skips # anything in media_cache.tmdb_override). if self._media_cache is not None: await self._media_cache.clear_file_tmdb(entry.id) await self._enrich_new_video_entries(indexer, updates) async def _enrich_new_audio_entries(self, indexer: DirectoryIndexer, entries: list) -> None: """ Music app (docs/MESHBAY_DESIGN.md §9.8): fire (never await further) tag/cover enrichment for unattempted audio entries under the group's configured audio_root — same gate as `_enrich_new_video_entries` above (the original "no root, whole shared tree" call turned out wrong against a real messy library: everything under every shared folder got mixed together with no way to scope it down). `_enriched_attempted` is shared with the video path — content-addressed ids never collide across the two. """ if not self._audio_enricher or not self._roster: return audio_dirs = await self._roster.app_directories(indexer.group_id, "music") if not audio_dirs: return # Resolved once per directory, not once per file: a library is # thousands of entries and this is a filesystem call each time. boundaries = {d: indexer.roots.resolve(d, require_available=False) for d in audio_dirs} for entry in entries: if entry.type != "audio" or (indexer.group_id, entry.id) in self._enriched_attempted: continue owner = _owning_directory(entry.path, audio_dirs) if owner is None: continue file_path = entry_abs_path(indexer.roots, entry) if not file_path or not file_path.exists(): continue self._enriched_attempted.add((indexer.group_id, entry.id)) async def on_done(file_id: str, fields: dict, _indexer=indexer) -> None: await self._on_enriched(_indexer, file_id, fields) # The boundary is *the configured directory this file is under*, # not the shared root it lives in — so the ancestor walk # (enrich_audio._artist_album_from_ancestors) treats a flat # top-level folder right under the configured Music directory as # ambiguous (artist-or-release, docs/MESHBAY_DESIGN.md §9.8), # rather than one level too shallow when that directory is # itself a subfolder. With several configured, each file is # measured against its own: # a single shared boundary would be wrong for all but one of them. self._audio_enricher.spawn(entry, file_path, on_done, boundaries.get(owner)) async def _enrich_audio_root_now(self, group_id: str) -> None: """ Music app: sweep a group's existing index right after its audio root is set or changed. Mirrors `_enrich_video_root_now` exactly — the ordinary path above only ever looks at entries new since the last broadcast, so a folder that already had files in it before it became the audio_root would otherwise never get enriched at all. """ indexer = self._state.get("indexers", {}).get(group_id) if not indexer: return await self._enrich_new_audio_entries(indexer, list(indexer.index.entries)) async def _reenrich_renamed_audio_entries( self, indexer: DirectoryIndexer, updates: list, previous: GroupIndex, ) -> None: """ Music app equivalent of `_reenrich_renamed_video_entries` — a rename can change the filename-parse fallback (title/track_no) even though embedded tags, when present, are unaffected. Re-running the whole pass on a rename is redundant work for a tagged file and a real fix for an untagged one, and renames are rare enough not to need a cheaper, tags-only special case. """ for entry in updates: if entry.type != "audio": continue old = previous.get_entry(entry.id) if old is None or (old.name == entry.name and old.path == entry.path): continue self._enriched_attempted.discard((indexer.group_id, entry.id)) await self._enrich_new_audio_entries(indexer, updates) async def _enrich_new_photo_entries(self, indexer: DirectoryIndexer, entries: list) -> None: """ Photos app (docs/MESHBAY_DESIGN.md §9.9): fire (never await further) thumbnail/EXIF enrichment for unattempted image entries under any of the group's configured photo_roots. Same gate shape as `_enrich_new_video_entries`/`_enrich_new_audio_entries` — no root configured yet means no work, since thumbnailing every image in a whole shared tree before the operator has chosen which folders are actually photo albums would burn CPU on files never meant to be in the Photos app at all. `_enriched_attempted` is shared with the video/audio paths — content-addressed ids never collide across them. """ if not self._photo_enricher or not self._roster: return photo_dirs = await self._roster.app_directories(indexer.group_id, "photo") if not photo_dirs: return for entry in entries: if entry.type != "image" or (indexer.group_id, entry.id) in self._enriched_attempted: continue if not _under_any_directory(entry.path, photo_dirs): continue file_path = entry_abs_path(indexer.roots, entry) if not file_path or not file_path.exists(): continue self._enriched_attempted.add((indexer.group_id, entry.id)) async def on_done(file_id: str, fields: dict, _indexer=indexer) -> None: await self._on_enriched(_indexer, file_id, fields) self._photo_enricher.spawn(entry, file_path, on_done) async def _enrich_photo_roots_now(self, group_id: str) -> None: """ Photos app: sweep a group's existing index right after its photo roots change. Mirrors `_enrich_video_root_now`/`_enrich_audio_root_now` — the ordinary path above only ever looks at entries new since the last broadcast, so a folder that already had photos in it before it was added to photo_roots would otherwise never get enriched at all. Also covers a root being *removed*: nothing un-enriches on removal (the cache entry is harmless, just unused — the media cache is disposable and tied to the index, docs/MESHBAY_DESIGN.md §6.5), so re-sweeping the new set is enough. """ indexer = self._state.get("indexers", {}).get(group_id) if not indexer: return await self._enrich_new_photo_entries(indexer, list(indexer.index.entries)) async def _reenrich_renamed_photo_entries( self, indexer: DirectoryIndexer, updates: list, previous: GroupIndex, ) -> None: """ Photos app equivalent of `_reenrich_renamed_video_entries` — a rename changes nothing about the image's own bytes (thumbnail, EXIF fields are content-derived, not name-derived), so this exists only for consistency/symmetry with Videos/Music and to catch the case of a file moving *into* a newly-covered photo_roots subtree via a rename rather than a fresh add. Re-running enrichment on an unchanged file is redundant work, not a correctness issue. """ for entry in updates: if entry.type != "image": continue old = previous.get_entry(entry.id) if old is None or (old.name == entry.name and old.path == entry.path): continue self._enriched_attempted.discard((indexer.group_id, entry.id)) await self._enrich_new_photo_entries(indexer, updates) async def _on_enriched(self, indexer: DirectoryIndexer, file_id: str, fields: dict) -> None: """ Merge enrichment fields into the live index and re-trigger a broadcast so they reach connected clients as an INDEX_DELTA update (GroupIndex.diff's `updates`, not `additions` — same id, new fields). Builds a *new* IndexEntry via dataclasses.replace rather than mutating the existing one in place: the diff mechanism compares against a shallow snapshot of entry *references*, so an in-place mutation would silently also change what "previous" looks like, and the change would never show up as a diff (see group_index.py's diff() docstring). """ idx = indexer.index entry = idx.get_entry(file_id) if entry is None: return # removed from the index while enrichment was in flight idx.add_entry(replace(entry, **fields)) await self._on_index_change(indexer)