mttw/orpheusdl-tidal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'develop'

Browse Source
This commit is contained in:
Dniel97
2023-11-23 23:39:40 +01:00
parent 8ef1a90ba5 85f106c757
commit c8a1620601
2 changed files with 151 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files

224
interface.py
View File

@@ -23,7 +23,7 @@ module_information = ModuleInformation(
global_settings={
'tv_token': '7m7Ap0JC9j1cOM3n',
'tv_secret': 'vRAdA108tlvkJpTsGZS8rGZ7xTlbJ0qaZ2K9saEzsgY=',
'mobile_atmos_token': 'dN2N95wCyEBTllu4',
'mobile_atmos_token': 'km8T1xS355y7dd3H',
'mobile_default_token': 'WAU9gXp3tHhK4Nns',
'enable_mobile': True,
'force_non_spatial': False,
@@ -74,86 +74,81 @@ class ModuleInterface:
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:
if not self.settings['enable_mobile']:
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'])
while True:
login_session = None
if session_type in saved_sessions:
logging.debug(f'{module_information.service_name}: {session_type} session found, loading')
# load the dictionary from the temporary_settings_controller inside the TidalSession class
sessions[session_type].set_storage(saved_sessions[session_type])
else:
logging.debug(f'{module_information.service_name}: No {session_type} session found, creating new one')
if session_type == SessionType.TV.name:
self.print(f'{module_information.service_name}: Creating a TV session')
sessions[session_type].auth()
else:
if not username or not password:
self.print(f'{module_information.service_name}: Creating a Mobile session')
self.print(f'{module_information.service_name}: Enter your TIDAL username and password:')
self.print(f'{module_information.service_name}: (password will not be echoed)')
username = input(' Username: ')
password = getpass(' Password: ')
sessions[session_type].auth(username, password)
self.print(f'Successfully logged in, using {session_type} token!')
def auth_and_save_session(session, session_type):
session = self.auth_session(session, session_type, login_session)
# get the dict representation from the TidalSession object and save it into saved_session/loginstorage
saved_sessions[session_type] = sessions[session_type].get_storage()
saved_sessions[session_type] = session.get_storage()
module_controller.temporary_settings_controller.set('sessions', saved_sessions)
return session
# 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)
# 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')
login_session = auth_and_save_session(self.init_session(login_session_type), login_session_type)
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)
while True:
# check for a valid subscription
subscription = self.check_subscription(sessions[session_type].get_subscription())
if 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 = {}
break
confirm = input(' Do you want to create a new session? [Y/n]: ')
if not login_session:
login_session = sessions[session_type]
if confirm.upper() == 'N':
self.print('Exiting...')
exit()
# create a new session finally
if session_type == SessionType.TV.name:
self.print(f'{module_information.service_name}: Recreating a TV session')
sessions[session_type].auth()
else:
self.print(f'{module_information.service_name}: Recreating a Mobile session')
self.print(f'{module_information.service_name}: Enter your TIDAL username and password:')
self.print(f'{module_information.service_name}: (password will not be echoed)')
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
if saved_sessions:
break
# only needed for region locked albums where the track is available but force_album_format is used
self.album_cache = {}
@@ -161,6 +156,39 @@ class ModuleInterface:
# 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_token'], self.settings['tv_secret'])
elif session_type == SessionType.MOBILE_ATMOS.name:
session = TidalMobileSession(self.settings['mobile_atmos_token'])
else:
session = TidalMobileSession(self.settings['mobile_default_token'])
return session
def auth_session(self, session, session_type, login_session):
if 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_PLUS'}:
@@ -185,10 +213,13 @@ class ModuleInterface:
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)
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[query_type.name + 's'].get('items'):
for i in results.get('items'):
duration, name = None, None
if query_type is DownloadTypeEnum.artist:
name = i.get('name')
@@ -424,26 +455,43 @@ class ModuleInterface:
# add the region locked album to the cache in order to properly use it later (force_album_format)
self.album_cache = {album_id: album_data}
# check if album is only available in LOSSLESS and STEREO, so it switches to the MOBILE_DEFAULT which will
# get FLACs faster, instead of using MPEG-DASH
# lmao what did I smoke when I wrote this, track_data and not album_data!
if (self.settings['force_non_spatial'] or (
(quality_tier is QualityEnum.LOSSLESS or track_data.get('audioQuality') == 'LOSSLESS')
and track_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 ('DOLBY_ATMOS' in track_data.get('audioModes') and self.settings['prefer_ac4'])) \
and SessionType.MOBILE_ATMOS.name in self.available_sessions:
self.session.default = SessionType.MOBILE_ATMOS
media_tags = track_data['mediaMetadata']['tags']
format = None
if 'HIRES_LOSSLESS' in media_tags and quality_tier is QualityEnum.HIFI:
format = 'flac_hires'
if 'SONY_360RA' in media_tags and not format and not self.settings['force_non_spatial']:
format = '360ra'
if 'DOLBY_ATMOS' in media_tags and not format and not self.settings['force_non_spatial']:
if self.settings['prefer_ac4']:
format = 'ac4'
else:
format = 'ac3'
session = {
'flac_hires': SessionType.MOBILE_ATMOS,
'360ra': SessionType.MOBILE_ATMOS,
'ac4': SessionType.MOBILE_ATMOS,
'ac3': SessionType.TV,
# MOBILE_DEFAULT is used whenever possible to avoid MPEG-DASH, which slows downloading
None: SessionType.MOBILE_DEFAULT,
}[format]
if not format and 'SONY_360RA' in media_tags:
# if 360RA is available, we don't use the mobile session here because that will get 360RA
# there are no tracks with both 360RA and atmos afaik,
# so this shouldn't be an issue for now
session = SessionType.TV
if session.name in self.available_sessions:
self.session.default = session
else:
self.session.default = SessionType.TV
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])
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
@@ -489,7 +537,8 @@ class ModuleInterface:
download_args = {'file_url': manifest['urls'][0]}
# https://en.wikipedia.org/wiki/Audio_bit_depth#cite_ref-1
bit_depth = 16 if track_codec in {CodecEnum.FLAC, CodecEnum.ALAC} else None
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:
@@ -498,7 +547,8 @@ class ModuleInterface:
'LOW': 96,
'HIGH': 320,
'LOSSLESS': 1411,
'HI_RES': None
'HI_RES': None,
'HI_RES_LOSSLESS': None
}[stream_data['audioQuality']]
# manually set bitrate for immersive formats
@@ -514,6 +564,8 @@ class ModuleInterface:
# 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:

