♻️ Some refactoring started within download.py.

✨ Track extension can be ignored while skipping files. Closes #11.
✨ Started to implement MPEG DASH download support.

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

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,

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)

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"

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.exceptions import MediaMissing, MediaUnknown, UnknownManifestFormat
-from tidal_dl_ng.helper.path import check_file_exists, format_path_media, path_validate
+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,
+        file_template: str,
-        id_media: str = None,
+        fn_logger: Callable,
-        file_template: str = None,
         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 only a media_id is provided, we need to create the media instance.
-            if media_type == MediaType.Track:
+        if media_id and media_type:
-                media = Track(self.session, id_media)
+            media = self.instantiate_media(self.session, media_type, media_id)
-            elif media_type == MediaType.Video:
+        elif not media:
-                media = Video(self.session, id_media)
+            raise MediaMissing
 
-                # If video download is not allowed
+        # If video download is not allowed end here
-                if not video_download:
+        if not video_download:
-                    return False, ""
+            fn_logger.info(
-            else:
+                f"Video downloads are deactivated (see settings). Skipping video: {name_builder_item(media)}"
-                raise MediaUnknown
+            )
-        else:
-            media = media
 
-        if file_template:
+            return False, ""
-            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)))
 
+        # 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 = self.stream_manifest_parse(stream.manifest)
-            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"
 
-        path_file = sanitize_filepath(path_file + file_extension)
+        # Sanitize final path_file to fit into OS boundaries.
-        # Check if path & filename longer than the OS allows. Shorten if necessary.
+        path_file = path_file_sanitize(path_file, adapt=True)
-        validation_result, path_file = path_validate(path_file, True)
 
-        download_skip = check_file_exists(path_file) if self.skip_existing else False
+        # 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
+
+            # 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.
+                    tmp_path_file = self._audio_stream(
-                    if progress_gui is None:
+                        fn_logger, media, progress, progress_gui, stream_manifest, tmp_path_file
-                        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)
                 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:
+    def is_encrypted(self, encryption_type: str) -> bool:
-        result = manifest["encryptionType"] != "NONE"
+        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

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

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

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:
+def path_file_sanitize(path_file: str, adapt: bool = False) -> (bool, str):
-    system = platform.system()
+    # Split into path and filename
-    result: int = 255
+    pathname, filename = os.path.split(path_file)
 
+    # Sanitize path
     try:
-        if system in ["Darwin", "Linux"]:
+        pathname_sanitized = sanitize_filepath(pathname, replacement_text=" ", validate_after_sanitize=True)
-            result: int = os.pathconf("/", "PC_NAME_MAX")
+    except ValidationError as e:
-    except Exception as e:
+        # If adaption of path is allowed in case of an error set path to HOME.
-        # TODO: Implement propper logging.
+        if adapt:
-        print(e)
+            pathname_sanitized = Path.home()
-
+        else:
-    return result
+            raise e
-
-
-def length_max_name_path() -> int:
-    system = platform.system()
-    result: int = 255
 
+    # Sanitize filename
     try:
-        if system in ["Darwin", "Linux"]:
+        filename_sanitized = sanitize_filename(path_file, replacement_text=" ", validate_after_sanitize=True)
-            result: int = os.pathconf("/", "PC_PATH_MAX")
+        filename_sanitized_extension = Path(filename_sanitized).suffix
-    except Exception as e:
+
-        # TODO: Implement propper logging.
+        # Check if the file extension was removed by shortening the filename length
-        print(e)
+        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
+
+    # Join path and filename
+    result = os.path.join(pathname_sanitized, filename_sanitized)
 
     return result
 
 
-def path_validate(path_file: str, adapt: bool = False) -> (bool, str):
+def check_file_exists(path_file: str, extension_ignore: bool = False):
-    result: bool = False
+    if extension_ignore:
-    length_max_path: int = length_max_name_path()
+        path_file = Path(path_file).stem + ".*"
-    length_max_file: int = length_max_name_file()
-    path, file = os.path.split(path_file)
-    filename, extension = os.path.splitext(file)
 
-    if len(path) >= length_max_path:
+    result = True if glob.glob(path_file) else False
-        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, path_file
-
-
-def check_file_exists(path_file: str):
-    result = pathlib.Path(path_file).is_file()
 
     return result

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)}"

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 = "",

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."

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