return await loop.run_in_executor(None, _get_song)
-async def get_library_artists(headers: Dict[str, str]) -> Dict[str, str]:
+async def get_library_artists(headers: Dict[str, str], username: str) -> Dict[str, str]:
"""Async wrapper around the ytmusicapi get_library_artists function."""
def _get_library_artists():
- ytm = ytmusicapi.YTMusic(auth=json.dumps(headers))
+ user = username if is_brand_account(username) else None
+ ytm = ytmusicapi.YTMusic(auth=json.dumps(headers), user=user)
artists = ytm.get_library_artists(limit=9999)
# Sync properties with uniformal artist object
for artist in artists:
return await loop.run_in_executor(None, _get_library_artists)
-async def get_library_albums(headers: Dict[str, str]) -> Dict[str, str]:
+async def get_library_albums(headers: Dict[str, str], username: str) -> Dict[str, str]:
"""Async wrapper around the ytmusicapi get_library_albums function."""
def _get_library_albums():
- ytm = ytmusicapi.YTMusic(auth=json.dumps(headers))
+ user = username if is_brand_account(username) else None
+ ytm = ytmusicapi.YTMusic(auth=json.dumps(headers), user=user)
return ytm.get_library_albums(limit=9999)
loop = asyncio.get_running_loop()
return await loop.run_in_executor(None, _get_library_albums)
-async def get_library_playlists(headers: Dict[str, str]) -> Dict[str, str]:
+async def get_library_playlists(
+ headers: Dict[str, str], username: str
+) -> Dict[str, str]:
"""Async wrapper around the ytmusicapi get_library_playlists function."""
def _get_library_playlists():
- ytm = ytmusicapi.YTMusic(auth=json.dumps(headers))
+ user = username if is_brand_account(username) else None
+ ytm = ytmusicapi.YTMusic(auth=json.dumps(headers), user=user)
playlists = ytm.get_library_playlists(limit=9999)
# Sync properties with uniformal playlist object
for playlist in playlists:
return await loop.run_in_executor(None, _get_library_playlists)
-async def get_library_tracks(headers: Dict[str, str]) -> Dict[str, str]:
+async def get_library_tracks(headers: Dict[str, str], username: str) -> Dict[str, str]:
"""Async wrapper around the ytmusicapi get_library_tracks function."""
def _get_library_tracks():
- ytm = ytmusicapi.YTMusic(auth=json.dumps(headers))
+ user = username if is_brand_account(username) else None
+ ytm = ytmusicapi.YTMusic(auth=json.dumps(headers), user=user)
tracks = ytm.get_library_songs(limit=9999)
return tracks
if key in playlist_obj:
return playlist_obj[key]
return str(int(time()))
+
+
+def is_brand_account(username: str) -> bool:
+ """Check if the provided username is a brand-account."""
+ return len(username) == 21 and username.isdigit()
async def get_library_artists(self) -> AsyncGenerator[Artist, None]:
"""Retrieve all library artists from Youtube Music."""
- artists_obj = await get_library_artists(headers=self._headers)
+ artists_obj = await get_library_artists(
+ headers=self._headers, username=self.config.username
+ )
for artist in artists_obj:
yield await self._parse_artist(artist)
async def get_library_albums(self) -> AsyncGenerator[Album, None]:
"""Retrieve all library albums from Youtube Music."""
- albums_obj = await get_library_albums(headers=self._headers)
+ albums_obj = await get_library_albums(
+ headers=self._headers, username=self.config.username
+ )
for album in albums_obj:
yield await self._parse_album(album, album["browseId"])
async def get_library_playlists(self) -> AsyncGenerator[Playlist, None]:
"""Retrieve all library playlists from the provider."""
- playlists_obj = await get_library_playlists(headers=self._headers)
+ playlists_obj = await get_library_playlists(
+ headers=self._headers, username=self.config.username
+ )
for playlist in playlists_obj:
yield await self._parse_playlist(playlist)
async def get_library_tracks(self) -> AsyncGenerator[Track, None]:
"""Retrieve library tracks from Youtube Music."""
- tracks_obj = await get_library_tracks(headers=self._headers)
+ tracks_obj = await get_library_tracks(
+ headers=self._headers, username=self.config.username
+ )
for track in tracks_obj:
# Library tracks sometimes do not have a valid artist id
# In that case, call the API for track details based on track id