Merge pull request #34 from oskvr37/dev

setup.py

@@ -2,7 +2,7 @@ from setuptools import setup, find_packages
 
 setup(
     name="tiddl",
-    version="1.6.0",
+    version="1.7.0",
     description="TIDDL (Tidal Downloader) is a Python CLI application that allows downloading Tidal tracks.",
     long_description=open('README.md', encoding="utf-8").read(),
     long_description_content_type='text/markdown',

tiddl/__init__.py

@@ -6,7 +6,7 @@ from random import randint
 from .api import TidalApi
 from .auth import getDeviceAuth, getToken, refreshToken
 from .config import Config, HOME_DIRECTORY
-from .download import downloadTrackStream, downloadCover
+from .download import downloadTrackStream, Cover
 from .parser import QUALITY_ARGS, parser
 from .types import TRACK_QUALITY, TrackQuality, Track
 from .utils import (
@@ -19,6 +19,8 @@ from .utils import (
     initLogging,
 )
 
+SAVE_COVER = True
+
 
 def main():
     args = parser.parse_args()
@@ -121,13 +123,15 @@ def main():
     )
 
     def downloadTrack(
-        track: Track, file_template: str, skip_existing=True, sleep=False, playlist=""
+        track: Track,
+        file_template: str,
+        skip_existing=True,
+        sleep=False,
+        playlist="",
+        cover_data=b"",
     ) -> tuple[str, str]:
         file_dir, file_name = formatFilename(file_template, track, playlist)
 
-        # it will stop detecting existing file for other extensions.
-        # we need to store track `id + quality` in metadata to differentiate tracks
-        # TODO: create better existing file detecting ✨
         file_path = f"{download_path}/{file_dir}/{file_name}"
         if skip_existing and (
             os.path.isfile(file_path + ".m4a") or os.path.isfile(file_path + ".flac")
@@ -138,7 +142,11 @@ def main():
         if sleep:
             sleep_time = randint(5, 15) / 10 + 1
             logger.info(f"sleeping for {sleep_time}s")
-            time.sleep(sleep_time)
+            try:
+                time.sleep(sleep_time)
+            except KeyboardInterrupt:
+                logger.info("stopping...")
+                exit()
 
         stream = api.getTrackStream(track["id"], track_quality)
         quality = TRACK_QUALITY[stream["audioQuality"]]
@@ -162,13 +170,18 @@ def main():
             stream["manifest"],
             stream["manifestMimeType"],
         )
-        try:
-            setMetadata(track_path, track)
-        except ValueError as e:
-            logger.error(f"setMetadata error: {e}")
 
         track_path = convertToFlac(track_path)
 
+        if not cover_data:
+            cover = Cover(track["album"]["cover"])
+            cover_data = cover.content
+
+        try:
+            setMetadata(track_path, track, cover_data)
+        except ValueError as e:
+            logger.error(f"could not set metadata. {e}")
+
         logger.info(f"track saved as {track_path}")
 
         return file_dir, file_name
@@ -180,7 +193,7 @@ def main():
         # i dont know if limit 100 is suspicious
         # but i will leave it here
         album_items = api.getAlbumItems(album_id, limit=100)
-        file_dir = ""
+        album_cover = Cover(album["cover"])
 
         for item in album_items["items"]:
             track = item["item"]
@@ -189,10 +202,11 @@ def main():
                 file_template=config["settings"]["album_template"],
                 skip_existing=skip_existing,
                 sleep=True,
+                cover_data=album_cover.content,
             )
 
-        if file_dir:
-            downloadCover(album["cover"], f"{download_path}/{file_dir}")
+            if SAVE_COVER:
+                album_cover.save(f"{download_path}/{file_dir}")
 
     skip_existing = not args.no_skip
     failed_input = []
@@ -215,9 +229,13 @@ def main():
         match input_type:
             case "track":
                 track = api.getTrack(input_id)
+
                 downloadTrack(
-                    track, file_template=track_template, skip_existing=skip_existing
+                    track,
+                    file_template=track_template,
+                    skip_existing=skip_existing,
                 )
+
                 continue
 
             case "album":
@@ -243,16 +261,26 @@ def main():
                 playlist = api.getPlaylist(input_id)
                 logger.info(f"playlist: {playlist['title']} ({playlist['url']})")
 
+                playlist_cover = Cover(
+                    playlist["squareImage"], 1080
+                )  # playlists have 1080x1080 size
+
                 playlist_items = api.getPlaylistItems(input_id)
+
                 for item in playlist_items["items"]:
