Merge pull request #76 from nathom/improve_metadata

Added support for embedded mp3 covers, other types of Qobuz urls, copyright tags

qobuz_dl/core.py

@@ -21,7 +21,8 @@ WEB_URL = "https://play.qobuz.com/"
 ARTISTS_SELECTOR = "td.chartlist-artist > a"
 TITLE_SELECTOR = "td.chartlist-name > a"
 EXTENSIONS = (".mp3", ".flac")
-QUALITIES = {5: "5 - MP3", 6: "6 - FLAC", 7: "7 - 24B<96kHz", 27: "27 - 24B>96kHz"}
+QUALITIES = {5: "5 - MP3", 6: "6 - FLAC",
+             7: "7 - 24B<96kHz", 27: "27 - 24B>96kHz"}
 
 logger = logging.getLogger(__name__)
 
@@ -32,7 +33,8 @@ class PartialFormatter(string.Formatter):
 
     def get_field(self, field_name, args, kwargs):
         try:
-            val = super(PartialFormatter, self).get_field(field_name, args, kwargs)
+            val = super(PartialFormatter, self).get_field(field_name,
+                                                          args, kwargs)
         except (KeyError, AttributeError):
             val = None, field_name
         return val
@@ -95,12 +97,29 @@ class QobuzDL:
 
     def get_id(self, url):
         return re.match(
-            r"https?://(?:w{0,3}|play|open)\.qobuz\.com/(?:(?:album|track|artist"
+            r"https?://(?:w{0,3}|play|open)\.qobuz\.com/(?:(?:album|track"
-            r"|playlist|label)/|[a-z]{2}-[a-z]{2}/album/-?\w+(?:-\w+)*-?/|user/"
+            r"|artist|playlist|label)/|[a-z]{2}-[a-z]{2}/album/-?\w+(?:-\w+)*"
-            r"library/favorites/)(\w+)",
+            r"-?/|user/library/favorites/)(\w+)",
             url,
         ).group(1)
 
+    def get_type(self, url):
+        if re.match(r'https?', url) is not None:
+            url_type = url.split('/')[3]
+            if url_type not in ['album', 'artist', 'playlist',
+                                'track', 'label']:
+                if url_type == "user":
+                    url_type = url.split('/')[-1]
+                else:
+                    # url is from Qobuz store
+                    # e.g. "https://www.qobuz.com/us-en/album/..."
+                    url_type = url.split('/')[4]
+        else:
+            # url missing base
+            # e.g. "/us-en/album/{artist}/{id}"
+            url_type = url.split('/')[2]
+        return url_type
+
     def download_from_id(self, item_id, album=True, alt_path=None):
         if handle_download_id(self.downloads_db, item_id, add_id=False):
             logger.info(
@@ -144,24 +163,27 @@ class QobuzDL:
             "track": {"album": False, "func": None, "iterable_key": None},
         }
         try:
-            url_type = url.split("/")[3]
+            url_type = self.get_type(url)
             type_dict = possibles[url_type]
             item_id = self.get_id(url)
         except (KeyError, IndexError):
             logger.info(
-                f'{RED}Invalid url: "{url}". Use urls from https://play.qobuz.com!'
+                f'{RED}Invalid url: "{url}". Use urls from '
+                'https://play.qobuz.com!'
             )
             return
         if type_dict["func"]:
             content = [item for item in type_dict["func"](item_id)]
             content_name = content[0]["name"]
             logger.info(
-                f"{YELLOW}Downloading all the music from {content_name} ({url_type})!"
+                f"{YELLOW}Downloading all the music from {content_name} "
+                f"({url_type})!"
             )
             new_path = self.create_dir(
                 os.path.join(self.directory, sanitize_filename(content_name))
             )
-            items = [item[type_dict["iterable_key"]]["items"] for item in content][0]
+            items = [item[type_dict["iterable_key"]]["items"]
+                     for item in content][0]
             logger.info(f"{YELLOW}{len(items)} downloads in queue")
             for item in items:
                 self.download_from_id(
@@ -210,9 +232,11 @@ class QobuzDL:
 
         logger.info(
             f'{YELLOW}Searching {self.lucky_type}s for "{query}".\n'
-            f"{YELLOW}qobuz-dl will attempt to download the first {self.lucky_limit} results."
+            f"{YELLOW}qobuz-dl will attempt to download the first "
+            f"{self.lucky_limit} results."
         )
-        results = self.search_by_type(query, self.lucky_type, self.lucky_limit, True)
+        results = self.search_by_type(query, self.lucky_type,
+                                      self.lucky_limit, True)
 
         if download:
             self.download_list_of_urls(results)
@@ -275,7 +299,8 @@ class QobuzDL:
                     )
 
                 url = "{}{}/{}".format(WEB_URL, item_type, i.get("id", ""))
