diff --git a/pyproject.toml b/pyproject.toml index cfe89eb..9300f87 100644 --- a/pyproject.toml +++ b/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] diff --git a/requirements.txt b/requirements.txt index d76bf5e..440848e 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,2 @@ yandex-music -eyed3 +mutagen diff --git a/ymd/api.py b/ymd/api.py new file mode 100644 index 0000000..1a25b00 --- /dev/null +++ b/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"] + ) diff --git a/ymd/cli.py b/ymd/cli.py index 5204eb8..fb76210 100644 --- a/ymd/cli.py +++ b/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, - help="Качество трека:\n0 - Низкое (mp3 128kbps)\n1 - Высокое (mp3 320kbps)\n(по умолчанию: %(default)s)", + type=quality_arg, + 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, - target_path=save_path, - quality=args.quality, + track_info=downloadable, 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) diff --git a/ymd/core.py b/ymd/core.py index 9cc384a..f4848e8 100644 --- a/ymd/core.py +++ b/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 -from eyed3.id3.frames import ImageFrame -from yandex_music import Track, YandexMusicObject +import mutagen +from mutagen.flac import FLAC, Picture +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: - datetime = dt.fromisoformat(album.release_date) - release_date = eyed3.core.Date(*datetime.timetuple()[:6]) - else: - release_date = album.year - audiofile = eyed3.load(path) - assert audiofile - - 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) + track_artists = [a.name for a in track.artists if a.name] + album_artists = [a.name for a in album.artists if a.name] + tag = mutagen.File(path, [MP3, MP4, FLAC]) # type: ignore + album_title = full_title(album) + track_title = full_title(track) + track_number = None + disc_number = None if position := album.track_position: - tag.track_num = position.index - tag.disc_num = position.volume - tag.release_date = tag.original_release_date = release_date - tag.encoded_by = ENCODED_BY - tag.audio_file_url = f"https://music.yandex.ru/album/{album.id}/track/{track.id}" + track_number = position.index + disc_number = position.volume + iso8601_release_date = None + if album.release_date is not None: + 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: - tag.lyrics.set(lyrics) - if album_cover is not None: - tag.images.set(ImageFrame.FRONT_COVER, album_cover, "image/jpeg") + if isinstance(tag, MP3): + tag["TIT2"] = TIT2(encoding=3, text=track_title) + tag["TALB"] = TALB(encoding=3, text=album_title) + 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, - target_path: Path, + track_info: DownloadableTrack, 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() - 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) + client.request.download(track_info.url, str(target_path)) 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, + )