diff options
| -rw-r--r-- | pyproject.toml | 2 | ||||
| -rw-r--r-- | tidal_dl_ng/cli.py | 2 | ||||
| -rw-r--r-- | tidal_dl_ng/config.py | 2 | ||||
| -rw-r--r-- | tidal_dl_ng/constants.py | 6 | ||||
| -rw-r--r-- | tidal_dl_ng/download.py | 255 | ||||
| -rw-r--r-- | tidal_dl_ng/helper/exceptions.py | 8 | ||||
| -rw-r--r-- | tidal_dl_ng/helper/format.py | 18 | ||||
| -rw-r--r-- | tidal_dl_ng/helper/path.py | 78 | ||||
| -rw-r--r-- | tidal_dl_ng/helper/tidal.py | 13 | ||||
| -rw-r--r-- | tidal_dl_ng/metadata.py | 4 | ||||
| -rw-r--r-- | tidal_dl_ng/model/cfg.py | 6 | ||||
| -rw-r--r-- | tidal_dl_ng/model/tidal.py | 11 |
12 files changed, 267 insertions, 138 deletions
diff --git a/pyproject.toml b/pyproject.toml index 6dc1d2d..1bff7bc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -170,7 +170,7 @@ profile = "black" multi_line_output = 3 indent = 4 color_output = true -known_third_party = ["Crypto", "PySide6", "coloredlogs", "constants", "dataclasses_json", "ffmpeg", "m3u8", "mutagen", "pathvalidate", "requests", "rich", "tidalapi", "typer"] +known_third_party = ["Crypto", "PySide6", "coloredlogs", "constants", "dataclasses_json", "ffmpeg", "helper", "m3u8", "model", "mutagen", "pathvalidate", "requests", "rich", "tidalapi", "typer"] [tool.mypy] # https://mypy.readthedocs.io/en/latest/config_file.html#using-a-pyproject-toml-file diff --git a/tidal_dl_ng/cli.py b/tidal_dl_ng/cli.py index f5704b8..6f7925e 100644 --- a/tidal_dl_ng/cli.py +++ b/tidal_dl_ng/cli.py @@ -169,7 +169,7 @@ def download( media_type = MediaType.Video dl.item( - id_media=id_item, + media_id=id_item, media_type=media_type, path_base=settings.data.download_base_path, file_template=file_template, diff --git a/tidal_dl_ng/config.py b/tidal_dl_ng/config.py index aeb5492..0906655 100644 --- a/tidal_dl_ng/config.py +++ b/tidal_dl_ng/config.py @@ -62,6 +62,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.video_quality = tidalapi.VideoQuality.high self.file_path = path_file_token() self.token_from_storage = self.read(self.file_path) diff --git a/tidal_dl_ng/constants.py b/tidal_dl_ng/constants.py index d2c9e5b..1980db5 100644 --- a/tidal_dl_ng/constants.py +++ b/tidal_dl_ng/constants.py @@ -17,3 +17,9 @@ class MediaType(Enum): Playlist = "playlist" Album = "album" Mix = "mix" + + +class SkipExisting(Enum): + Disabled = False + Filename = "exact" + ExtensionIgnore = "extension_ignore" diff --git a/tidal_dl_ng/download.py b/tidal_dl_ng/download.py index a84996f..350150b 100644 --- a/tidal_dl_ng/download.py +++ b/tidal_dl_ng/download.py @@ -5,22 +5,26 @@ import random import shutil import tempfile import time +import xml.etree.ElementTree as ET +from collections.abc import Callable from logging import Logger from uuid import uuid4 import ffmpeg import m3u8 import requests -from pathvalidate import sanitize_filepath +from helper.format import is_json, is_xml +from helper.tidal import name_builder_item +from model.tidal import StreamManifest from requests.exceptions import HTTPError from rich.progress import Progress from tidalapi import Album, Mix, Playlist, Session, Track, UserPlaylist, Video from tidal_dl_ng.config import Settings -from tidal_dl_ng.constants import REQUESTS_TIMEOUT_SEC, MediaType +from tidal_dl_ng.constants import REQUESTS_TIMEOUT_SEC, MediaType, SkipExisting from tidal_dl_ng.helper.decryption import decrypt_file, decrypt_security_token -from tidal_dl_ng.helper.exceptions import MediaUnknown -from tidal_dl_ng.helper.path import check_file_exists, format_path_media, path_validate +from tidal_dl_ng.helper.exceptions import MediaMissing, MediaUnknown, UnknownManifestFormat +from tidal_dl_ng.helper.path import check_file_exists, format_path_media, path_file_sanitize from tidal_dl_ng.helper.wrapper import WrapperLogger from tidal_dl_ng.metadata import Metadata from tidal_dl_ng.model.gui_data import ProgressBars @@ -43,12 +47,15 @@ class RequestsClient: class Download: # TODO: Implement download cover 1280. session: Session = None - skip_existing: bool = False + skip_existing: SkipExisting = False - def __init__(self, session: Session, skip_existing: bool = False): + def __init__(self, session: Session, skip_existing: SkipExisting = SkipExisting.Disabled): self.session = session self.skip_existing = skip_existing + def _audio_mpeg_dash(self, audio: Track, path_file: str) -> str | None: + pass + def _video(self, video: Video, path_file: str) -> str | None: result: str | None = None m3u8_variant: m3u8.M3U8 = m3u8.load(video.get_url()) @@ -77,116 +84,157 @@ class Download: return result + def instantiate_media( + self, session: Session, media_type: MediaType.Track | MediaType.Video, id_media: str + ) -> Track | Video: + if media_type == MediaType.Track: + media = Track(session, id_media) + elif media_type == MediaType.Video: + media = Video(session, id_media) + else: + raise MediaUnknown + + return media + def item( self, path_base: str, - fn_logger: Logger | WrapperLogger, - id_media: str = None, - file_template: str = None, + file_template: str, + fn_logger: Callable, media: Track | Video = None, + media_id: str = None, media_type: MediaType = None, video_download: bool = True, progress_gui: ProgressBars = None, progress: Progress = None, ) -> (bool, str): - if id_media: - if media_type == MediaType.Track: - media = Track(self.session, id_media) - elif media_type == MediaType.Video: - media = Video(self.session, id_media) + # If only a media_id is provided, we need to create the media instance. + if media_id and media_type: + media = self.instantiate_media(self.session, media_type, media_id) + elif not media: + raise MediaMissing - # If video download is not allowed - if not video_download: - return False, "" - else: - raise MediaUnknown - else: - media = media + # If video download is not allowed end here + if not video_download: + fn_logger.info( + f"Video downloads are deactivated (see settings). Skipping video: {name_builder_item(media)}" + ) - if file_template: - file_name_relative = format_path_media(file_template, media) - path_file = os.path.abspath(os.path.normpath(os.path.join(path_base, file_name_relative))) - else: - path_file = os.path.abspath(os.path.normpath(format_path_media(path_base, media))) + return False, "" + + # Create file name and path + file_name_relative = format_path_media(file_template, media) + path_file = os.path.abspath(os.path.normpath(os.path.join(path_base, file_name_relative))) + # Compute the file extension + # TODO: Move further down? if isinstance(media, Track): stream = media.stream() - # TODO: Check for `manifest_mime_type'. It could be also xml. - stream_manifest = json.loads(base64.b64decode(stream.manifest).decode("utf-8")) - # TODO: Handle more than one dowload URL - stream_url = stream_manifest["urls"][0] - file_extension = self.get_file_extension(stream_url, stream_manifest["codecs"]) - elif isinstance(media, Video): - file_extension = ".ts" + stream_manifest = self.stream_manifest_parse(stream.manifest) + + # Sanitize final path_file to fit into OS boundaries. + path_file = path_file_sanitize(path_file, adapt=True) - path_file = sanitize_filepath(path_file + file_extension) - # Check if path & filename longer than the OS allows. Shorten if necessary. - validation_result, path_file = path_validate(path_file, True) + # Compute if and how downloads need to be skipped. + if self.skip_existing: + if self.skip_existing == SkipExisting.ExtensionIgnore: + extension_ignore = True + else: + extension_ignore = False - download_skip = check_file_exists(path_file) if self.skip_existing else False + # TODO: Check if extension is already in `path_file` or not. + download_skip = check_file_exists(path_file, extension_ignore=extension_ignore) + else: + download_skip = False if not download_skip: + # Create a temp directory and file. with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmp_path_dir: - tmp_path_file = os.path.join(tmp_path_dir, str(uuid4()) + file_extension) + tmp_path_file = os.path.join(tmp_path_dir, str(uuid4())) if isinstance(media, Track): - # TODO: Refactor to separate method. - if progress_gui is None: - progress_stdout: bool = True - else: - progress_stdout: bool = False - progress_gui.item_name.emit(media.name) - - try: - # Streaming, so we can iterate over the response. - r = requests.get(stream_url, stream=True, timeout=REQUESTS_TIMEOUT_SEC) - r.raise_for_status() - total_size_in_bytes = int(r.headers.get("content-length", 0)) - block_size = 4096 - p_task = progress.add_task( - f"[blue]Item '{media.name[:30]}'", - total=total_size_in_bytes / block_size, - visible=progress_stdout, - ) - - while not progress.tasks[p_task].finished: - with open(tmp_path_file, "wb") as f: - for data in r.iter_content(chunk_size=block_size): - f.write(data) - progress.advance(p_task, advance=1) - - if not progress_stdout: - progress_gui.item.emit(progress.tasks[p_task].percentage) - except HTTPError: - # TODO: Handle Exception... - pass - - needs_decryption = self.is_encrypted(stream_manifest) - - if needs_decryption: - key, nonce = decrypt_security_token(stream_manifest["encryptionKey"]) - tmp_path_file_decrypted = tmp_path_file + "_decrypted" - decrypt_file(tmp_path_file, tmp_path_file_decrypted, key, nonce) - else: - tmp_path_file_decrypted = tmp_path_file - - self.metadata_write(media, tmp_path_file_decrypted) + tmp_path_file = self._audio_stream( + fn_logger, media, progress, progress_gui, stream_manifest, tmp_path_file + ) elif isinstance(media, Video): - tmp_path_file_decrypted = self._video(media, tmp_path_file) + tmp_path_file = self._video(media, tmp_path_file) # TODO: Check if is possible to write metadata to MPEG Transport Stream files. # TODO: Make optional. + # Convert `*.ts` file to `*.mp4` using ffmpeg if True: - tmp_path_file_decrypted = self._video_convert(tmp_path_file_decrypted) + tmp_path_file = self._video_convert(tmp_path_file) path_file = os.path.splitext(path_file)[0] + ".mp4" + # Move final file to the configured destination directory. os.makedirs(os.path.dirname(path_file), exist_ok=True) - shutil.move(tmp_path_file_decrypted, path_file) + shutil.move(tmp_path_file, path_file) else: fn_logger.debug(f"Download skipped, since file exists: '{path_file}'") return not download_skip, path_file + def _audio_stream( + self, + fn_logger: Callable, + media: Track, + progress: Progress, + progress_gui: ProgressBars, + stream_manifest: StreamManifest, + path_file: str, + ): + # Set the correct progress output channel. + if progress_gui is None: + progress_stdout: bool = True + else: + progress_stdout: bool = False + progress_gui.item_name.emit(media.name) + + try: + # Download the media as stream, so we can iterate over the response. + r = requests.get(stream_manifest.stream_url, stream=True, timeout=REQUESTS_TIMEOUT_SEC) + + r.raise_for_status() + + # Get file size and compute progress steps + total_size_in_bytes = int(r.headers.get("content-length", 0)) + block_size = 4096 + p_task = progress.add_task( + f"[blue]Item '{media.name[:30]}'", + total=total_size_in_bytes / block_size, + visible=progress_stdout, + ) + + # Write content to file until progress is finished. + while not progress.tasks[p_task].finished: + with open(path_file, "wb") as f: + for data in r.iter_content(chunk_size=block_size): + f.write(data) + # Advance progress bar. + progress.advance(p_task) + + # To send the progress to the GUI, we need to emit the percentage. + if not progress_stdout: + progress_gui.item.emit(progress.tasks[p_task].percentage) + except HTTPError as e: + # TODO: Handle Exception... + fn_logger(e) + + # Check if file is encrypted. + needs_decryption = self.is_encrypted(stream_manifest.encryption_type) + + if needs_decryption: + key, nonce = decrypt_security_token(stream_manifest.encryption_key) + tmp_path_file_decrypted = path_file + "_decrypted" + decrypt_file(path_file, tmp_path_file_decrypted, key, nonce) + else: + tmp_path_file_decrypted = path_file + + # Write metadata to file. + self.metadata_write(media, tmp_path_file_decrypted) + + return tmp_path_file_decrypted + def cover_url(self, sid: str, width: int = 320, height: int = 320): if sid is None: return "" @@ -303,8 +351,8 @@ class Download: fn_logger.debug(f"Next download will start in {time_sleep} seconds.") time.sleep(time_sleep) - def is_encrypted(self, manifest: dict) -> bool: - result = manifest["encryptionType"] != "NONE" + def is_encrypted(self, encryption_type: str) -> bool: + result = encryption_type != "NONE" return result @@ -330,3 +378,42 @@ class Download: result, _ = ffmpeg.input(path_file).output(path_file_out, map=0, c="copy").run() return path_file_out + + def stream_manifest_parse(self, manifest: str) -> StreamManifest: + # Stream Manifest is base64 encoded. + manifest_parsed: str = base64.b64decode(manifest).decode("utf-8") + + if is_xml(manifest_parsed): + root = ET.fromstring(manifest_parsed) + stream_url: str = root[0][0][0][0].attrib["media"] + codecs: str = root[0][0][0].attrib["codecs"] + mime_type: str = root[0][0].attrib["mimeType"] + file_extension: str = self.get_file_extension(stream_url, codecs) + # TODO: Handle encryption key. But I have never seen an encrypted file so far. + encryption_type: str = "NONE" + encryption_key: str | None = None + elif is_json(manifest_parsed): + # JSON string to object. + stream_manifest = json.loads(manifest_parsed) + # TODO: Handle more than one dowload URL + stream_url: str = stream_manifest["urls"][0] + codecs: str = stream_manifest["codecs"] + mime_type: str = stream_manifest["mimeType"] + file_extension: str = self.get_file_extension(stream_url, codecs) + encryption_type: str = stream_manifest["encryptionType"] + encryption_key: str | None = ( + stream_manifest["encryptionKey"] if self.is_encrypted(encryption_type) else None + ) + else: + raise UnknownManifestFormat + + result: StreamManifest = StreamManifest( + stream_url=stream_url, + codecs=codecs, + file_extension=file_extension, + encryption_type=encryption_type, + encryption_key=encryption_key, + mime_type=mime_type, + ) + + return result diff --git a/tidal_dl_ng/helper/exceptions.py b/tidal_dl_ng/helper/exceptions.py index 9ef6ddc..be1bdee 100644 --- a/tidal_dl_ng/helper/exceptions.py +++ b/tidal_dl_ng/helper/exceptions.py @@ -4,3 +4,11 @@ class LoginError(Exception): class MediaUnknown(Exception): pass + + +class UnknownManifestFormat(Exception): + pass + + +class MediaMissing(Exception): + pass diff --git a/tidal_dl_ng/helper/format.py b/tidal_dl_ng/helper/format.py new file mode 100644 index 0000000..ab62a69 --- /dev/null +++ b/tidal_dl_ng/helper/format.py @@ -0,0 +1,18 @@ +def is_xml(value: str) -> bool: + result = False + + if value: + if value.startswith("<?xml"): + result = True + + return result + + +def is_json(value: str) -> bool: + result = False + + if value: + if value.startswith("{"): + result = True + + return result diff --git a/tidal_dl_ng/helper/path.py b/tidal_dl_ng/helper/path.py index 3061160..a907985 100644 --- a/tidal_dl_ng/helper/path.py +++ b/tidal_dl_ng/helper/path.py @@ -1,11 +1,11 @@ +import glob import os -import pathlib -import platform import re from collections.abc import Callable from pathlib import Path -from pathvalidate import sanitize_filename +from pathvalidate import sanitize_filename, sanitize_filepath +from pathvalidate.error import ValidationError from tidalapi import Album, Mix, Playlist, Track, UserPlaylist, Video @@ -121,59 +121,43 @@ def get_fn_format(media: Track | Album | Playlist | UserPlaylist | Video | Mix) return result -def length_max_name_file() -> int: - system = platform.system() - result: int = 255 +def path_file_sanitize(path_file: str, adapt: bool = False) -> (bool, str): + # Split into path and filename + pathname, filename = os.path.split(path_file) + # Sanitize path try: - if system in ["Darwin", "Linux"]: - result: int = os.pathconf("/", "PC_NAME_MAX") - except Exception as e: - # TODO: Implement propper logging. - print(e) - - return result - - -def length_max_name_path() -> int: - system = platform.system() - result: int = 255 + pathname_sanitized = sanitize_filepath(pathname, replacement_text=" ", validate_after_sanitize=True) + except ValidationError as e: + # If adaption of path is allowed in case of an error set path to HOME. + if adapt: + pathname_sanitized = Path.home() + else: + raise e + # Sanitize filename try: - if system in ["Darwin", "Linux"]: - result: int = os.pathconf("/", "PC_PATH_MAX") - except Exception as e: - # TODO: Implement propper logging. - print(e) - - return result + filename_sanitized = sanitize_filename(path_file, replacement_text=" ", validate_after_sanitize=True) + filename_sanitized_extension = Path(filename_sanitized).suffix + # Check if the file extension was removed by shortening the filename length + if filename_sanitized_extension == "": + # Add the original file extension + file_extension = "_" + Path(path_file).suffix + filename_sanitized = filename_sanitized[: -len(file_extension)] + file_extension + except ValidationError as e: + raise e -def path_validate(path_file: str, adapt: bool = False) -> (bool, str): - result: bool = False - length_max_path: int = length_max_name_path() - length_max_file: int = length_max_name_file() - path, file = os.path.split(path_file) - filename, extension = os.path.splitext(file) + # Join path and filename + result = os.path.join(pathname_sanitized, filename_sanitized) - if len(path) >= length_max_path: - result = False - - if adapt: - path = Path.home() - - if len(file) >= length_max_file: - result = False - - if adapt: - file = f'{filename[:length_max_file - len(extension + "_")]}_{extension}' - - path_file = os.path.join(path, file) + return result - return result, path_file +def check_file_exists(path_file: str, extension_ignore: bool = False): + if extension_ignore: + path_file = Path(path_file).stem + ".*" -def check_file_exists(path_file: str): - result = pathlib.Path(path_file).is_file() + result = True if glob.glob(path_file) else False return result diff --git a/tidal_dl_ng/helper/tidal.py b/tidal_dl_ng/helper/tidal.py new file mode 100644 index 0000000..78434b6 --- /dev/null +++ b/tidal_dl_ng/helper/tidal.py @@ -0,0 +1,13 @@ +from tidalapi import Track + + +def name_builder_artist(media: Track) -> str: + return ", ".join(artist.name for artist in media.artists) + + +def name_builder_title(media: Track) -> str: + return media.name + + +def name_builder_item(media: Track) -> str: + return f"{name_builder_artist(media)} - {name_builder_title(media)}" diff --git a/tidal_dl_ng/metadata.py b/tidal_dl_ng/metadata.py index f23c41d..651d8dd 100644 --- a/tidal_dl_ng/metadata.py +++ b/tidal_dl_ng/metadata.py @@ -34,13 +34,13 @@ class Metadata: type_audio: str = "", album: str = "", title: str = "", - artists: [str] | None = None, + artists: list[str] | None = None, copy_right: str = "", tracknumber: int = 0, discnumber: int = 0, totaltrack: int = 0, totaldisc: int = 0, - composer: [str] | None = None, + composer: list[str] | None = None, isrc: str = "", albumartist: str = "", date: str = "", diff --git a/tidal_dl_ng/model/cfg.py b/tidal_dl_ng/model/cfg.py index 74e0305..ee0a856 100644 --- a/tidal_dl_ng/model/cfg.py +++ b/tidal_dl_ng/model/cfg.py @@ -3,13 +3,14 @@ from dataclasses import dataclass from dataclasses_json import dataclass_json from tidalapi import Quality -from tidal_dl_ng.constants import QualityVideo +from tidal_dl_ng.constants import QualityVideo, SkipExisting @dataclass_json @dataclass class Settings: - skip_existing: bool = False + skip_existing: SkipExisting = SkipExisting.Disabled + skip_existing_ignore_extension: bool = False album_cover_save: bool = True lyrics_save: bool = False # TODO: Implement API KEY selection. @@ -32,7 +33,6 @@ class Settings: metadata_cover_height: int = 320 -@dataclass_json @dataclass class HelpSettings: skip_existing: str = "Do not download, if file already exists." diff --git a/tidal_dl_ng/model/tidal.py b/tidal_dl_ng/model/tidal.py new file mode 100644 index 0000000..a5fbae9 --- /dev/null +++ b/tidal_dl_ng/model/tidal.py @@ -0,0 +1,11 @@ +from dataclasses import dataclass + + +@dataclass +class StreamManifest: + stream_url: str + codecs: str + mime_type: str + file_extension: str + encryption_type: str | None = None + encryption_key: str | None = None |