from typing import Dict, Any, Mapping, Iterable from server.artist.models import ArtistSearchData from server.core.helpers.discovery import DiscoveryHelper from server.artist.utils import ImageService from server.db.constants import ARTIST_SEARCH_INDEX, DSP_TIKTOK, DSP_YOUTUBE, DSP_SPOTIFY, DSP_SOUNDCLOUD, ALL_DSPS __all__ = ["ArtistDiscoveryHelper"] class _ArtistDiscoveryHelper(DiscoveryHelper): _index_name: str = ARTIST_SEARCH_INDEX _data_model = ArtistSearchData _dsp_sort_mapping: Mapping[str, str] = { DSP_TIKTOK: "followers", DSP_SPOTIFY: "listeners", DSP_YOUTUBE: "subscribers", DSP_SOUNDCLOUD: "followers", } _dsp_default_charts: Mapping[str, str] = { DSP_TIKTOK: "followers", DSP_YOUTUBE: "subscribers", DSP_SPOTIFY: "followers", DSP_SOUNDCLOUD: "followers", } def _get_sort(self, params: Dict[str, Any]) -> Dict[str, Any]: source = {} sort = {} for dsp in ALL_DSPS: if dsp in params.keys(): source[dsp] = f"{dsp}.{self._dsp_sort_mapping.get(dsp, {})}_change" if source.get(DSP_TIKTOK): sort = {f"{source[DSP_TIKTOK]}": "desc"} elif source.get(DSP_SPOTIFY): sort = {f"{source[DSP_SPOTIFY]}": "desc"} elif source.get(DSP_YOUTUBE): sort = {f"{source[DSP_YOUTUBE]}": "desc"} elif source.get(DSP_SOUNDCLOUD): sort = {f"{source[DSP_SOUNDCLOUD]}": "desc"} return sort def _hydrate_search(self, *args, **kwargs) -> Iterable[Dict[str, Any]]: for row in super()._hydrate_search(*args, **kwargs): spotify_artist_id = row.pop("spotify_artist_id") row["image_url"] = ImageService.get_img_url(spotify_artist_id) row["artwork_url_alt"] = row.get("artwork_url") row["artwork_url"] = ImageService.get_img_url(spotify_artist_id) yield row def _hydrate_get(self, *args, **kwargs) -> Dict[str, Any]: result = super()._hydrate_get(*args, **kwargs) spotify_artist_id = result.pop("spotify_artist_id") result["image_url"] = ImageService.get_img_url(spotify_artist_id) result["artwork_url_alt"] = result.get("artwork_url") result["artwork_url"] = ImageService.get_img_url(spotify_artist_id) return result ArtistDiscoveryHelper = _ArtistDiscoveryHelper()