-                item_list.append({"text": text, "url": url} if not lucky else url)
+                item_list.append({"text": text, "url": url} if not lucky
+                                 else url)
             return item_list
         except (KeyError, IndexError):
             logger.info(f"{RED}Invalid type: {item_type}")
@@ -287,7 +312,8 @@ class QobuzDL:
         except (ImportError, ModuleNotFoundError):
             if os.name == "nt":
                 sys.exit(
-                    'Please install curses with "pip3 install windows-curses" to continue'
+                    'Please install curses with '
+                    '"pip3 install windows-curses" to continue'
                 )
             raise
 
@@ -306,13 +332,15 @@ class QobuzDL:
 
         try:
             item_types = ["Albums", "Tracks", "Artists", "Playlists"]
-            selected_type = pick(item_types, "I'll search for:\n[press Intro]")[0][
+            selected_type = pick(item_types,
-                :-1
+                                 "I'll search for:\n[press Intro]"
-            ].lower()
+                                 )[0][:-1].lower()
-            logger.info(f"{YELLOW}Ok, we'll search for {selected_type}s{RESET}")
+            logger.info(f"{YELLOW}Ok, we'll search for "
+                        f"{selected_type}s{RESET}")
             final_url_list = []
             while True:
-                query = input(f"{CYAN}Enter your search: [Ctrl + c to quit]\n-{DF} ")
+                query = input(f"{CYAN}Enter your search: [Ctrl + c to quit]\n"
+                              f"-{DF} ")
                 logger.info(f"{YELLOW}Searching...{RESET}")
                 options = self.search_by_type(
                     query, selected_type, self.interactive_limit
@@ -334,10 +362,12 @@ class QobuzDL:
                     options_map_func=get_title_text,
                 )
                 if len(selected_items) > 0:
-                    [final_url_list.append(i[0]["url"]) for i in selected_items]
+                    [final_url_list.append(i[0]["url"])
+                     for i in selected_items]
                     y_n = pick(
                         ["Yes", "No"],
-                        "Items were added to queue to be downloaded. Keep searching?",
+                        "Items were added to queue to be downloaded. "
+                        "Keep searching?",
                     )
                     if y_n[0][0] == "N":
                         break
