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.exceptions import InvalidInput 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_atmos_token": "cgiF7TQuB97BUIu3", "tv_atmos_secret": "1nqpgx8uvBdZigrx4hUPDV2hOwgYAAAG5DYXOr6uNf8=", "mobile_atmos_hires_token": "km8T1xS355y7dd3H", "mobile_hires_token": "6BDSRdpK9hqEBTgU", "enable_mobile": True, "prefer_ac4": False, "fix_mqa": False, 'tidal_token': '' }, # currently too broken to keep it, cover needs to be jpg else crash, problems on termux due to pillow # flags=ModuleFlags.needs_cover_resize, session_storage_variables=["sessions"], netlocation_constant="tidal", test_url="https://tidal.com/browse/track/92265335", url_decoding=ManualEnum.manual, ) @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" ) tidal_token = module_controller.module_settings.get('tidal_token') if not saved_sessions: saved_sessions = {} if not self.settings["enable_mobile"]: self.available_sessions = [SessionType.TV.name] while True: login_session = None relogin_requested = False def auth_and_save_session(session, session_type, tidal_token=None): session = self.auth_session(session, session_type, login_session, tidal_token) # get the dict representation from the TidalSession object and save it into saved_session/loginstorage saved_sessions[session_type] = session.get_storage() module_controller.temporary_settings_controller.set( "sessions", saved_sessions ) return session # ask for login if there are no saved sessions if not saved_sessions: login_session_type = None if len(self.available_sessions) == 1: login_session_type = self.available_sessions[0] else: self.print( f"{module_information.service_name}: Choose a login method:" ) self.print(f"{module_information.service_name}: 1. TV (browser)") self.print( f"{module_information.service_name}: 2. Mobile (username and password, choose TV if this doesn't work)" ) while not login_session_type: input_str = input(" Login method: ") try: login_session_type = { "1": SessionType.TV.name, "tv": SessionType.TV.name, "2": SessionType.MOBILE_DEFAULT.name, "mobile": SessionType.MOBILE_DEFAULT.name, }[input_str.lower()] except KeyError: self.print(f'{module_information.service_name}: Invalid choice, try again') # auto use tidal_token if present in config.json else prompt it while True: reply = input(' Use a tidal_token instead of email/password? (Y/n): ').strip().lower() if reply in ("y", "yes"): if not tidal_token: tidal_token = input(' No stored tidal_token detected, paste it here: ').strip() break elif reply in ("n", "no"): tidal_token = None break else: self.print(f'{module_information.service_name}: Reply with (Y)es or (N)o') login_session = auth_and_save_session( self.init_session(login_session_type), login_session_type, tidal_token ) self.print(f"Account Country: {login_session.country_code}") for session_type in self.available_sessions: sessions[session_type] = self.init_session(session_type) 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" ) sessions[session_type] = auth_and_save_session( sessions[session_type], session_type ) # 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 ) # check for a valid subscription subscription = self.check_subscription( sessions[session_type].get_subscription() ) if not subscription: confirm = input(" Do you want to relogin? [Y/n]: ") if confirm.upper() == "N": self.print("Exiting...") exit() # reset saved sessions and loop back to login saved_sessions = {} # flag used to not ask 2 times in a row to relogin relogin_requested = True break if not login_session: login_session = sessions[session_type] # check if tidal_token in config.json matches any saved refresh_token to swap between them easily if tidal_token and not relogin_requested and tidal_token not in [v.get('refresh_token') for v in saved_sessions.values()]: confirm = input(' Stored tidal_token differs from setting.json - Do you want to relogin? [Y/n]: ').strip().lower() if confirm in ('y', 'yes'): saved_sessions = {} continue # Restart the while loop to re-authenticate if saved_sessions: # print available stored sessions and their country code country_codes = set(entry['country_code'] for entry in saved_sessions.values()) sessions_list = ' - '.join(saved_sessions.keys()) if len(country_codes) == 1: result = f"Using saved sessions: {sessions_list} ({country_codes.pop()})" else: result = "Using saved sessions: " + " - ".join(f"{key} ({entry['country_code']})" for key, entry in saved_sessions.items()) self.print(result) break # 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 init_session(self, session_type): session = None # initialize session with the needed API keys if session_type == SessionType.TV.name: session = TidalTvSession( self.settings["tv_atmos_token"], self.settings["tv_atmos_secret"] ) elif session_type == SessionType.MOBILE_ATMOS.name: session = TidalMobileSession(self.settings["mobile_atmos_hires_token"]) else: session = TidalMobileSession(self.settings["mobile_hires_token"]) return session def auth_session(self, session, session_type, login_session, tidal_token=None): if tidal_token: session.refresh_token = tidal_token session.refresh() elif login_session: # refresh tokens can be used with any client id # this can be used to switch to any client type from an existing session session.refresh_token = login_session.refresh_token session.user_id = login_session.user_id session.country_code = login_session.country_code session.refresh() elif session_type == SessionType.TV.name: self.print(f"{module_information.service_name}: Creating a TV session") session.auth() else: self.print(f"{module_information.service_name}: Creating a Mobile session") self.print( f"{module_information.service_name}: Enter your TIDAL username and password:" ) self.print( f"{module_information.service_name}: (password will not be echoed)" ) username = input(" Username: ") password = getpass(" Password: ") session.auth(username, password) self.print(f"Successfully logged in, using {session_type} token!") return session 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", "PREMIUM_PLUS", }: self.print( f"{module_information.service_name}: Account does not have a HiFi (Plus) subscription, " 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 custom_url_parse(self, link: str): # the most beautiful regex ever written match = re.search( r"https?://tidal\.com/(?:browse/)?(?Ptrack|album|playlist|artist)/" r"(?P[A-Za-z0-9-]+)", link, ) # so parse the regex "match" to the actual DownloadTypeEnum media_types = { "track": DownloadTypeEnum.track, "album": DownloadTypeEnum.album, "artist": DownloadTypeEnum.artist, "playlist": DownloadTypeEnum.playlist, } if not match: self.print(f"Unsupported URL: {link}") exit() return MediaIdentification( media_type=media_types[match.group("media_type")], media_id=match.group("media_id"), ) def search( self, query_type: DownloadTypeEnum, query: str, track_info: TrackInfo = None, limit: int = 20, ): if track_info and track_info.tags.isrc: results = self.session.get_tracks_by_isrc(track_info.tags.isrc) else: results = self.session.get_search_data(query, limit=limit)[ query_type.name + "s" ] items = [] for i in results.get("items"): duration, name = None, None 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"] duration = i.get("duration") # 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] if i.get("album").get("releaseDate") else None ) duration = i.get("duration") elif query_type is DownloadTypeEnum.album: artists = [j.get("name") for j in i.get("artists")] duration = i.get("duration") 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 not in {DownloadTypeEnum.artist, DownloadTypeEnum.playlist}: 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"), duration=duration, 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" if playlist_data.get("squareImage"): cover_url = self._generate_artwork_url( playlist_data["squareImage"], size=self.cover_size, max_size=1080 ) cover_type = ImageFileTypeEnum.jpg else: # fallback to defaultPlaylistImage cover_url = "https://tidal.com/browse/assets/images/defaultImages/defaultPlaylistImage.png" cover_type = ImageFileTypeEnum.png return PlaylistInfo( name=playlist_data.get("title"), creator=creator_name, tracks=tracks, release_year=playlist_data.get("created")[:4], duration=playlist_data.get("duration"), creator_id=playlist_data["creator"].get("id"), cover_url=cover_url, cover_type=cover_type, 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] if album_data.get("cover"): cover_url = self._generate_artwork_url( album_data.get("cover"), size=self.cover_size ) cover_type = ImageFileTypeEnum.jpg else: # fallback to defaultAlbumImage cover_url = "https://tidal.com/browse/assets/images/defaultImages/defaultAlbumImage.png" cover_type = ImageFileTypeEnum.png return AlbumInfo( name=album_data.get("title"), release_year=release_year, explicit=album_data.get("explicit"), quality=quality, upc=album_data.get("upc"), duration=album_data.get("duration"), cover_url=cover_url, cover_type=cover_type, 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} media_tags = track_data["mediaMetadata"]["tags"] format = None if codec_options.spatial_codecs: if "SONY_360RA" in media_tags: format = "360ra" elif "DOLBY_ATMOS" in media_tags: if self.settings["prefer_ac4"]: format = "ac4" else: format = "ac3" if ( "HIRES_LOSSLESS" in media_tags and not format and quality_tier is QualityEnum.HIFI ): format = "flac_hires" session = { "flac_hires": SessionType.MOBILE_DEFAULT, "360ra": SessionType.MOBILE_DEFAULT, "ac4": SessionType.MOBILE_ATMOS, "ac3": SessionType.TV, # TV is used whenever possible to avoid MPEG-DASH, which slows downloading None: SessionType.TV, }[format] if not format and "DOLBY_ATMOS" in media_tags: # if atmos is available, we don't use the TV session here because that will get atmos everytime # there are no tracks with both 360RA and atmos afaik, # so this shouldn't be an issue for now session = SessionType.MOBILE_DEFAULT if session.name in self.available_sessions: self.session.default = session else: format = None # 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] if format != "flac_hires" else "HI_RES_LOSSLESS" ), ) 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 = ( ( 24 if stream_data and stream_data["audioQuality"] == "HI_RES_LOSSLESS" else 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, "HI_RES_LOSSLESS": 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 if stream_data["audioQuality"] == "HI_RES_LOSSLESS": sample_rate = audio_track.sample_rate / 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 "" ) if track_data["album"].get("cover"): cover_url = self._generate_artwork_url( track_data["album"].get("cover"), size=self.cover_size ) else: # fallback to defaultTrackImage, no cover_type flag? Might crash in the future cover_url = "https://tidal.com/browse/assets/images/defaultImages/defaultTrackImage.png" 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.get("streamStartDate") else ( track_data.get("dateAdded")[:4] if track_data.get("dateAdded") else None ) ), bit_depth=bit_depth, sample_rate=sample_rate, bitrate=bitrate, duration=track_data.get("duration"), cover_url=cover_url, 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 error is not None: track_info.error = f"Error: {error}" return track_info @staticmethod def download_temp_header(file_url: str, chunk_size: int = 32768) -> 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: 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") if cover_id: return CoverInfo( url=self._generate_artwork_url(cover_id, size=cover_options.resolution), file_type=ImageFileTypeEnum.jpg, ) return CoverInfo( url="https://tidal.com/browse/assets/images/defaultImages/defaultTrackImage.png", file_type=ImageFileTypeEnum.png, ) 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, # regex to remove the space after the timestamp "[mm:ss.xx] " to "[mm:ss.xx]" synced=( re.sub(r"(\[\d{2}:\d{2}.\d{2,3}])(?: )", r"\1", synced) if synced else None ), ) 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, )