diff --git a/CHANGELOG.md b/CHANGELOG.md index dc50f31..28cbadb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [0.6.1] - Upcoming +### Added + +- Added a migration example to fully copy a playlist from an arbitrary service to another. (#60) + ### Fixed - When syncing playlists to traktor, we now insert missing tracks into the library collection. This avoids those track to disappear from the playlist when launching Traktor after the sync. (#54) diff --git a/examples/community/migrate.py b/examples/community/migrate.py new file mode 100644 index 0000000..718dcbf --- /dev/null +++ b/examples/community/migrate.py @@ -0,0 +1,173 @@ +"""Migrate your playlists from one service to another. + +Copies all playlists from one service to another. This +allows to quickly migrate to another music service +provider. +""" + +import sys +from typing import Annotated, NamedTuple + +import typer + +from plistsync.core import Library, Matches, ServicePlaylist, Track +from plistsync.logger import log +from plistsync.services.spotify import SpotifyLibrary +from plistsync.services.tidal import TidalLibrary + + +class MigrationContext(NamedTuple): + overwrite: bool + skip_empty: bool + + +def migrate_playlist( + from_playlist: ServicePlaylist, + to_playlist: ServicePlaylist, + to_library: Library, +): + """Migrate a single playlist from one service to another. + + Needs to_library to match tracks in the given library. + This will overwrite all existing tracks in the destination playlist. + """ + log.info(f"Transferring {from_playlist.name!r} with {len(from_playlist)} tracks.") + + log.debug(f"Matching tracks on {to_library.name!r}...") + matches: list[Matches[Track]] = [ + to_library.match(t) # TODO: Multi match would be nice + for t in from_playlist.tracks + ] + log.debug("Finished matching track.") + + for match in matches: + if match.best_match is None: + log.warning( + f"Couldn't find '{match.truth.title} " + f"- {match.truth.primary_artist}' " + f"on {to_library.name!r}" + ) + + with to_playlist.edit(): + to_playlist.tracks = [ + match.best_match for match in matches if match.best_match is not None + ] + + log.info(f"Successfully migrated {from_playlist.name!r}.") + + +def migrate_library( + from_library: Library, + to_library: Library, + context: MigrationContext, +): + # Construct mapping of all playlists + existing_playlists_to_service = {pl.name: pl for pl in to_library.playlists} + + # TODO: It would be nice to have a playlist picker here + for from_playlist in from_library.playlists: + to_playlist = existing_playlists_to_service.get(from_playlist.name) + + if not isinstance(from_playlist, ServicePlaylist): + raise NotImplementedError( + "Migration not supported for {from_library.name!r} playlists. " + ) + + if context.skip_empty and len(from_playlist) == 0: + log.info(f"Skipping empty playlist {from_playlist.name!r}.") + continue + + if ( + to_playlist is not None + and not context.overwrite + and not typer.prompt( + f"Found existing {to_playlist.name!r} on {to_library.name!r}." + "Overwrite?", + type=bool, + default=True, + ) + ): + log.warning( + f"Not overwriting {to_playlist.name!r} on {to_library.name!r}. " + "This will yield two playlists with the same name." + ) + to_playlist = to_library.create_playlist( + from_playlist.name, + from_playlist.description, + ) + else: + to_playlist = to_library.create_playlist( + from_playlist.name, + from_playlist.description, + ) + + if not isinstance(to_playlist, ServicePlaylist): + raise NotImplementedError( + "Migration not supported for {to_playlist.name!r} playlists. " + ) + + migrate_playlist( + from_playlist, + to_playlist, + to_library, + ) + + +service_mapping: dict[str, type[SpotifyLibrary] | type[TidalLibrary]] = { + "spotify": SpotifyLibrary, + "tidal": TidalLibrary, +} + + +def main( + from_service: Annotated[ + str, + typer.Argument( + help="Source of the playlists, either 'spotify' or 'tidal'", + ), + ], + to_service: Annotated[ + str, + typer.Argument( + help="Destination of the playlists, either 'spotify' or 'tidal'.", + ), + ], + overwrite: Annotated[ + bool, + typer.Option( + help="Overwrite playlists if found by name in 'to_service'", + ), + ] = False, + skip_empty: Annotated[ + bool, + typer.Option( + help="Skip empty playlist in migration.", + ), + ] = True, +): + if not (from_library := service_mapping.get(from_service.lower())): + log.error( + f"Invalid from_service {from_service!r}." + f"Pick one of {service_mapping.keys()}" + ) + sys.exit(1) + if not (to_library := service_mapping.get(to_service.lower())): + log.error( + f"Invalid to_service {to_service!r}. " + f"Pick one of {list(service_mapping.keys())}." + ) + sys.exit(1) + + if from_library == to_library: + raise ValueError("from_service and to_service must be different!") + + migrate_library( + from_library(), + to_library(), + MigrationContext(overwrite=overwrite, skip_empty=skip_empty), + ) + + +main.__doc__ = __doc__ # use module docstring as help +if __name__ == "__main__": + typer.run(main) diff --git a/plistsync/core/__init__.py b/plistsync/core/__init__.py index aba666a..62a85c6 100644 --- a/plistsync/core/__init__.py +++ b/plistsync/core/__init__.py @@ -7,6 +7,8 @@ """ from .collection import Collection, Library +from .matching import Matches +from .playlist import Playlist, ServicePlaylist from .rewrite import PathRewrite from .track import GlobalTrackIDs, Track @@ -16,4 +18,7 @@ "PathRewrite", "Track", "GlobalTrackIDs", + "Playlist", + "ServicePlaylist", + "Matches", ] diff --git a/plistsync/core/collection.py b/plistsync/core/collection.py index 408fe5e..fd2274c 100644 --- a/plistsync/core/collection.py +++ b/plistsync/core/collection.py @@ -67,7 +67,7 @@ class MyTrackCollection(Collection, GlobalLookup, LocalLookup, TrackStream): if TYPE_CHECKING: from .playlist import Playlist, PlaylistIDs -Plist = TypeVar("Plist", bound="Playlist", default="Playlist") +Plist = TypeVar("Plist", bound="Playlist", default="Playlist", covariant=True) @runtime_checkable @@ -393,6 +393,11 @@ class Library(Generic[T, Plist], Collection[T]): to implement its specifics. """ + @property + def name(self) -> str: + """Name of the library, typically the service name.""" + return type(self).__name__.replace("Library", "") + @property @abstractmethod def playlists(self) -> Iterable[Plist]: diff --git a/plistsync/core/playlist.py b/plistsync/core/playlist.py index 72e43fb..5e0fa8d 100644 --- a/plistsync/core/playlist.py +++ b/plistsync/core/playlist.py @@ -248,8 +248,8 @@ class ServicePlaylist(Generic[T], Playlist[T], ABC): an `OfflinePlaylist` to retain the data. Provides two synchronization strategies: - - `edit()` – transactional edits with automatic local rollback - - `update()` – bulk update by comparing remote and local snapshots + - `edit()` - transactional edits with automatic local rollback + - `update()` - bulk update by comparing remote and local snapshots """ library: Library[Track, Self] diff --git a/plistsync/services/spotify/api.py b/plistsync/services/spotify/api.py index 0ef5d82..61ee2b3 100644 --- a/plistsync/services/spotify/api.py +++ b/plistsync/services/spotify/api.py @@ -187,7 +187,7 @@ def _load_tracks( """Resolve the track pagination.""" all_items: list[SpotifyApiPlaylistTrack] = data.get("items", []) # type: ignore [assignment] - next_page = data.get("next") + next_page = data.get("next", data["href"]) if force or len(all_items) == 0: all_items = [] next_page = data["href"] diff --git a/uv.lock b/uv.lock index fc60a23..96bd130 100644 --- a/uv.lock +++ b/uv.lock @@ -1801,7 +1801,7 @@ wheels = [ [[package]] name = "plistsync" -version = "0.5.1" +version = "0.6.0" source = { editable = "." } dependencies = [ { name = "eyconf", extra = ["cli"] },