Учет домена

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,

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)
+    url = client.get_track_download_url(track, hq)
-    http_utils.download_file(session, url, target_path)
+    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)

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:
+class YandexMusicApi:
-    resp = session.get(
+    session: Session
-        "https://music.yandex.ru/api/v2.1/handlers/track"
+    domain: str
-        f"/{track.id}:{track.album.id}"
-        "/web-album_track-track-track-main/download/m"
-        f"?hq={int(hq)}"
-    )
-    url_info_src = resp.json()["src"]
 
-    resp = session.get("https:" + url_info_src)
+    def __init__(self, session: Session, domain: str) -> None:
-    url_info = ET.fromstring(resp.text)
+        self.session = session
-    path = url_info.find("path").text[1:]
+        self.domain = domain
-    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_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]:
+        resp = self.session.get("https:" + url_info_src)
-    params = {"track": track_id, "lang": "ru"}
+        url_info = ET.fromstring(resp.text)
-    resp = session.get("https://music.yandex.ru/handlers/track.jsx", params=params)
+        path = url_info.find("path").text[1:]
-    return FullTrackInfo.from_json(resp.json())
+        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:
+    def get_full_album_info(self, album_id: str) -> FullAlbumInfo:
-    params = {"album": album_id, "lang": "ru"}
+        params = {"album": album_id, "lang": "ru"}
-    resp = session.get("https://music.yandex.ru/handlers/album.jsx", params=params)
+        resp = self.session.get(
-    return FullAlbumInfo.from_json(resp.json())
+            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:
+    def get_playlist(self, playlist: PlaylistId) -> list[BasicTrackInfo]:
-    params = {"artist": artist_id, "what": "albums", "lang": "ru"}
+        params = {"owner": playlist.owner, "kinds": playlist.kind, "lang": "ru"}
-    resp = session.get("https://music.yandex.ru/handlers/artist.jsx", params=params)
+        resp = self.session.get(
-    return FullArtistInfo.from_json(resp.json())
+            f"https://{self.domain}/handlers/playlist.jsx", params=params
-
+        )
-
+        raw_tracks = resp.json()["playlist"].get("tracks", [])
-def get_playlist(session: Session, playlist: PlaylistId) -> list[BasicTrackInfo]:
+        tracks = map(BasicTrackInfo.from_json, raw_tracks)
-    params = {"owner": playlist.owner, "kinds": playlist.kind, "lang": "ru"}
+        tracks = [t for t in tracks if t is not None]
-    resp = session.get("https://music.yandex.ru/handlers/playlist.jsx", params=params)
+        return tracks
-    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

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