From 8cd62a686c9e8b86fd544b3f210eb8581cfb1d35 Mon Sep 17 00:00:00 2001 From: uh wot Date: Sat, 25 Dec 2021 22:38:24 +0100 Subject: [PATCH] initial commit --- .gitignore | 152 ++++++++++++++++++++++++++++++ __init__.py | 0 dzapi.py | 206 ++++++++++++++++++++++++++++++++++++++++ interface.py | 260 +++++++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 618 insertions(+) create mode 100644 .gitignore create mode 100644 __init__.py create mode 100644 dzapi.py create mode 100755 interface.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..d9005f2 --- /dev/null +++ b/.gitignore @@ -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/ diff --git a/__init__.py b/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/dzapi.py b/dzapi.py new file mode 100644 index 0000000..cbe76db --- /dev/null +++ b/dzapi.py @@ -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) diff --git a/interface.py b/interface.py new file mode 100755 index 0000000..0edf4eb --- /dev/null +++ b/interface.py @@ -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}' \ No newline at end of file