Import Playlists from SoundCloud (#143)

* added soundcloud import

* added soundcloud import

* added soundcloud import

* added title spliter

* added title spliter

* removed logs and cleaned the code

* added rate limit to apple/soundcloud endpoints

* defined rate limit for better clarification

* added http session with retry && refactored API calls

* minor cleanup

fix authorization header, formatting fixes, and remove unused dependency

---------

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

Author: 3 people

troi/patches/playlist_from_ms.py

@@ -6,6 +6,7 @@
 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
+from troi.tools.soundcloud_lookup import get_tracks_from_soundcloud_playlist
 
 
 class ImportPlaylistPatch(Patch):
@@ -46,18 +47,20 @@ def create(self, inputs):
         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)
+        elif music_service == "soundcloud":
+            tracks, name, desc = get_tracks_from_soundcloud_playlist(ms_token, playlist_id)
 
         source = RecordingsFromMusicServiceElement(token=ms_token, playlist_id=playlist_id, music_service=music_service, apple_user_token=apple_user_token)
 

troi/tools/apple_lookup.py

@@ -1,31 +1,30 @@
-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
+from .utils import create_http_session
 
+APPLE_MUSIC_URL = f"https://api.music.apple.com/"
 
 def get_tracks_from_apple_playlist(developer_token, user_token, playlist_id):
     """ Get tracks from the Apple Music playlist.
     """
+    http = create_http_session()
+
     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"].get("description", {}).get("standard", "")
-    else:
-        response.raise_for_status()
-    return tracks, name, description
+    response = http.get(APPLE_MUSIC_URL+f"v1/me/library/playlists/{playlist_id}?include=tracks", headers=headers)
+    response.raise_for_status()
+
+    response = response.json()
+    tracks = response["data"][0]["relationships"]["tracks"]["data"]
+    name = response["data"][0]["attributes"]["name"]
+    description = response["data"][0]["attributes"].get("description", {}).get("standard", "")
+
+    mapped_tracks = [
+        {
+            "recording_name": track['attributes']['name'],
+            "artist_name": track['attributes']['artistName']
+        }
+        for track in tracks
+    ]
+
+    return mapped_tracks, name, description

troi/tools/common_lookup.py

@@ -3,8 +3,9 @@
 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
+from troi.tools.apple_lookup import get_tracks_from_apple_playlist
+from troi.tools.spotify_lookup import get_tracks_from_spotify_playlist
+from troi.tools.soundcloud_lookup import get_tracks_from_soundcloud_playlist
 
 MAX_LOOKUPS_PER_POST = 50
 MBID_LOOKUP_URL = "https://api.listenbrainz.org/1/metadata/lookup/"
@@ -16,19 +17,19 @@ def music_service_tracks_to_mbid(token, playlist_id, music_service, apple_user_t
     """ 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)
+        tracks, name, desc = get_tracks_from_spotify_playlist(token, playlist_id)
     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)
+        tracks, name, desc = get_tracks_from_apple_playlist(token, apple_user_token, playlist_id)
+    elif music_service == "soundcloud":
+        tracks, name, desc = get_tracks_from_soundcloud_playlist(token, playlist_id)
     else:
         raise ValueError("Unknown music service")
 
     track_lists = list(chunked(tracks, MAX_LOOKUPS_PER_POST))
-    return mbid_mapping_spotify(track_lists)
+    return mbid_mapping_tracks(track_lists)
 
 
-def mbid_mapping_spotify(track_lists):
+def mbid_mapping_tracks(track_lists):
     """ Given a track_name and artist_name, try to find MBID for these tracks from mbid lookup.
     """
     track_mbids = []

troi/tools/soundcloud_lookup.py

@@ -0,0 +1,29 @@
+from .utils import create_http_session
+
+SOUNDCLOUD_URL = f"https://api.soundcloud.com/"
+
+def get_tracks_from_soundcloud_playlist(developer_token, playlist_id):
+    """ Get tracks from the Soundcloud playlist.
+    """
+    http = create_http_session()
+
+    headers = {
+        "Authorization": f"OAuth {developer_token}",
+    }
+    response = http.get(f"{SOUNDCLOUD_URL}/playlists/{playlist_id}", headers=headers)
+    response.raise_for_status()
+
+    response = response.json()
+    tracks = response["tracks"]
+    name = response["title"]
+    description = response["description"]
+
+    mapped_tracks = [
+        {
+            "recording_name": track['title'].split(" - ")[1] if " - " in track['title'] else track['title'],
+            "artist_name": track['title'].split(" - ")[0] if " - " in track['title'] else track['user']['username']
+        }
+        for track in tracks
+    ]
+
+    return mapped_tracks, name, description

troi/tools/spotify_lookup.py

@@ -182,6 +182,7 @@ def get_tracks_from_spotify_playlist(spotify_token, playlist_id):
     name = playlist_info["name"]
     description = playlist_info["description"]
 
+    tracks = convert_spotify_tracks_to_json(tracks)
     return tracks, name, description
 
 

troi/tools/utils.py

@@ -0,0 +1,20 @@
+import requests
+from requests.adapters import HTTPAdapter
+from urllib3.util.retry import Retry
+
+
+def create_http_session():
+    """ Create an HTTP session with retry strategy for handling rate limits and server errors.
+    """
+    retry_strategy = Retry(
+        total=3,
+        status_forcelist=[429, 500, 502, 503, 504], 
+        allowed_methods=["HEAD", "GET", "OPTIONS"], 
+        backoff_factor=1
+    )
+
+    adapter = HTTPAdapter(max_retries=retry_strategy)
+    http = requests.Session()
+    http.mount("https://", adapter)
+    http.mount("http://", adapter)
+    return http