import asyncio from apollo_utils.core.utils.dispatchers.dsp_dispatcher import dsp_dispatch from http import HTTPStatus from typing import Dict, Iterable, List from server.client import services from server.constants import DSP, SPOTIFY_PLAYLIST_IMAGE_URL_MASK, SPOTIFY_PLAYLIST_URI_PREFIX from server.constants.common import SortOrder from server.constants.playlists.v0.misc import ElasticPlaylistsVendorsData from server.utils.common import prefix_str DSP_NEEDS_SORTING = (DSP.AMAZON.value,) DSP_NEEDS_ADDITIONAL_ITEMS = (DSP.SPOTIFY.value,) ADDITIONAL_ITEMS = 5 def get_items_size(size: int, dsp: DSP): return size if dsp.value not in DSP_NEEDS_ADDITIONAL_ITEMS else size + ADDITIONAL_ITEMS class ElasticsearchQuery: """Builds and returns Elasticsearch (ES) request body for playlists search functionality. Attr: query (str): Search string. field_name (str): ES source field for querying in. size (int): Num of items in ES response. """ factor_field = ElasticPlaylistsVendorsData.FACTOR_FIELDS sorting_field = ElasticPlaylistsVendorsData.SORTING_FIELDS def __init__(self, dsp: DSP, query: str, field_name: str = "name", items_size: int = 30): self.dsp = dsp self.query = query.split() self.field_name = field_name self.items_size = items_size def _get_wildcard_query(self): result = { "query_string": { "query": f"{'* AND '.join(self.query)}*", "default_field": self.field_name, "boost": 2, "analyze_wildcard": True, } } return result def _get_terms_query(self): result = { "query_string": {"query": f"{' AND '.join(self.query)}", "default_field": self.field_name, "boost": 10} } return result def _get_fuzzy_query(self): result = { "query_string": { "query": f"{'~ AND '.join(self.query)}~", "default_field": self.field_name, "fuzziness": "auto", } } return result def _get_fuzzy_wildcard_query(self): result = { # fuzzy wildcard query "query_string": { "query": f"{'*~ AND '.join(self.query)}*~", "default_field": self.field_name, "fuzziness": "auto", } } return result def _get_fuzzy_wildcard_trailing_query(self): result = { "query_string": { "query": f"{'~ AND '.join(self.query)}*~", "default_field": self.field_name, "fuzziness": "auto", } } return result def get_playlists_search_body_query(self): should_query_list = [ self._get_wildcard_query(), self._get_terms_query(), self._get_fuzzy_query(), self._get_fuzzy_wildcard_query(), ] if len(self.query) > 1: should_query_list.append(self._get_fuzzy_wildcard_trailing_query()) query = { "query": { "function_score": { "query": {"bool": {"should": should_query_list}}, "field_value_factor": {"field": self.factor_field[self.dsp], "modifier": "log1p"}, } }, "size": self.items_size, } if self.dsp.value in DSP_NEEDS_SORTING: query["sort"] = [{self.sorting_field[self.dsp]: {"order": SortOrder.ASC.value}}] return query async def get_spotify_elasticsearch_playlists_search_response( dsp: DSP, data: List[Dict], items_size: int, *args, **kwargs ) -> Iterable[Dict]: """ Get spotify elasticsearch playlists search response by provided data Args: dsp (DSP): DSP.SPOTIFY data (List[Dict]): playlists data from elasticsearch database items_size: count of items to return Return: Iterable[Dict] """ preliminary_playlists, playlists_ids, result = {}, [], [] for item in data: item_source = item.get("_source") if not item_source: continue pl_id = item_source["playlistUri"].replace(SPOTIFY_PLAYLIST_URI_PREFIX, "") playlists_ids.append(pl_id) preliminary_playlists[pl_id] = { "owner": item_source.get("userDisplayName") or item_source.get("user"), "image_url": SPOTIFY_PLAYLIST_IMAGE_URL_MASK.format(playlist_id=pl_id) if pl_id else None, "id": pl_id, "country_code": item_source.get("countryCode"), "personalized": False, "vendor": dsp.value, "followers": item_source.get("followers"), "name": item_source.get("name"), } tasks = [ services.dsp.get_public_playlists_info( playlist_id=prefix_str(v=pl, prefix=dsp.value, delimiter="_"), allowed_status_codes=(HTTPStatus.NOT_FOUND,) ) for pl in playlists_ids ] playlists_data = await asyncio.gather(*tasks) existing_playlists_mapping = { item["dsp_playlist_id"]: item for item in playlists_data if item.get("dsp_playlist_id") } for key, value in preliminary_playlists.items(): if key in existing_playlists_mapping.keys(): value["personalized"] = existing_playlists_mapping[key]["is_personalised"] result.append(value) return result[: items_size - ADDITIONAL_ITEMS] async def get_apple_elasticsearch_playlists_search_response( dsp: DSP, data: List[Dict], image_size: int, *args, **kwargs ) -> List[Dict]: """ Get apple elasticsearch playlists search response by provided data Args: dsp (DSP): DSP.APPLE data (List[Dict]): playlists data from elasticsearch database image_size (int): Track image height/width value. Return: List[Dict] """ result = [] for item in data: item_source = item.get("_source") if not item_source: continue image = item_source.get("artworkUrl") result.append( { "owner": item_source.get("curator"), "image_url": image.format(w=image_size, h=image_size) if image else None, "id": item_source["playlistId"], "country_code": item_source.get("countryCode"), "vendor": dsp.value, "name": item_source.get("name"), } ) return result async def get_amazon_elasticsearch_playlists_search_response(dsp: DSP, data: List[Dict], *args, **kwargs): """ Get amazon elasticsearch playlists search response by provided data Args: dsp (DSP): DSP.APPLE data (List[Dict]): playlists data from elasticsearch database Return: List[Dict] """ result = [] for item in data: item_source = item.get("_source", {}) amazon_pl_id = item_source.get("amazonPlaylistId") if not item_source or not amazon_pl_id: continue result.append( { "image_url": item_source.get("imageUri"), "id": f"{amazon_pl_id}:{item_source['playlistId'].replace('amazon_', '')}", "country_code": item_source.get("countryCode"), "vendor": dsp.value, "name": item_source.get("name"), } ) return result @dsp_dispatch( (DSP.SPOTIFY, get_spotify_elasticsearch_playlists_search_response), (DSP.APPLE, get_apple_elasticsearch_playlists_search_response), (DSP.AMAZON, get_amazon_elasticsearch_playlists_search_response), ) async def get_dsp_elasticsearch_response(dsp: DSP, data: List[Dict], items_size: int, image_size: int = None): """ Get elasticsearch playlists search response by provided data and dsp (V1) Args: dsp (DSP): DSP.APPLE or DSP.SPOTIFY data (List[Dict]): playlists data from elasticsearch database items_size: count of items to request image_size (int): Track image height/width value. Return: Iterable[Dict] """ raise NotImplemented("Unknown dsp")