20
tidal_api.py
View File

@@ -13,7 +13,7 @@ import requests
import urllib3
import urllib.parse as urlparse
from urllib.parse import parse_qs
from urllib.parse import parse_qs, quote
from datetime import datetime, timedelta
from utils.utils import create_requests_session
@@ -110,7 +110,7 @@ class TidalApi(object):
return resp_json
def get_stream_url(self, track_id, quality):
return self._get('tracks/' + str(track_id) + '/playbackinfopostpaywall', {
return self._get('tracks/' + str(track_id) + '/playbackinfopostpaywall/v4', {
'playbackmode': 'STREAM',
'assetpresentation': 'FULL',
'audioquality': quality,
@@ -174,6 +174,11 @@ class TidalApi(object):
def get_video(self, video_id):
return self._get('videos/' + str(video_id))
def get_tracks_by_isrc(self, isrc):
return self._get('tracks', params={
'isrc': isrc
})
def get_favorite_tracks(self, user_id):
return self._get('users/' + str(user_id) + '/favorites/tracks')
@@ -323,8 +328,8 @@ class TidalMobileSession(TidalSession):
self.code_verifier = base64.urlsafe_b64encode(secrets.token_bytes(32)).rstrip(b'=')
self.code_challenge = base64.urlsafe_b64encode(hashlib.sha256(self.code_verifier).digest()).rstrip(b'=')
self.client_unique_key = secrets.token_hex(8)
self.user_agent = 'Mozilla/5.0 (Linux; Android 13; Pixel 7 Build/TD1A.221105.001; wv) AppleWebKit/537.36' \
'(KHTML, like Gecko) Version/4.0 Chrome/109.0.5414.80 Mobile Safari/537.36'
self.user_agent = 'Mozilla/5.0 (Linux; Android 13; Pixel 8 Build/TQ2A.230505.002; wv) AppleWebKit/537.36 ' \
'(KHTML, like Gecko) Version/4.0 Chrome/119.0.6045.163 Mobile Safari/537.36'
def auth(self, username: str, password: str):
s = requests.Session()
@@ -355,10 +360,11 @@ class TidalMobileSession(TidalSession):
# try Tidal DataDome cookie request
r = s.post('https://dd.tidal.com/js/', data={
'jsData': f'{{"opts":"endpoint,ajaxListenerPath","ua":"{self.user_agent}"}}',
'ddk': '1F633CDD8EF22541BD6D9B1B8EF13A', # API Key (required)
'Referer': r.url, # Referer authorize link (required)
'Referer': quote(r.url), # Referer authorize link (required)
'responsePage': 'origin', # useless?
'ddv': '4.4.7' # useless?
'ddv': '4.17.0' # useless?
}, headers={
'user-agent': self.user_agent,
'content-type': 'application/x-www-form-urlencoded'
@@ -408,7 +414,7 @@ class TidalMobileSession(TidalSession):
raise TidalAuthError(r.text)
# retrieve access code
r = s.get('https://login.tidal.com/success?lang=en', allow_redirects=False, headers={
r = s.get('https://login.tidal.com/success', allow_redirects=False, headers={
'user-agent': self.user_agent,
'accept-language': 'en-US',
'x-requested-with': 'com.aspiro.tidal'