from typing import Dict, List, Union from server.client import services from server.constants import DSP from server.scenarios.tracks.get_tracks import get_tracks from server.utils.isrc_relation import isrc_to_id_mapper async def get_isrc_to_id_map(isrc_list: List[str], market: str) -> Dict[str, Union[str, int]]: """ Return isrc to track_id dictionary for particular market and some isrc. Searching priorities: Apollo database, Spotify API, Apple Music API. """ isrc_upper_set = {isrc.upper() for isrc in set(isrc_list)} apollo_result = await services.apollo.get_isrc_to_id_relation(isrc_list=list(isrc_upper_set)) # to avoid duplicates in upper/down cases results = {key.upper(): value for key, value in apollo_result.items()} spotify_id_data = isrc_to_id_mapper(await get_tracks(track_ids=results.values(), dsp=DSP.SPOTIFY, market=market)) results.update(spotify_id_data) apple_isrc_data = isrc_to_id_mapper( await get_tracks(track_ids=isrc_upper_set - set(results.keys()), dsp=DSP.APPLE, market=market, by_isrc=True) ) results.update(apple_isrc_data) return results async def get_isrc_relation(market: str or None = None, isrc: str or None = None): """ Get ISRC(s) relation(s) for specific market Args: market: Market country code isrc: ISRC Returns: List[Dict] with main isrc and track_id and related isrc(s) """ params = {"market": market, "isrc": isrc} isrc_to_id = {} isrc_relation_data = await services.apollo.get_v1_isrc_relation(**params) if not isrc_relation_data: return [{}] if market: isrc_to_id = await get_isrc_to_id_map(isrc_relation_data.keys(), market) result = [{"main": {"isrc": k, "track_id": isrc_to_id.get(k)}, "related": v} for k, v in isrc_relation_data.items()] return result