ovosimpatico/orpheusdl-deezer
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

initial commit

Browse Source
This commit is contained in:
uh wot
2021-12-25 22:38:24 +01:00
commit 8cd62a686c
4 changed files with 618 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

152
.gitignore vendored Normal file
View File

@@ -0,0 +1,152 @@
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
cover/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
.pybuilder/
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
# For a library or package, you might want to ignore these files since the code is
# intended to run in multiple environments; otherwise, check them in:
# .python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don't work, or not
# install all needed dependencies.
#Pipfile.lock
# poetry
# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
#poetry.lock
# PEP 582; used by e.g. github.com/David-OConnor/pyflow
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# pytype static type analyzer
.pytype/
# Cython debug symbols
cython_debug/
# PyCharm
# JetBrains specific template is maintainted in a separate JetBrains.gitignore that can
# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
# and can be added to the global gitignore or merged into this file. For a more nuclear
# option (not recommended) you can uncomment the following to ignore the entire idea folder.
#.idea/

0
__init__.py Normal file
View File

206
dzapi.py Normal file
View File

@@ -0,0 +1,206 @@
from hashlib import md5
from random import randint
from time import time
from math import ceil
from Cryptodome.Hash import MD5
from Cryptodome.Cipher import Blowfish, AES
from utils.utils import create_requests_session
class DeezerAPI:
def __init__(self, exception, client_id, client_secret, bf_secret, track_url_key):
self.gw_light_url = 'https://www.deezer.com/ajax/gw-light.php'
self.api_token = ''
self.exception = exception
self.client_id = client_id
self.client_secret = client_secret
self.legacy_url_cipher = AES.new(track_url_key.encode('ascii'), AES.MODE_ECB)
self.bf_secret = bf_secret.encode('ascii')
self.s = create_requests_session()
self.s.headers.update({
'accept': '*/*',
'user-agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/96.0.4664.110 Safari/537.36',
'content-type': 'text/plain;charset=UTF-8',
'origin': 'https://www.deezer.com',
'sec-fetch-site': 'same-origin',
'sec-fetch-mode': 'same-origin',
'sec-fetch-dest': 'empty',
'referer': 'https://www.deezer.com/',
'accept-language': 'en-US,en;q=0.9',
})
def _api_call(self, method, payload={}):
api_token = self.api_token if method not in ('deezer.getUserData', 'user.getArl') else ''
params = {
'method': method,
'input': 3,
'api_version': 1.0,
'api_token': api_token,
'cid': randint(0, 1e9),
}
resp = self.s.post(self.gw_light_url, params=params, json=payload).json()
if resp['error']:
key = list(resp['error'].keys())[0]
msg = list(resp['error'].values())[0]
raise self.exception((key, msg))
if method == 'deezer.getUserData':
self.api_token = resp['results']['checkForm']
self.country = resp['results']['COUNTRY']
self.license_token = resp['results']['USER']['OPTIONS']['license_token']
self.renew_timestamp = ceil(time())
self.language = resp['results']['USER']['SETTING']['global']['language']
self.available_formats = ['MP3_128']
format_dict = {'web_hq': 'MP3_320', 'web_lossless': 'FLAC'}
for k, v in format_dict.items():
if resp['results']['USER']['OPTIONS'][k]:
self.available_formats.append(v)
return resp['results']
def login_via_email(self, email, password):
password = md5(password.encode()).hexdigest()
params = {
'app_id': self.client_id,
'login': email,
'password': password,
'hash': md5((self.client_id + email + password + self.client_secret).encode()).hexdigest(),
}
json = self.s.get('https://connect.deezer.com/oauth/user_auth.php', params=params).json()
if 'error' in json:
raise self.exception('Error while getting access token, check your credentials')
headers = {'Authorization': f'Bearer {json["access_token"]}'}
# server sends set-cookie header with account sid
self.s.get('https://api.deezer.com/platform/generic/track/80085', headers=headers)
arl = self._api_call('user.getArl')
return arl, self.login_via_arl(arl)
def login_via_arl(self, arl):
self.s.cookies.set('arl', arl, domain='.deezer.com')
user_data = self._api_call('deezer.getUserData')
if not user_data['USER']['USER_ID']:
self.s.cookies.clear()
raise self.exception('Invalid arl')
return user_data
def get_track(self, id):
return self._api_call('deezer.pageTrack', {'sng_id': id})
def get_track_lyrics(self, id):
return self._api_call('song.getLyrics', {'sng_id': id})
def get_track_contributors(self, id):
return self._api_call('song.getData', {'sng_id': id, 'array_default': ['SNG_CONTRIBUTORS']})['SNG_CONTRIBUTORS']
def get_track_cover(self, id):
return self._api_call('song.getData', {'sng_id': id, 'array_default': ['ALB_PICTURE']})['ALB_PICTURE']
def get_track_data_by_isrc(self, isrc):
resp = self.s.get(f'https://api.deezer.com/track/isrc:{isrc}').json()
if 'error' in resp:
raise self.exception((resp['error']['type'], resp['error']['message'], resp['error']['code']))
return self._api_call('song.getData', {'sng_id': resp['id']})
def get_album(self, id):
return self._api_call('deezer.pageAlbum', {'alb_id': id, 'lang': self.language})
def get_playlist(self, id, nb, start):
return self._api_call('deezer.pagePlaylist', {'nb': nb, 'start': start, 'playlist_id': id, 'lang': self.language, 'tab': 0, 'tags': True, 'header': True})
def get_artist_name(self, id):
return self._api_call('artist.getData', {'art_id': id, 'array_default': ['ART_NAME']})['ART_NAME']
def search(self, query, type, start, nb):
return self._api_call('search.music', {'query': query, 'start': start, 'nb': nb, 'filter': 'ALL', 'output': type.upper()})
def get_artist_album_ids(self, id, start, nb, credited_albums):
payload = {
'art_id': id,
'start': start,
'nb': nb,
'filter_role_id': [0,5] if credited_albums else [0],
'nb_songs': 0,
'discography_mode': 'all' if credited_albums else None,
'array_default': ['ALB_ID']
}
resp = self._api_call('album.getDiscography', payload)
return [a['ALB_ID'] for a in resp['data']]
def get_track_url(self, id, track_token, track_token_expiry, format):
# renews license token
if time() - self.renew_timestamp >= 3600:
self._api_call('deezer.getUserData')
# renews track token
if time() - track_token_expiry >= 0:
track_token = self._api_call('song.getData', {'sng_id': id, 'array_default': ['TRACK_TOKEN']})['TRACK_TOKEN']
json = {
'license_token': self.license_token,
'media': [
{
'type': 'FULL',
'formats': [{'cipher': 'BF_CBC_STRIPE', 'format': format}]
}
],
'track_tokens': [track_token]
}
resp = self.s.post('https://media.deezer.com/v1/get_url', json=json).json()
return resp['data'][0]['media'][0]['sources'][0]['url']
def get_legacy_track_url(self, md5_origin, format, id, media_version):
# mashing a bunch of metadata and hashing it with MD5
info = b"\xa4".join([i.encode() for i in [
md5_origin, format, str(id), str(media_version)
]])
hash = MD5.new(info).hexdigest()
# hash + metadata
hash_metadata = hash.encode() + b"\xa4" + info + b"\xa4"
# padding
while len(hash_metadata) % 16 > 0:
hash_metadata += b"\0"
# AES encryption
result = self.legacy_url_cipher.encrypt(hash_metadata).hex()
# getting url
return f"https://cdns-proxy-{md5_origin[0]}.dzcdn.net/mobile/1/{result}"
def _get_blowfish_key(self, track_id):
# yeah, you use the bytes of the hex digest of the hash. bruh moment
md5_id = MD5.new(str(track_id).encode()).hexdigest().encode('ascii')
key = bytes([md5_id[i] ^ md5_id[i + 16] ^ self.bf_secret[i] for i in range(16)])
return key
def dl_track(self, id, url, path):
bf_key = self._get_blowfish_key(id)
req = self.s.get(url, stream=True)
req.raise_for_status()
with open(path, "ab") as file:
for i, chunk in enumerate(req.iter_content(2048)):
# every 3rd chunk is encrypted
if i % 3 == 0 and len(chunk) == 2048:
# yes, the cipher has to be reset on every chunk.
# those deezer devs were prob smoking crack when they made this DRM
cipher = Blowfish.new(bf_key, Blowfish.MODE_CBC, b"\x00\x01\x02\x03\x04\x05\x06\x07")
chunk = cipher.decrypt(chunk)
file.write(chunk)

