mopidy-smartplaylists/mopidy_smartplaylists/library.py

import logging
from mopidy.backend import LibraryProvider
from mopidy.models import Ref

logger = logging.getLogger(__name__)

class SmartLibrary(LibraryProvider):
    def __init__(self, backend, config):
        super().__init__(backend)
        self.config = config
        self.max_tracks = int(config.get("max_tracks") or 50)
        raw_genres = config.get("genres") or ""
        self.fixed_genres = [g.strip() for g in raw_genres.split(",") if g.strip()]
        self.case_insensitive = bool(config.get("case_insensitive"))

    # Browsing: return Ref objects for directories/tracks
    def browse(self, uri):
        if uri in (None, "smart:", "smart://"):
            # top-level
            return [Ref.directory(uri="smart:genres", name="Genres")]

        if uri == "smart:genres":
            return [
                Ref.directory(uri=f"smart:genre:{g}", name=g.title())
                for g in self.fixed_genres
            ]

        if uri.startswith("smart:genre:"):
            genre = uri.split(":", 2)[2]
            tracks = self._tracks_for_genre(genre)
            return [Ref.track(uri=t.uri, name=t.name) for t in tracks]

        return []

    # Lookup: return full Track models for playback
    def lookup(self, uri):
        if uri.startswith("smart:genre:"):
            genre = uri.split(":", 2)[2]
            return self._tracks_for_genre(genre)
        return []

    # Internal: perform the search via Mopidy core/library API
    def _search_core(self, query):
        """Return list-like of SearchResult objects or [] if unavailable."""
        # Preferred: self.backend.core (available when Mopidy injects core)
        try:
            core = getattr(self.backend, "core", None)
            if core is not None:
                return core.library.search(query)
        except Exception:
            logger.debug("backend.core.library.search failed, falling back to actor RPC", exc_info=True)

        # Fallback: actor RPC (synchronous)
        try:
            actor = getattr(self.backend, "actor", None)
            if actor is not None and hasattr(actor, "root"):
                return actor.root.actor.library.search(query)
        except Exception:
            logger.exception("actor RPC search failed")

        return []

    def _tracks_for_genre(self, genre):
        qval = genre
        if self.case_insensitive:
            # Some backends store lowercase genres; do exact matching here but allow fallback
            qval = genre.lower()

        # Primary query: genre exact
        query = {"genre": [genre]}

        tracks = []
        try:
            results = self._search_core(query)
            for res in results:
                # res may be a SearchResult-like object (has .tracks)
                if hasattr(res, "tracks"):
                    for t in res.tracks:
                        tracks.append(t)
                        if len(tracks) >= self.max_tracks:
                            break
                if len(tracks) >= self.max_tracks:
                    break

            # If case_insensitive and not enough results, try lowercase variant
            if self.case_insensitive and len(tracks) < self.max_tracks:
                query2 = {"genre": [qval]}
                if query2 != query:
                    results2 = self._search_core(query2)
                    for res in results2:
                        if hasattr(res, "tracks"):
                            for t in res.tracks:
                                if t not in tracks:
                                    tracks.append(t)
                                    if len(tracks) >= self.max_tracks:
                                        break
                        if len(tracks) >= self.max_tracks:
                            break
        except Exception:
            logger.exception("Failed to search core for genre %s", genre)

        return tracks[: self.max_tracks]