Загрузка зашифрованных файлов

This commit is contained in:
Lev Plyusnin 2025-04-26 15:37:20 +07:00
parent c6818a3f1f
commit c64e2afbe1
No known key found for this signature in database
GPG key ID: 21C6C2C9C0A4460D
6 changed files with 111 additions and 80 deletions

View file

@ -86,7 +86,7 @@ options:
Общие параметры:
--quality <Качество> Качество трека:
0 - Низкое (AAC 64kbps)
1 - Высокое (MP3 320kbps)
1 - Оптимальное (AAC 192kbps)
2 - Лучшее (FLAC)
(по умолчанию: 0)
--skip-existing Пропускать уже загруженные треки

View file

@ -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]

View file

@ -1,3 +0,0 @@
yandex-music
mutagen
StrEnum

View file

@ -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)

View file

@ -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 += "]"

View file

@ -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)
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
download_info = get_download_info(track, api_quality)
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) -> 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.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,