260
interface.py Executable file
View File

@@ -0,0 +1,260 @@
from enum import Enum, auto
from utils.models import *
from utils.utils import create_temp_filename
from .dzapi import DeezerAPI
module_information = ModuleInformation(
service_name = 'Deezer',
module_supported_modes = ModuleModes.download | ModuleModes.lyrics | ModuleModes.covers | ModuleModes.credits,
global_settings = {'client_id': '447462', 'client_secret': 'a83bf7f38ad2f137e444727cfc3775cf', 'bf_secret': '', 'track_url_key': ''},
session_settings = {'email': '', 'password': ''},
session_storage_variables = ['arl'],
netlocation_constant = 'deezer',
test_url = 'https://www.deezer.com/track/3135556',
)
class ImageType(Enum):
cover = auto(),
artist = auto(),
playlist = auto(),
user = auto(),
misc = auto(),
talk = auto()
class ModuleInterface:
def __init__(self, module_controller: ModuleController):
settings = module_controller.module_settings
self.exception = module_controller.module_error
self.tsc = module_controller.temporary_settings_controller
self.default_cover = module_controller.orpheus_options.default_cover_options
if self.default_cover.file_type is ImageFileTypeEnum.webp:
self.default_cover.file_type = ImageFileTypeEnum.jpg
self.session = DeezerAPI(self.exception, settings['client_id'], settings['client_secret'], settings['bf_secret'], settings['track_url_key'])
arl = module_controller.temporary_settings_controller.read('arl')
if arl:
try:
self.session.login_via_arl(arl)
except self.exception:
self.login(settings['email'], settings['password'])
self.quality_parse = {
QualityEnum.LOW: 'MP3_128',
QualityEnum.MEDIUM: 'MP3_320',
QualityEnum.HIGH: 'MP3_320',
QualityEnum.LOSSLESS: 'FLAC',
QualityEnum.HIFI: 'FLAC'
}
self.compression_nums = {
CoverCompressionEnum.high: 80,
CoverCompressionEnum.low: 50
}
if arl and not module_controller.orpheus_options.disable_subscription_check and (self.quality_parse[module_controller.orpheus_options.quality_tier] not in self.session.available_formats):
print('Deezer: quality set in the settings is not accessible by the current subscription')
def login(self, email: str, password: str): # Called automatically by Orpheus when standard_login is flagged, otherwise optional
arl, _ = self.session.login_via_email(email, password)
self.tsc.set('arl', arl)
def get_track_info(self, track_id: str, quality_tier: QualityEnum, codec_options: CodecOptions, data={}) -> TrackInfo: # Mandatory
format = self.quality_parse[quality_tier]
track = data[track_id] if data and track_id in data else self.session.get_track(track_id)
t_data = track['DATA']
if 'FALLBACK' in t_data:
t_data = t_data['FALLBACK']
tags = Tags(
album_artist = t_data['ART_NAME'],
track_number = t_data['TRACK_NUMBER'],
copyright = t_data['COPYRIGHT'],
isrc = t_data['ISRC'],
disc_number = t_data['DISK_NUMBER'],
replay_gain = t_data.get('GAIN'),
release_date = t_data['PHYSICAL_RELEASE_DATE']
)
countries = t_data['AVAILABLE_COUNTRIES']['STREAM_ADS']
can_download = (t_data[f'FILESIZE_{format}'] != '0' and self.session.country in countries) or (format not in ('MP3_320', 'FLAC') and countries)
error = None
if not can_download:
error = 'Cannot download track'
codec = {
'MP3_128': CodecEnum.MP3,
'MP3_320': CodecEnum.MP3,
'FLAC': CodecEnum.FLAC,
}[format]
bitrate = {
'MP3_128': 128,
'MP3_320': 320,
'FLAC': 1411,
}[format]
download_extra_kwargs = {
'id': t_data['SNG_ID'],
'track_token': t_data['TRACK_TOKEN'],
'track_token_expiry': t_data['TRACK_TOKEN_EXPIRE'],
'format': format,
'md5_origin': t_data['MD5_ORIGIN'],
'media_version': t_data['MEDIA_VERSION']
}
return TrackInfo(
name = t_data['SNG_TITLE'] if not t_data.get('VERSION') else f'{t_data["SNG_TITLE"]} {t_data["VERSION"]}',
album_id = t_data['ALB_ID'],
album = t_data['ALB_TITLE'],
artists = [a['ART_NAME'] for a in t_data['ARTISTS']],
tags = tags,
codec = codec,
cover_url = self.get_image_url(t_data['ALB_PICTURE'], ImageType.cover, self.default_cover.file_type, self.default_cover.resolution, self.compression_nums[self.default_cover.compression]), # make sure to check module_controller.orpheus_options.default_cover_options
release_year = t_data['PHYSICAL_RELEASE_DATE'].split('-')[0],
explicit = t_data['EXPLICIT_LYRICS'] == '1',
artist_id = t_data['ART_ID'],
bitrate = bitrate,
download_extra_kwargs = download_extra_kwargs,
cover_extra_kwargs = {'data': {track_id: t_data['ALB_PICTURE']}},
credits_extra_kwargs = {'data': {track_id: t_data['SNG_CONTRIBUTORS']}},
lyrics_extra_kwargs = {'data': {track_id: track.get('LYRICS')}},
error = error
)
def get_track_download(self, id, track_token, track_token_expiry, format, md5_origin, media_version):
path = create_temp_filename()
# legacy urls don't have country restrictions, but aren't available for 320 and flac
# you can still get shit like 360RA with those though. bruh moment
if format in ('MP3_320', 'FLAC'):
url = self.session.get_track_url(id, track_token, track_token_expiry, format)
else:
url = self.session.get_legacy_track_url(md5_origin, '1', id, media_version)
self.session.dl_track(id, url, path)
return TrackDownloadInfo(
download_type = DownloadEnum.TEMP_FILE_PATH,
temp_file_path = path
)
def get_album_info(self, album_id: str, data={}) -> Optional[AlbumInfo]: # Mandatory if ModuleModes.download
album = data[album_id] if album_id in data else self.session.get_album(album_id)
a_data = album['DATA']
return AlbumInfo(
name = a_data['ALB_TITLE'],
artist = a_data['ART_NAME'],
tracks = [track['SNG_ID'] for track in album['SONGS']['data']],
release_year = a_data['PHYSICAL_RELEASE_DATE'].split('-')[0],
explicit = a_data['EXPLICIT_ALBUM_CONTENT']['EXPLICIT_LYRICS_STATUS'] in (1, 4),
artist_id = a_data['ART_ID'],
cover_url = self.get_image_url(a_data['ALB_PICTURE'], ImageType.cover, self.default_cover.file_type, self.default_cover.resolution, self.compression_nums[self.default_cover.compression]),
cover_type = self.default_cover.file_type,
all_track_cover_jpg_url = self.get_image_url(a_data['ALB_PICTURE'], ImageType.cover, ImageFileTypeEnum.jpg, self.default_cover.resolution, self.compression_nums[self.default_cover.compression]),
#track_extra_kwargs = {'data': ''} # optional, whatever you want
)
def get_playlist_info(self, playlist_id: str, data={}) -> PlaylistInfo:
playlist = data[playlist_id] if playlist_id in data else self.session.get_playlist(playlist_id, -1, 0)
p_data = playlist['DATA']
return PlaylistInfo(
name = p_data['TITLE'],
creator = p_data['PARENT_USERNAME'],
tracks = [t['SNG_ID'] for t in playlist['SONGS']['data']],
release_year = p_data['DATE_ADD'].split('-')[0],
creator_id = p_data['PARENT_USER_ID'],
cover_url = self.get_image_url(p_data['PLAYLIST_PICTURE'], ImageType.playlist, self.default_cover.file_type, self.default_cover.resolution, self.compression_nums[self.default_cover.compression]), # optional
cover_type = self.default_cover.file_type,
description = p_data['DESCRIPTION'],
#track_extra_kwargs = {'data': ''} # optional, whatever you want
)
def get_artist_info(self, artist_id: str, get_credited_albums: bool, artist_name = None) -> ArtistInfo:
name = artist_name if artist_name else self.session.get_artist_name(artist_id)
return ArtistInfo(
name = name,
albums = self.session.get_artist_album_ids(artist_id, 0, -1, get_credited_albums),
#album_extra_kwargs = {'data': ''}, # optional, whatever you want
)
def get_track_credits(self, track_id: str, data={}):
credits = data[track_id] if track_id in data else self.session.get_track_contributors(track_id)
if not credits:
return []
return [CreditsInfo(k, v) for k, v in credits.items()]
def get_track_cover(self, track_id: str, cover_options: CoverOptions, data={}) -> CoverInfo:
cover_md5 = data[track_id] if track_id in data else self.session.get_track_cover(track_id)
if cover_options.file_type is ImageFileTypeEnum.webp:
cover_options.file_type = ImageFileTypeEnum.jpg
url = self.get_image_url(cover_md5, ImageType.cover, cover_options.file_type, cover_options.resolution, self.compression_nums[cover_options.compression])
return CoverInfo(url=url, file_type=cover_options.file_type)
def get_track_lyrics(self, track_id: str, data={}) -> LyricsInfo:
try:
lyrics = data[track_id] if track_id in data else self.session.get_track_lyrics(track_id)
except self.exception:
return LyricsInfo()
if not lyrics:
return LyricsInfo()
synced_text = None
if 'LYRICS_SYNC_JSON' in lyrics:
synced_text = ''
for line in lyrics['LYRICS_SYNC_JSON']:
if 'lrc_timestamp' in line:
synced_text += f'{line["lrc_timestamp"]}{line["line"]}\n'
else:
synced_text += '\n'
return LyricsInfo(embedded=lyrics['LYRICS_TEXT'], synced=synced_text)
def search(self, query_type: DownloadTypeEnum, query: str, track_info: TrackInfo = None, limit: int = 10):
results = {}
if track_info and track_info.tags.isrc:
results = [self.session.get_track_data_by_isrc(track_info.tags.isrc)]
if not results:
results = self.session.search(query, query_type.name, 0, limit)['data']
if query_type is DownloadTypeEnum.track:
return [SearchResult(
result_id = i['SNG_ID'],
name = i['SNG_TITLE'] if not i.get('VERSION') else f'{i["SNG_TITLE"]} {i["VERSION"]}',
artists = [a['ART_NAME'] for a in i['ARTISTS']],
explicit = i['EXPLICIT_LYRICS'] == '1',
additional = [i["ALB_TITLE"]]
) for i in results]
elif query_type is DownloadTypeEnum.album:
return [SearchResult(
result_id = i['ALB_ID'],
name = i['ALB_TITLE'],
artists = [a['ART_NAME'] for a in i['ARTISTS']],
year = i['PHYSICAL_RELEASE_DATE'].split('-')[0],
explicit = i['EXPLICIT_ALBUM_CONTENT']['EXPLICIT_LYRICS_STATUS'] in (1, 4),
additional = [i["NUMBER_TRACK"]]
) for i in results]
elif query_type is DownloadTypeEnum.artist:
return [SearchResult(
result_id = i['ART_ID'],
name = i['ART_NAME'],
extra_kwargs = {'artist_name': i['ART_NAME']}
) for i in results]
elif query_type is DownloadTypeEnum.playlist:
return [SearchResult(
result_id = i['PLAYLIST_ID'],
name = i['TITLE'],
artists = [i['PARENT_USERNAME']],
additional = [i["NB_SONG"]]
) for i in results]
def get_image_url(self, md5, img_type: ImageType, file_type: ImageFileTypeEnum, res, compression):
filename = {
ImageFileTypeEnum.jpg: f'{res}x{res}-000000-{compression}-0-0.jpg',
ImageFileTypeEnum.png: f'{res}x{res}-none-100-0-0.png'
}[file_type]
return f'https://cdns-images.dzcdn.net/images/{img_type.name}/{md5}/{filename}'