♻️ Refactored download for CLI.
This commit is contained in:
@@ -4,5 +4,7 @@ import importlib.metadata
|
||||
try:
|
||||
# __package__ allows for the case where __name__ is "__main__"
|
||||
__version__ = importlib.metadata.version(__package__ or __name__)
|
||||
__name_display__ = importlib.metadata.metadata(__package__ or __name__).json["name"]
|
||||
except importlib.metadata.PackageNotFoundError:
|
||||
__version__ = "0.0.0"
|
||||
__name_display__ = __package__ or __name__
|
||||
|
||||
+28
-40
@@ -3,6 +3,7 @@ from pathlib import Path
|
||||
from typing import Annotated, Optional
|
||||
|
||||
import typer
|
||||
from helper.tidal import get_tidal_media_id, get_tidal_media_type
|
||||
from rich.live import Live
|
||||
from rich.panel import Panel
|
||||
from rich.progress import BarColumn, Console, Progress, SpinnerColumn, TextColumn
|
||||
@@ -12,8 +13,7 @@ from tidal_dl_ng import __version__
|
||||
from tidal_dl_ng.config import Settings, Tidal
|
||||
from tidal_dl_ng.constants import CTX_TIDAL, MediaType
|
||||
from tidal_dl_ng.download import Download
|
||||
from tidal_dl_ng.helper.path import path_file_settings
|
||||
from tidal_dl_ng.helper.url import get_tidal_media_id, get_tidal_media_type
|
||||
from tidal_dl_ng.helper.path import get_format_template, path_file_settings
|
||||
from tidal_dl_ng.helper.wrapper import WrapperLogger
|
||||
from tidal_dl_ng.model.cfg import HelpSettings
|
||||
|
||||
@@ -104,7 +104,7 @@ def login(ctx: typer.Context) -> bool:
|
||||
@app.command(name="dl")
|
||||
def download(
|
||||
ctx: typer.Context,
|
||||
urls_or_ids: Annotated[Optional[list[str]], typer.Argument()] = None,
|
||||
urls: Annotated[Optional[list[str]], typer.Argument()] = None,
|
||||
list_urls: Annotated[
|
||||
Optional[Path],
|
||||
typer.Option(
|
||||
@@ -120,77 +120,64 @@ def download(
|
||||
),
|
||||
] = None,
|
||||
):
|
||||
if not urls_or_ids:
|
||||
if not urls:
|
||||
# Read the text file provided.
|
||||
if list_urls:
|
||||
text = list_urls.read_text()
|
||||
urls_or_ids = text.splitlines()
|
||||
urls = text.splitlines()
|
||||
else:
|
||||
print("Provide either URLs, IDs or a file containing URLs (one per line).")
|
||||
|
||||
raise typer.Abort()
|
||||
|
||||
# Call login method to validate the token.
|
||||
ctx.invoke(login, ctx)
|
||||
|
||||
dl = Download(ctx.obj[CTX_TIDAL].session, ctx.obj[CTX_TIDAL].settings.data.skip_existing)
|
||||
# Create initial objects.
|
||||
settings: Settings = Settings()
|
||||
media_type: MediaType = None
|
||||
dl = Download(ctx.obj[CTX_TIDAL].session, ctx.obj[CTX_TIDAL].settings.data.skip_existing)
|
||||
progress: Progress = Progress(
|
||||
"{task.description}",
|
||||
SpinnerColumn(),
|
||||
BarColumn(),
|
||||
TextColumn("[progress.percentage]{task.percentage:>3.0f}%"),
|
||||
)
|
||||
progress_table = Table.grid()
|
||||
fn_logger = WrapperLogger(progress.print)
|
||||
progress_table = Table.grid()
|
||||
|
||||
# Style Progress display.
|
||||
progress_table.add_row(Panel.fit(progress, title="Download Progress", border_style="green", padding=(2, 2)))
|
||||
|
||||
for item in urls_or_ids:
|
||||
if "http" in item:
|
||||
media_name = get_tidal_media_type(item)
|
||||
id_item = get_tidal_media_id(item)
|
||||
else:
|
||||
media_name = False
|
||||
id_item = item
|
||||
for item in urls:
|
||||
media_type: str | bool = False
|
||||
|
||||
if not media_name:
|
||||
# Extract media name and id from link.
|
||||
if "http" in item:
|
||||
media_type = get_tidal_media_type(item)
|
||||
item_id = get_tidal_media_id(item)
|
||||
file_template = get_format_template(media_type, settings)
|
||||
|
||||
# If url is invalid skip to next url in list.
|
||||
if not media_type:
|
||||
print(f"It seems like that you have supplied an invalid URL: {item}")
|
||||
|
||||
continue
|
||||
|
||||
# Create Live display for Progress.
|
||||
with Live(progress_table, refresh_per_second=10):
|
||||
# TODO: Fix media type
|
||||
if media_name in ["track", "video"]:
|
||||
if media_name == "track":
|
||||
file_template = settings.data.format_track
|
||||
media_type = MediaType.Track
|
||||
elif media_name == "video":
|
||||
file_template = settings.data.format_video
|
||||
media_type = MediaType.Video
|
||||
|
||||
# Download media.
|
||||
if media_type in [MediaType.Track, MediaType.Video]:
|
||||
dl.item(
|
||||
media_id=id_item,
|
||||
media_id=item_id,
|
||||
media_type=media_type,
|
||||
path_base=settings.data.download_base_path,
|
||||
file_template=file_template,
|
||||
progress=progress,
|
||||
fn_logger=fn_logger,
|
||||
)
|
||||
elif media_name in ["album", "playlist", "mix"]:
|
||||
file_template: str | bool = False
|
||||
|
||||
if media_name == "album":
|
||||
file_template = settings.data.format_album
|
||||
media_type = MediaType.Album
|
||||
elif media_name == "playlist":
|
||||
file_template = settings.data.format_playlist
|
||||
media_type = MediaType.Playlist
|
||||
elif media_name == "mix":
|
||||
file_template = settings.data.format_mix
|
||||
media_type = MediaType.Mix
|
||||
|
||||
elif media_type in [MediaType.Album, MediaType.Playlist, MediaType.Mix]:
|
||||
dl.items(
|
||||
media_id=id_item,
|
||||
media_id=item_id,
|
||||
media_type=media_type,
|
||||
path_base=settings.data.download_base_path,
|
||||
file_template=file_template,
|
||||
@@ -200,6 +187,7 @@ def download(
|
||||
fn_logger=fn_logger,
|
||||
)
|
||||
|
||||
# Stop Progress display.
|
||||
progress.stop()
|
||||
|
||||
return True
|
||||
|
||||
@@ -8,7 +8,7 @@ import tidalapi
|
||||
from requests import HTTPError
|
||||
|
||||
from tidal_dl_ng.helper.decorator import SingletonMeta
|
||||
from tidal_dl_ng.helper.path import path_file_settings, path_file_token
|
||||
from tidal_dl_ng.helper.path import path_base, path_file_settings, path_file_token
|
||||
from tidal_dl_ng.model.cfg import Settings as ModelSettings
|
||||
from tidal_dl_ng.model.cfg import Token as ModelToken
|
||||
|
||||
@@ -17,10 +17,14 @@ class BaseConfig:
|
||||
data: ModelSettings | ModelToken = None
|
||||
file_path: str = None
|
||||
cls_model: object = None
|
||||
path_base: str = path_base()
|
||||
|
||||
def save(self) -> None:
|
||||
data_json = self.data.to_json()
|
||||
|
||||
# Try to create the base folder.
|
||||
os.makedirs(self.path_base, exist_ok=True)
|
||||
|
||||
with open(self.file_path, encoding="utf-8", mode="w") as f:
|
||||
f.write(data_json)
|
||||
|
||||
@@ -79,8 +83,8 @@ class Tidal(BaseConfig, metaclass=SingletonMeta):
|
||||
|
||||
def __init__(self, settings: Settings = None):
|
||||
self.session = tidalapi.Session()
|
||||
self.session.config.client_id = "6BDSRdpK9hqEBTgU"
|
||||
self.session.config.client_secret = "xeuPmY7nbpZ9IIbLAcQ93shka1VNheUAqN6IcszjTG8="
|
||||
# self.session.config.client_id = "km8T1xS355y7dd3H"
|
||||
# self.session.config.client_secret = "vcmeGW1OuZ0fWYMCSZ6vNvSLJlT3XEpW0ambgYt5ZuI="
|
||||
self.session.video_quality = tidalapi.VideoQuality.high
|
||||
self.file_path = path_file_token()
|
||||
self.token_from_storage = self.read(self.file_path)
|
||||
|
||||
@@ -267,7 +267,9 @@ class Download:
|
||||
totaltrack=track.album.num_tracks if track.album and track.album.num_tracks else 1,
|
||||
totaldisc=track.album.num_volumes if track.album and track.album.num_volumes else 1,
|
||||
discnumber=track.volume_num if track.volume_num else 1,
|
||||
url_cover=self.cover_url(track.album.cover, self.settings.data.metadata_cover_dimension) if track.album else "",
|
||||
url_cover=(
|
||||
self.cover_url(track.album.cover, self.settings.data.metadata_cover_dimension) if track.album else ""
|
||||
),
|
||||
)
|
||||
|
||||
m.save()
|
||||
|
||||
+4
-18
@@ -2,6 +2,8 @@ import math
|
||||
import sys
|
||||
from collections.abc import Callable
|
||||
|
||||
from helper.path import get_format_template
|
||||
|
||||
try:
|
||||
import qdarktheme
|
||||
from PySide6 import QtCore, QtGui, QtWidgets
|
||||
@@ -409,13 +411,9 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
|
||||
item=self.s_item_advance, list_item=self.s_list_advance, item_name=self.s_item_name
|
||||
)
|
||||
progress: Progress = Progress()
|
||||
file_template = get_format_template(media, self.settings)
|
||||
|
||||
if isinstance(media, Track | Video):
|
||||
if isinstance(media, Track):
|
||||
file_template: str = self.settings.data.format_track
|
||||
elif isinstance(media, Video):
|
||||
file_template: str = self.settings.data.format_video
|
||||
|
||||
result_download, download_path_file = dl.item(
|
||||
media=media,
|
||||
path_base=self.settings.data.download_base_path,
|
||||
@@ -430,19 +428,7 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
|
||||
else:
|
||||
logger_gui.info(f"Download skipped (file exists): {download_path_file}")
|
||||
elif isinstance(media, Album | Playlist | Mix):
|
||||
file_template: str | bool = False
|
||||
|
||||
if isinstance(media, Album):
|
||||
file_template: str = self.settings.data.format_album
|
||||
progress_name = media.name
|
||||
elif isinstance(media, Playlist):
|
||||
file_template: str = self.settings.data.format_playlist
|
||||
progress_name = media.name
|
||||
elif isinstance(media, Mix):
|
||||
file_template: str = self.settings.data.format_mix
|
||||
progress_name = media.title
|
||||
else:
|
||||
progress_name = "List N/A"
|
||||
progress_name = media.name if media.name else media.title if media.title else "List N/A"
|
||||
|
||||
self.progress_list_name(progress_name)
|
||||
|
||||
|
||||
@@ -8,8 +8,11 @@ from pathvalidate import sanitize_filename, sanitize_filepath
|
||||
from pathvalidate.error import ValidationError
|
||||
from tidalapi import Album, Mix, Playlist, Track, UserPlaylist, Video
|
||||
|
||||
from tidal_dl_ng import __name_display__
|
||||
from tidal_dl_ng.constants import MediaType
|
||||
|
||||
def path_base():
|
||||
|
||||
def path_home() -> str:
|
||||
if "XDG_CONFIG_HOME" in os.environ:
|
||||
return os.environ["XDG_CONFIG_HOME"]
|
||||
elif "HOME" in os.environ:
|
||||
@@ -20,16 +23,41 @@ def path_base():
|
||||
return os.path.abspath("./")
|
||||
|
||||
|
||||
def path_file_log():
|
||||
return os.path.join(path_base(), ".tidal-dl-ng.log")
|
||||
def path_base() -> str:
|
||||
path_config: str = ".config"
|
||||
path_base: str = os.path.join(path_home(), path_config, __name_display__)
|
||||
|
||||
return path_base
|
||||
|
||||
|
||||
def path_file_token():
|
||||
return os.path.join(path_base(), ".tidal-dl-ng_token.json")
|
||||
def path_file_log() -> str:
|
||||
# TODO: Remove this soon. Only for migration to new dir.
|
||||
old = os.path.join(path_home(), ".tidal-dl-ng.log")
|
||||
if os.path.isfile(old):
|
||||
os.makedirs(path_base(), exist_ok=True)
|
||||
os.rename(old, os.path.join(path_base(), "app.log"))
|
||||
|
||||
return os.path.join(path_base(), "app.log")
|
||||
|
||||
|
||||
def path_file_settings():
|
||||
return os.path.join(path_base(), ".tidal-dl-ng_settings.json")
|
||||
def path_file_token() -> str:
|
||||
# TODO: Remove this soon. Only for migration to new dir.
|
||||
old = os.path.join(path_home(), ".tidal-dl-ng_token.json")
|
||||
if os.path.isfile(old):
|
||||
os.makedirs(path_base(), exist_ok=True)
|
||||
os.rename(old, os.path.join(path_base(), "token.json"))
|
||||
|
||||
return os.path.join(path_base(), "token.json")
|
||||
|
||||
|
||||
def path_file_settings() -> str:
|
||||
# TODO: Remove this soon. Only for migration to new dir.
|
||||
old = os.path.join(path_home(), ".tidal-dl-ng_settings.json")
|
||||
if os.path.isfile(old):
|
||||
os.makedirs(path_base(), exist_ok=True)
|
||||
os.rename(old, os.path.join(path_base(), "settings.json"))
|
||||
|
||||
return os.path.join(path_base(), "settings.json")
|
||||
|
||||
|
||||
def format_path_media(fmt_template: str, media: Track | Album | Playlist | UserPlaylist | Video | Mix) -> str:
|
||||
@@ -38,7 +66,7 @@ def format_path_media(fmt_template: str, media: Track | Album | Playlist | UserP
|
||||
# Search track format template for placeholder.
|
||||
regex = r"\{(.+?)\}"
|
||||
matches = re.finditer(regex, fmt_template, re.MULTILINE)
|
||||
fn_format = get_fn_format(media)
|
||||
fn_format = get_format_fn(media)
|
||||
|
||||
for _matchNum, match in enumerate(matches, start=1):
|
||||
template_str = match.group()
|
||||
@@ -104,7 +132,7 @@ def format_str_video(name: str, media: Video) -> str | bool:
|
||||
return result
|
||||
|
||||
|
||||
def get_fn_format(media: Track | Album | Playlist | UserPlaylist | Video | Mix) -> Callable:
|
||||
def get_format_fn(media: Track | Album | Playlist | UserPlaylist | Video | Mix) -> Callable:
|
||||
result = None
|
||||
|
||||
if isinstance(media, Track):
|
||||
@@ -121,6 +149,25 @@ def get_fn_format(media: Track | Album | Playlist | UserPlaylist | Video | Mix)
|
||||
return result
|
||||
|
||||
|
||||
def get_format_template(
|
||||
media: Track | Album | Playlist | UserPlaylist | Video | Mix | MediaType, settings
|
||||
) -> str | bool:
|
||||
result = False
|
||||
|
||||
if isinstance(media, Track) or media == MediaType.Track:
|
||||
result = settings.data.format_track
|
||||
elif isinstance(media, Album) or media == MediaType.Album:
|
||||
result = settings.data.format_album
|
||||
elif isinstance(media, Playlist | UserPlaylist) or media == MediaType.Playlist:
|
||||
result = settings.data.format_playlist
|
||||
elif isinstance(media, Mix) or media == MediaType.Mix:
|
||||
result = settings.data.format_mix
|
||||
elif isinstance(media, Video) or media == MediaType.Video:
|
||||
result = settings.data.format_video
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def path_file_sanitize(path_file: str, adapt: bool = False) -> (bool, str):
|
||||
# Split into path and filename
|
||||
pathname, filename = os.path.split(path_file)
|
||||
|
||||
@@ -1,5 +1,7 @@
|
||||
from tidalapi import Track
|
||||
|
||||
from tidal_dl_ng.constants import MediaType
|
||||
|
||||
|
||||
def name_builder_artist(media: Track) -> str:
|
||||
return ", ".join(artist.name for artist in media.artists)
|
||||
@@ -11,3 +13,31 @@ def name_builder_title(media: Track) -> str:
|
||||
|
||||
def name_builder_item(media: Track) -> str:
|
||||
return f"{name_builder_artist(media)} - {name_builder_title(media)}"
|
||||
|
||||
|
||||
def get_tidal_media_id(url_or_id_media: str) -> str:
|
||||
id_dirty = url_or_id_media.rsplit("/", 1)[-1]
|
||||
id_media = id_dirty.rsplit("?", 1)[0]
|
||||
|
||||
return id_media
|
||||
|
||||
|
||||
def get_tidal_media_type(url_media: str) -> MediaType | bool:
|
||||
result: MediaType | bool = False
|
||||
url_split = url_media.split("/")[-2]
|
||||
|
||||
if len(url_split) > 1:
|
||||
media_name = url_media.split("/")[-2]
|
||||
|
||||
if media_name == "track":
|
||||
result = MediaType.Track
|
||||
elif media_name == "video":
|
||||
result = MediaType.Video
|
||||
elif media_name == "album":
|
||||
result = MediaType.Album
|
||||
elif media_name == "playlist":
|
||||
result = MediaType.Playlist
|
||||
elif media_name == "mix":
|
||||
result = MediaType.Mix
|
||||
|
||||
return result
|
||||
|
||||
@@ -1,15 +0,0 @@
|
||||
def get_tidal_media_id(url_or_id_media: str) -> str:
|
||||
id_dirty = url_or_id_media.rsplit("/", 1)[-1]
|
||||
id_media = id_dirty.rsplit("?", 1)[0]
|
||||
|
||||
return id_media
|
||||
|
||||
|
||||
def get_tidal_media_type(url_media: str) -> str | bool:
|
||||
result = False
|
||||
url_split = url_media.split("/")[-2]
|
||||
|
||||
if len(url_split) > 1:
|
||||
result = url_media.split("/")[-2]
|
||||
|
||||
return result
|
||||
@@ -11,15 +11,15 @@ from tidal_dl_ng.constants import CoverDimensions, QualityVideo, SkipExisting
|
||||
class Settings:
|
||||
skip_existing: SkipExisting = SkipExisting.Disabled
|
||||
# TODO: Implement cover download to a separate file.
|
||||
#album_cover_save: bool = True
|
||||
# album_cover_save: bool = True
|
||||
lyrics_save: bool = False
|
||||
# TODO: Implement API KEY selection.
|
||||
#api_key_index: bool = 0
|
||||
# api_key_index: bool = 0
|
||||
# TODO: Implement album info download to separate file.
|
||||
#album_info_save: bool = False
|
||||
# album_info_save: bool = False
|
||||
video_download: bool = True
|
||||
# TODO: Implement multi threading for downloads.
|
||||
#multi_thread: bool = False
|
||||
# multi_thread: bool = False
|
||||
download_delay: bool = True
|
||||
download_base_path: str = "./download"
|
||||
quality_audio: Quality = Quality.low_320k
|
||||
|
||||
Reference in New Issue
Block a user