diff --git a/ymd/cli.py b/ymd/cli.py index af662c2..9c6556f 100644 --- a/ymd/cli.py +++ b/ymd/cli.py @@ -8,16 +8,18 @@ import tempfile import time from pathlib import Path from typing import Optional +from urllib.parse import urlparse import browser_cookie3 from browser_cookie3 import BrowserCookieError from requests import Session from ymd import core -from ymd.ym_api import BasicTrackInfo, PlaylistId, api +from ymd.ym_api import BasicTrackInfo, PlaylistId, YandexMusicApi, api DEFAULT_USER_AGENT = "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/106.0.0.0 Safari/537.36" DEFAULT_DELAY = 3 +DEFAULT_DOMAIN = "music.yandex.ru" SUPPORTED_BROWSERS = [ "chrome", "opera", @@ -33,7 +35,7 @@ SUPPORTED_BROWSERS = [ CACHE_EXPIRE_AFTER = dt.timedelta(hours=8) CACHE_DIR = Path(tempfile.gettempdir()) / "ymd" -TRACK_RE = re.compile(r"track/(\d+)$") +TRACK_RE = re.compile(r"track/(\d+)") ALBUM_RE = re.compile(r"album/(\d+)$") ARTIST_RE = re.compile(r"artist/(\d+)$") PLAYLIST_RE = re.compile(r"([\w\-._]+)/playlists/(\d+)$") @@ -185,29 +187,33 @@ def main(): print(f"Не удалось получить cookies для браузера {args.browser}") return 1 - session = Session() - core.setup_session(session, cookies, DEFAULT_USER_AGENT) - session.hooks = {"response": response_hook} - result_tracks: list[BasicTrackInfo] = [] + domain = DEFAULT_DOMAIN if args.url is not None: - if match := ARTIST_RE.search(args.url): + parsed_url = urlparse(args.url) + path = parsed_url.path + if match := ARTIST_RE.search(path): args.artist_id = match.group(1) - elif match := ALBUM_RE.search(args.url): + elif match := ALBUM_RE.search(path): args.album_id = match.group(1) - elif match := TRACK_RE.search(args.url): + elif match := TRACK_RE.search(path): args.track_id = match.group(1) - elif match := PLAYLIST_RE.search(args.url): + elif match := PLAYLIST_RE.search(path): args.playlist_id = PlaylistId( owner=match.group(1), kind=int(match.group(2)) ) else: print("Параметер url указан в неверном формате") return 1 + domain = parsed_url.hostname + session = Session() + core.setup_session(session, cookies, DEFAULT_USER_AGENT, domain) + session.hooks = {"response": response_hook} + client = YandexMusicApi(session, domain) if args.artist_id is not None: - artist_info = api.get_artist_info(session, args.artist_id) + artist_info = client.get_artist_info(args.artist_id) albums_count = 0 for album in artist_info.albums: if args.stick_to_artist and album.artists[0].name != artist_info.name: @@ -220,24 +226,24 @@ def main(): f'Альбом "{album.title}" пропущен' " т.к. не является музыкальным" ) continue - full_album = api.get_full_album_info(session, album.id) + full_album = client.get_full_album_info(album.id) result_tracks.extend(full_album.tracks) albums_count += 1 print(artist_info.name) print(f"Альбомов: {albums_count}") elif args.album_id is not None: - album = api.get_full_album_info(session, args.album_id) + album = client.get_full_album_info(args.album_id) print(album.title) result_tracks = album.tracks elif args.track_id is not None: - track = api.get_full_track_info(session, args.track_id) + track = client.get_full_track_info(args.track_id) if track is not None: result_tracks = [track] else: logger.info("Трек не доступен для скачивания") return 1 elif args.playlist_id is not None: - result_tracks = api.get_playlist(session, args.playlist_id) + result_tracks = client.get_playlist(args.playlist_id) print(f"Треков: {len(result_tracks)}") @@ -261,7 +267,7 @@ def main(): print(f"Загружается {save_path}") core.download_track( - session=session, + client=client, track=track, target_path=save_path, hq=args.hq, diff --git a/ymd/core.py b/ymd/core.py index 71bcf39..e66e1bf 100644 --- a/ymd/core.py +++ b/ymd/core.py @@ -9,7 +9,8 @@ from eyed3.id3.frames import ImageFrame from requests import Session from ymd import http_utils -from ymd.ym_api import BasicTrackInfo, FullTrackInfo, api +from ymd.ym_api import BasicTrackInfo, FullTrackInfo +from ymd.ym_api.api import YandexMusicApi ENCODED_BY = "https://github.com/llistochek/yandex-music-downloader" FILENAME_CLEAR_RE = re.compile(r"[^\w\-\'() ]+") @@ -53,6 +54,7 @@ def set_id3_tags( track: BasicTrackInfo, lyrics: Optional[str], album_cover: Optional[bytes], + domain: str, ) -> None: if track.album.release_date is not None: release_date = eyed3.core.Date(*track.album.release_date.timetuple()[:6]) @@ -71,7 +73,7 @@ def set_id3_tags( tag.disc_num = track.disc_number tag.release_date = tag.original_release_date = release_date tag.encoded_by = ENCODED_BY - tag.audio_file_url = track.url + tag.audio_file_url = f"https://{domain}/album/{track.album.id}/track/{track.id}" if lyrics is not None: tag.lyrics.set(lyrics) @@ -81,15 +83,17 @@ def set_id3_tags( tag.save() -def setup_session(session: Session, cookie_jar: CookieJar, user_agent: str) -> Session: +def setup_session( + session: Session, cookie_jar: CookieJar, user_agent: str, domain: str +) -> Session: session.cookies = cookie_jar # type: ignore session.headers["User-Agent"] = user_agent - session.headers["X-Retpath-Y"] = urllib.parse.quote_plus("https://music.yandex.ru") + session.headers["X-Retpath-Y"] = urllib.parse.quote_plus(f"https://{domain}") return session def download_track( - session: Session, + client: YandexMusicApi, track: BasicTrackInfo, target_path: Path, covers_cache: dict[str, bytes], @@ -100,15 +104,15 @@ def download_track( ): album = track.album - url = api.get_track_download_url(session, track, hq) - http_utils.download_file(session, url, target_path) + url = client.get_track_download_url(track, hq) + http_utils.download_file(client.session, url, target_path) lyrics = None if add_lyrics and track.has_lyrics: if isinstance(track, FullTrackInfo): lyrics = track.lyrics else: - full_track = api.get_full_track_info(session, track.id) + full_track = client.get_full_track_info(track.id) if full_track is not None: lyrics = full_track.lyrics @@ -120,11 +124,11 @@ def download_track( cover = cached_cover else: cover = covers_cache[album.id] = http_utils.download_bytes( - session, cover_url + client.session, cover_url ) else: cover_path = target_path.parent / "cover.jpg" if not cover_path.is_file(): - http_utils.download_file(session, cover_url, cover_path) + http_utils.download_file(client.session, cover_url, cover_path) - set_id3_tags(target_path, track, lyrics, cover) + set_id3_tags(target_path, track, lyrics, cover, client.domain) diff --git a/ymd/ym_api/api.py b/ymd/ym_api/api.py index 88f9d11..9b6bd8c 100644 --- a/ymd/ym_api/api.py +++ b/ymd/ym_api/api.py @@ -8,47 +8,59 @@ from .models import * MD5_SALT = "XGRlBW9FXlekgbPrRHuSiA" -def get_track_download_url(session: Session, track: BasicTrackInfo, hq: bool) -> str: - resp = session.get( - "https://music.yandex.ru/api/v2.1/handlers/track" - f"/{track.id}:{track.album.id}" - "/web-album_track-track-track-main/download/m" - f"?hq={int(hq)}" - ) - url_info_src = resp.json()["src"] +class YandexMusicApi: + session: Session + domain: str - resp = session.get("https:" + url_info_src) - url_info = ET.fromstring(resp.text) - path = url_info.find("path").text[1:] - s = url_info.find("s").text - ts = url_info.find("ts").text - host = url_info.find("host").text - path_hash = hashlib.md5((MD5_SALT + path + s).encode()).hexdigest() - return f"https://{host}/get-mp3/{path_hash}/{ts}/{path}?track-id={track.id}" + def __init__(self, session: Session, domain: str) -> None: + self.session = session + self.domain = domain + def get_track_download_url(self, track: BasicTrackInfo, hq: bool) -> str: + resp = self.session.get( + f"https://{self.domain}/api/v2.1/handlers/track" + f"/{track.id}:{track.album.id}" + "/web-album_track-track-track-main/download/m" + f"?hq={int(hq)}" + ) + url_info_src = resp.json()["src"] -def get_full_track_info(session: Session, track_id: str) -> Optional[FullTrackInfo]: - params = {"track": track_id, "lang": "ru"} - resp = session.get("https://music.yandex.ru/handlers/track.jsx", params=params) - return FullTrackInfo.from_json(resp.json()) + resp = self.session.get("https:" + url_info_src) + url_info = ET.fromstring(resp.text) + path = url_info.find("path").text[1:] + s = url_info.find("s").text + ts = url_info.find("ts").text + host = url_info.find("host").text + path_hash = hashlib.md5((MD5_SALT + path + s).encode()).hexdigest() + return f"https://{host}/get-mp3/{path_hash}/{ts}/{path}?track-id={track.id}" + def get_full_track_info(self, track_id: str) -> Optional[FullTrackInfo]: + params = {"track": track_id, "lang": "ru"} + resp = self.session.get( + f"https://{self.domain}/handlers/track.jsx", params=params + ) + return FullTrackInfo.from_json(resp.json()) -def get_full_album_info(session: Session, album_id: str) -> FullAlbumInfo: - params = {"album": album_id, "lang": "ru"} - resp = session.get("https://music.yandex.ru/handlers/album.jsx", params=params) - return FullAlbumInfo.from_json(resp.json()) + def get_full_album_info(self, album_id: str) -> FullAlbumInfo: + params = {"album": album_id, "lang": "ru"} + resp = self.session.get( + f"https://{self.domain}/handlers/album.jsx", params=params + ) + return FullAlbumInfo.from_json(resp.json()) + def get_artist_info(self, artist_id: str) -> FullArtistInfo: + params = {"artist": artist_id, "what": "albums", "lang": "ru"} + resp = self.session.get( + f"https://{self.domain}/handlers/artist.jsx", params=params + ) + return FullArtistInfo.from_json(resp.json()) -def get_artist_info(session: Session, artist_id: str) -> FullArtistInfo: - params = {"artist": artist_id, "what": "albums", "lang": "ru"} - resp = session.get("https://music.yandex.ru/handlers/artist.jsx", params=params) - return FullArtistInfo.from_json(resp.json()) - - -def get_playlist(session: Session, playlist: PlaylistId) -> list[BasicTrackInfo]: - params = {"owner": playlist.owner, "kinds": playlist.kind, "lang": "ru"} - resp = session.get("https://music.yandex.ru/handlers/playlist.jsx", params=params) - raw_tracks = resp.json()["playlist"].get("tracks", []) - tracks = map(BasicTrackInfo.from_json, raw_tracks) - tracks = [t for t in tracks if t is not None] - return tracks + def get_playlist(self, playlist: PlaylistId) -> list[BasicTrackInfo]: + params = {"owner": playlist.owner, "kinds": playlist.kind, "lang": "ru"} + resp = self.session.get( + f"https://{self.domain}/handlers/playlist.jsx", params=params + ) + raw_tracks = resp.json()["playlist"].get("tracks", []) + tracks = map(BasicTrackInfo.from_json, raw_tracks) + tracks = [t for t in tracks if t is not None] + return tracks diff --git a/ymd/ym_api/models.py b/ymd/ym_api/models.py index 6ed56c6..8ab03a6 100644 --- a/ymd/ym_api/models.py +++ b/ymd/ym_api/models.py @@ -128,10 +128,6 @@ class BasicTrackInfo: cover_info=cover_info, ) - @property - def url(self) -> str: - return f"https://music.yandex.ru/album/{self.album.id}/track/{self.id}" - @dataclass class FullTrackInfo(BasicTrackInfo):