From 387138410532daad174ed602fd03f4f979cd3d81 Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Thu, 24 Sep 2026 18:30:10 +0200 Subject: refactor(node): move per-app enrichment out of NodeDaemon The _enrich_*/_reenrich_* methods and _on_enriched become EnrichmentMixin in meshbay_node/enrichment.py, with the two directory helpers only they use. APP_DIR_KEYS stays on NodeDaemon. Co-Authored-By: Claude Opus 5.5 --- .../meshbay-node/src/meshbay_node/enrichment.py | 298 +++++++++++++++++++++ 1 file changed, 298 insertions(+) create mode 100644 packages/meshbay-node/src/meshbay_node/enrichment.py (limited to 'packages/meshbay-node/src/meshbay_node/enrichment.py') diff --git a/packages/meshbay-node/src/meshbay_node/enrichment.py b/packages/meshbay-node/src/meshbay_node/enrichment.py new file mode 100644 index 0000000..66b69b9 --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/enrichment.py @@ -0,0 +1,298 @@ +"""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) -- cgit v1.2.3