724 lines
33 KiB
Python
724 lines
33 KiB
Python
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 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.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:
|
|
if 'name' in i.get('creator'):
|
|
artists = [i.get('creator').get('name')]
|
|
elif i.get('type') == 'EDITORIAL':
|
|
artists = ['TIDAL']
|
|
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 = '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].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 = {}
|
|
|
|
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.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']
|
|
|
|
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 (
|
|
(quality_tier is QualityEnum.LOSSLESS 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
|
|
|
|
# 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]}
|
|
|
|
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
|
|
|
|
if stream_data:
|
|
# fallback bitrate
|
|
bitrate = {
|
|
'LOW': 96,
|
|
'HIGH': 320,
|
|
'LOSSLESS': 1411,
|
|
'HI_RES': None
|
|
}[stream_data['audioQuality']]
|
|
|
|
# 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,
|
|
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 not stream_data:
|
|
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
|
|
)
|