From c64e2afbe1017b40ccbbbb6447a2f97ae3dc92cf Mon Sep 17 00:00:00 2001 From: Lev Plyusnin Date: Sat, 26 Apr 2025 15:37:20 +0700 Subject: [PATCH] =?UTF-8?q?=D0=97=D0=B0=D0=B3=D1=80=D1=83=D0=B7=D0=BA?= =?UTF-8?q?=D0=B0=20=D0=B7=D0=B0=D1=88=D0=B8=D1=84=D1=80=D0=BE=D0=B2=D0=B0?= =?UTF-8?q?=D0=BD=D0=BD=D1=8B=D1=85=20=D1=84=D0=B0=D0=B9=D0=BB=D0=BE=D0=B2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 2 +- pyproject.toml | 5 ++- requirements.txt | 3 -- ymd/api.py | 72 +++++++++++++++++++++++++++++++----- ymd/cli.py | 13 +++---- ymd/core.py | 96 +++++++++++++++++++----------------------------- 6 files changed, 111 insertions(+), 80 deletions(-) delete mode 100644 requirements.txt diff --git a/README.md b/README.md index f971a0f..1a37f7c 100644 --- a/README.md +++ b/README.md @@ -86,7 +86,7 @@ options: Общие параметры: --quality <Качество> Качество трека: 0 - Низкое (AAC 64kbps) - 1 - Высокое (MP3 320kbps) + 1 - Оптимальное (AAC 192kbps) 2 - Лучшее (FLAC) (по умолчанию: 0) --skip-existing Пропускать уже загруженные треки diff --git a/pyproject.toml b/pyproject.toml index 193a1b0..343b971 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -7,14 +7,15 @@ build-backend = "setuptools.build_meta" [project] name = "yandex-music-downloader" -version = "3.4.3b3" +version = "3.4.4b0" description = "Загрузчик музыки с сервиса Яндекс.Музыка" requires-python = ">=3.9" readme = "README.md" dependencies = [ "yandex-music @ https://github.com/llistochek/yandex-music-api/archive/9623fbca7704f47766614efe51d66c9fd496714c.zip", "mutagen>=1.47.0", - "StrEnum" + "StrEnum", + "pycryptodome" ] [project.scripts] diff --git a/requirements.txt b/requirements.txt deleted file mode 100644 index 16446f7..0000000 --- a/requirements.txt +++ /dev/null @@ -1,3 +0,0 @@ -yandex-music -mutagen -StrEnum diff --git a/ymd/api.py b/ymd/api.py index 1a25b00..8938fb4 100644 --- a/ymd/api.py +++ b/ymd/api.py @@ -1,32 +1,61 @@ import base64 import hashlib import hmac +import itertools +import random import time import typing +from collections.abc import Iterable from dataclasses import dataclass +from enum import Enum -from yandex_music import Track +from Crypto.Cipher import AES +from strenum import StrEnum +from yandex_music import Client, Track from yandex_music.utils.sign_request import DEFAULT_SIGN_KEY +class Codec(Enum): + FLAC = ["flac", "flac-mp4"] + MP3 = ["mp3"] + AAC = ["aac", "he-aac", "aac-mp4", "he-aac-mp4"] + + @classmethod + def get_all_codecs(cls) -> Iterable[str]: + return itertools.chain.from_iterable(e.value for e in cls) + + @classmethod + def from_codec_str(cls, codec_str: str) -> typing.Optional["Codec"]: + for codec in cls: + if any(e == codec_str for e in codec.value): + return codec + + +class ApiTrackQuality(StrEnum): + LOW = "lq" + NORMAL = "nq" + LOSSLESS = "lossless" + + @dataclass -class LosslessDownloadInfo: +class CustomDownloadInfo: quality: str - codec: str + codec: Codec urls: list[str] + decryption_key: str bitrate: int -def get_lossless_info(track: Track) -> LosslessDownloadInfo: +def get_download_info(track: Track, quality: ApiTrackQuality) -> CustomDownloadInfo: 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", + "quality": quality, + "codecs": ",".join(Codec.get_all_codecs()), + "transports": "encraw", } hmac_sign = hmac.new( DEFAULT_SIGN_KEY.encode(), @@ -41,6 +70,31 @@ def get_lossless_info(track: Track) -> LosslessDownloadInfo: ) resp = typing.cast(dict, resp) e = resp["download_info"] - return LosslessDownloadInfo( - quality=e["quality"], codec=e["codec"], urls=e["urls"], bitrate=e["bitrate"] + raw_codec = e["codec"] + codec = Codec.from_codec_str(raw_codec) + if codec is None: + raise ValueError(f"Unknown codec: {raw_codec}") + return CustomDownloadInfo( + quality=e["quality"], + codec=codec, + urls=e["urls"], + bitrate=e["bitrate"], + decryption_key=e.get("key"), ) + + +def download_track(client: Client, download_info: CustomDownloadInfo) -> bytes: + data = client.request.retrieve(random.choice(download_info.urls)) + if decryption_key := download_info.decryption_key: + data = decrypt_data(data, decryption_key) + return data + + +def decrypt_data(data: bytes, key: str) -> bytes: + aes = AES.new( + key=bytes.fromhex(key), + nonce=bytes(12), + mode=AES.MODE_CTR, + ) + + return aes.decrypt(data) diff --git a/ymd/cli.py b/ymd/cli.py index 02d2d18..6ee04ab 100644 --- a/ymd/cli.py +++ b/ymd/cli.py @@ -82,7 +82,7 @@ def main(): metavar="<Качество>", default=0, type=quality_arg, - help="Качество трека:\n0 - Низкое (AAC 64kbps)\n1 - Высокое (MP3 320kbps)\n2 - Лучшее (FLAC)\n(по умолчанию: %(default)s)", + help="Качество трека:\n0 - Низкое (AAC 64kbps)\n1 - Оптимальное (AAC 192kbps)\n2 - Лучшее (FLAC)\n(по умолчанию: %(default)s)", ) common_group.add_argument( "--skip-existing", action="store_true", help="Пропускать уже загруженные треки" @@ -124,8 +124,7 @@ def main(): common_group.add_argument( "--only-music", action="store_true", - help="Загружать только музыкальные альбомы" - " (пропускать подкасты и аудиокниги)", + help="Загружать только музыкальные альбомы (пропускать подкасты и аудиокниги)", ) common_group.add_argument( "--compatibility-level", @@ -236,9 +235,9 @@ def main(): if album.id is None or not album.available: print(f'Альбом "{title}" не доступен для скачивания') elif args.only_music and album.meta_type != "music": - print(f'Альбом "{title}" пропущен' " т.к. не является музыкальным") + print(f'Альбом "{title}" пропущен т.к. не является музыкальным') elif args.stick_to_artist and album.artists[0].id != int(args.artist_id): - print(f'Альбом "{title}" пропущен' " из-за флага --stick-to-artist") + print(f'Альбом "{title}" пропущен из-за флага --stick-to-artist') else: return True return False @@ -298,8 +297,8 @@ def main(): save_dir.mkdir(parents=True) downloadable = core.to_downloadable_track(track, args.quality, save_path) - bitrate = downloadable.bitrate - format_info = "[" + downloadable.codec.upper() + bitrate = downloadable.download_info.bitrate + format_info = "[" + downloadable.download_info.codec.name if bitrate > 0: format_info += f" {bitrate}kbps" format_info += "]" diff --git a/ymd/core.py b/ymd/core.py index 572f37b..a07af8a 100644 --- a/ymd/core.py +++ b/ymd/core.py @@ -1,10 +1,9 @@ import datetime as dt -import random import re import typing from collections.abc import Callable from dataclasses import dataclass -from enum import auto +from enum import IntEnum, auto from pathlib import Path from typing import Optional, Union @@ -28,12 +27,17 @@ from mutagen.mp4 import MP4, MP4Cover from strenum import LowercaseStrEnum from yandex_music import ( Client, - DownloadInfo, Track, YandexMusicModel, ) -from ymd.api import get_lossless_info +from ymd import api +from ymd.api import ( + ApiTrackQuality, + Codec, + CustomDownloadInfo, + get_download_info, +) from ymd.mime_utils import MimeType, guess_mime_type UNSAFE_PATH_CLEAR_RE = re.compile(r"[/\\]+") @@ -49,6 +53,12 @@ AUDIO_FILE_SUFFIXES = {".mp3", ".flac", ".m4a"} TEMPORARY_FILE_NAME_TEMPLATE = ".yandex-music-downloader.{}.tmp" +class CoreTrackQuality(IntEnum): + LOW = 0 + NORMAL = auto() + LOSSLESS = auto() + + class LyricsFormat(LowercaseStrEnum): NONE = auto() TEXT = auto() @@ -57,9 +67,7 @@ class LyricsFormat(LowercaseStrEnum): @dataclass class DownloadableTrack: - url: str - bitrate: int - codec: str + download_info: CustomDownloadInfo path: Path track: Track @@ -265,7 +273,8 @@ def download_track( if not lrc_path.is_file() and ( track_lyrics := track.get_lyrics(format_="LRC") ): - download_via_temporary_file(client, track_lyrics.download_url, lrc_path) + lyrics = track_lyrics.fetch_lyrics() + write_via_temporary_file(lyrics.encode("utf-8"), lrc_path) elif lyrics_info.has_available_text_lyrics: if track_lyrics := track.get_lyrics(format_="TEXT"): text_lyrics = track_lyrics.fetch_lyrics() @@ -297,78 +306,49 @@ def download_track( if not cover_path.is_file(): write_via_temporary_file(album_cover.data, cover_path) - download_via_temporary_file( - client, - track_info.url, + download_info = track_info.download_info + track_data = api.download_track(client, download_info) + + write_via_temporary_file( + track_data, target_path, - post_download_hook=lambda tmp_path: set_tags( + temporary_file_hook=lambda tmp_path: set_tags( tmp_path, track, text_lyrics, cover, compatibility_level ), ) def to_downloadable_track( - track: Track, quality: int, base_path: Path + track: Track, quality: CoreTrackQuality, 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")] + api_quality = ApiTrackQuality.NORMAL + if quality == CoreTrackQuality.LOW: + api_quality = ApiTrackQuality.LOW + elif quality == CoreTrackQuality.NORMAL: + api_quality = ApiTrackQuality.NORMAL + elif quality == CoreTrackQuality.LOSSLESS: + api_quality = ApiTrackQuality.LOSSLESS - def sort_key(e: DownloadInfo) -> Union[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 = get_download_info(track, api_quality) + codec = download_info.codec - 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": + if codec == Codec.MP3: suffix = ".mp3" - elif codec == "aac" or codec == "he-aac": + elif codec == Codec.AAC: suffix = ".m4a" - elif codec == "flac": + elif codec == Codec.FLAC: suffix = ".flac" else: raise RuntimeError("Unknown codec") + target_path = str(base_path) + suffix return DownloadableTrack( - url=url, + download_info=download_info, track=track, - bitrate=bitrate, - codec=codec, path=Path(target_path), ) -def download_via_temporary_file( - client: Client, - url: str, - target_path: Path, - post_download_hook: Optional[Callable[[Path], None]] = None, -) -> Path: - data = client.request.retrieve(url) - return write_via_temporary_file(data, target_path, post_download_hook) - - def write_via_temporary_file( data: bytes, target_path: Path,