From dcf2f2d2ce0ab4ed5d983bf1dea4b2f594b17398 Mon Sep 17 00:00:00 2001 From: Luca Bilke Date: Thu, 22 Jun 2023 17:39:20 +0200 Subject: [PATCH] minor refactor, functions to retrieve information --- __init__.py | 9 +++-- objects.py | 94 ++++++++++++++++++++++------------------------------- scraper.py | 92 ++++++++++++++++++++++++++++++++++++++++----------- 3 files changed, 117 insertions(+), 78 deletions(-) diff --git a/__init__.py b/__init__.py index 82b8ee1..98c6fbf 100644 --- a/__init__.py +++ b/__init__.py @@ -1,5 +1,8 @@ from scraper import scraper -import os -authfile = os.environ["XDG_CACHE_HOME"] + "/auth.json" -s = scraper(authfile) +NORMAL = "LOW" +HIGH = "HIGH" +HIFI = "LOSSLESS" +MASTER = "HI_RES" + +s = scraper(quality = HIFI) diff --git a/objects.py b/objects.py index 4122ac5..ccc3d69 100644 --- a/objects.py +++ b/objects.py @@ -21,70 +21,52 @@ class Auth: @dataclass class Artist: - id: int | None - name: str | None - type: str | None - picture: str | None + id: int + name: str + type: str + picture: str @dataclass class Album: - id: int | None - title: str | None - duration: int | None - numberOfTracks: int | None - numberOfVideos: int | None - numberOfVolumes: int | None - releaseDate: str | None - type: str | None - version: str | None - cover: str | None - explicit: bool | None - audioQuality: str | None - audioModes: str | None - artist: Artist | None - artists: Artist | None - - -@dataclass -class Playlist: - uuid: str | None - title: str | None - numberOfTracks: int | None - numberOfVideos: int | None - description: str | None - duration: int | None - image: str | None - squareImage: str | None + id: int + title: str + duration: int + numberOfTracks: int + numberOfVolumes: int + releaseDate: str + type: str + version: str + cover: str + explicit: bool + audioQuality: str + audioModes: str + artist: Artist + artists: list @dataclass class Track: - id: int | None - title: str | None - duration: int | None - trackNumber: int | None - volumeNumber: int | None - trackNumberOnPlaylist: int | None - version: str | None - isrc: str | None - explicit: bool | None - audioQuality: str | None - copyRight: str | None - artist: Artist | None - artists: Artist | None - album: Album | None - allowStreaming: bool | None - playlist: Playlist | None + id: int + title: str + duration: int + number: int + volumeNumber: int + version: str + isrc: str + explicit: bool + audioQuality: str + copyRight: str + artist: Artist + artists: Artist + album: Album + allowStreaming: bool @dataclass -class StreamResponse: - trackid: int | None - streamType: str | None - assetPresentation: str | None - audioMode: str | None - audioQuality: str | None - videoQuality: str | None - manifestMimeType: str | None - manifest: str | None +class StreamInfo: + trackId: int + audioQuality: str + codecs: str + encryptionKey: str + url: str diff --git a/scraper.py b/scraper.py index c9002d9..7e1af74 100644 --- a/scraper.py +++ b/scraper.py @@ -1,18 +1,36 @@ CLIENT_ID = "zU4XHVVkc2tDPo4t" CLIENT_SECRET = "VJKhDFqJPqvsPVNBV6ukXTJmwlvbttP7wlMlrc72se4" -API_URL_BASE = "https://api.tidalhifi.com/v1" -AUTH_URL_BASE = "https://auth.tidal.com/v1/oauth2" - from objects import * +from typing import Tuple import requests import time import random import json +import base64 +import os class scraper: - def __init__(self, authfile: str): + def __init__( + self, + quality: str, + redownload: bool = False, + authUrlBase: str = "https://auth.tidal.com/v1/oauth2", + apiUrlBase: str = "https://api.tidalhifi.com/v1", + clientToken: tuple = (CLIENT_ID, CLIENT_SECRET), + downloadPath: str = "~/Downloads", + cachePath: str = "~/.cache", + ): + self.quality = quality + self.redownload = redownload + self.authUrlBase = authUrlBase + self.apiUrlBase = apiUrlBase + self.apiUrlBase = apiUrlBase + self.clientToken = clientToken + self.downloadPath = downloadPath + self.cachePath = cachePath + authfile = self.cachePath + "/auth.json" try: with open(authfile, "rb") as f: a = json.load(f) @@ -29,7 +47,10 @@ class scraper: json.dump(self.auth.__dict__, f) def post(self, url: str, data: dict) -> dict: - return requests.post(url, data=data, auth=(CLIENT_ID, CLIENT_SECRET)).json() + return requests.post(url, data=data, auth=self.clientToken).json() + + def retrieve(self, url: str, path: str) -> None: + # TODO: Write function to retrieve stream def get(self, url: str, params: dict = {}) -> dict: headers = {"authorization": f"Bearer {self.auth.accessToken}"} @@ -78,8 +99,8 @@ class scraper: def loginByWeb(self) -> Auth: result = self.post( - f"{AUTH_URL_BASE}/device_authorization", - {"client_id": CLIENT_ID, "scope": "r_usr+w_usr+w_sub"}, + f"{self.authUrlBase}/device_authorization", + {"client_id": self.clientToken[0], "scope": "r_usr+w_usr+w_sub"}, ) if "status" in result and result["status"] != 200: raise Exception("Client ID not accepted by Tidal") @@ -99,9 +120,9 @@ class scraper: while elapsed < timeout and not auth: elapsed = time.time() - start result = self.post( - f"{AUTH_URL_BASE}/token", + f"{self.authUrlBase}/token", { - "client_id": CLIENT_ID, + "client_id": self.clientToken[0], "device_code": login.deviceCode, "grant_type": "urn:ietf:params:oauth:grant-type:device_code", "scope": "r_usr+w_usr+w_sub", @@ -126,16 +147,49 @@ class scraper: return auth raise Exception("Failed to log in") - def getTracks(self, obj) -> list: - url = API_URL_BASE - if type(obj) is Album: - url += f"/albums/{str(obj.id)}/items" - elif type(obj) is Playlist: - url += f"/playlists/{obj.uuid}/items" - else: - raise Exception("Tried to get tracks from incorrect object") - return self.getItems(url) + def getTracks(self, album: Album) -> list: + return self.getItems(f"{self.apiUrlBase}/albums/{str(album.id)}/items") + + def getAlbumFsPath(self, album: Album) -> str: + return self.downloadPath + album.title + + def getTrackFsPath(self, track) -> str: + return f"{self.downloadPath}/{self.getAlbumFsPath(track.album)}/[{track.number}] {track.title}" + + def getStreamInfo(self, track: Track) -> StreamInfo: + response = self.get( + f"tracks/{str(track.id)}/playbackinfopostpaywall", + { + "audioquality": self.quality, + "playbackmode": "STREAM", + "assetpresentation": "FULL", + }, + ) + if "vnd.tidal.bt" in response["manifestMimeType"]: + manifest = json.loads( + base64.b64decode(response["manifest"]).decode("utf-8") + ) + return StreamInfo( + response["trackid"], + response["audioQuality"], + manifest["codecs"], + manifest["keyId"] if "keyId" in manifest else "", + manifest["urls"][0], + ) + raise Exception("Can't read manifest of type {response['manifestMimeType']}") + + def downloadTrack(self, track, partSize=1048576) -> Tuple[bool, str]: + try: + stream = self.getStreamInfo(track.id) + path = self.getTrackFsPath(track) + print(f"Starting download of track \"{track.title}\"") + if not self.redownload and os.path.exists(path): + print(f"Skipping download, \"{track.title}\" already exists") + return True, "exists" + + def downloadAlbum(self, album: Album): tracks = self.getTracks(album) - # TODO: Continue working here + for i, track in enumerate(tracks): + self.downloadTrack(track)