Поддержка lossless

pyproject.toml

@@ -7,13 +7,13 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "yandex-music-downloader"
-version = "3.0.0b1"
+version = "3.1.0b0"
 description = "Загрузчик музыки с сервиса Яндекс.Музыка"
 requires-python = ">=3.9"
 readme = "README.md"
 dependencies = [
     "yandex-music",
-    "eyed3"
+    "mutagen"
 ]
 
 [project.scripts]

requirements.txt

@@ -1,2 +1,2 @@
 yandex-music
-eyed3
+mutagen

ymd/api.py

@@ -0,0 +1,46 @@
+import base64
+import hashlib
+import hmac
+import time
+import typing
+from dataclasses import dataclass
+
+from yandex_music import Track
+from yandex_music.utils.sign_request import DEFAULT_SIGN_KEY
+
+
+@dataclass
+class LosslessDownloadInfo:
+    quality: str
+    codec: str
+    urls: list[str]
+    bitrate: int
+
+
+def get_lossless_info(track: Track) -> LosslessDownloadInfo:
+    client = track.client
+    assert client
+    timestamp = int(time.time())
+    params = {
+        "ts": timestamp,
+        "trackId": track.id,
+        "quality": "lossless",
+        "codecs": "flac,aac,he-aac,mp3",
+        "transports": "raw",
+    }
+    hmac_sign = hmac.new(
+        DEFAULT_SIGN_KEY.encode(),
+        "".join(str(e) for e in params.values()).replace(",", "").encode(),
+        hashlib.sha256,
+    )
+    sign = base64.b64encode(hmac_sign.digest()).decode()[:-1]
+    params["sign"] = sign
+
+    resp = client.request.get(
+        "https://api.music.yandex.net/get-file-info", params=params
+    )
+    resp = typing.cast(dict, resp)
+    e = resp["download_info"]
+    return LosslessDownloadInfo(
+        quality=e["quality"], codec=e["codec"], urls=e["urls"], bitrate=e["bitrate"]
+    )

ymd/cli.py

@@ -10,7 +10,7 @@ from pathlib import Path
 from typing import Optional, Union
 from urllib.parse import urlparse
 
-from yandex_music import Album, Client, Playlist, Track
+from yandex_music import Album, Playlist, Track
 
 from ymd import core
 
@@ -31,6 +31,24 @@ def show_default(text: Optional[str] = None) -> str:
     return f"{text} ({default})"
 
 
+def quality_arg(astr: str) -> int:
+    aint = int(astr)
+    if 0 <= aint <= 2:
+        return aint
+    raise argparse.ArgumentTypeError("Значение должно быть в промежутке от 0 до 2")
+
+
+def compatibility_level_arg(astr: str) -> int:
+    aint = int(astr)
+    min_val = core.MIN_COMPATIBILITY_LEVEL
+    max_val = core.MAX_COMPATIBILITY_LEVEL
+    if min_val <= aint <= max_val:
+        return aint
+    raise argparse.ArgumentTypeError(
+        f"Значение должен быть в промежутке от {min_val} до {max_val}"
+    )
+
+
 def main():
     parser = argparse.ArgumentParser(
         description="Загрузчик музыки с сервиса Яндекс.Музыка",
@@ -42,8 +60,8 @@ def main():
         "--quality",
         metavar="<Качество>",
         default=0,
-        type=int,
+        type=quality_arg,
-        help="Качество трека:\n0 - Низкое (mp3 128kbps)\n1 - Высокое (mp3 320kbps)\n(по умолчанию: %(default)s)",
+        help="Качество трека:\n0 - Низкое (AAC 64kbps)\n1 - Высокое (MP3 320kbps)\n2 - Лучшее (FLAC)\n(по умолчанию: %(default)s)",
     )
     common_group.add_argument(
         "--skip-existing", action="store_true", help="Пропускать уже загруженные треки"
@@ -52,7 +70,7 @@ def main():
         "--add-lyrics", action="store_true", help="Загружать тексты песен"
     )
     common_group.add_argument(
-        "--embed-cover", action="store_true", help="Встраивать обложку в .mp3 файл"
+        "--embed-cover", action="store_true", help="Встраивать обложку в аудиофайл"
     )
     common_group.add_argument(
         "--cover-resolution",
@@ -79,6 +97,15 @@ def main():
         help="Загружать только музыкальные альбомы"
         " (пропускать подкасты и аудиокниги)",
     )
