"""The library-provider registry — the plugin extension point for song sources. `LocalLibraryProvider` wraps the local `MetadataDB`; third-party plugins register their own providers (duck-typed: any object with the advertised methods) through `LibraryProviderRegistry`, and smart collections are surfaced as `SmartCollectionProvider`s over the local one. server.py constructs the singleton (`library_providers`), injects it + the local provider into appstate, and exposes `register_library_provider`/`unregister_library_provider` to plugins via plugin_context (with per-plugin ownership scoping in plugins/__init__.py). Moved verbatim out of server.py (R3). The shared query/collection helpers live here too so routers/library.py can import them without reaching into server. """ import re import threading from typing import ClassVar import appstate from metadata_db import ( MetadataDB, _effective_tuning_cols_sql, _perspective_is_inferred_sql, _tuning_group_key_sql, ) import tunings as tunings_mod from tunings import DEFAULT_PERSPECTIVE, PERSPECTIVES from routers import art as art_router import logging log = logging.getLogger("feedBack.server") def _safe_art_redirect_url(url: str) -> str | None: """Return the URL if it is safe to redirect to (http/https only), else None.""" from urllib.parse import urlparse if not url or not isinstance(url, str): return None try: parsed = urlparse(url) if parsed.scheme.lower() not in ("http", "https"): return None if not parsed.hostname: return None return url except Exception: return None class LocalLibraryProvider: id = "local" label = "My Library" kind = "local" capabilities = ( "library.read", "art.read", "song.play", "favorite.write", "metadata.write", ) def __init__(self, db: MetadataDB): self._db = db def query_page(self, **kwargs) -> tuple[list[dict], int]: return self._db.query_page(**kwargs) def query_artists(self, **kwargs) -> tuple[list[dict], int]: return self._db.query_artists(**kwargs) def query_albums(self, **kwargs) -> tuple[list[dict], int]: return self._db.query_albums(**kwargs) def query_stats(self, **kwargs) -> dict: return self._db.query_stats(**kwargs) def tuning_names(self, instrument: str = DEFAULT_PERSPECTIVE) -> dict: # Group custom tunings on their raw offsets so distinct ones stay # distinct (tuning_name collapses them all to "Custom Tuning"); named # tunings keep grouping by name (stable across the rescan boundary, no # offsets/name split). `key` is the value the client sends back as the # filter selector — equal to the name for named tunings, the offsets # string for customs; offsets also feed the client's custom-pill label. # # `instrument=bass` swaps every column for its effective bass-facing # expression (bass arrangement's tuning, guitar fallback) — the SAME # expressions _build_intrinsic_where filters on, so a facet entry # always selects exactly the songs it counted. name_sql, offsets_sql, sort_sql = _effective_tuning_cols_sql("songs", instrument) gkey_sql = _tuning_group_key_sql("songs", instrument) # How many of a row's songs are showing an INFERRED tuning — i.e. have # no bass chart of their own and are falling back to the guitar-derived # one. Reported per entry so the UI can be honest about it instead of # presenting a borrowed tuning as a measured one. Always 0 for guitar. inferred_sql = f"SUM({_perspective_is_inferred_sql('songs', instrument)})" with self._db._lock: rows = self._db.conn.execute( f"SELECT {name_sql}, {gkey_sql} AS gkey, " f"MIN({sort_sql}), COUNT(*), MIN({offsets_sql}), {inferred_sql} " f"FROM songs WHERE title != '' AND COALESCE({name_sql}, '') != '' " "GROUP BY gkey COLLATE NOCASE " f"ORDER BY ABS(COALESCE(MIN({sort_sql}), 0)), " f"COALESCE(MIN({sort_sql}), 0) ASC, " f"{name_sql} COLLATE NOCASE" ).fetchall() return { "instrument": instrument, "tunings": [ {"name": name, "key": gkey, "offsets": offs or "", "sort_key": int(sk or 0), "count": count, # Portion of `count` borrowed from the guitar chart. "inferred_count": int(inferred or 0)} for name, gkey, sk, count, offs, inferred in rows ], } async def get_art(self, song_id: str): return await art_router.get_song_art(song_id) class LibraryProviderRegistry: # Methods required per declared capability — only validated when the # provider advertises the corresponding capability so action-only providers # (e.g. art.read + song.sync without library.read) don't need to implement # unused stubs. _CAPABILITY_METHODS: ClassVar[dict[str, tuple[str, ...]]] = { "library.read": ("query_page", "query_artists", "query_stats", "tuning_names"), "art.read": ("get_art",), "song.sync": ("sync_song",), } _ID_RE: ClassVar[re.Pattern[str]] = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{0,127}$") def __init__(self): self._providers: dict[str, object] = {} # Capabilities inferred at registration for legacy providers that omit # the `capabilities` field. Merged with provider_capabilities() so that # runtime capability checks see the complete effective capability set. self._inferred_caps: dict[str, set[str]] = {} self._owner_plugin_ids: dict[str, str] = {} self._lock = threading.RLock() def register(self, provider: object, *, replace: bool = False, owner_plugin_id: str | None = None) -> object: provider_id = self.provider_id(provider) if not self._ID_RE.match(provider_id): raise ValueError( "library provider id must start with an alphanumeric character " "and contain only letters, digits, _, ., :, or -" ) if not self.provider_label(provider): raise ValueError("library provider label must be a non-empty string") # Use declared-only caps during validation — never include stale inferred # caps from a previous provider registered under the same id (replace=True). caps = self._declared_capabilities(provider) # Backward compatibility: providers that predate explicit capability # declarations may omit `capabilities` entirely. If the browse methods # are all present, infer `library.read` so they still work unchanged. # If capabilities are absent but the browse surface is also absent, # raise a clear error rather than letting the provider register and # then fail on every API call with a late 501. inferred: set[str] = set() if not caps: browse_methods = self._CAPABILITY_METHODS["library.read"] if all(callable(self.provider_method(provider, m)) for m in browse_methods): # Legacy provider without explicit capabilities — infer library.read # from the presence of all browse methods. Store in _inferred_caps # so that runtime capability checks see the full effective set. inferred = {"library.read"} caps = inferred else: raise TypeError( f"library provider {provider_id!r} must declare at least one capability " f"(or implement the {browse_methods!r} browse methods for backward compatibility)" ) for cap, methods in self._CAPABILITY_METHODS.items(): if cap not in caps: continue for method_name in methods: if not callable(self.provider_method(provider, method_name)): raise TypeError(f"library provider {provider_id!r} declares {cap!r} but is missing callable {method_name}()") with self._lock: if provider_id == "local" and provider_id in self._providers and self._providers[provider_id] is not provider: raise ValueError("the local library provider cannot be replaced") if provider_id in self._providers and not replace: raise ValueError(f"library provider {provider_id!r} is already registered") self._providers[provider_id] = provider # owner_plugin_id is attribution that flows into the browser # capability participant id. The scoped register_library_provider # wrappers force it to the trusted loading plugin id, so the spoof # vector is closed there. Here we only normalize: trim and require a # non-empty string. We deliberately do NOT apply the provider-id # grammar (_ID_RE) — plugin ids aren't constrained to it at load # time, so that would silently drop attribution for valid plugins. owner = owner_plugin_id.strip() if isinstance(owner_plugin_id, str) else "" owner = owner or None if owner: self._owner_plugin_ids[provider_id] = owner else: self._owner_plugin_ids.pop(provider_id, None) if inferred: self._inferred_caps[provider_id] = inferred else: self._inferred_caps.pop(provider_id, None) return provider def unregister(self, provider_id: str) -> bool: if provider_id == "local": raise ValueError("the local library provider cannot be unregistered") with self._lock: self._inferred_caps.pop(provider_id, None) self._owner_plugin_ids.pop(provider_id, None) return self._providers.pop(provider_id, None) is not None def get(self, provider_id: str = "local") -> object | None: with self._lock: return self._providers.get(provider_id or "local") def list(self) -> list[dict]: with self._lock: providers = list(self._providers.values()) return [self.describe(provider) for provider in providers] def describe(self, provider: object) -> dict: provider_id = self.provider_id(provider) with self._lock: owner_plugin_id = self._owner_plugin_ids.get(provider_id) return { "id": provider_id, "label": self.provider_label(provider), "kind": self.provider_field(provider, "kind", "local" if provider_id == "local" else "remote"), "capabilities": sorted(self.provider_capabilities(provider)), "owner_plugin_id": owner_plugin_id, "default": provider_id == "local", } def provider_field(self, provider: object, name: str, default=None): if isinstance(provider, dict): return provider.get(name, default) return getattr(provider, name, default) def provider_id(self, provider: object) -> str: provider_id = self.provider_field(provider, "id", "") if not isinstance(provider_id, str) or not provider_id: raise ValueError("library provider id must be a non-empty string") return provider_id def provider_label(self, provider: object) -> str: label = self.provider_field(provider, "label", self.provider_field(provider, "name", "")) if not isinstance(label, str): return "" return label.strip() def _declared_capabilities(self, provider: object) -> set[str]: """Return only the capabilities explicitly declared on the provider object.""" raw = self.provider_field(provider, "capabilities", ()) if raw is None: raw = () if isinstance(raw, str): raw = (raw,) if raw else () return {str(cap) for cap in raw if cap} def provider_capabilities(self, provider: object) -> set[str]: # Guard against a common plugin authoring mistake: passing a single string # instead of a list/tuple. Iterating a string produces individual characters, # none of which would match a valid capability name. declared = self._declared_capabilities(provider) # Merge with any capabilities inferred at registration time for legacy # providers that omit the `capabilities` field but implement browse methods. provider_id = self.provider_id(provider) with self._lock: inferred = self._inferred_caps.get(provider_id, set()) return declared | inferred def provider_method(self, provider: object, name: str): if isinstance(provider, dict): return provider.get(name) return getattr(provider, name, None) # Keys `_library_filter_args` (and a smart collection's stored `rules`) accept. _LIBRARY_FILTER_PARAM_KEYS = frozenset(( "q", "favorites", "format", "artist", "album", "arrangements_has", "arrangements_lacks", "stems_has", "stems_lacks", "has_lyrics", "tunings", )) # Rules mirror the raw /api/library query params (so the provider can feed them # straight through `_library_filter_args`, and the frontend can build a rule from # the same query string it already constructs). Multi-value filters are CSV # strings; `favorites` is 0/1; the rest are plain strings. _RULE_CSV_KEYS = frozenset(( "tunings", "arrangements_has", "arrangements_lacks", "stems_has", "stems_lacks", )) _RULE_STR_KEYS = frozenset(("q", "format", "artist", "album", "has_lyrics", "sort")) def _sanitize_collection_rules(raw) -> dict: """Normalize rules to the raw query-param format, keeping only known keys. A list for a multi-value filter is joined to CSV; `favorites` becomes 0/1. Unknown keys are dropped so a rule survives a filter-vocab change rather than 500-ing. Applied at API ingress AND when a provider loads a persisted row, so a hand-edited / imported bad value (e.g. an int where a string is expected, or a list for `sort`) can never crash a query.""" if not isinstance(raw, dict): return {} out: dict = {} for k, v in raw.items(): if k in _RULE_CSV_KEYS: if isinstance(v, list): vals = [str(x) for x in v if isinstance(x, (str, int)) and not isinstance(x, bool)] elif isinstance(v, str): vals = [s for s in (p.strip() for p in v.split(",")) if s] else: continue if vals: out[k] = ",".join(vals) elif k == "favorites": if v: out[k] = 1 elif k in _RULE_STR_KEYS: if isinstance(v, (str, int)) and not isinstance(v, bool): s = str(v).strip() if s: out[k] = s return out class SmartCollectionProvider: """A saved library filter, surfaced as a source (#636 item 2). Browse/stats delegate to the local DB with the collection's stored `rules` applied — so selecting it in the v3 source picker shows exactly that filtered slice with the whole Songs UI (paging, stats, A–Z rail, art) for free. P1: the rules ARE the query (live in-collection search is a P2 nicety). The matched songs are local rows, so `kind="local"` keeps the client's play/art paths on the local (not remote-sync) branch and art delegates straight through.""" kind = "local" capabilities = ("library.read", "art.read") def __init__(self, collection: dict, local: "LocalLibraryProvider"): self._local = local self.update(collection) def update(self, collection: dict) -> None: self.id = f"collection:{collection['id']}" self.collection_id = collection["id"] self.label = collection.get("name") or "Collection" # Re-sanitize on load: persisted JSON may predate the current vocab or # have been hand-edited; never let a bad value reach a query. self._rules = _sanitize_collection_rules(collection.get("rules") or {}) def _filter_kwargs(self, instrument: str = "", playable_from_pitch=None) -> dict: # `instrument` is the CALLER's play perspective (rides every request), # never part of the saved rules — a collection saved by a guitarist # must still read in bass tunings for a bass player, and vice versa. args = _library_filter_args(**{k: v for k, v in self._rules.items() if k in _LIBRARY_FILTER_PARAM_KEYS}) args["instrument"] = _normalize_instrument(instrument) # The caller's CURRENT tuning is likewise per-request, never a saved rule. args["playable_from_pitch"] = playable_from_pitch return args def _sort(self, fallback: str) -> str: # A collection may pin its own sort (e.g. "recently added"); query_page # falls back safely for an unknown value, so no validation needed here. return self._rules.get("sort") or fallback def query_page(self, *, page=0, size=24, sort="artist", direction="asc", naming_mode="legacy", instrument="", playable_from_pitch=None, **_ignore): return self._local._db.query_page( page=page, size=size, sort=self._sort(sort), direction=direction, naming_mode=naming_mode, **self._filter_kwargs(instrument, playable_from_pitch)) def query_artists(self, *, letter="", page=0, size=50, naming_mode="legacy", instrument="", playable_from_pitch=None, **_ignore): return self._local._db.query_artists( letter=letter, page=page, size=size, naming_mode=naming_mode, **self._filter_kwargs(instrument, playable_from_pitch)) def query_albums(self, *, page=0, size=120, naming_mode="legacy", instrument="", playable_from_pitch=None, **_ignore): return self._local._db.query_albums( page=page, size=size, naming_mode=naming_mode, **self._filter_kwargs(instrument, playable_from_pitch)) def query_stats(self, *, sort="artist", want_sort_letters=False, naming_mode="legacy", instrument="", playable_from_pitch=None, **_ignore): return self._local._db.query_stats( sort=self._sort(sort), want_sort_letters=want_sort_letters, naming_mode=naming_mode, **self._filter_kwargs(instrument, playable_from_pitch)) def tuning_names(self, instrument: str = "guitar"): return self._local.tuning_names(instrument=_normalize_instrument(instrument)) async def get_art(self, song_id: str): return await self._local.get_art(song_id) def _split_csv(raw: str) -> list[str]: """Parse a comma-separated query-string list. Empty / whitespace-only entries are dropped so `arrangements_has=` (no value) and `arrangements_has=,` both mean 'no filter'.""" if not raw: return [] return [s.strip() for s in raw.split(",") if s.strip()] def _parse_has_lyrics(raw: str) -> int | None: """Tri-state parse for has_lyrics. `1` → require, `0` → exclude, anything else (including empty) → no filter.""" if raw == "1": return 1 if raw == "0": return 0 return None def _library_filter_args(q: str = "", favorites: int = 0, format: str = "", artist: str = "", album: str = "", arrangements_has: str = "", arrangements_lacks: str = "", stems_has: str = "", stems_lacks: str = "", has_lyrics: str = "", tunings: str = "", instrument: str = "", tuning_match: str = "", playable_offsets: str = "", playable_instrument: str = "", playable_string_count: str = "") -> dict: fmt = format if format in ("archive", "sloppak", "loose") else "" return { "q": q, "favorites_only": bool(favorites), "format_filter": fmt, "artist_filter": (artist or "").strip(), "album_filter": (album or "").strip(), "arrangements_has": _split_csv(arrangements_has), "arrangements_lacks": _split_csv(arrangements_lacks), "stems_has": _split_csv(stems_has), "stems_lacks": _split_csv(stems_lacks), "has_lyrics": _parse_has_lyrics(has_lyrics), "tunings": _split_csv(tunings), # Which perspective the tuning facet/filter/sort speaks for (the # caller's play role, NOT a saved rule — see _sanitize_collection_rules). "instrument": _normalize_instrument(instrument), # "Playable without retuning" mode: the caller's CURRENT tuning, # resolved to the one number the comparison needs. None = exact-match # mode (the default), so the tuning pills behave exactly as before. "playable_from_pitch": ( _playable_from_pitch(playable_offsets, playable_instrument, playable_string_count) if tuning_match == "playable" else None), } def _playable_from_pitch(offsets_csv: str, instrument: str, string_count: str): """Lowest open-string MIDI pitch of the CALLER's current tuning. The client sends its live working tuning (offsets + instrument + string count) rather than a precomputed pitch, so the pitch tables stay in one place (lib/tunings.py) instead of being duplicated in JS. Returns None for anything unusable — the caller then applies NO playable filter at all. That is the neutral state, not a claim: a malformed tuning must not silently assert that everything is playable OR that nothing is. """ try: offsets = [int(x) for x in _split_csv(offsets_csv)] except (TypeError, ValueError): return None if not offsets: return None inst = "bass" if instrument == "bass" else "guitar" try: sc = int(string_count) except (TypeError, ValueError): sc = len(offsets) key = tunings_mod.instrument_key(inst, sc) if key not in tunings_mod.STANDARD_OPEN_MIDIS or len(offsets) != sc: return None midis = tunings_mod.tuning_midis_from_offsets(key, offsets) return min(midis) if midis else None def _normalize_instrument(raw: str) -> str: """Resolve a tuning PERSPECTIVE id (guitar-lead | guitar-rhythm | bass). Tolerates the legacy two-valued vocabulary ("guitar" -> guitar-lead) and falls back to the default for anything unknown — an unrecognised value must never silently change filter semantics.""" return raw if raw in PERSPECTIVES else ( DEFAULT_PERSPECTIVE if raw != "bass" else "bass") def _sync_collection_provider(collection: dict) -> None: """Register (or replace) the provider for one collection.""" appstate.library_providers.register( SmartCollectionProvider(collection, appstate.local_library_provider), replace=True) def _unregister_collection_provider(pid: int) -> None: appstate.library_providers.unregister(f"collection:{pid}")