Add support for syncing from YT to Spotify

README.md

@@ -2,7 +2,7 @@
 
 Just a shitty and possibly over engineered Python script for syncing
 playlists between music streaming services. Right now it only supports
-syncing from Spotify to YTMusic because that's all I need.
+syncing between Spotify and YTMusic because that's all I need.
 
 This is only on GitHub because I spent multiple hours on this for no
 reason. If it's actually helpful (or could be) and you want me to finish

playlist_sync/services/base.py

@@ -30,5 +30,11 @@ def add_to_playlist(self, url: str, tracks: list[Track]):
     def _extract_track_metadata(self, track: Track) -> Any | None:
         return track._service_metadata.get(str(self))
 
+    def _remove_duplicates(self, tracks: list[str]) -> list[str]:
+        seen = set()
+        seen_add = seen.add
+        return [x for x in tracks if not (x in seen or seen_add(x))]
+
+
     def __str__(self) -> str:
         return self.__class__.__name__

playlist_sync/services/spotify.py

@@ -5,6 +5,7 @@
 from __future__ import annotations
 
 import logging
+import shlex
 from typing import TYPE_CHECKING, Any
 
 from playlist_sync.services.base import BaseService
@@ -21,7 +22,7 @@ def __init__(self, api: spotipy.Spotify):
         self.api = api
 
     FETCH_PLAYLIST_KWARGS: dict[str, Any] = dict(
-        fields='total,limit,items(track(name,artists))',
+        fields='total,limit,items(track(name,id,artists))',
         additional_types=('track',),
     )
 
@@ -61,20 +62,63 @@ def fetch_playlist(self, url: str) -> list[Track]:
             sp_track = item['track']
             artists = ', '.join(a['name'] for a in sp_track['artists'])
             track = Track(
-                title=sp_track['name'], artist=artists, service=str(self), metadata=item['track']
+                title=sp_track['name'], artist=artists, service=str(self), metadata=sp_track
             )
             tracks.append(track)
 
         return tracks
 
     def clear_playlist(self, url: str):
-        raise NotImplementedError
+        log.info(f'Fetching playlist {url}')
+        tracks = self.fetch_playlist(url)
+
+        if not tracks:
+            log.info('No songs to clear in playlist')
+            return
+
+        track_ids = [t._service_metadata[str(self)]['id'] for t in tracks]
+        result = self.api.playlist_remove_all_occurrences_of_items(url, track_ids)
+
+        if result is None:
+            log.info(f'Failed to remove')
 
     def search_track_id(self, track: Track) -> str:
-        raise NotImplementedError
+        log.info(f'Searching for id for {track}')
+        query = f'track:{shlex.quote(track.title)} artist:{shlex.quote(track.artist)}'
+        print(query)
+        result = self.api.search(query, limit=1, type='track')
+
+        if not result or result['tracks']['total'] == 0:
+            log.info('Failed to find a corresponding track id, trying with broader query')
+            result = self.api.search(str(track), limit=1, type='track')
+
+            if not result or result['tracks']['total'] == 0:
+                log.info('Failed to find a corresponding song')
+                raise RuntimeError('TODO')
+
+        track_id = result['tracks']['items'][0]['id']
+        log.info(f'Found id for {track}: {track_id}')
+        return track_id
 
     def remove_from_playlist(self, url: str, tracks: list[Track]):
         raise NotImplementedError
 
     def add_to_playlist(self, url: str, tracks: list[Track]):
-        raise NotImplementedError
+        log.info('Resolving corresponding track IDs')
+        track_ids = []
+
+        for track in tracks:
+            if str(self) in track._service_metadata:
+                # we already have the track ID for this one
+                track_ids.append(track._service_metadata[str(self)]['id'])
+            else:
+                track_ids.append(self.search_track_id(track))
+
+        # remove duplicates
+        track_ids = self._remove_duplicates(track_ids)
+
+        log.info('Adding tracks to playlist')
+        result = self.api.playlist_add_items(url, track_ids)
+
+        if result is None:
+            print('Failed to add tracks')

playlist_sync/services/ytmusic.py

@@ -31,8 +31,9 @@ def fetch_playlist(self, url: str) -> list[Track]:
         tracks = []
 
         for yt_track in yt_tracks:
+            artists = ', '.join(a['name'] for a in yt_track['artists'])
             track = Track(
-                title=yt_track['title'], artist='', service=str(self), metadata=yt_track
+                title=yt_track['title'], artist=artists, service=str(self), metadata=yt_track
             )  # TODO: artist
             tracks.append(track)
 
@@ -72,17 +73,16 @@ def remove_from_playlist(self, url: str, tracks: list[Track]):
         # self.api.remove_playlist_items(url, track_ids)
         pass
 
-    def _remove_duplicates(self, tracks: list[str]) -> list[str]:
-        seen = set()
-        seen_add = seen.add
-        return [x for x in tracks if not (x in seen or seen_add(x))]
-
     def add_to_playlist(self, url: str, tracks: list[Track]):
-        log.info('Searching for corresponding track IDs')
+        log.info('Resolving corresponding track IDs')
         track_ids = []
 
         for track in tracks:
-            track_ids.append(self.search_track_id(track))
+            if str(self) in track._service_metadata:
+                # we already have the track ID for this one
+                track_ids.append(track._service_metadata[str(self)]['videoId'])
+            else:
+                track_ids.append(self.search_track_id(track))
 
         # remove duplicates
         track_ids = self._remove_duplicates(track_ids)