@@ -347,7 +377,8 @@ class QobuzDL:
             if final_url_list:
                 desc = (
                     "Select [intro] the quality (the quality will "
-                    "be automatically\ndowngraded if the selected is not found)"
+                    "be automatically\ndowngraded if the selected "
+                    "is not found)"
                 )
                 self.quality = pick(
                     qualities,
@@ -389,11 +420,13 @@ class QobuzDL:
         pl_title = sanitize_filename(soup.select_one("h1").text)
         pl_directory = os.path.join(self.directory, pl_title)
         logger.info(
-            f"{YELLOW}Downloading playlist: {pl_title} ({len(track_list)} tracks)"
+            f"{YELLOW}Downloading playlist: {pl_title} "
+            f"({len(track_list)} tracks)"
         )
 
         for i in track_list:
-            track_id = self.get_id(self.search_by_type(i, "track", 1, lucky=True)[0])
+            track_id = self.get_id(self.search_by_type(i, "track", 1,
+                                                       lucky=True)[0])
             if track_id:
                 self.download_from_id(track_id, False, pl_directory)
 
@@ -410,7 +443,9 @@ class QobuzDL:
             dirs.sort()
             audio_rel_files = [
                 # os.path.abspath(os.path.join(local, file_))
-                # os.path.join(rel_folder, os.path.basename(os.path.normpath(local)), file_)
+                # os.path.join(rel_folder,
+                #              os.path.basename(os.path.normpath(local)),
+                #              file_)
                 os.path.join(os.path.basename(os.path.normpath(local)), file_)
                 for file_ in files
                 if os.path.splitext(file_)[-1] in EXTENSIONS
@@ -423,7 +458,8 @@ class QobuzDL:
             if not audio_files or len(audio_files) != len(audio_rel_files):
                 continue
 
-            for audio_rel_file, audio_file in zip(audio_rel_files, audio_files):
+            for audio_rel_file, audio_file in zip(audio_rel_files,
+                                                  audio_files):
                 try:
                     pl_item = (
                         EasyMP3(audio_file)

qobuz_dl/metadata.py

@@ -2,11 +2,16 @@ import os
 import logging
 
 from mutagen.flac import FLAC, Picture
-from mutagen.mp3 import EasyMP3
+import mutagen.id3 as id3
+from mutagen.id3 import ID3NoHeaderError
 
 logger = logging.getLogger(__name__)
 
 
+# unicode symbols
+COPYRIGHT, PHON_COPYRIGHT = '\u2117', '\u00a9'
+
+
 def get_title(track_dict):
     title = track_dict["title"]
     version = track_dict.get("version")
@@ -19,8 +24,15 @@ def get_title(track_dict):
     return title
 
 
+def _format_copyright(s: str) -> str:
+    s = s.replace('(P)', PHON_COPYRIGHT)
+    s = s.replace('(C)', COPYRIGHT)
+    return s
+
+
 # Use KeyError catching instead of dict.get to avoid empty tags
-def tag_flac(filename, root_dir, final_name, d, album, istrack=True, em_image=False):
+def tag_flac(filename, root_dir, final_name, d, album,
+             istrack=True, em_image=False):
     """
     Tag a FLAC file
 
@@ -61,23 +73,29 @@ def tag_flac(filename, root_dir, final_name, d, album, istrack=True, em_image=Fa
 
     if istrack:
         audio["GENRE"] = ", ".join(d["album"]["genres_list"])  # GENRE
-        audio["ALBUMARTIST"] = d["album"]["artist"]["name"]  # ALBUM ARTIST
+        audio["ALBUMARTIST"] = d["album"]["artist"]["name"]    # ALBUM ARTIST
         audio["TRACKTOTAL"] = str(d["album"]["tracks_count"])  # TRACK TOTAL
-        audio["ALBUM"] = d["album"]["title"]  # ALBUM TITLE
+        audio["ALBUM"] = d["album"]["title"]                   # ALBUM TITLE
         audio["DATE"] = d["album"]["release_date_original"]
+        audio["COPYRIGHT"] = _format_copyright(d["copyright"])
     else:
         audio["GENRE"] = ", ".join(album["genres_list"])  # GENRE
-        audio["ALBUMARTIST"] = album["artist"]["name"]  # ALBUM ARTIST
+        audio["ALBUMARTIST"] = album["artist"]["name"]    # ALBUM ARTIST
         audio["TRACKTOTAL"] = str(album["tracks_count"])  # TRACK TOTAL
-        audio["ALBUM"] = album["title"]  # ALBUM TITLE
+        audio["ALBUM"] = album["title"]                   # ALBUM TITLE
         audio["DATE"] = album["release_date_original"]
+        audio["COPYRIGHT"] = _format_copyright(album["copyright"])
 
     if em_image:
         emb_image = os.path.join(root_dir, "cover.jpg")
         multi_emb_image = os.path.join(
             os.path.abspath(os.path.join(root_dir, os.pardir)), "cover.jpg"
         )
-        cover_image = emb_image if os.path.isfile(emb_image) else multi_emb_image
+        if os.path.isfile(emb_image):
+            cover_image = emb_image
+        else:
+            cover_image = multi_emb_image
+
         try:
             image = Picture()
             image.type = 3
@@ -93,50 +111,94 @@ def tag_flac(filename, root_dir, final_name, d, album, istrack=True, em_image=Fa
     os.rename(filename, final_name)
 
 
-def tag_mp3(filename, root_dir, final_name, d, album, istrack=True, em_image=False):
+def tag_mp3(filename, root_dir, final_name, d, album,
+            istrack=True, em_image=False):
     """
-    Tag a mp3 file
+    Tag an mp3 file
 
-    :param str filename: mp3 file path
+    :param str filename: mp3 temporary file path
     :param str root_dir: Root dir used to get the cover art
     :param str final_name: Final name of the mp3 file (complete path)
     :param dict d: Track dictionary from Qobuz_client
-    :param bool istrack: Embed cover art into file
+    :param bool istrack
     :param bool em_image: Embed cover art into file
     """
-    # TODO: add embedded cover art support for mp3
-    audio = EasyMP3(filename)
-
-    audio["title"] = get_title(d)
-
-    audio["tracknumber"] = str(d["track_number"])
-
-    if "Disc " in final_name:
-        audio["discnumber"] = str(d["media_number"])
 
+    id3_legend = {
+        "album": id3.TALB,
+        "albumartist": id3.TPE2,
+        "artist": id3.TPE1,
+        "comment": id3.COMM,
+        "composer": id3.TCOM,
+        "copyright": id3.TCOP,
+        "date": id3.TDAT,
+        "genre": id3.TCON,
+        "isrc": id3.TSRC,
+        "label": id3.TPUB,
+        "performer": id3.TOPE,
+        "title": id3.TIT2,
+        "year": id3.TYER
+    }
     try:
-        audio["composer"] = d["composer"]["name"]
+        audio = id3.ID3(filename)
+    except ID3NoHeaderError:
+        audio = id3.ID3()
+
+    # temporarily holds metadata
+    tags = dict()
+    tags['title'] = get_title(d)
+    try:
+        tags['label'] = album["label"]["name"]
     except KeyError:
         pass
 
     try:
-        audio["artist"] = d["performer"]["name"]  # TRACK ARTIST
+        tags['artist'] = d["performer"]["name"]
     except KeyError:
         if istrack:
-            audio["artist"] = d["album"]["artist"]["name"]  # TRACK ARTIST
+            tags['artist'] = d["album"]["artist"]["name"]
         else:
-            audio["artist"] = album["artist"]["name"]
+            tags['artist'] = album["artist"]["name"]
 
     if istrack:
-        audio["genre"] = ", ".join(d["album"]["genres_list"])  # GENRE
+        tags["genre"] = ", ".join(d["album"]["genres_list"])
-        audio["albumartist"] = d["album"]["artist"]["name"]  # ALBUM ARTIST
+        tags["albumartist"] = d["album"]["artist"]["name"]
-        audio["album"] = d["album"]["title"]  # ALBUM TITLE
+        tags["album"] = d["album"]["title"]
-        audio["date"] = d["album"]["release_date_original"]
+        tags["date"] = d["album"]["release_date_original"]
+        tags["copyright"] = _format_copyright(d["copyright"])
+        tracktotal = str(d["album"]["tracks_count"])
     else:
-        audio["genre"] = ", ".join(album["genres_list"])  # GENRE
+        tags["genre"] = ", ".join(album["genres_list"])
-        audio["albumartist"] = album["artist"]["name"]  # ALBUM ARTIST
+        tags["albumartist"] = album["artist"]["name"]
-        audio["album"] = album["title"]  # ALBUM TITLE
+        tags["album"] = album["title"]
-        audio["date"] = album["release_date_original"]
+        tags["date"] = album["release_date_original"]
+        tags["copyright"] = _format_copyright(album["copyright"])
+        tracktotal = str(album["tracks_count"])
 
-    audio.save()
+    tags['year'] = tags['date'][:4]
+
+    audio['TRCK'] = id3.TRCK(encoding=3,
+                             text=f'{d["track_number"]}/{tracktotal}')
+    audio['TPOS'] = id3.TPOS(encoding=3,
+                             text=str(d["media_number"]))
+
+    # write metadata in `tags` to file
+    for k, v in tags.items():
+        id3tag = id3_legend[k]
+        audio[id3tag.__name__] = id3tag(encoding=3, text=v)
+
+    if em_image:
+        emb_image = os.path.join(root_dir, "cover.jpg")
+        multi_emb_image = os.path.join(
+            os.path.abspath(os.path.join(root_dir, os.pardir)), "cover.jpg"
+        )
+        if os.path.isfile(emb_image):
+            cover_image = emb_image
+        else:
+            cover_image = multi_emb_image
+
+        with open(cover_image, 'rb') as cover:
+            audio.add(id3.APIC(3, 'image/jpeg', 3, '', cover.read()))
+
+    audio.save(filename, 'v2_version=3')
     os.rename(filename, final_name)