feat: implement librespot-powered search

This commit is contained in:
Xoconoch
2025-08-26 22:09:07 -06:00
parent c38f10957c
commit 6c795d8d92
3 changed files with 76 additions and 3 deletions

View File

@@ -482,9 +482,22 @@ class Spo:
@classmethod
def search(cls, query, search_type='track', limit=10, country: Optional[str] = None, locale: Optional[str] = None, catalogue: Optional[str] = None, image_size: Optional[str] = None, client_id=None, client_secret=None):
cls.__check_initialized()
# Map simple type value; librespot returns a combined JSON-like response
# Preferred path: use LibrespotClient for consistent defaults and options
if cls.__client is not None:
res = cls.__client.search(
query=query,
limit=limit,
country=country or cls.__get_session_country_code(),
locale=locale,
catalogue=catalogue,
image_size=image_size,
)
# Optionally filter by type if requested (best-effort; librespot returns mixed)
if search_type and isinstance(res, dict) and search_type in res:
return {search_type: res.get(search_type)}
return res
# Fallback: direct SearchManager
req = SearchManager.SearchRequest(query).set_limit(limit)
# Country precedence: explicit country > session country
if country:
req.set_country(country)
else:
@@ -498,4 +511,6 @@ class Spo:
if image_size:
req.set_image_size(image_size)
res = cls.__session.search().request(req) # type: ignore[union-attr]
if search_type and isinstance(res, dict) and search_type in res:
return {search_type: res.get(search_type)}
return res

View File

@@ -8,7 +8,7 @@ from typing import Any, Dict, List, Optional, Union
from google.protobuf.descriptor import FieldDescriptor
from google.protobuf.message import Message
from librespot.core import Session
from librespot.core import Session, SearchManager
from librespot.metadata import AlbumId, ArtistId, PlaylistId, TrackId
from librespot import util
from librespot.proto import Metadata_pb2 as Metadata
@@ -68,6 +68,34 @@ class LibrespotClient:
playlist_proto = self._session.api().get_playlist(playlist_id)
return self._playlist_proto_to_object(playlist_proto, include_track_objects=expand_items)
def search(
self,
query: str,
limit: int = 10,
country: Optional[str] = None,
locale: Optional[str] = None,
catalogue: Optional[str] = None,
image_size: Optional[str] = None,
) -> Dict[str, Any]:
"""Perform a full-featured search using librespot's SearchManager.
- country precedence: explicit country > session country code > unset
- returns the raw JSON-like mapping response provided by librespot
"""
req = SearchManager.SearchRequest(query).set_limit(limit)
# Country precedence
cc = country or self._get_session_country_code()
if cc:
req.set_country(cc)
if locale:
req.set_locale(locale)
if catalogue:
req.set_catalogue(catalogue)
if image_size:
req.set_image_size(image_size)
res = self._session.search().request(req)
return res
# ---------- ID parsing helpers ----------
@staticmethod
@@ -118,6 +146,18 @@ class LibrespotClient:
builder.stored_file(stored_credentials_path)
return builder.create()
def _get_session_country_code(self) -> str:
try:
cc = getattr(self._session, "_Session__country_code", None)
if isinstance(cc, str) and len(cc) == 2:
return cc
cc2 = getattr(self._session, "country_code", None)
if isinstance(cc2, str) and len(cc2) == 2:
return cc2
except Exception:
pass
return ""
# ---------- Private: ID coercion ----------
def _ensure_track_id(self, v: Union[str, TrackId]) -> TrackId: