feat: download artist

src/api.py

@@ -138,3 +138,46 @@ async def get_song_lyrics(song_id: str, storefront: str, token: str, dsid: str,
                            cookies={f"mz_at_ssl-{dsid}": account_token})
     result = SongLyrics.model_validate(req.json())
     return result.data[0].attributes.ttml
+
+
+@retry(retry=retry_if_exception_type((httpx.TimeoutException, httpcore.ConnectError, SSLError, FileNotFoundError)),
+       stop=stop_after_attempt(5),
+       before_sleep=before_sleep_log(logger, logging.WARN))
+async def get_albums_from_artist(artist_id: str, storefront: str, token: str, lang: str, offset: int = 0):
+    resp = await client.get(f"https://amp-api.music.apple.com/v1/catalog/{storefront}/artists/{artist_id}/albums",
+                            params={"l": lang},
+                            headers={"Authorization": f"Bearer {token}", "User-Agent": user_agent_browser,
+                                     "Origin": "https://music.apple.com"})
+    artist_album = ArtistAlbums.parse_obj(resp.json())
+    albums = [album.attributes.url for album in artist_album.data]
+    if artist_album.next:
+        next_albums = await get_albums_from_artist(artist_id, storefront, token, lang, offset + 25)
+        albums.extend(next_albums)
+    return list(set(albums))
+
+
+@retry(retry=retry_if_exception_type((httpx.TimeoutException, httpcore.ConnectError, SSLError, FileNotFoundError)),
+       stop=stop_after_attempt(5),
+       before_sleep=before_sleep_log(logger, logging.WARN))
+async def get_songs_from_artist(artist_id: str, storefront: str, token: str, lang: str, offset: int = 0):
+    resp = await client.get(f"https://amp-api.music.apple.com/v1/catalog/{storefront}/artists/{artist_id}/songs",
+                            params={"l": lang},
+                            headers={"Authorization": f"Bearer {token}", "User-Agent": user_agent_browser,
+                                     "Origin": "https://music.apple.com"})
+    artist_song = ArtistSongs.parse_obj(resp.json())
+    songs = [song.attributes.url for song in artist_song.data]
+    if artist_song.next:
+        next_songs = await get_songs_from_artist(artist_id, storefront, token, lang, offset + 20)
+        songs.extend(next_songs)
+    return list[set(songs)]
+
+
+@retry(retry=retry_if_exception_type((httpx.TimeoutException, httpcore.ConnectError, SSLError, FileNotFoundError)),
+       stop=stop_after_attempt(5),
+       before_sleep=before_sleep_log(logger, logging.WARN))
+async def get_artist_info(artist_id: str, storefront: str, token: str, lang: str):
+    resp = await client.get(f"https://amp-api.music.apple.com/v1/catalog/{storefront}/artists/{artist_id}",
+                            params={"l": lang},
+                            headers={"Authorization": f"Bearer {token}", "User-Agent": user_agent_browser,
+                                     "Origin": "https://music.apple.com"})
+    return ArtistInfo.parse_obj(resp.json())

src/cmd.py

@@ -11,7 +11,7 @@ from prompt_toolkit.patch_stdout import patch_stdout
 from src.adb import Device
 from src.api import get_token, init_client_and_lock, upload_m3u8_to_api, get_info_from_adam
 from src.config import Config
-from src.rip import rip_song, rip_album
+from src.rip import rip_song, rip_album, rip_artist
 from src.types import GlobalAuthParams
 from src.url import AppleMusicURL, URLType, Song
 from src.utils import get_song_id_from_m3u8
@@ -41,6 +41,7 @@ class NewInteractiveShell:
                                      choices=["alac", "ec3", "aac", "aac-binaural", "aac-downmix", "ac3"],
                                      default="alac")
         download_parser.add_argument("-f", "--force", type=bool, default=False)
+        download_parser.add_argument("--include-participate-songs", type=bool, default=False, dest="include")
         m3u8_parser = subparser.add_parser("m3u8")
         m3u8_parser.add_argument("url", type=str)
         m3u8_parser.add_argument("-c", "--codec",
@@ -79,7 +80,7 @@ class NewInteractiveShell:
             return
         match cmds[0]:
             case "download":
-                await self.do_download(args.url, args.codec, args.force)
+                await self.do_download(args.url, args.codec, args.force, args.include)
             case "m3u8":
                 await self.do_m3u8(args.url, args.codec, args.force)
             case "mitm":
@@ -88,17 +89,26 @@ class NewInteractiveShell:
                 self.loop.stop()
                 sys.exit()
 
-    async def do_download(self, raw_url: str, codec: str, force_download: bool):
+    async def do_download(self, raw_url: str, codec: str, force_download: bool, include: bool = False):
         url = AppleMusicURL.parse_url(raw_url)
         available_device = await self._get_available_device(url.storefront)
         global_auth_param = GlobalAuthParams.from_auth_params_and_token(available_device.get_auth_params(),
                                                                         self.anonymous_access_token)
+        tasks = set()
         match url.type:
             case URLType.Song:
                 task = self.loop.create_task(
                     rip_song(url, global_auth_param, codec, self.config, available_device, force_download))
             case URLType.Album:
                 task = self.loop.create_task(rip_album(url, global_auth_param, codec, self.config, available_device))
+            case URLType.Artist:
+                task = self.loop.create_task(rip_artist(url, global_auth_param, codec, self.config, available_device,
+                                                        force_download, include))
+            case _:
+                logger.error("Unsupported URLType")
+                return
+        tasks.add(task)
+        task.add_done_callback(tasks.remove)
 
     async def do_m3u8(self, m3u8_url: str, codec: str, force_download: bool):
         song_id = get_song_id_from_m3u8(m3u8_url)

src/models/__init__.py

@@ -3,3 +3,6 @@ from src.models.playlist_meta import PlaylistMeta
 from src.models.song_data import SongData
 from src.models.song_lyrics import SongLyrics
 from src.models.tracks_meta import TracksMeta
+from src.models.artist_albums import ArtistAlbums
+from src.models.artist_songs import ArtistSongs
+from src.models.artist_info import ArtistInfo

src/models/artist_albums.py

@@ -0,0 +1,71 @@
+from __future__ import annotations
+
+from typing import List, Optional
+
+from pydantic import BaseModel
+
+
+class Artwork(BaseModel):
+    width: Optional[int] = None
+    url: Optional[str] = None
+    height: Optional[int] = None
+    textColor3: Optional[str] = None
+    textColor2: Optional[str] = None
+    textColor4: Optional[str] = None
+    textColor1: Optional[str] = None
+    bgColor: Optional[str] = None
+    hasP3: bool
+
+
+class PlayParams(BaseModel):
+    id: Optional[str] = None
+    kind: Optional[str] = None
+
+
+class EditorialNotes(BaseModel):
+    short: Optional[str] = None
+    standard: Optional[str] = None
+    name: Optional[str] = None
+
+
+class Attributes(BaseModel):
+    copyright: Optional[str] = None
+    genreNames: List[str]
+    releaseDate: Optional[str] = None
+    isMasteredForItunes: bool
+    upc: Optional[str] = None
+    artwork: Artwork
+    url: Optional[str] = None
+    playParams: PlayParams
+    recordLabel: Optional[str] = None
+    trackCount: Optional[int] = None
+    isCompilation: bool
+    isPrerelease: bool
+    audioTraits: List[str]
+    isSingle: bool
+    name: Optional[str] = None
+    artistName: Optional[str] = None
+    isComplete: bool
+    editorialNotes: Optional[EditorialNotes] = None
+
+
+class ContentVersion(BaseModel):
+    MZ_INDEXER: Optional[int] = None
+    RTCI: Optional[int] = None
+
+
+class Meta(BaseModel):
+    contentVersion: ContentVersion
+
+
+class Datum(BaseModel):
+    id: Optional[str] = None
+    type: Optional[str] = None
+    href: Optional[str] = None
+    attributes: Attributes
+    meta: Meta
+
+
+class ArtistAlbums(BaseModel):
+    next: Optional[str] = None
+    data: List[Datum]

src/models/artist_info.py

@@ -0,0 +1,53 @@
+from __future__ import annotations
+
+from typing import List, Optional
+
+from pydantic import BaseModel
+
+
+class Artwork(BaseModel):
+    width: Optional[int] = None
+    url: Optional[str] = None
+    height: Optional[int] = None
+    textColor3: Optional[str] = None
+    textColor2: Optional[str] = None
+    textColor4: Optional[str] = None
+    textColor1: Optional[str] = None
+    bgColor: Optional[str] = None
+    hasP3: bool
+
+
+class Attributes(BaseModel):
+    genreNames: List[Optional[str]]  = None
+    name: Optional[str] = None
+    artwork: Artwork
+    classicalUrl: Optional[str] = None
+    url: Optional[str] = None
+
+
+class Datum1(BaseModel):
+    id: Optional[str] = None
+    type: Optional[str] = None
+    href: Optional[str] = None
+
+
+class Albums(BaseModel):
+    href: Optional[str] = None
+    next: Optional[str] = None
+    data: List[Datum1]
+
+
+class Relationships(BaseModel):
+    albums: Albums
+
+
+class Datum(BaseModel):
+    id: Optional[str] = None
+    type: Optional[str] = None
+    href: Optional[str] = None
+    attributes: Attributes
+    relationships: Relationships
+
+
+class ArtistInfo(BaseModel):
+    data: List[Datum]

src/models/artist_songs.py

@@ -0,0 +1,73 @@
+from __future__ import annotations
+
+from typing import List, Optional
+
+from pydantic import BaseModel
+
+
+class Artwork(BaseModel):
+    width: Optional[int] = None
+    url: Optional[str] = None
+    height: Optional[int] = None
+    textColor3: Optional[str] = None
+    textColor2: Optional[str] = None
+    textColor4: Optional[str] = None
+    textColor1: Optional[str] = None
+    bgColor: Optional[str] = None
+    hasP3: bool
+
+
+class PlayParams(BaseModel):
+    id: Optional[str] = None
+    kind: Optional[str] = None
+
+
+class Preview(BaseModel):
+    url: Optional[str] = None
+
+
+class Attributes(BaseModel):
+    hasTimeSyncedLyrics: bool
+    albumName: Optional[str] = None
+    genreNames: List[str]
+    trackNumber: Optional[int] = None
+    releaseDate: Optional[str] = None
+    durationInMillis: Optional[int] = None
+    isVocalAttenuationAllowed: bool
+    isMasteredForItunes: bool
+    isrc: Optional[str] = None
+    artwork: Artwork
+    audioLocale: Optional[str] = None
+    composerName: Optional[str] = None
+    url: Optional[str] = None
+    playParams: PlayParams
+    discNumber: Optional[int] = None
+    hasCredits: bool
+    hasLyrics: bool
+    isAppleDigitalMaster: bool
+    audioTraits: List[str]
+    name: Optional[str] = None
+    previews: List[Preview]
+    artistName: Optional[str] = None
+
+
+class ContentVersion(BaseModel):
+    RTCI: Optional[int] = None
+    MZ_INDEXER: Optional[int] = None
+
+
+class Meta(BaseModel):
+    contentVersion: ContentVersion
+
+
+class Datum(BaseModel):
+    id: Optional[str] = None
+    type: Optional[str] = None
+    href: Optional[str] = None
+    attributes: Attributes
+    meta: Meta
+
+
+class ArtistSongs(BaseModel):
+    next: Optional[str] = None
+    data: List[Datum]

src/rip.py

@@ -3,14 +3,15 @@ import subprocess
 
 from loguru import logger
 
-from src.api import get_info_from_adam, get_song_lyrics, get_meta, download_song, get_m3u8_from_api
+from src.api import (get_info_from_adam, get_song_lyrics, get_meta, download_song,
+                     get_m3u8_from_api, get_artist_info, get_songs_from_artist, get_albums_from_artist)
 from src.config import Config, Device
 from src.decrypt import decrypt
 from src.metadata import SongMetadata
 from src.mp4 import extract_media, extract_song, encapsulate, write_metadata
 from src.save import save
 from src.types import GlobalAuthParams, Codec
-from src.url import Song, Album, URLType
+from src.url import Song, Album, URLType, Artist
 from src.utils import check_song_exists
 
 
@@ -74,5 +75,17 @@ async def rip_playlist():
     pass
 
 
-async def rip_artist():
+async def rip_artist(artist: Artist, auth_params: GlobalAuthParams, codec: str, config: Config, device: Device,
-    pass
+                     force_save: bool = False, include_participate_in_works: bool = False):
+    artist_info = await get_artist_info(artist.id, artist.storefront, auth_params.anonymousAccessToken, config.region.language)
+    logger.info(f"Ripping Artist: {artist_info.data[0].attributes.name}")
+    async with asyncio.TaskGroup() as tg:
+        if include_participate_in_works:
+            songs = await get_songs_from_artist(artist.id, artist.storefront, auth_params.anonymousAccessToken, config.region.language)
+            for song_url in songs:
+                tg.create_task(rip_song(Song.parse_url(song_url), auth_params, codec, config, device, force_save))
+        else:
+            albums = await get_albums_from_artist(artist.id, artist.storefront, auth_params.anonymousAccessToken, config.region.language)
+            for album_url in albums:
+                tg.create_task(rip_album(Album.parse_url(album_url), auth_params, codec, config, device, force_save))
+    logger.info(f"Artist: {artist_info.data[0].attributes.name} finished ripping")