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)