-                    downloadTrack(
-                        item["item"],
+                    track = item["item"]
+
+                    file_dir, file_name = downloadTrack(
+                        track,
                         file_template=config["settings"]["playlist_template"],
                         skip_existing=skip_existing,
                         sleep=True,
                         playlist=playlist["title"],
                     )
 
+                    if SAVE_COVER:
+                        playlist_cover.save(f"{download_path}/{file_dir}")
+
                 continue
 
             case _:

tiddl/download.py

@@ -226,7 +226,13 @@ def downloadTrackStream(
     return file_path
 
 
-def downloadCover(uid: str, path: str, size=640):
+def downloadCover(uid: str, path: str, size=1280):
+    file = f"{path}/cover.jpg"
+
+    if os.path.isfile(file):
+        logger.debug(f"cover already exists ({file})")
+        return
+
     formatted_uid = uid.replace("-", "/")
     url = f"https://resources.tidal.com/images/{formatted_uid}/{size}x{size}.jpg"
 
@@ -236,10 +242,56 @@ def downloadCover(uid: str, path: str, size=640):
         logger.error(f"could not download cover. ({req.status_code}) {url}")
         return
 
-    file = f"{path}/cover.jpg"
-
     try:
         with open(file, "wb") as f:
             f.write(req.content)
     except FileNotFoundError as e:
         logger.error(f"could not save cover. {file} -> {e}")
+
+
+class Cover:
+    def __init__(self, uid: str, size=1280) -> None:
+        if size > 1280:
+            logger.warning(
+                f"can not set cover size higher than 1280 (user set: {size})"
+            )
+            size = 1280
+
+        self.uid = uid
+
+        formatted_uid = uid.replace("-", "/")
+        self.url = (
+            f"https://resources.tidal.com/images/{formatted_uid}/{size}x{size}.jpg"
+        )
+
+        logger.debug((self.uid, self.url))
+        self.content = self.get()
+
+    def get(self) -> bytes:
+        req = requests.get(self.url)
+
+        if req.status_code != 200:
+            logger.error(f"could not download cover. ({req.status_code}) {self.url}")
+            return b""
+
+        logger.debug("got cover")
+
+        return req.content
+
+    def save(self, path: str):
+        if not self.content:
+            logger.error("cover file content is empty")
+            return
+
+        file = f"{path}/cover.jpg"
+
+        if os.path.isfile(file):
+            logger.debug(f"cover already exists ({file})")
+            return
+
+        try:
+            with open(file, "wb") as f:
+                logger.debug(file)
+                f.write(self.content)
+        except FileNotFoundError as e:
+            logger.error(f"could not save cover. {file} -> {e}")

tiddl/utils.py

@@ -4,7 +4,7 @@ import logging
 import subprocess
 
 from typing import TypedDict, Literal, List, get_args
-from mutagen.flac import FLAC as MutagenFLAC
+from mutagen.flac import FLAC as MutagenFLAC, Picture
 from mutagen.easymp4 import EasyMP4 as MutagenMP4
 
 from .types.track import Track
@@ -58,7 +58,7 @@ def formatFilename(template: str, track: Track, playlist=""):
     }
 
     dirs = template.split("/")
-    filename = dirs.pop().format(**formatted_track)
+    filename = sanitizeFileName(dirs.pop().format(**formatted_track))
 
     template_without_filename = "/".join(dirs)
     formatted_dir = template_without_filename.format(**formatted_track)
@@ -70,11 +70,20 @@ def formatFilename(template: str, track: Track, playlist=""):
 
 def sanitizeDirName(dir_name: str):
     # replace invalid characters with an underscore
-    sanitized_dir = re.sub(r'[<>:"|?*]', "_", dir_name)
+    sanitized = re.sub(r'[<>:"|?*]', "_", dir_name)
     # strip whitespace
-    sanitized_dir = sanitized_dir.strip()
+    sanitized = sanitized.strip()
 
-    return sanitized_dir
+    return sanitized
+
+
+def sanitizeFileName(file_name: str):
+    # replace invalid characters with an underscore
+    sanitized = re.sub(r'[<>:"|?*/\\]', "_", file_name)
+    # strip whitespace
+    sanitized = sanitized.strip()
+
+    return sanitized
 
 
 def loadingSymbol(i: int, text: str):
@@ -83,13 +92,19 @@ def loadingSymbol(i: int, text: str):
     print(f"\r{text} {symbol}", end="\r")
 
 
-def setMetadata(file_path: str, track: Track):
+def setMetadata(file_path: str, track: Track, cover_data=b""):
     _, extension = os.path.splitext(file_path)
 
     if extension == ".flac":
         metadata = MutagenFLAC(file_path)
+        if cover_data:
+            picture = Picture()
+            picture.data = cover_data
+            picture.mime = "image/jpeg"
+            metadata.add_picture(picture)
     elif extension == ".m4a":
         metadata = MutagenMP4(file_path)
+        # i dont know if there is a way to add cover for m4a file
     else:
         raise ValueError(f"Unknown file extension: {extension}")
 
@@ -124,7 +139,7 @@ def convertToFlac(source_path: str, remove_source=True):
     if source_extension != ".m4a":
         return source_path
 
-    logger.info(f"converting `{source_path}` to FLAC")
+    logger.debug(f"converting `{source_path}` to FLAC")
     command = ["ffmpeg", "-i", source_path, dest_path]
     result = subprocess.run(
         command, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL