Merge pull request #60 from oskvr37/1.9.0

1.9.0

requirements.txt

@@ -1,2 +1,3 @@
 requests>=2.20.0
-mutagen>=1.47.0
+mutagen>=1.47.0
+ffmpeg-python>=0.2.0

setup.py

@@ -2,7 +2,7 @@
 
 setup(
     name="tiddl",
-    version="1.8.3",
+    version="1.9.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

@@ -3,7 +3,7 @@
 import logging
 from random import randint
 
-from .api import TidalApi
+from .api import TidalApi, ApiError
 from .auth import getDeviceAuth, getToken, refreshToken
 from .config import Config
 from .download import downloadTrackStream, Cover
@@ -16,7 +16,7 @@
     formatFilename,
     loadingSymbol,
     setMetadata,
-    convertToFlac,
+    convertFileExtension,
     initLogging,
     parseFileInput,
 )
@@ -162,6 +162,8 @@ def downloadTrack(
                 exit()
 
         stream = api.getTrackStream(track["id"], track_quality)
+        logger.debug({"stream": stream})
+
         quality = TRACK_QUALITY[stream["audioQuality"]]
 
         MASTER_QUALITIES: list[TrackQuality] = ["HI_RES_LOSSLESS", "LOSSLESS"]
@@ -177,28 +179,30 @@ def downloadTrack(
 
         logger.info(f"{file_name} :: {quality['name']} Quality - {details}")
 
-        track_path = downloadTrackStream(
-            f"{download_path}/{file_dir}",
-            file_name,
+        track_data, extension = downloadTrackStream(
             stream["manifest"],
             stream["manifestMimeType"],
         )
 
-        if file_extension:
-            track_path = convertToFlac(
-                source_path=track_path, file_extension=file_extension
-            )
+        os.makedirs(file_dir, exist_ok=True)
+
+        file_path = f"{download_path}/{file_dir}/{file_name}.{extension}"
+
+        with open(file_path, "wb+") as f:
+            f.write(track_data)
 
         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}")
+        setMetadata(file_path, extension, track, cover_data)
 
-        logger.info(f"track saved as {track_path}")
+        if file_extension:
+            file_path = convertFileExtension(
+                source_path=file_path, file_extension=file_extension
+            )
+
+        logger.info(f"track saved as {file_path}")
 
         return file_dir, file_name
 
@@ -249,16 +253,17 @@ def downloadAlbum(album_id: str | int, skip_existing: bool):
 
         match input_type:
             case "track":
-                track = api.getTrack(input_id)
-
                 try:
-                    downloadTrack(
-                        track,
-                        file_template=track_template,
-                        skip_existing=skip_existing,
-                    )
-                except ValueError as e:
-                    logger.warning(f"track unavailable")
+                    track = api.getTrack(input_id)
+                except ApiError as e:
+                    logger.warning(f"{e.error['userMessage']} ({e.error['status']})")
+                    continue
+
+                downloadTrack(
+                    track,
+                    file_template=track_template,
+                    skip_existing=skip_existing,
+                )
 
                 continue
 

tiddl/api.py

@@ -1,7 +1,9 @@
 import logging
 from requests import Session
+from typing import TypedDict
 
 from .types import (
+    ErrorResponse,
     SessionResponse,
     TrackQuality,
     Track,
@@ -22,6 +24,12 @@
 PLAYLIST_LIMIT = 50
 
 
+class ApiError(Exception):
+    def __init__(self, message: str, error: ErrorResponse):
+        super().__init__(message)
+        self.error = error
+
+
 class TidalApi:
     def __init__(self, token: str, user_id: str, country_code: str) -> None:
         self.token = token
@@ -38,9 +46,12 @@ def _request(self, endpoint: str, params={}):
             method="GET", url=f"{API_URL}/{endpoint}", params=params
         )
 
-        # TODO: endpoints error handling ✨
+        data = req.json()
+
+        if req.status_code != 200:
+            raise ApiError(req.text, data)
 
-        return req.json()
+        return data
 
     def getSession(self) -> SessionResponse:
         return self._request(

tiddl/download.py

@@ -2,6 +2,7 @@
 import requests
 import json
 import os
+import ffmpeg
 
 from queue import Queue
 from threading import Thread
@@ -168,12 +169,25 @@ def threadDownload(urls: list[str]) -> bytes:
     return data
 
 
+def toFlac(track_data: bytes) -> bytes:
+    process = (
+        ffmpeg.input("pipe:0")
+        .output("pipe:1", format="flac", codec="copy")
+        .run_async(pipe_stdin=True, pipe_stdout=True, pipe_stderr=True)
+    )
+
+    flac_data, stderr = process.communicate(input=track_data)
+
+    if process.returncode != 0:
+        raise RuntimeError(f"FFmpeg failed: {stderr.decode()}")
+
+    return flac_data
+
+
 def downloadTrackStream(
-    file_dir: str,
-    file_name: str,
     encoded_manifest: str,
     mime_type: ManifestMimeType,
-):
+) -> tuple[bytes, str]:
     logger.debug(f"mime_type: {mime_type}")
     manifest = decodeManifest(encoded_manifest)
 
@@ -192,7 +206,7 @@ def downloadTrackStream(
         track_data = download(track_urls[0])
     else:
         track_data = threadDownload(track_urls)
-        codecs = "mp4a"  # for tracks that have mulitple `mp4` fragments
+        track_data = toFlac(track_data)
 
     """
     known codecs
@@ -204,26 +218,16 @@ def downloadTrackStream(
     if codecs is None:
         raise Exception("Missing codecs")
 
-    if codecs == "flac":
-        extension = "flac"
-    elif codecs.startswith("mp4a"):
+    extension = "flac"
+
+    if codecs.startswith("mp4a"):
         extension = "m4a"
-    else:
-        extension = "flac"
+    elif codecs != "flac":
         logger.warning(
             f'unknown file codecs: "{codecs}", please submit this as issue on GitHub'
         )
 
-    logger.debug((file_dir, file_name))
-
-    os.makedirs(file_dir, exist_ok=True)
-
-    file_path = f"{file_dir}/{file_name}.{extension}"
-
-    with open(file_path, "wb+") as f:
-        f.write(track_data)
-
-    return file_path
+    return track_data, extension
 
 
 def downloadCover(uid: str, path: str, size=1280):

tiddl/utils.py

@@ -7,7 +7,8 @@
 from datetime import datetime
 from typing import TypedDict, Literal, List, get_args
 from mutagen.flac import FLAC as MutagenFLAC, Picture
-from mutagen.easymp4 import EasyMP4 as MutagenMP4
+from mutagen.easymp4 import EasyMP4 as MutagenEasyMP4
+from mutagen.mp4 import MP4Cover, MP4 as MutagenMP4
 
 from .types.track import Track
 from .config import HOME_DIRECTORY
@@ -91,7 +92,7 @@ def formatFilename(template: str, track: Track, playlist=""):
         "playlist": playlist.strip(),
         "released": release_date.strftime("%m-%d-%Y"),
         "year": release_date.strftime("%Y"),
-        "playlist_number": str(track.get("playlistNumber", ""))
+        "playlist_number": str(track.get("playlistNumber", "")),
     }
 
     dirs = template.split("/")
@@ -127,19 +128,20 @@ def loadingSymbol(i: int, text: str):
     print(f"\r{text} {symbol}", end="\r")
 
 
-def setMetadata(file_path: str, track: Track, cover_data=b""):
-    _, extension = os.path.splitext(file_path)
-
-    if extension == ".flac":
-        metadata = MutagenFLAC(file_path)
+def setMetadata(file: str, extension: str, track: Track, cover_data=b""):
+    if extension == "flac":
+        metadata = MutagenFLAC(file)
         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
+    elif extension == "m4a":
+        if cover_data:
+            metadata = MutagenMP4(file)
+            metadata["covr"] = [MP4Cover(cover_data, imageformat=MP4Cover.FORMAT_JPEG)]
+            metadata.save(file)
+        metadata = MutagenEasyMP4(file)
     else:
         raise ValueError(f"Unknown file extension: {extension}")
 
@@ -148,24 +150,21 @@ def setMetadata(file_path: str, track: Track, cover_data=b""):
         "trackNumber": str(track["trackNumber"]),
         "discnumber": str(track["volumeNumber"]),
         "copyright": track["copyright"],
-        "artist": track["artist"]["name"],
+        "albumartist": track["artist"]["name"],
+        "artist": ";".join([artist["name"].strip() for artist in track["artists"]]),
         "album": track["album"]["title"],
         "date": track["streamStartDate"][:10],
-        # "tags": track["audioQuality"],
-        # "id": str(track["id"]),
-        # "url": track["url"],
     }
 
+    metadata.update(new_metadata)
+
     try:
-        metadata.update(new_metadata)
+        metadata.save(file)
     except Exception as e:
-        logger.error(e)
-        return
-
-    metadata.save()
+        logger.error(f"Failed to set metadata for {extension}: {e}")
 
 
-def convertToFlac(source_path: str, file_extension: str, remove_source=True):
+def convertFileExtension(source_path: str, file_extension: str, remove_source=True):
     source_dir, source_extension = os.path.splitext(source_path)
     dest_path = f"{source_dir}.{file_extension}"