A new Patch for importing playlists from Spotify (#141)

* added new patch for import

* fixed token

* fixed token

* fixed token

* fixed token

* fixed track fetching

* added comments

* TEST: added chunking

* fixed chunking

* integrated  apple music

* fixed typo

* fixed missing parameter

* Refactor lookup functions

---------

Co-authored-by: Rimma Kubanova <[email protected]>
Co-authored-by: Kartik Ohri <[email protected]>

troi/patches/playlist_from_ms.py

@@ -0,0 +1,70 @@
+import json
+
+from troi import Playlist
+from troi.patch import Patch
+from troi.playlist import RecordingsFromMusicServiceElement, PlaylistMakerElement
+from troi.musicbrainz.recording_lookup import RecordingLookupElement
+from troi.tools.apple_lookup import get_tracks_from_apple_playlist
+from troi.tools.spotify_lookup import get_tracks_from_spotify_playlist
+
+
+class ImportPlaylistPatch(Patch):
+
+    @staticmethod
+    def inputs():
+        """
+        A patch that retrieves an existing playlist from Spotify for use in Troi.
+
+        \b
+        MS_TOKEN is the music service token from which the playlist is retrieved. For now, only Spotify tokens are accepted. 
+        PLAYLIST_ID is the playlist id to retrieve the tracks from it.
+        MUSIC_SERVICE is the music service from which the playlist is retrieved
+        APPLE_USER_TOKEN is the apple user token. Optional, if music services is not Apple Music
+        """
+        return [
+            {"type": "argument", "args": ["ms_token"], "kwargs": {"required": False}},
+            {"type": "argument", "args": ["playlist_id"], "kwargs": {"required": False}},
+            {"type": "argument", "args": ["music_service"], "kwargs": {"required": False}},
+            {"type": "argument", "args": ["apple_user_token"], "kwargs": {"required": False}},
+        ]
+
+    @staticmethod
+    def outputs():
+        return [Playlist]
+
+    @staticmethod
+    def slug():
+        return "import-playlist"
+
+    @staticmethod
+    def description():
+        return "Retrieve a playlist from the Music Services (Spotify/Apple Music/SoundCloud)"
+
+    def create(self, inputs):
+
+        ms_token = inputs["ms_token"]
+        playlist_id = inputs["playlist_id"]
+        music_service = inputs["music_service"]
+        apple_user_token = inputs["apple_user_token"]
+
+        if apple_user_token == "":
+            apple_user_token = None
+
+        if music_service == "apple_music" and apple_user_token is None:
+            raise RuntimeError("Authentication is required")
+
+        # this one only used to get track name and desc
+        if music_service == "spotify":
+            tracks, name, desc = get_tracks_from_spotify_playlist(ms_token, playlist_id)
+        elif music_service == "apple_music":
+            tracks, name, desc = get_tracks_from_apple_playlist(ms_token, apple_user_token, playlist_id)
+
+        source = RecordingsFromMusicServiceElement(token=ms_token, playlist_id=playlist_id, music_service=music_service, apple_user_token=apple_user_token)
+
+        rec_lookup = RecordingLookupElement()
+        rec_lookup.set_sources(source)
+
+        pl_maker = PlaylistMakerElement(name, desc, patch_slug=self.slug())
+        pl_maker.set_sources(rec_lookup)
+
+        return pl_maker

troi/playlist.py

@@ -9,6 +9,7 @@
 from troi import Recording, Playlist, PipelineError, Element, Artist, ArtistCredit, Release
 from troi.operations import is_homogeneous
 from troi.print_recording import PrintRecordingList
+from troi.tools.common_lookup import music_service_tracks_to_mbid
 from troi.tools.spotify_lookup import submit_to_spotify
 
 logger = logging.getLogger(__name__)
@@ -523,6 +524,39 @@ def read(self, inputs):
         return [playlist]
 
 
+class RecordingsFromMusicServiceElement(Element):
+    """ Create a troi.Playlist entity from track and artist names."""
+
+    def __init__(self, token=None, playlist_id=None, music_service=None, apple_user_token=None):
+        """
+            Args:
+                playlist_id: id of the Spotify playlist to be used for creating the playlist element
+                token: the Spotify token to fetch the playlist tracks
+                music_service: the name of the music service to be used for fetching the playlist data
+                apple_music_token (optional): the user token for Apple Music API 
+        """
+        super().__init__()
+        self.token = token
+        self.playlist_id = playlist_id
+        self.music_service = music_service
+        self.apple_user_token = apple_user_token
+
+
+    @staticmethod
+    def outputs():
+        return [ Recording ]
+
+    def read(self, inputs):
+        recordings = []
+
+        mbid_mapped_tracks = music_service_tracks_to_mbid(self.token, self.playlist_id, self.music_service, self.apple_user_token)
+        if mbid_mapped_tracks:
+            for mbid in mbid_mapped_tracks:
+                recordings.append(Recording(mbid=mbid))
+
+        return recordings
+
+
 class PlaylistFromJSPFElement(Element):
     """ Create a troi.Playlist entity from a ListenBrainz JSPF playlist or LB playlist."""
 

troi/tools/apple_lookup.py

@@ -0,0 +1,31 @@
+import requests
+
+from troi.tools.spotify_lookup import APPLE_MUSIC_URL
+
+
+def convert_apple_tracks_to_json(apple_tracks):
+    tracks= []
+    for track in apple_tracks:
+        tracks.append({
+            "recording_name": track['attributes']['name'],
+            "artist_name": track['attributes']['artistName'],
+        })
+    return tracks
+
+
+def get_tracks_from_apple_playlist(developer_token, user_token, playlist_id):
+    """ Get tracks from the Apple Music playlist.
+    """
+    headers = {
+        "Authorization": f"Bearer {developer_token}",
+        "Music-User-Token": user_token
+    }
+    response = requests.get(APPLE_MUSIC_URL+f"v1/me/library/playlists/{playlist_id}?include=tracks", headers=headers)
+    if response.status_code == 200:
+        response = response.json()
+        tracks = response["data"][0]["relationships"]["tracks"]["data"]
+        name = response["data"][0]["attributes"]["name"]
+        description = response["data"][0]["attributes"]["description"]["standard"]
+    else:
+        response.raise_for_status()
+    return tracks, name, description

troi/tools/common_lookup.py

@@ -0,0 +1,47 @@
+import logging
+
+import requests
+from more_itertools import chunked
+
+from troi.tools.apple_lookup import get_tracks_from_apple_playlist, convert_apple_tracks_to_json
+from troi.tools.spotify_lookup import get_tracks_from_spotify_playlist, convert_spotify_tracks_to_json
+
+MAX_LOOKUPS_PER_POST = 50
+MBID_LOOKUP_URL = "https://api.listenbrainz.org/1/metadata/lookup/"
+
+logger = logging.getLogger(__name__)
+
+
+def music_service_tracks_to_mbid(token, playlist_id, music_service, apple_user_token=None):
+    """ Convert Spotify playlist tracks to a list of MBID tracks.
+    """
+    if music_service == "spotify":
+        tracks_from_playlist, name, desc = get_tracks_from_spotify_playlist(token, playlist_id)
+        tracks = convert_spotify_tracks_to_json(tracks_from_playlist)
+    elif music_service == "apple_music":
+        tracks_from_playlist, name, desc = get_tracks_from_apple_playlist(token, apple_user_token, playlist_id)
+        tracks = convert_apple_tracks_to_json(tracks_from_playlist)
+    else:
+        raise ValueError("Unknown music service")
+
+    track_lists = list(chunked(tracks, MAX_LOOKUPS_PER_POST))
+    return mbid_mapping_spotify(track_lists)
+
+
+def mbid_mapping_spotify(track_lists):
+    """ Given a track_name and artist_name, try to find MBID for these tracks from mbid lookup.
+    """
+    track_mbids = []
+    for tracks in track_lists:
+        params = {
+            "recordings": tracks
+        }
+        response = requests.post(MBID_LOOKUP_URL, json=params)
+        if response.status_code == 200:
+            data = response.json()
+            for d in data:
+                if d is not None and "recording_mbid" in d:
+                    track_mbids.append(d["recording_mbid"])
+        else:
+            logger.error("Error occurred: %s", response.text)
+    return track_mbids

troi/tools/spotify_lookup.py

@@ -8,6 +8,7 @@
 
 logger = logging.getLogger(__name__)
 
+APPLE_MUSIC_URL = f"https://api.music.apple.com/"
 SPOTIFY_IDS_LOOKUP_URL = "https://labs.api.listenbrainz.org/spotify-id-from-mbid/json"
 
 
@@ -170,3 +171,32 @@ def submit_to_spotify(spotify, playlist, spotify_user_id: str, is_public: bool =
     playlist.add_metadata({"external_urls": {"spotify": playlist_url}})
 
     return playlist_url, playlist_id
+
+
+def get_tracks_from_spotify_playlist(spotify_token, playlist_id):
+    """ Get tracks from the Spotify playlist.
+    """
+    sp = spotipy.Spotify(auth=spotify_token, requests_timeout=10, retries=10)
+    playlist_info = sp.playlist(playlist_id)
+    tracks = sp.playlist_items(playlist_id, limit=100)
+    name = playlist_info["name"]
+    description = playlist_info["description"]
+
+    return tracks, name, description
+
+
+def convert_spotify_tracks_to_json(spotify_tracks):
+    tracks = []
+    for track in spotify_tracks["items"]:
+        artists = track["track"].get("artists", [])
+        artist_names = []
+        for a in artists:
+            name = a.get("name")
+            if name is not None:
+                artist_names.append(name)
+        artist_name = ", ".join(artist_names)
+        tracks.append({
+            "recording_name": track["track"]["name"],
+            "artist_name": artist_name,
+        })
+    return tracks