[ie/yandexmusic] Implement track info cache to avoid Too many requests error

Signed-off-by: Terekhin Alexandr <alex@bearns.me>

yt_dlp/extractor/yandexmusic.py

@@ -8,9 +8,6 @@ import urllib
 import uuid
 from typing import Any
 
-from sphinx.ext import duration
-from wx.lib.agw.ribbon import art_internal
-
 from .common import InfoExtractor
 from ..postprocessor.decrypt import DecryptPP
 from ..utils import (
@@ -23,6 +20,7 @@ from ..utils import (
 
 class YandexMusicBaseIE(InfoExtractor):
     _VALID_URL_BASE = r'https?://music\.yandex\.(?P<tld>ru|kz|ua|by|com)'
+    _track_info_cache = {}
 
     @staticmethod
     def _handle_error(response):
@@ -67,6 +65,62 @@ class YandexMusicBaseIE(InfoExtractor):
             },
             query=query)
 
+    def _get_user_id(self) -> str | None:
+        cookies = self._get_cookies('https://music.yandex.ru')
+        user_id = None
+        if cookies:
+            user_id = cookies['Session_id'].value
+            match = re.search(r"\|(?P<target>\d+)", user_id)
+            if match:
+                user_id = match.group("target")
+            else:
+                user_id = None
+        return user_id
+
+    def _get_tracks_info(self, tld, track_id, track_ids) -> Any:
+        payload = [("trackIds", str(tid)) for tid in track_ids]
+
+        payload.append(('removeDuplicates', True))
+        payload.append(('withProgress', False))
+        payload.append(('withMixData', False))
+
+        data = urllib.parse.urlencode(query=payload).encode('utf-8')
+
+        json_response = self._download_json(
+            url_or_request=f'https://api.music.yandex.{tld}/tracks',
+            video_id=track_id,
+            data=data,
+            headers={
+                'x-request-id': uuid.uuid4(),
+                'X-Requested-With': 'XMLHttpRequest',
+                'X-Retpath-Y': track_id,
+                'x-yandex-music-client': 'YandexMusicWebNext/1.0.0',
+                'x-yandex-music-multi-auth-user-id': self._get_user_id(),
+                'x-yandex-music-without-invocation-info': 1,
+            },
+        )
+        return json_response
+
+    def _get_cashed_tracks_info(self, tld, track_id, track_ids) -> Any:
+        result = []
+        query = []
+
+        for tid in track_ids:
+            cached = YandexMusicBaseIE._track_info_cache.get(tid)
+            if cached:
+                result.append(cached)
+            else:
+                query.append(tid)
+
+        if len(query) > 0:
+            track_infos = self._get_tracks_info(tld, track_id, query)
+            result.extend(track_infos)
+            for item in query:
+                fetched = [track for track in track_infos if _track_match(track, item)]
+                if fetched is not None and len(fetched) > 0:
+                    YandexMusicBaseIE._track_info_cache[item] = fetched[0]
+
+        return result
 
 def _get_signature(track_id) -> tuple[int, str]:
     SECRET_KEY = "7tvSmFbyf5hJnIHhCimDDD"
@@ -110,6 +164,14 @@ def _get_artist(item) -> str | None:
     item = item[0] if item and len(item) > 0 else None
     return item.get('name') if item else None
 
+def _track_match(track_info, track_id) -> bool:
+    id = track_info.get('id')
+    for album in track_info.get('albums'):
+        album_id = album.get('id')
+        if track_id == f'{id}:{album_id}':
+            return True
+    return False
+
 
 class YandexMusicTrackIE(YandexMusicBaseIE):
     IE_NAME = 'yandexmusic:track'
@@ -161,7 +223,7 @@ class YandexMusicTrackIE(YandexMusicBaseIE):
         mobj = self._match_valid_url(url)
         tld, album_id, track_id = mobj.group('tld'), mobj.group('album_id'), mobj.group('id')
 
-        json_response = self._get_tracks_info(album_id, tld, track_id)
+        json_response = self._get_cashed_tracks_info(tld, url, [f'{track_id}:{album_id}'])
 
         track_title = duration_ms = artist = album = year = genre = None
 
@@ -290,29 +352,6 @@ class YandexMusicTrackIE(YandexMusicBaseIE):
 
         return track_info
 
-    def _get_tracks_info(self, album_id, tld, track_id) -> Any:
-        form_fields = {
-            'trackIds': f'{track_id}:{album_id}',
-            'removeDuplicates': True,
-            'withProgress': False,
-            'withMixData': False,
-        }
-
-        json_response = self._download_json(
-            url_or_request=f'https://api.music.yandex.{tld}/tracks',
-            video_id=track_id,
-            data=urllib.parse.urlencode(form_fields).encode('utf-8'),
-            headers={
-                'x-request-id': uuid.uuid4(),
-                'X-Requested-With': 'XMLHttpRequest',
-                'X-Retpath-Y': f'https://music.yandex.ru/album/{album_id}/track/{track_id}',
-                'x-yandex-music-client': 'YandexMusicWebNext/1.0.0',
-                'x-yandex-music-multi-auth-user-id': self._get_user_id(),
-                'x-yandex-music-without-invocation-info': 1,
-            },
-        )
-        return json_response
-
     def _get_download_info(self, tld, album_id, track_id) -> Any:
         sign_base64, timestamp = _get_signature(track_id)
         user_id = self._get_user_id()
@@ -341,18 +380,6 @@ class YandexMusicTrackIE(YandexMusicBaseIE):
         )
         return playlist['downloadInfo']
 
-    def _get_user_id(self) -> str | None:
-        cookies = self._get_cookies('https://music.yandex.ru')
-        user_id = None
-        if cookies:
-            user_id = cookies['Session_id'].value
-            match = re.search(r"\|(?P<target>\d+)", user_id)
-            if match:
-                user_id = match.group("target")
-            else:
-                user_id = None
-        return user_id
-
 
 class YandexMusicPlaylistBaseIE(YandexMusicBaseIE):
     def _extract_tracks(self, source, item_id, url, tld):
@@ -528,6 +555,12 @@ class YandexMusicPlaylistIE(YandexMusicPlaylistBaseIE):
                     'albums': albums
                 })
 
+            query = [f'{track.get("id")}:{track.get("albums")[0].get("id")}' for track in tracks]
+            batch_size = 30
+            for i in range(0, len(query), batch_size):
+                current_batch = query[i: i + batch_size]
+                self._get_cashed_tracks_info(tld, url, current_batch)
+
             playlist = {
                 'trackIds': track_ids,
                 'tracks': tracks