snailed/tidal-scraper
2
0
Fork 0
Code Activity

readme, bugfixes better UX

Browse source
This commit is contained in:
Luca Bilke 2023-07-18 14:25:14 +02:00
commit 312f0ce9d8
No known key found for this signature in database
GPG key ID: 7B77C51E8C779E75
9 changed files with 223 additions and 121 deletions

73
tidal_scraper/cli.py
View file

@ -1,34 +1,47 @@
from tidalapi import Album, Track, Artist, Playlist
from tidal_scraper.state import State
from tidal_scraper.helper import get_conf, write_conf
from tidal_scraper.download import (
download_album,
download_playlist,
download_track,
download_artist,
)
from tidal_scraper.state import State
from tidal_scraper.helper import get_conf, human_sleep
import sys
from tidalapi import Album, Track, Artist, Playlist
from argparse import ArgumentParser, Namespace
from pathlib import Path
def save_state(
state: State, obj: Playlist | Track | Album | Artist, state_file: str
) -> None:
state.set_dl_state(obj, True)
state.write_dl_state(state_file)
def handle_favorites(state: State, conf: dict, args: Namespace) -> None:
match args.obj:
case "album":
for album in state.favorites.albums():
download_album(album, conf)
human_sleep()
if not state.state_downloaded(album):
download_album(album, conf)
save_state(state, album, conf["state_dir"] + "/state.json")
case "track":
for track in state.favorites.tracks():
download_track(track, conf)
human_sleep()
if not state.state_downloaded(track):
download_track(track, True, conf)
save_state(state, track, conf["state_dir"] + "/state.json")
case "artist":
for artist in state.favorites.artists():
download_artist(artist, conf)
human_sleep()
if not state.state_downloaded(artist):
download_artist(artist, conf)
save_state(state, artist, conf["state_dir"] + "/state.json")
case "playlist":
for playlist in state.favorites.playlists():
download_playlist(playlist, conf)
human_sleep()
if not state.state_downloaded(playlist):
download_playlist(playlist, conf)
save_state(state, playlist, conf["state_dir"] + "/state.json")
def handle_id(state: State, conf: dict, args: Namespace) -> None:
@ -36,15 +49,19 @@ def handle_id(state: State, conf: dict, args: Namespace) -> None:
case "album":
album = Album(state.session, args.id)
download_album(album, conf)
state.set_dl_state(album, True)
case "track":
track = Track(state.session, args.id)
download_track(track, conf)
download_track(track, False, conf)
state.set_dl_state(track, True)
case "artist":
artist = Artist(state.session, args.id)
download_artist(artist, conf)
state.set_dl_state(artist, True)
case "playlist":
playlist = Playlist(state.session, args.id)
download_playlist(playlist, conf)
state.set_dl_state(playlist, True)
def run():
@ -55,6 +72,13 @@ def run():
parser.add_argument(
"-s", "--state", help="Directory to keep state in", type=Path, dest="state_dir"
)
parser.add_argument(
"-C",
"--create-conf",
action="store_true",
help="Create config file and exit",
dest="create_conf",
)
obj = parser.add_mutually_exclusive_group(required=True)
obj.add_argument(
"-a",
@ -105,13 +129,22 @@ def run():
)
args = parser.parse_args()
conf = get_conf(args.state_dir, args.conf_file)
state = State(conf["user_id"], conf["quality"], conf["state_dir"])
state.login(conf["state_dir"] + "auth.json")
try:
conf = get_conf(args.state_dir, args.conf_file)
except FileNotFoundError:
write_conf(notify=True)
sys.exit()
if args.favorite:
handle_favorites(state, conf, args)
elif args.id is not None:
handle_id(state, conf, args)
state = State(conf)
state.load_dl_state(conf["state_dir"] + "state.json")
state.login()
state.write_dl_state(conf["state_dir"] + "state.json")
try:
if args.create_conf:
write_conf(notify=True)
elif args.favorite:
handle_favorites(state, conf, args)
elif args.id is not None:
handle_id(state, conf, args)
except KeyboardInterrupt:
print("Bye bye! Come back again!")

34
tidal_scraper/download.py
View file

