path: root/packages/meshbay-node/src/meshbay_node/enrichment.py

"""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)