orpheusdl-tidal/interface.py

import base64
import json
import logging
import re
from datetime import datetime
from getpass import getpass
from dataclasses import dataclass
from shutil import copyfileobj
from xml.etree import ElementTree

import ffmpeg
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 import MqaIdentifier
from .tidal_api import TidalTvSession, TidalApi, SessionStorage, TidalMobileSession, SessionType, TidalError

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.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'Tidal: {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'Tidal: No {session_type} session found, creating new one')
                if session_type == SessionType.TV.name:
                    self.print('Tidal: Creating a TV session')
                    sessions[session_type].auth()
                else:
                    if not username or not password:
                        self.print('Tidal: Creating a Mobile session')
                        self.print('Tidal: 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('Tidal: Recreating a TV session')
                    sessions[session_type].auth()
                else:
                    self.print('Tidal: Recreating a Mobile session')
                    self.print('Tidal: 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

        # 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'Tidal: Account is not a HiFi Plus account, 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:
                name = i.get('title')
                artists = [i.get('creator').get('name')]
                year = ""
            elif query_type is DownloadTypeEnum.track:
                name = i.get('title')
                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:
                name = i.get('title')
                artists = [j.get('name') for j in i.get('artists')]
                year = i.get('releaseDate')[:4]
            else:
                raise Exception('Query type is invalid')

            additional = None
            if query_type is not DownloadTypeEnum.artist:
                if i.get('audioModes') == ['DOLBY_ATMOS']:
                    additional = "Dolby Atmos"
                elif i.get('audioModes') == ['SONY_360RA']:
                    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')),
                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('creator').get('id') == 0:
            creator_name = 'TIDAL'
        else:
            creator_name = 'Unknown'

        return PlaylistInfo(
            name=playlist_data.get('title'),
            creator=creator_name,
            tracks=tracks,
            # TODO: Use playlist creation date or lastUpdated?
            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]['pagedList']
            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 = {}

        album_data = data[album_id] if album_id in data else self.session.get_album(album_id)

        # get all album tracks with corresponding credits with a limit of 100
        limit = 100
        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

        # 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']

        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'

        return AlbumInfo(
            name=album_data.get('title'),
            release_year=album_data.get('releaseDate')[:4],
            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'Tidal: {e}, trying anyway', drop_level=1)
            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 (
                album_data.get('audioQuality') == 'LOSSLESS' and album_data.get('audioModes') == ['STEREO'])) and \
            SessionType.MOBILE_DEFAULT.name in self.available_sessions:
            self.session.default = SessionType.MOBILE_DEFAULT
        elif (track_data.get('audioModes') == ['SONY_360RA']
              or (track_data.get('audioModes') == ['DOLBY_ATMOS'] 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

        stream_data = self.session.get_stream_url(track_id, self.quality_parse[quality_tier])
        # only needed for MPEG-DASH
        audio_track = 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()]

        track_name = track_data.get('title')
        track_name += f' ({track_data.get("version")})' if track_data.get("version") else ''

        mqa_file = None
        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]}

        bit_depth = 24 if track_codec in [CodecEnum.EAC3, CodecEnum.MHA1] else 16
        sample_rate = 48 if track_codec in [CodecEnum.EAC3, CodecEnum.MHA1, CodecEnum.AC4] else 44.1

        # 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_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,
            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 = 'Spatial codecs are disabled, if you want to download it, set "spatial_codecs": true'

        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) -> LyricsInfo:
        # get lyrics data for current track id
        lyrics_data = self.session.get_lyrics(track_id)

        if 'error' in lyrics_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
        )