@ -1,11 +1,13 @@
import tidal_scraper.metadata as metadata
from tidal_scraper.helper import extensions, clean_template, log_error, human_sleep
from tidal_scraper.helper import extensions, clean_template, log_error
import tidalapi
import os
import json
import requests
import io
import time
import random
from tqdm import tqdm
from base64 import b64decode
from Crypto.Cipher import AES
@ -55,21 +57,23 @@ def __download_file(url: str, fp: BinaryIO) -> str:
def download_track(
track: tidalapi.Track,
sleep: bool,
conf: dict | None = None,
name_template: str | None = None,
dest_dir: str | None = None,
skip_dl: bool | None = None,
skip_downloaded: bool | None = None,
errorfile: str | None = None,
) -> None:
album = track.album
assert album
assert album.artist
if conf is None:
assert skip_dl is not None
assert skip_downloaded is not None
assert errorfile is not None
assert name_template is not None
assert dest_dir is not None
else:
skip_dl = skip_dl or conf["skip_downloaded"]
skip_downloaded = skip_downloaded or conf["skip_downloaded"]
errorfile = errorfile or conf["error_log"]
dest_dir = dest_dir or conf["dest_dir"]
name_template = name_template or conf["track_name"]
@ -79,7 +83,6 @@ def download_track(
http_failures = 0
while http_failures <= 3:
try:
print("running")
stream = track.stream()
manifest = json.loads(b64decode(stream.manifest))
url = manifest["urls"][0]
@ -94,8 +97,11 @@ def download_track(
if ext in url:
dest += ext
break
if os.path.exists(dest) and skip_dl:
print("Skipping track")
if os.path.exists(dest) and skip_downloaded:
print("Skipping track\n")
if sleep:
t = random.randrange(750, 1500) / 1000
time.sleep(t)
return
assert track.name and album.name
@ -118,12 +124,17 @@ def download_track(
data = b.getvalue()
f.write(data)
print()
if sleep:
t = random.randrange(1000, 5000) / 1000
time.sleep(t)
break
except requests.HTTPError:
http_failures += 1
t = random.randrange(10000, 20000) / 1000
time.sleep(t)
except KeyboardInterrupt as e:
raise e
except:
except Exception:
log_error(
errorfile or "error.log",
"Failure while downloading {artist} - {track}",
@ -179,8 +190,7 @@ def download_album(album: tidalapi.Album, conf: dict) -> None:
download_cover(album, conf)
tracks = album.tracks()
for track in tracks:
download_track(track, conf, dest_dir=dest)
human_sleep()
download_track(track, True, conf, dest_dir=dest)
def download_playlist(playlist: tidalapi.Playlist, conf: dict) -> None:
@ -192,12 +202,10 @@ def download_playlist(playlist: tidalapi.Playlist, conf: dict) -> None:
download_cover(playlist, conf)
tracks = playlist.tracks()
for track in tracks:
download_track(track, conf, dest_dir=dest)
human_sleep()
download_track(track, True, conf, dest_dir=dest)
def download_artist(artist: tidalapi.Artist, conf: dict) -> None:
albums = artist.get_albums()
for album in albums:
download_album(album, conf)
human_sleep()

79
tidal_scraper/helper.py
View file

@ -1,33 +1,77 @@
import re
import os
import time
import random
import tomllib
import sys
import traceback
from pprint import pp
extensions = [".flac", ".mp4", ".m4a", ""]
conf_text = """# All options postfixed "_dir" are expected to have a trailing slash!
skip_downloaded = true
error_log = "error.log"
# Quality can be one of "master, lossless, high and low"
# Keep in mind that you can only get the quality included in your tidal sub
quality = "lossless"
user_id =
dest_dir = "./downloads/"
# Leave empty for default state location
# Default location is the first valid path in this order:
# XDG_STATE_HOME/tidal-scraper/
# XDG_CACHE_HOME/tidal-scraper/
# $HOME/.cache/tidal-scraper/
state_dir =
# These templates are passed their respective tidalapi objects
# Possible attributes can be found here: https://tidalapi.netlify.app/api.html
album_dir = "{album.artist.name}/{album.name}/"
playlist_dir = "{playlist.name}/"
# Track receives "albumartist" and "trackartist", but no "artist"
track_name = "{track.track_num}: {track.artist.name} - {track.name}"
# One of 160, 320, 480, 640, 750, 1080
playlist_image_size = 1080
# One of 80, 160, 320, 640, 1280
album_image_size = 1280
"""
def get_conf(state_dir: str | None = None, conf_file: str | None = None) -> dict:
def default_conf_file() -> str:
home = os.getenv("HOME")
assert home
conf_file = (
return (
os.getenv("XDG_CONFIG_HOME") or home + "/.config"
) + "/tidal-scraper/conf.toml"
def default_state_file() -> str:
home = os.getenv("HOME")
assert home
return (
os.getenv("XDG_STATE_HOME")
or os.getenv("XDG_CACHE_HOME")
or home + "/.cache"
) + "/tidal-scraper/"
def write_conf(conf_file: str | None = None, notify: bool = False) -> None:
if not conf_file:
conf_file = default_conf_file()
with open(conf_file, "w") as f:
f.write(conf_text)
if notify:
print(f"Config file created at {conf_file}. Make sure to set your user ID!")
def get_conf(state_dir: str | None = None, conf_file: str | None = None) -> dict:
if not conf_file:
conf_file = default_conf_file()
with open(conf_file, "rb") as f:
conf = tomllib.load(f)
if not state_dir:
state_dir = (
os.getenv("XDG_STATE_HOME")
or os.getenv("XDG_CACHE_HOME")
or home + "/.cache"
)
state_dir += "/tidal-scraper"
conf["state_dir"] = state_dir
state_dir = default_state_file()
conf["state_dir"] = state_dir
return conf
@ -39,14 +83,9 @@ def clean_template(path: str, **kwargs) -> str:
return "/".join(cleaned_split)
def log_error(logfile: str, template: str, **kwargs):
def log_error(logfile: str, template: str, **kwargs) -> None:
with open(logfile, "a") as f:
msg = template.format(**kwargs)
f.write(msg + "\n")
traceback.format_exception(*sys.exc_info())
pp(traceback.format_exception(*sys.exc_info()), stream=f, indent=4)
f.write("\n\n")
def human_sleep() -> None:
t = random.randrange(1000, 5000) / 1000
time.sleep(t)

2
tidal_scraper/metadata.py
View file

@ -1,8 +1,6 @@
from mutagen import flac, mp4
from mutagen.mp4 import MP4Tags
from mutagen._vorbis import VCommentDict
from typing import BinaryIO

100
tidal_scraper/state.py
View file

@ -1,54 +1,59 @@
from tidal_scraper.helper import log_error
import json
from datetime import datetime
from tidalapi import session, user, playlist, media, album, artist
from helper import log_error
from tidalapi import session, user, playlist, media, album, artist, Quality
class State:
def __init__(
self,
user_id: int,
quality: str,
dl_state_path: str,
conf: dict | None = None,
state_dir: str | None = None,
user_id: int | None = None,
quality: str | None = None,
errorfile: str | None = None,
):
if conf is None:
assert user_id is not None
assert quality is not None
assert state_dir is not None
assert errorfile is not None
else:
user_id = user_id or conf["user_id"]
quality = quality or conf["quality"]
state_dir = state_dir or conf["state_dir"]
errorfile = errorfile or conf["error_log"]
match quality:
case "master":
q = session.Quality.master
q = Quality.master
case "lossless":
q = session.Quality.lossless
q = Quality.lossless
case "high":
q = session.Quality.high
q = Quality.high
case "low":
q = session.Quality.low
q = Quality.low
case _:
raise Exception("Bad Quality String")
config = session.Config(quality=q)
api_config = session.Config(quality=q)
self.conf = conf
self.user_id = user_id
self.session = session.Session(config)
self.session = session.Session(api_config)
self.favorites = user.Favorites(self.session, user_id)
try:
self.load_dl_state(dl_state_path)
except (FileNotFoundError, IndexError, AssertionError):
log_error(
errorfile or "error.log",
f"Could not find state file at {dl_state_path}",
)
self._state = {
"albums": {},
"artists": {},
"playlists": {},
"tracks": {},
}
self.errorfile = errorfile
self._state = {
"albums": {},
"artists": {},
"playlists": {},
"tracks": {},
}
def login(self, auth_file: str | None = None) -> None:
s = self.session
if auth_file is None:
assert self.conf is not None
auth_file = self.conf["state_dir"] + "auth.json"
try:
assert auth_file
with open(auth_file, "r") as f:
@ -75,7 +80,7 @@ class State:
"expiry_time": s.expiry_time.timestamp(),
}
with open(auth_file, "w") as f:
json.dump(data, f)
json.dump(data, fp=f, indent=4)
assert self.session.check_login()
@ -94,19 +99,46 @@ class State:
case media.Track:
t = "tracks"
case _:
raise Exception("Incorrect object type received")
raise Exception("Object of incorrect type received")
self._state[t][obj.id] = downloaded
def state_downloaded(
self, obj: playlist.Playlist | media.Track | album.Album | artist.Artist
) -> bool:
match type(obj):
case album.Album:
t = "albums"
case artist.Artist:
t = "artists"
case playlist.Playlist:
t = "playlists"
case media.Track:
t = "tracks"
case _:
raise Exception("Object of incorrect type received")
return self._state[t].get(str(obj.id), False)
def write_dl_state(self, statefile: str) -> None:
with open(statefile, "w") as f:
json.dump(self._state, f)
json.dump(self._state, fp=f, indent=4)
def load_dl_state(self, statefile: str) -> None:
with open(statefile, "r") as f:
self._state = json.load(f)
def load_dl_state(self, state_file: str) -> None:
try:
with open(state_file, "r") as f:
self._state = json.load(f)
assert type(self._state["albums"]) is dict[int, bool]
assert type(self._state["artists"]) is dict[int, bool]
assert type(self._state["playlists"]) is dict[int, bool]
assert type(self._state["tracks"]) is dict[int, bool]
for t in self._state.values():
for k, v in t.items():
assert isinstance(k, (str, type(None)))
assert isinstance(v, (bool, type(None)))
except (FileNotFoundError, IndexError, KeyError):
log_error(
self.errorfile or "error.log",
f"Could not find state file at {state_file}",
)
except (json.decoder.JSONDecodeError, AssertionError):
log_error(
self.errorfile or "error.log",
f"Statefile at {state_file} is malformed",
)