+    common_group.add_argument(
+        "--compatibility-level",
+        metavar="<Уровень совместимости>",
+        default=1,
+        type=compatibility_level_arg,
+        help=show_default(
+            f"Уровень совместимости, от {core.MIN_COMPATIBILITY_LEVEL} до {core.MAX_COMPATIBILITY_LEVEL}. См. README для подробного описания"
+        ),
+    )
     common_group.add_argument("--debug", action="store_true", help=argparse.SUPPRESS)
 
     id_group_meta = parser.add_argument_group("ID")
@@ -148,7 +175,7 @@ def main():
             print("Параметер url указан в неверном формате")
             return 1
 
-    client = Client(args.token).init()
+    client = core.init_client(args.token)
     result_tracks: Iterable[Track] = []
 
     def album_tracks_gen(album_ids: Iterable[Union[int, str]]) -> Generator[Track]:
@@ -210,7 +237,7 @@ def main():
             print(f"Трек {track.title} не доступен для скачивания")
             continue
 
-        save_path = args.dir / core.prepare_track_path(
+        save_path = args.dir / core.prepare_base_path(
             args.path_pattern,
             track,
             args.unsafe_path,
@@ -222,15 +249,20 @@ def main():
         if not save_dir.is_dir():
             save_dir.mkdir(parents=True)
 
-        print(f"Загружается {save_path}")
+        downloadable = core.to_downloadable_track(track, args.quality, save_path)
+        bitrate = downloadable.bitrate
+        format_info = "[" + downloadable.codec.upper()
+        if bitrate > 0:
+            format_info += f" {bitrate}kbps"
+        format_info += "]"
+        print(f"{format_info} Загружается {downloadable.path}")
         core.download_track(
-            track=track,
+            track_info=downloadable,
-            target_path=save_path,
-            quality=args.quality,
             add_lyrics=args.add_lyrics,
             embed_cover=args.embed_cover,
             cover_resolution=args.cover_resolution,
             covers_cache=covers_cache,
+            compatibility_level=args.compatibility_level,
         )
         if args.delay > 0:
             time.sleep(args.delay)

ymd/core.py

@@ -1,19 +1,53 @@
+import datetime as dt
+import random
 import re
 import typing
-from datetime import datetime as dt
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Optional, Union
 
-import eyed3
+import mutagen
-from eyed3.id3.frames import ImageFrame
+from mutagen.flac import FLAC, Picture
-from yandex_music import Track, YandexMusicObject
+from mutagen.id3._frames import (
+    APIC,
+    TALB,
+    TDRC,
+    TIT2,
+    TPE1,
+    TPE2,
+    TPOS,
+    TRCK,
+    USLT,
+    WOAF,
+)
+from mutagen.id3._specs import ID3TimeStamp, PictureType
+from mutagen.mp3 import MP3
+from mutagen.mp4 import MP4, MP4Cover
+from yandex_music import Client, DownloadInfo, Track, YandexMusicObject
+
+from ymd.api import get_lossless_info
 
-ENCODED_BY = "https://github.com/llistochek/yandex-music-downloader"
 FILENAME_CLEAR_RE = re.compile(r"[^\w\-\'() ]+")
 
 DEFAULT_PATH_PATTERN = Path("#album-artist", "#album", "#number - #title")
 DEFAULT_COVER_RESOLUTION = 400
 
+MIN_COMPATIBILITY_LEVEL = 0
+MAX_COMPATIBILITY_LEVEL = 1
+
+
+@dataclass
+class DownloadableTrack:
+    url: str
+    bitrate: int
+    codec: str
+    path: Path
+    track: Track
+
+
+def init_client(token: str) -> Client:
+    return Client(token).init()
+
 
 def full_title(obj: YandexMusicObject) -> Optional[str]:
     result = obj["title"]
@@ -24,7 +58,7 @@ def full_title(obj: YandexMusicObject) -> Optional[str]:
     return result
 
 
-def prepare_track_path(
+def prepare_base_path(
     path_pattern: Path, track: Track, unsafe_path: bool = False
 ) -> Path:
     path_str = str(path_pattern)
@@ -50,65 +84,131 @@ def prepare_track_path(
         if not unsafe_path:
             replacement = FILENAME_CLEAR_RE.sub("_", replacement)
         path_str = path_str.replace(placeholder, replacement)
-    path_str += ".mp3"
     return Path(path_str)
 
 
-def set_id3_tags(
+def set_tags(
     path: Path,
     track: Track,
     lyrics: Optional[str],
     album_cover: Optional[bytes],
+    compatibility_level: int,
 ) -> None:
     album = track.albums[0]
-    if album.release_date is not None:
+    track_artists = [a.name for a in track.artists if a.name]
-        datetime = dt.fromisoformat(album.release_date)
+    album_artists = [a.name for a in album.artists if a.name]
-        release_date = eyed3.core.Date(*datetime.timetuple()[:6])
+    tag = mutagen.File(path, [MP3, MP4, FLAC])  # type: ignore
-    else:
+    album_title = full_title(album)
-        release_date = album.year
+    track_title = full_title(track)
-    audiofile = eyed3.load(path)
+    track_number = None
-    assert audiofile
+    disc_number = None
-
-    tag = audiofile.initTag()
-
-    tag.artist = chr(0).join(a.name for a in track.artists if a.name)
-    tag.album_artist = album.artists[0].name
-    tag.album = full_title(album)
-    tag.title = full_title(track)
     if position := album.track_position:
-        tag.track_num = position.index
+        track_number = position.index
-        tag.disc_num = position.volume
+        disc_number = position.volume
-    tag.release_date = tag.original_release_date = release_date
+    iso8601_release_date = None
-    tag.encoded_by = ENCODED_BY
+    if album.release_date is not None:
-    tag.audio_file_url = f"https://music.yandex.ru/album/{album.id}/track/{track.id}"
+        iso8601_release_date = dt.datetime.fromisoformat(album.release_date).astimezone(
+            dt.timezone.utc
+        )
+        iso8601_release_date = iso8601_release_date.strftime("%Y-%m-%d %H:%M:%S")
+    release_year = None
+    if year := album.year:
+        release_year = str(year)
+    track_url = f"https://music.yandex.ru/album/{album.id}/track/{track.id}"
 
-    if lyrics is not None:
+    if isinstance(tag, MP3):
-        tag.lyrics.set(lyrics)
+        tag["TIT2"] = TIT2(encoding=3, text=track_title)
-    if album_cover is not None:
+        tag["TALB"] = TALB(encoding=3, text=album_title)
-        tag.images.set(ImageFrame.FRONT_COVER, album_cover, "image/jpeg")
+        tag["TPE1"] = TPE1(encoding=3, text=track_artists)
+        tag["TPE2"] = TPE2(encoding=3, text=album_artists)
+
+        if tdrc_text := iso8601_release_date or release_year:
+            tag["TDRC"] = TDRC(encoding=3, text=[ID3TimeStamp(tdrc_text)])
+        if track_number:
+            tag["TRCK"] = TRCK(encoding=3, text=str(track_number))
+        if disc_number:
+            tag["TPOS"] = TPOS(encoding=3, text=str(disc_number))
+
+        if lyrics:
+            tag["USLT"] = USLT(encoding=3, text=lyrics)
+        if album_cover:
+            tag["APIC"] = APIC(encoding=3, mime="image/jpeg", type=3, data=album_cover)
+
+        tag["WOAF"] = WOAF(
+            encoding=3,
+            text=track_url,
+        )
+    elif isinstance(tag, MP4):
+        tag["\xa9nam"] = track_title
+        tag["\xa9alb"] = album_title
+        artists_value = track_artists
+        album_artists_value = album_artists
+        if compatibility_level == 1:
+            artists_value = "; ".join(track_artists)
+            album_artists_value = "; ".join(album_artists)
+        tag["\xa9ART"] = artists_value
+        tag["aART"] = album_artists_value
+
+        if iso8601_release_date is not None:
+            tag["rldt"] = iso8601_release_date
+        elif release_year is not None:
+            tag["\xa9day"] = release_year
+        if track_number:
+            tag["trkn"] = [(track_number, 0)]
+        if disc_number:
+            tag["disk"] = [(disc_number, 0)]
+
+        if lyrics:
+            tag["\xa9lyr"] = lyrics
+        if album_cover:
+            tag["covr"] = [MP4Cover(album_cover, imageformat=MP4Cover.FORMAT_JPEG)]
+        tag["\xa9cmt"] = track_url
+    elif isinstance(tag, FLAC):
+        tag["title"] = track_title
+        tag["album"] = album_title
+        tag["artist"] = track_artists
+        tag["albumartist"] = album_artists
+
+        if date_text := iso8601_release_date or release_year:
+            tag["date"] = date_text
+        if track_number:
+            tag["tracknumber"] = str(track_number)
+        if disc_number:
+            tag["discnumber"] = str(disc_number)
+
+        if lyrics:
+            tag["lyrics"] = lyrics
+        if album_cover is not None:
+            pic = Picture()
+            pic.type = PictureType.COVER_FRONT
+            pic.data = album_cover
+            pic.mime = "image/jpeg"
+            tag.add_picture(pic)
+        tag["comment"] = track_url
+    else:
+        raise RuntimeError("Unknown file format")
 
     tag.save()
 
 
 def download_track(
-    track: Track,
+    track_info: DownloadableTrack,
-    target_path: Path,
     cover_resolution: int = DEFAULT_COVER_RESOLUTION,
-    quality: int = 0,
     add_lyrics: bool = False,
     embed_cover: bool = False,
     covers_cache: Optional[dict[int, bytes]] = None,
+    compatibility_level: int = 1,
 ):
     if embed_cover and covers_cache is None:
         raise RuntimeError("covers_cache isn't provided")
     covers_cache = typing.cast(dict[int, bytes], covers_cache)
+    target_path = track_info.path
+    track = track_info.track
+    client = track.client
+    assert client
     album = track.albums[0]
 
-    download_info = track.get_download_info()
+    client.request.download(track_info.url, str(target_path))
-    download_info = [e for e in download_info if e.codec == "mp3"]
-    download_info.sort(key=lambda e: e.bitrate_in_kbps)
-    target_info = download_info[-1] if quality == 1 else download_info[0]
-    track.download(str(target_path), bitrate_in_kbps=target_info.bitrate_in_kbps)
 
     lyrics = None
     if add_lyrics and (lyrics_info := track.lyrics_info):
@@ -132,4 +232,56 @@ def download_track(
             if not cover_path.is_file():
                 track.download_cover(str(cover_path), cover_size)
 
-    set_id3_tags(target_path, track, lyrics, cover)
+    set_tags(target_path, track, lyrics, cover, compatibility_level)
+
+
+def to_downloadable_track(
+    track: Track, quality: int, base_path: Path
+) -> DownloadableTrack:
+    url: str
+    codec: str
+    bitrate: int
+    codec: str
+    if quality == 2:
+        download_info = get_lossless_info(track)
+        codec = download_info.codec
+        url = random.choice(download_info.urls)
+        bitrate = download_info.bitrate
+    else:
+        download_info = track.get_download_info(get_direct_links=True)
+        download_info = [e for e in download_info if e.codec in ("mp3", "aac")]
+
+        def sort_key(e: DownloadInfo) -> int | float:
+            aac_multiplier = 1.5
+            bitrate = e.bitrate_in_kbps
+            if bitrate <= 192:
+                aac_multiplier = 0.5
+            if e.codec == "aac":
+                bitrate *= aac_multiplier
+            return bitrate
+
+        download_info.sort(
+            key=sort_key,
+            reverse=quality == 0,
+        )
+        target_info = download_info[-1]
+        url = typing.cast(str, target_info.direct_link)
+        bitrate = target_info.bitrate_in_kbps
+        codec = target_info.codec
+
+    if codec == "mp3":
+        suffix = ".mp3"
+    elif codec == "aac" or codec == "he-aac":
+        suffix = ".m4a"
+    elif codec == "flac":
+        suffix = ".flac"
+    else:
+        raise RuntimeError("Unknown codec")
+    target_path = base_path.with_suffix(suffix)
+    return DownloadableTrack(
+        url=url,
+        track=track,
+        bitrate=bitrate,
+        codec=codec,
+        path=target_path,
+    )