import base64 import json import logging import re import ffmpeg from datetime import datetime from getpass import getpass from dataclasses import dataclass from shutil import copyfileobj from xml.etree import ElementTree from tqdm import tqdm from utils.models import * from utils.utils import sanitise_name, silentremove, download_to_temp, create_temp_filename, create_requests_session from .mqa_identifier_python.mqa_identifier_python.mqa_identifier import MqaIdentifier from .tidal_api import TidalTvSession, TidalApi, TidalMobileSession, SessionType, TidalError, TidalRequestError module_information = ModuleInformation( service_name='TIDAL', module_supported_modes=ModuleModes.download | ModuleModes.credits | ModuleModes.covers | ModuleModes.lyrics, login_behaviour=ManualEnum.manual, global_settings={ 'tv_token': '7m7Ap0JC9j1cOM3n', 'tv_secret': 'vRAdA108tlvkJpTsGZS8rGZ7xTlbJ0qaZ2K9saEzsgY=', 'mobile_atmos_token': 'dN2N95wCyEBTllu4', 'mobile_default_token': 'WAU9gXp3tHhK4Nns', 'enable_mobile': True, 'force_non_spatial': False, 'prefer_ac4': False, 'fix_mqa': True }, session_storage_variables=['sessions'], netlocation_constant='tidal', test_url='https://tidal.com/browse/track/92265335' ) @dataclass class AudioTrack: codec: CodecEnum sample_rate: int bitrate: int urls: list class ModuleInterface: # noinspection PyTypeChecker def __init__(self, module_controller: ModuleController): self.cover_size = module_controller.orpheus_options.default_cover_options.resolution self.oprinter = module_controller.printer_controller self.print = module_controller.printer_controller.oprint self.disable_subscription_check = module_controller.orpheus_options.disable_subscription_check self.settings = module_controller.module_settings # LOW = 96kbit/s AAC, HIGH = 320kbit/s AAC, LOSSLESS = 44.1/16 FLAC, HI_RES <= 48/24 FLAC with MQA self.quality_parse = { QualityEnum.MINIMUM: 'LOW', QualityEnum.LOW: 'LOW', QualityEnum.MEDIUM: 'HIGH', QualityEnum.HIGH: 'HIGH', QualityEnum.LOSSLESS: 'LOSSLESS', QualityEnum.HIFI: 'HI_RES' } # save all the TidalSession objects sessions = {} self.available_sessions = [SessionType.TV.name, SessionType.MOBILE_DEFAULT.name, SessionType.MOBILE_ATMOS.name] # load all saved sessions (TV, Mobile Atmos, Mobile Default) saved_sessions = module_controller.temporary_settings_controller.read('sessions') if not saved_sessions: saved_sessions = {} if self.settings['enable_mobile']: # check all saved session for a session starting with "MOBILE" if not any(session for session in saved_sessions.keys() if session[:6] == 'MOBILE'): confirm = input(' "enable_mobile" is enabled but no MOBILE session was found. Do you want to create a ' 'MOBILE session (used for AC-4/360RA) [Y/n]? ') if confirm.upper() == 'N': self.available_sessions = [SessionType.TV.name] else: self.available_sessions = [SessionType.TV.name] username, password = None, None for session_type in self.available_sessions: # create all sessions with the needed API keys if session_type == SessionType.TV.name: sessions[session_type] = TidalTvSession(self.settings['tv_token'], self.settings['tv_secret']) elif session_type == SessionType.MOBILE_ATMOS.name: sessions[session_type] = TidalMobileSession(self.settings['mobile_atmos_token']) else: sessions[session_type] = TidalMobileSession(self.settings['mobile_default_token']) if session_type in saved_sessions: logging.debug(f'{module_information.service_name}: {session_type} session found, loading') # load the dictionary from the temporary_settings_controller inside the TidalSession class sessions[session_type].set_storage(saved_sessions[session_type]) else: logging.debug(f'{module_information.service_name}: No {session_type} session found, creating new one') if session_type == SessionType.TV.name: self.print(f'{module_information.service_name}: Creating a TV session') sessions[session_type].auth() else: if not username or not password: self.print(f'{module_information.service_name}: Creating a Mobile session') self.print(f'{module_information.service_name}: Enter your Tidal username and password:') username = input(' Username: ') password = getpass(' Password: ') sessions[session_type].auth(username, password) self.print(f'Successfully logged in, using {session_type} token!') # get the dict representation from the TidalSession object and save it into saved_session/loginstorage saved_sessions[session_type] = sessions[session_type].get_storage() module_controller.temporary_settings_controller.set('sessions', saved_sessions) # always try to refresh session if not sessions[session_type].valid(): sessions[session_type].refresh() # Save the refreshed session in the temporary settings saved_sessions[session_type] = sessions[session_type].get_storage() module_controller.temporary_settings_controller.set('sessions', saved_sessions) while True: # check for a valid subscription subscription = self.check_subscription(sessions[session_type].get_subscription()) if subscription: break confirm = input(' Do you want to create a new session? [Y/n]: ') if confirm.upper() == 'N': self.print('Exiting...') exit() # create a new session finally if session_type == SessionType.TV.name: self.print(f'{module_information.service_name}: Recreating a TV session') sessions[session_type].auth() else: self.print(f'{module_information.service_name}: Recreating a Mobile session') self.print(f'{module_information.service_name}: Enter your Tidal username and password:') username = input('Username: ') password = getpass('Password: ') sessions[session_type].auth(username, password) saved_sessions[session_type] = sessions[session_type].get_storage() module_controller.temporary_settings_controller.set('sessions', saved_sessions) # reset username and password username, password = None, None # only needed for region locked albums where the track is available but force_album_format is used self.album_cache = {} # load the Tidal session with all saved sessions (TV, Mobile Atmos, Mobile Default) self.session: TidalApi = TidalApi(sessions) def check_subscription(self, subscription: str) -> bool: # returns true if "disable_subscription_checks" is enabled or subscription is HIFI Plus if not self.disable_subscription_check and subscription not in {'HIFI', 'PREMIUM_PLUS'}: self.print(f'{module_information.service_name}: Account is not a HiFi Plus account, ' f'detected subscription: {subscription}') return False return True @staticmethod def generate_artwork_url(cover_id: str, size: int, max_size: int = 1280): # not the best idea, but it rounds the self.cover_size to the nearest number in supported_sizes, 1281 is needed # for the "uncompressed" cover supported_sizes = [80, 160, 320, 480, 640, 1080, 1280, 1281] best_size = min(supported_sizes, key=lambda x: abs(x - size)) # only supports 80x80, 160x160, 320x320, 480x480, 640x640, 1080x1080 and 1280x1280 only for non playlists # return "uncompressed" cover if self.cover_resolution > max_size image_name = '{0}x{0}.jpg'.format(best_size) if best_size <= max_size else 'origin.jpg' return f'https://resources.tidal.com/images/{cover_id.replace("-", "/")}/{image_name}' @staticmethod def generate_animated_artwork_url(cover_id: str, size=1280): return 'https://resources.tidal.com/videos/{0}/{1}x{1}.mp4'.format(cover_id.replace('-', '/'), size) def search(self, query_type: DownloadTypeEnum, query: str, track_info: TrackInfo = None, limit: int = 20): results = self.session.get_search_data(query, limit=limit) items = [] for i in results[query_type.name + 's'].get('items'): if query_type is DownloadTypeEnum.artist: name = i.get('name') artists = None year = None elif query_type is DownloadTypeEnum.playlist: if 'name' in i.get('creator'): artists = [i.get('creator').get('name')] elif i.get('type') == 'EDITORIAL': artists = [module_information.service_name] else: artists = ['Unknown'] # TODO: Use playlist creation date or lastUpdated? year = i.get('created')[:4] elif query_type is DownloadTypeEnum.track: artists = [j.get('name') for j in i.get('artists')] # Getting the year from the album? year = i.get('album').get('releaseDate')[:4] elif query_type is DownloadTypeEnum.album: artists = [j.get('name') for j in i.get('artists')] year = i.get('releaseDate')[:4] else: raise Exception('Query type is invalid') if query_type is not DownloadTypeEnum.artist: name = i.get('title') name += f' ({i.get("version")})' if i.get("version") else '' additional = None if query_type is not DownloadTypeEnum.artist: if 'DOLBY_ATMOS' in i.get('audioModes'): additional = "Dolby Atmos" elif 'SONY_360RA' in i.get('audioModes'): additional = "360 Reality Audio" elif i.get('audioQuality') == 'HI_RES': additional = "MQA" else: additional = 'HiFi' item = SearchResult( name=name, artists=artists, year=year, result_id=str(i.get('id')) if query_type is not DownloadTypeEnum.playlist else i.get('uuid'), explicit=i.get('explicit'), additional=[additional] if additional else None ) items.append(item) return items def get_playlist_info(self, playlist_id: str) -> PlaylistInfo: playlist_data = self.session.get_playlist(playlist_id) playlist_tracks = self.session.get_playlist_items(playlist_id) tracks = [track.get('item').get('id') for track in playlist_tracks.get('items') if track.get('type') == 'track'] if 'name' in playlist_data.get('creator'): creator_name = playlist_data.get('creator').get('name') elif playlist_data.get('type') == 'EDITORIAL': creator_name = module_information.service_name else: creator_name = 'Unknown' return PlaylistInfo( name=playlist_data.get('title'), creator=creator_name, tracks=tracks, release_year=playlist_data.get('created')[:4], creator_id=playlist_data['creator'].get('id'), cover_url=self.generate_artwork_url(playlist_data['squareImage'], size=self.cover_size, max_size=1080) if playlist_data['squareImage'] else None, track_extra_kwargs={ 'data': {track.get('item').get('id'): track.get('item') for track in playlist_tracks.get('items')} } ) def get_artist_info(self, artist_id: str, get_credited_albums: bool) -> ArtistInfo: artist_data = self.session.get_artist(artist_id) artist_albums = self.session.get_artist_albums(artist_id).get('items') artist_singles = self.session.get_artist_albums_ep_singles(artist_id).get('items') # Only works with a mobile session, annoying, never do this again credit_albums = [] if get_credited_albums and SessionType.MOBILE_DEFAULT.name in self.available_sessions: self.session.default = SessionType.MOBILE_DEFAULT credited_albums_page = self.session.get_page('contributor', params={'artistId': artist_id}) # This is so retarded page_list = credited_albums_page['rows'][-1]['modules'][0].get('pagedList') if page_list: total_items = page_list['totalNumberOfItems'] more_items_link = page_list['dataApiPath'][6:] # Now fetch all the found total_items items = [] for offset in range(0, total_items // 50 + 1): print(f'Fetching {offset * 50}/{total_items}', end='\r') items += self.session.get_page(more_items_link, params={'limit': 50, 'offset': offset * 50})['items'] credit_albums = [item.get('item').get('album') for item in items] self.session.default = SessionType.TV # use set to filter out duplicate album ids albums = {str(album.get('id')) for album in artist_albums + artist_singles + credit_albums} return ArtistInfo( name=artist_data.get('name'), albums=list(albums), album_extra_kwargs={'data': {str(album.get('id')): album for album in artist_albums + artist_singles}} ) def get_album_info(self, album_id: str, data=None) -> AlbumInfo: # check if album is already in album cache, add it if data is None: data = {} if data.get(album_id): album_data = data[album_id] elif self.album_cache.get(album_id): album_data = self.album_cache[album_id] else: album_data = self.session.get_album(album_id) # get all album tracks with corresponding credits with a limit of 100 limit = 100 cache = {'data': {}} try: tracks_data = self.session.get_album_contributors(album_id, limit=limit) total_tracks = tracks_data.get('totalNumberOfItems') # round total_tracks to the next 100 and loop over the offset, that's hideous for offset in range(limit, ((total_tracks // limit) + 1) * limit, limit): # fetch the new album tracks with the given offset track_items = self.session.get_album_contributors(album_id, offset=offset, limit=limit) # append those tracks to the album_data tracks_data['items'] += track_items.get('items') # add the track contributors to a new list called 'credits' cache = {'data': {}} for track in tracks_data.get('items'): track.get('item').update({'credits': track.get('credits')}) cache.get('data')[str(track.get('item').get('id'))] = track.get('item') # filter out video clips tracks = [str(track['item']['id']) for track in tracks_data.get('items') if track.get('type') == 'track'] except TidalError: tracks = [] quality = None if 'audioModes' in album_data: if album_data['audioModes'] == ['DOLBY_ATMOS']: quality = 'Dolby Atmos' elif album_data['audioModes'] == ['SONY_360RA']: quality = '360' elif album_data['audioQuality'] == 'HI_RES': quality = 'M' release_year = None if album_data.get('releaseDate'): release_year = album_data.get('releaseDate')[:4] elif album_data.get('streamStartDate'): release_year = album_data.get('streamStartDate')[:4] elif album_data.get('copyright'): # assume that every copyright includes the year release_year = [int(s) for s in album_data.get('copyright').split() if s.isdigit()] if len(release_year) > 0: release_year = release_year[0] return AlbumInfo( name=album_data.get('title'), release_year=release_year, explicit=album_data.get('explicit'), quality=quality, upc=album_data.get('upc'), cover_url=self.generate_artwork_url(album_data.get('cover'), size=self.cover_size) if album_data.get('cover') else None, animated_cover_url=self.generate_animated_artwork_url(album_data.get('videoCover')) if album_data.get( 'videoCover') else None, artist=album_data.get('artist').get('name'), artist_id=album_data.get('artist').get('id'), tracks=tracks, track_extra_kwargs=cache ) def get_track_info(self, track_id: str, quality_tier: QualityEnum, codec_options: CodecOptions, data=None) -> TrackInfo: if data is None: data = {} track_data = data[track_id] if track_id in data else self.session.get_track(track_id) album_id = str(track_data.get('album').get('id')) # check if album is already in album cache, get it try: album_data = data[album_id] if album_id in data else self.session.get_album(album_id) except TidalError as e: # if an error occurs, catch it and set the album_data to an empty dict to catch it self.print(f'{module_information.service_name}: {e} Trying workaround ...', drop_level=1) album_data = track_data.get('album') album_data.update({ 'artist': track_data.get('artist'), 'numberOfVolumes': 1, 'audioQuality': 'LOSSLESS', 'audioModes': ['STEREO'] }) # add the region locked album to the cache in order to properly use it later (force_album_format) self.album_cache = {album_id: album_data} # check if album is only available in LOSSLESS and STEREO, so it switches to the MOBILE_DEFAULT which will # get FLACs faster if (self.settings['force_non_spatial'] or ( (quality_tier is QualityEnum.LOSSLESS or album_data.get('audioQuality') == 'LOSSLESS') and 'STEREO' in album_data.get('audioModes'))) and SessionType.MOBILE_DEFAULT.name in self.available_sessions: self.session.default = SessionType.MOBILE_DEFAULT elif (track_data.get('audioModes') == ['SONY_360RA'] or ('DOLBY_ATMOS' in track_data.get('audioModes') and self.settings['prefer_ac4'])) \ and SessionType.MOBILE_ATMOS.name in self.available_sessions: self.session.default = SessionType.MOBILE_ATMOS else: self.session.default = SessionType.TV # define all default values in case the stream_data is None (region locked) audio_track, mqa_file, track_codec, bitrate, download_args, error = None, None, CodecEnum.FLAC, None, None, None try: stream_data = self.session.get_stream_url(track_id, self.quality_parse[quality_tier]) except TidalRequestError as e: error = e # definitely region locked if 'Asset is not ready for playback' in str(e): error = f'Track [{track_id}] is not available in your region' stream_data = None if stream_data is not None: if stream_data['manifestMimeType'] == 'application/dash+xml': manifest = base64.b64decode(stream_data['manifest']) audio_track = self.parse_mpd(manifest)[0] # Only one AudioTrack? track_codec = audio_track.codec else: manifest = json.loads(base64.b64decode(stream_data['manifest'])) track_codec = CodecEnum['AAC' if 'mp4a' in manifest['codecs'] else manifest['codecs'].upper()] if not codec_data[track_codec].spatial: if not codec_options.proprietary_codecs and codec_data[track_codec].proprietary: self.print(f'Proprietary codecs are disabled, if you want to download {track_codec.name}, ' f'set "proprietary_codecs": true', drop_level=1) stream_data = self.session.get_stream_url(track_id, 'LOSSLESS') if stream_data['manifestMimeType'] == 'application/dash+xml': manifest = base64.b64decode(stream_data['manifest']) audio_track = self.parse_mpd(manifest)[0] # Only one AudioTrack? track_codec = audio_track.codec else: manifest = json.loads(base64.b64decode(stream_data['manifest'])) track_codec = CodecEnum['AAC' if 'mp4a' in manifest['codecs'] else manifest['codecs'].upper()] if audio_track: download_args = {'audio_track': audio_track} else: # check if MQA if track_codec is CodecEnum.MQA and self.settings['fix_mqa']: # download the first chunk of the flac file to analyze it temp_file_path = self.download_temp_header(manifest['urls'][0]) # detect MQA file mqa_file = MqaIdentifier(temp_file_path) # add the file to download_args download_args = {'file_url': manifest['urls'][0]} # https://en.wikipedia.org/wiki/Audio_bit_depth#cite_ref-1 bit_depth = 16 if track_codec in {CodecEnum.FLAC, CodecEnum.ALAC} else None sample_rate = 48 if track_codec in {CodecEnum.EAC3, CodecEnum.MHA1, CodecEnum.AC4} else 44.1 if stream_data: # fallback bitrate bitrate = { 'LOW': 96, 'HIGH': 320, 'LOSSLESS': 1411, 'HI_RES': None }[stream_data['audioQuality']] # manually set bitrate for immersive formats if stream_data['audioMode'] == 'DOLBY_ATMOS': # check if the Dolby Atmos format is E-AC-3 JOC or AC-4 if track_codec == CodecEnum.EAC3: bitrate = 768 elif track_codec == CodecEnum.AC4: bitrate = 256 elif stream_data['audioMode'] == 'SONY_360RA': bitrate = 667 # more precise bitrate tidal uses MPEG-DASH if audio_track: bitrate = audio_track.bitrate // 1000 # now set everything for MQA if mqa_file is not None and mqa_file.is_mqa: bit_depth = mqa_file.bit_depth sample_rate = mqa_file.get_original_sample_rate() track_name = track_data.get('title') track_name += f' ({track_data.get("version")})' if track_data.get("version") else '' track_info = TrackInfo( name=track_name, album=album_data.get('title'), album_id=album_id, artists=[a.get('name') for a in track_data.get('artists')], artist_id=track_data['artist'].get('id'), release_year=track_data.get('streamStartDate')[:4] if track_data[ 'streamStartDate'] else track_data.get('dateAdded')[:4], bit_depth=bit_depth, sample_rate=sample_rate, bitrate=bitrate, # duration=track_data.get('duration'), cover_url=self.generate_artwork_url(track_data['album'].get('cover'), size=self.cover_size) if track_data['album'].get('cover') else None, explicit=track_data.get('explicit'), tags=self.convert_tags(track_data, album_data, mqa_file), codec=track_codec, download_extra_kwargs=download_args, lyrics_extra_kwargs={'track_data': track_data}, # check if 'credits' are present (only from get_album_data) credits_extra_kwargs={'data': {track_id: track_data['credits']} if 'credits' in track_data else {}} ) if not codec_options.spatial_codecs and codec_data[track_codec].spatial: track_info.error = 'Info: Spatial codecs are disabled, if you want to download it, set "spatial_codecs": ' \ 'true ' if error is not None: track_info.error = f'Error: {error}' return track_info @staticmethod def download_temp_header(file_url: str, chunk_size: int = 16384) -> str: # create flac temp_location temp_location = create_temp_filename() + '.flac' # create session and download the file to the temp_location r_session = create_requests_session() r = r_session.get(file_url, stream=True, verify=False) with open(temp_location, 'wb') as f: # only download the first chunk_size bytes for chunk in r.iter_content(chunk_size=chunk_size): if chunk: # filter out keep-alive new chunks f.write(chunk) break return temp_location @staticmethod def parse_mpd(xml: bytes) -> list: xml = xml.decode('UTF-8') # Removes default namespace definition, don't do that! xml = re.sub(r'xmlns="[^"]+"', '', xml, count=1) root = ElementTree.fromstring(xml) # List of AudioTracks tracks = [] for period in root.findall('Period'): for adaptation_set in period.findall('AdaptationSet'): for rep in adaptation_set.findall('Representation'): # Check if representation is audio content_type = adaptation_set.get('contentType') if content_type != 'audio': raise ValueError('Only supports audio MPDs!') # Codec checks codec = rep.get('codecs').upper() if codec.startswith('MP4A'): codec = 'AAC' # Segment template seg_template = rep.find('SegmentTemplate') # Add init file to track_urls track_urls = [seg_template.get('initialization')] start_number = int(seg_template.get('startNumber') or 1) # https://dashif-documents.azurewebsites.net/Guidelines-TimingModel/master/Guidelines-TimingModel.html#addressing-explicit # Also see example 9 seg_timeline = seg_template.find('SegmentTimeline') if seg_timeline is not None: seg_time_list = [] cur_time = 0 for s in seg_timeline.findall('S'): # Media segments start time if s.get('t'): cur_time = int(s.get('t')) # Segment reference for i in range((int(s.get('r') or 0) + 1)): seg_time_list.append(cur_time) # Add duration to current time cur_time += int(s.get('d')) # Create list with $Number$ indices seg_num_list = list(range(start_number, len(seg_time_list) + start_number)) # Replace $Number$ with all the seg_num_list indices track_urls += [seg_template.get('media').replace('$Number$', str(n)) for n in seg_num_list] tracks.append(AudioTrack( codec=CodecEnum[codec], sample_rate=int(rep.get('audioSamplingRate') or 0), bitrate=int(rep.get('bandwidth') or 0), urls=track_urls )) return tracks def get_track_download(self, file_url: str = None, audio_track: AudioTrack = None) \ -> TrackDownloadInfo: # only file_url or audio_track at a time # MHA1, EC-3 or MQA if file_url: return TrackDownloadInfo(download_type=DownloadEnum.URL, file_url=file_url) # MPEG-DASH # use the total_file size for a better progress bar? Is it even possible to calculate the total size from MPD? try: columns = os.get_terminal_size().columns if os.name == 'nt': bar = tqdm(audio_track.urls, ncols=(columns - self.oprinter.indent_number), bar_format=' ' * self.oprinter.indent_number + '{l_bar}{bar}{r_bar}') else: raise OSError except OSError: bar = tqdm(audio_track.urls, bar_format=' ' * self.oprinter.indent_number + '{l_bar}{bar}{r_bar}') # download all segments and save the locations inside temp_locations temp_locations = [] for download_url in bar: temp_locations.append(download_to_temp(download_url, extension='mp4')) # needed for bar indent bar.close() # concatenated/Merged .mp4 file merged_temp_location = create_temp_filename() + '.mp4' # actual converted .flac file output_location = create_temp_filename() + '.' + codec_data[audio_track.codec].container.name # download is finished, merge chunks into 1 file with open(merged_temp_location, 'wb') as dest_file: for temp_location in temp_locations: with open(temp_location, 'rb') as segment_file: copyfileobj(segment_file, dest_file) # convert .mp4 back to .flac try: ffmpeg.input(merged_temp_location, hide_banner=None, y=None).output(output_location, acodec='copy', loglevel='error').run() # Remove all files silentremove(merged_temp_location) for temp_location in temp_locations: silentremove(temp_location) except Exception: self.print('FFmpeg is not installed or working! Using fallback, may have errors') # return the MP4 temp file, but tell orpheus to change the container to .m4a (AAC) return TrackDownloadInfo( download_type=DownloadEnum.TEMP_FILE_PATH, temp_file_path=merged_temp_location, different_codec=CodecEnum.AAC ) # return the converted flac file now return TrackDownloadInfo( download_type=DownloadEnum.TEMP_FILE_PATH, temp_file_path=output_location, ) def get_track_cover(self, track_id: str, cover_options: CoverOptions, data=None) -> CoverInfo: if data is None: data = {} track_data = data[track_id] if track_id in data else self.session.get_track(track_id) cover_id = track_data['album'].get('cover') # Tidal don't support PNG, so it will always get JPG cover_url = self.generate_artwork_url(cover_id, size=cover_options.resolution) return CoverInfo(url=cover_url, file_type=ImageFileTypeEnum.jpg) def get_track_lyrics(self, track_id: str, track_data: dict = None) -> LyricsInfo: if not track_data: track_data = {} # get lyrics data for current track id lyrics_data = self.session.get_lyrics(track_id) if 'error' in lyrics_data and track_data: # search for title and artist to find a matching track (non Atmos) results = self.search( DownloadTypeEnum.track, f'{track_data.get("title")} {"".join(a.get("name") for a in track_data.get("artists"))}', limit=10) # check every result to find a matching result best_tracks = [r.result_id for r in results if r.name == track_data.get('title') and r.artists[0] == track_data.get('artist').get('name') and 'Dolby Atmos' not in r.additional] # retrieve the lyrics for the first one, otherwise return empty dict lyrics_data = self.session.get_lyrics(best_tracks[0]) if len(best_tracks) > 0 else {} embedded = lyrics_data.get('lyrics') synced = lyrics_data.get('subtitles') return LyricsInfo( embedded=embedded, synced=synced ) def get_track_credits(self, track_id: str, data=None) -> Optional[list]: if data is None: data = {} credits_dict = {} # fetch credits from cache if not fetch those credits if track_id in data: track_contributors = data[track_id] for contributor in track_contributors: credits_dict[contributor.get('type')] = [c.get('name') for c in contributor.get('contributors')] else: track_contributors = self.session.get_track_contributors(track_id).get('items') if len(track_contributors) > 0: for contributor in track_contributors: # check if the dict contains no list, create one if contributor.get('role') not in credits_dict: credits_dict[contributor.get('role')] = [] credits_dict[contributor.get('role')].append(contributor.get('name')) if len(credits_dict) > 0: # convert the dictionary back to a list of CreditsInfo return [CreditsInfo(sanitise_name(k), v) for k, v in credits_dict.items()] return None @staticmethod def convert_tags(track_data: dict, album_data: dict, mqa_file: MqaIdentifier = None) -> Tags: track_name = track_data.get('title') track_name += f' ({track_data.get("version")})' if track_data.get('version') else '' extra_tags = {} if mqa_file is not None: encoder_time = datetime.now().strftime("%b %d %Y %H:%M:%S") extra_tags = { 'ENCODER': f'MQAEncode v1.1, 2.4.0+0 (278f5dd), E24F1DE5-32F1-4930-8197-24954EB9D6F4, {encoder_time}', 'MQAENCODER': f'MQAEncode v1.1, 2.4.0+0 (278f5dd), E24F1DE5-32F1-4930-8197-24954EB9D6F4, {encoder_time}', 'ORIGINALSAMPLERATE': str(mqa_file.original_sample_rate) } return Tags( album_artist=album_data.get('artist').get('name') if 'artist' in album_data else None, track_number=track_data.get('trackNumber'), total_tracks=album_data.get('numberOfTracks'), disc_number=track_data.get('volumeNumber'), total_discs=album_data.get('numberOfVolumes'), isrc=track_data.get('isrc'), upc=album_data.get('upc'), release_date=album_data.get('releaseDate'), copyright=track_data.get('copyright'), replay_gain=track_data.get('replayGain'), replay_peak=track_data.get('peak'), extra_tags=extra_tags )