diff --git a/pyproject.toml b/pyproject.toml index 93c8559..fb386de 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -49,7 +49,7 @@ dependencies = [ "python-dateutil", "PyYAML", "requests", - "spotify-tools>=0.0.1", + "spotify-tools>=0.0.2", "tqdm", "youtube-dl" ] diff --git a/src/djtools/spotify/helpers.py b/src/djtools/spotify/helpers.py index 338795a..af7f61f 100644 --- a/src/djtools/spotify/helpers.py +++ b/src/djtools/spotify/helpers.py @@ -1,6 +1,6 @@ """This module contains helper functions used by the spotify module. -Most of the heavy lifting is now delegated to the spotify-tools library. +All Spotify API interactions are delegated to the spotify-tools library. This module provides DJ-Tools specific wrappers and configuration handling. """ @@ -21,16 +21,17 @@ import yaml from spotify_tools import ( Client, + PlaylistTrack, SpotifyConfig, - filter_tracks_by_similarity, - is_duplicate_track, + create_playlist, + get_playlist, + resolve_track_from_url, search_track_fuzzy, + update_playlist, ) logger = logging.getLogger(__name__) -# Threshold for fuzzy string matching (0-100 scale) -FUZZY_MATCH_THRESHOLD = 90 BaseConfig = Type["BaseConfig"] DJToolsSpotifyConfig = Type["SpotifyConfig"] SubredditConfig = Type["SubredditConfig"] @@ -103,71 +104,13 @@ def get_spotify_client( ) -def filter_results( - spotify: Client, - results: Dict, - threshold: float, - title: str, - artist: str, -) -> Tuple[Dict[str, Any], float]: - """Filter Spotify search results to find best matching track. - - Args: - spotify: Spotify client. - results: Spotify search results. - threshold: Minimum Levenshtein distance. - title: Potential title of a track. - artist: Potential artist of a track. - - Returns: - Tuple of track object (as dict) and similarity score. - """ - # Convert spotify-tools Track objects to dicts for backward compatibility - tracks = results.get("tracks", {}).get("items", []) - - # Use spotify-tools filtering - from spotify_tools.schemas import Track - - track_objects = [Track.model_validate(t) for t in tracks if t] - - matches = filter_tracks_by_similarity( - track_objects, title, artist, threshold - ) - - # Paginate through remaining results - while results.get("tracks", {}).get("next"): - try: - results = spotify.next(results["tracks"]) - if not results: - break - tracks = results.get("tracks", {}).get("items", []) - if not tracks: - tracks = results.get("items", []) - track_objects = [Track.model_validate(t) for t in tracks if t] - matches.extend( - filter_tracks_by_similarity( - track_objects, title, artist, threshold - ) - ) - except Exception: - logger.warning(f"Failed to get next tracks for {title, artist}") - break - - if not matches: - return {}, 0.0 - - best = max(matches, key=lambda m: m.score) - # Convert back to dict format for backward compatibility - return best.track.model_dump(), best.score - - async def get_subreddit_posts( spotify: Client, reddit: praw.Reddit, subreddit: SubredditConfig, config: BaseConfig, praw_cache: Dict[str, bool], -) -> Tuple[List[Tuple[str]], Dict[str, Union[str, int]]]: +) -> Tuple[List[Tuple[str, str]], Dict[str, Union[str, int]]]: """Filters subreddit submissions and resolves them to Spotify tracks. Args: @@ -209,7 +152,7 @@ async def get_subreddit_posts( submissions.append(submission) praw_cache[submission.id] = True - new_tracks = [] + new_tracks: List[Tuple[str, str]] = [] if submissions: msg = ( f"Searching Spotify for {len(submissions)} new submission(s) from " @@ -229,10 +172,11 @@ async def get_subreddit_posts( with tqdm(total=len(futures), desc=msg) as pbar: new_tracks = [] for future in as_completed(futures): - new_tracks.append(future.result()) + result = future.result() + if result: + new_tracks.append(result) pbar.update(1) - new_tracks = [track for track in new_tracks if track] logger.info( f"Got {len(new_tracks)} Spotify track(s) from new " f'"r/{subreddit.name}" posts' @@ -248,9 +192,9 @@ def populate_playlist( playlist_ids: Dict[str, str], spotify_username: str, spotify: Client, - tracks: List[Tuple[str]], + tracks: List[Tuple[str, str]], playlist_limit: Optional[int] = None, - verbosity: Optional[int] = 0, + verbosity: int = 0, ) -> Dict[str, str]: """Inserts tracks into either a new playlist or an existing one. @@ -267,31 +211,47 @@ def populate_playlist( Updated playlist IDs. """ playlist_id = playlist_ids.get(playlist_name) - playlist = None - if playlist_id and tracks: - playlist = _update_existing_playlist( + # Convert track tuples to PlaylistTrack objects, resolving URLs if needed + playlist_tracks = _resolve_tracks(spotify, tracks) + + if playlist_id and playlist_tracks: + # Update existing playlist using spotify-tools + result = update_playlist( spotify, playlist_id, - tracks, - playlist_limit, - verbosity, + playlist_tracks, + max_size=playlist_limit, + check_duplicates=True, + duplicate_threshold=90.0, + verbosity=verbosity, ) - elif tracks: + _log_update_result(result, verbosity) + playlist = get_playlist(spotify, playlist_id) + elif playlist_tracks: + # Create new playlist using spotify-tools logger.warning( f"Unable to get ID for {playlist_name}...creating a new playlist" ) - playlist = _build_new_playlist( - spotify, spotify_username, playlist_name, tracks + playlist = create_playlist( + spotify, + name=playlist_name.title(), + tracks=playlist_tracks, + public=True, + user_id=spotify_username, ) - playlist_ids[playlist_name] = playlist["id"] + if playlist and playlist.id: + playlist_ids[playlist_name] = playlist.id elif playlist_id: - playlist = spotify.playlist(playlist_id) + playlist = get_playlist(spotify, playlist_id) + else: + playlist = None if playlist: - logger.info( - f'"{playlist["name"]}": {playlist["external_urls"].get("spotify")}' - ) + url = "" + if playlist.external_urls: + url = playlist.external_urls.spotify or "" + logger.info(f'"{playlist.name}": {url}') return playlist_ids @@ -310,30 +270,36 @@ def write_playlist_ids(playlist_ids: Dict[str, str]): yaml.dump(playlist_ids, _file) -def _build_new_playlist( +def filter_results( spotify: Client, - username: str, - playlist_name: str, - new_tracks: List[Tuple[str]], -) -> Dict[str, Any]: - """Creates a new playlist from a list of track IDs / URLs. + results: Dict, + threshold: float, + title: str, + artist: str, +) -> Tuple[Dict[str, Any], float]: + """Filter Spotify search results to find best matching track. + + This function uses spotify-tools' search_track_fuzzy for matching. Args: spotify: Spotify client. - username: Spotify username. - playlist_name: Name for the new playlist. - new_tracks: List of (track_id, track_name) tuples. + results: Spotify search results (unused, kept for API compatibility). + threshold: Minimum Levenshtein distance. + title: Potential title of a track. + artist: Potential artist of a track. Returns: - Playlist object for the newly constructed playlist. + Tuple of track object (as dict) and similarity score. """ - ids = [t[0] for t in new_tracks] - playlist = spotify.user_playlist_create( - username, name=playlist_name.title() + # Use spotify-tools search with pagination for thorough matching + result = search_track_fuzzy( + spotify, title, artist, threshold=threshold, limit=50 ) - spotify.playlist_add_items(playlist["id"], ids) - return playlist + if result and result.track: + return result.track.model_dump(), result.score + + return {}, 0.0 async def _catch( @@ -358,66 +324,44 @@ async def _catch( continue -def _filter_tracks( - tracks: List[Dict], threshold: float, title: str, artist: str -) -> List[Tuple[Dict[str, Any], float]]: - """Filter tracks by Levenshtein distance on artist and name fields. +def _log_update_result(result, verbosity: int): + """Log the result of a playlist update operation. Args: - tracks: Spotify search results. - threshold: Minimum Levenshtein distance. - title: Potential title of a track. - artist: Potential artist of a track. - - Returns: - List of (track_dict, similarity_score) tuples. + result: UpdateResult from spotify-tools. + verbosity: Logging verbosity level. """ - from spotify_tools.schemas import Track - - track_objects = [Track.model_validate(t) for t in tracks if t] - matches = filter_tracks_by_similarity( - track_objects, title, artist, threshold - ) - return [(m.track.model_dump(), m.score) for m in matches] - - -def _fuzzy_match( - spotify: Client, title: str, threshold: float -) -> Optional[Tuple[str, str]]: - """Attempts to find a Spotify track matching the title. + if result.tracks_added: + logger.info(f"{len(result.tracks_added)} new tracks added") + if verbosity > 0: # pragma: no cover + for track in result.tracks_added: + logger.info(f"\t{track.display_name}") - Args: - spotify: Spotify client. - title: Submission title. - threshold: Minimum Levenshtein distance. + if result.tracks_removed: + logger.info(f"{len(result.tracks_removed)} old tracks removed") + if verbosity > 0: # pragma: no cover + for track in result.tracks_removed: + logger.info(f"\t{track.display_name}") - Returns: - Tuple of (track_id, "track_name - artists") or None. - """ - parts = _parse_title(title) - if not all(parts): - return None + if result.skipped_existing: + for track in result.skipped_existing: + logger.warning( + f'Candidate new track "{track.display_name}" is already in ' + "the playlist" + ) - # Try both orderings (title-artist and artist-title) - for track, artist in [parts, parts[::-1]]: - try: - result = search_track_fuzzy( - spotify, track, artist, threshold=threshold, limit=50 + if result.skipped_duplicates: + for track in result.skipped_duplicates: + logger.warning( + f'Candidate new track "{track.display_name}" is too similar ' + "to existing track" ) - if result and result.track: - track_obj = result.track - artists = ", ".join( - a.name for a in (track_obj.artists or []) if a.name - ) - return (track_obj.id, f"{track_obj.name} - {artists}") - except Exception as exc: - logger.error(f'Error searching for "{track} - {artist}": {exc}') - continue - return None + if not (result.tracks_added or result.tracks_removed): + logger.info("No tracks added or removed") -def _parse_title(title: str) -> List[str]: +def _parse_title(title: str) -> List[Optional[str]]: """Attempts to split submission title into (track name, artist(s)). Args: @@ -462,152 +406,60 @@ def _process( if "spotify.com/track/" in submission.url: return (submission.url, submission.title) - return _fuzzy_match(spotify, submission.title, threshold) - - -def _track_name_too_similar( - track: str, - playlist_track_names: set, -) -> bool: - """Check if a track is too similar to existing tracks. - - Args: - track: Track name to check. - playlist_track_names: Set of existing track names. + # Parse title into track/artist + parts = _parse_title(submission.title) + if not all(parts): + return None - Returns: - True if the track is a duplicate. - """ - if is_duplicate_track(track, playlist_track_names, threshold=90.0): - for other in playlist_track_names: - from fuzzywuzzy import fuzz - - if ( - fuzz.ratio(track.lower(), other.lower()) - > FUZZY_MATCH_THRESHOLD - ): - logger.warning( - f'Candidate new track "{track}" is too similar to ' - f'existing track "{other}"' + # Try both orderings (title-artist and artist-title) + for track, artist in [parts, parts[::-1]]: + try: + result = search_track_fuzzy( + spotify, track, artist, threshold=threshold, limit=50 + ) + if result and result.track: + track_obj = result.track + artists = ", ".join( + a.name for a in (track_obj.artists or []) if a.name ) - return True - return False + return (track_obj.id, f"{track_obj.name} - {artists}") + except Exception as exc: + logger.error(f'Error searching for "{track} - {artist}": {exc}') + continue + + return None -def _update_existing_playlist( +def _resolve_tracks( spotify: Client, - playlist_id: str, - new_tracks: List[Tuple[str, str]], - limit: Optional[int], - verbosity: int, -) -> Dict[str, Any]: - """Updates an existing playlist with new tracks (LIFO queue behavior). + tracks: List[Tuple[str, str]], +) -> List[PlaylistTrack]: + """Convert track tuples to PlaylistTrack objects. + + Resolves Spotify URLs to track IDs as needed. Args: spotify: Spotify client. - playlist_id: Spotify playlist ID. - new_tracks: List of (track_id, track_name) tuples. - limit: Maximum number of tracks in playlist. - verbosity: Verbosity level. + tracks: List of (track_id_or_url, track_name) tuples. Returns: - Playlist object. + List of PlaylistTrack objects. """ - import sys - - if limit is None: - limit = sys.maxsize # pragma: no cover - - # Get current playlist and tracks - playlist_object = spotify.playlist(playlist_id) - _playlist = playlist_object - tracks = _playlist["tracks"]["items"] - - # Paginate to get all tracks - try: - while _playlist["tracks"]["next"]: - _playlist = spotify.next(_playlist["tracks"]) - try: - tracks.extend(_playlist["tracks"]["items"]) - except KeyError: - tracks.extend(_playlist["items"]) - except KeyError: - tracks.extend(_playlist.get("items", [])) - except Exception as exc: - logger.error(f"Failed to get tracks from playlist: {exc}") - - # Build sets of existing track IDs and names - track_count = len(tracks) - track_index = 0 - add_payload = [] - tracks_added = [] - remove_payload = [] - tracks_removed = [] - ids = set() - playlist_track_names = set() - - for track in tracks: - track_data = track["track"] - ids.add(track_data["id"]) - artists = ", ".join([x["name"] for x in track_data["artists"]]) - playlist_track_names.add(f"{track_data['name']} - {artists}") - - # Process new tracks - for id_, track_name in new_tracks: - # Resolve URL to track ID if needed - track_id = id_ - display_name = track_name - if "spotify.com/track/" in id_: - resp = spotify.track(id_) - track_id = resp["id"] - artists = ", ".join([x["name"] for x in resp["artists"]]) - display_name = f"{resp['name']} - {artists}" - - if track_id in ids: - logger.warning( - f'Candidate new track "{display_name}" is already in the playlist' - ) - continue - - if _track_name_too_similar(display_name, playlist_track_names): - continue # pragma: no cover - - tracks_added.append(display_name) - add_payload.append(track_id) - - # Remove oldest track if we'd exceed limit - if track_count + len(tracks_added) > limit: - _track = tracks.pop(0)["track"] - artists = ", ".join([x["name"] for x in _track["artists"]]) - tracks_removed.append(f"{_track['name']} - {artists}") - remove_payload.append( - {"uri": _track["uri"], "positions": [track_index]} + playlist_tracks = [] + for id_or_url, name in tracks: + if "spotify.com/track/" in id_or_url: + # Resolve URL to PlaylistTrack + pt = resolve_track_from_url(spotify, id_or_url) + if pt: + playlist_tracks.append(pt) + else: + # Create PlaylistTrack from ID and name + playlist_tracks.append( + PlaylistTrack( + id=id_or_url, + uri=f"spotify:track:{id_or_url}", + name=name.split(" - ")[0] if " - " in name else name, + artists=name.split(" - ")[1] if " - " in name else "", + ) ) - track_index += 1 - track_count -= 1 - - # Log what we're doing - if tracks_added: - logger.info(f"{len(tracks_added)} new tracks added") - if verbosity > 0: - for track in tracks_added: - logger.info(f"\t{track}") - - if tracks_removed: - logger.info(f"{len(tracks_removed)} old tracks removed") - if verbosity > 0: # pragma: no cover - for track in tracks_removed: - logger.info(f"\t{track}") - - if not (tracks_added or tracks_removed): - logger.info("No tracks added or removed") - - # Execute the changes - if remove_payload: - spotify.playlist_remove_specific_occurrences_of_items( - playlist_id, remove_payload - ) - if add_payload: - spotify.playlist_add_items(playlist_id, add_payload) - - return playlist_object + return playlist_tracks diff --git a/tests/spotify/test_helpers.py b/tests/spotify/test_helpers.py index 5338fc9..55fd262 100644 --- a/tests/spotify/test_helpers.py +++ b/tests/spotify/test_helpers.py @@ -12,14 +12,11 @@ import yaml from djtools.spotify.helpers import ( - _build_new_playlist, _catch, - _filter_tracks, - _fuzzy_match, + _log_update_result, _parse_title, _process, - _track_name_too_similar, - _update_existing_playlist, + _resolve_tracks, filter_results, get_playlist_ids, get_reddit_client, @@ -39,18 +36,6 @@ async def _aiter(obj, num_subs): await asyncio.sleep(0.1) -@mock.patch("djtools.spotify.helpers.Client") -def test_build_new_playlist(mock_client): - """Test for the _build_new_playlist function.""" - mock_client.user_playlist_create.return_value = {"id": "test_id"} - new_tracks = [("test_id", "track title - artist name")] - ret = _build_new_playlist(mock_client, "test_user", "r/techno", new_tracks) - assert isinstance(ret, dict) - assert ret == {"id": "test_id"} - mock_client.user_playlist_create.assert_called_once() - mock_client.playlist_add_items.assert_called_once() - - @pytest.mark.asyncio @pytest.mark.parametrize("message", ["", "oops"]) async def test_catch(message, caplog): @@ -159,45 +144,58 @@ def test_parse_title(title): @pytest.mark.parametrize("playlist_ids", [{}, {"playlist": "id"}]) @pytest.mark.parametrize("tracks", [[], [("id", "title - artist")]]) +@mock.patch("djtools.spotify.helpers.update_playlist") +@mock.patch("djtools.spotify.helpers.create_playlist") +@mock.patch("djtools.spotify.helpers.get_playlist") @mock.patch("djtools.spotify.helpers.Client") def test_populate_playlist( mock_client, + mock_get_playlist, + mock_create_playlist, + mock_update_playlist, playlist_ids, tracks, caplog, ): """Test for the populate_playlist function.""" - ret_val = { - "name": "playlist", - "external_urls": {"spotify": "https://test-url.com"}, - "id": "test-id", - } - mock_client.playlist.return_value = ret_val - mock_client.user_playlist_create.return_value = ret_val + # Mock playlist response + mock_playlist = mock.MagicMock() + mock_playlist.id = "test-id" + mock_playlist.name = "playlist" + mock_playlist.external_urls = mock.MagicMock() + mock_playlist.external_urls.spotify = "https://test-url.com" + + mock_get_playlist.return_value = mock_playlist + mock_create_playlist.return_value = mock_playlist + + # Mock update result + mock_result = mock.MagicMock() + mock_result.tracks_added = [] + mock_result.tracks_removed = [] + mock_result.skipped_existing = [] + mock_result.skipped_duplicates = [] + mock_update_playlist.return_value = mock_result + caplog.set_level("INFO") - with mock.patch( - "djtools.spotify.helpers._update_existing_playlist", - return_value=ret_val, - ) as mock_update: - populate_playlist( - playlist_name="playlist", - playlist_ids=dict(playlist_ids), - spotify_username="test", - spotify=mock_client, - tracks=tracks, - playlist_limit=50, - ) + result = populate_playlist( + playlist_name="playlist", + playlist_ids=dict(playlist_ids), + spotify_username="test", + spotify=mock_client, + tracks=tracks, + playlist_limit=50, + ) if playlist_ids and tracks: - assert mock_update.call_count == 1 + mock_update_playlist.assert_called_once() elif tracks: assert "Unable to get ID for playlist" in caplog.text + mock_create_playlist.assert_called_once() elif playlist_ids: - mock_client.playlist.assert_called_once() + mock_get_playlist.assert_called() - if not (playlist_ids or tracks): - assert mock_update.call_count == 0 + assert isinstance(result, dict) @pytest.mark.parametrize( @@ -207,40 +205,34 @@ def test_populate_playlist( "https://some-other-url.com/some_id", ], ) +@mock.patch("djtools.spotify.helpers.search_track_fuzzy") @mock.patch("djtools.spotify.helpers.Client") @mock.patch("djtools.spotify.helpers.praw.models.Submission") -def test_process(mock_submission, mock_client, url): +def test_process(mock_submission, mock_client, mock_search, url): """Test for the _process function.""" title = "Arctic Oscillations - Fanu" mock_submission.url = url mock_submission.title = title - with mock.patch( - "djtools.spotify.helpers._fuzzy_match", return_value=(url, title) - ): - ret = _process(mock_submission, mock_client, 50) + # Mock search result + mock_track = mock.MagicMock() + mock_track.id = "track123" + mock_track.name = "Arctic Oscillations" + mock_artist = mock.MagicMock() + mock_artist.name = "Fanu" + mock_track.artists = [mock_artist] - assert isinstance(ret, tuple) - assert ret == (url, title) + mock_result = mock.MagicMock() + mock_result.track = mock_track + mock_search.return_value = mock_result + ret = _process(mock_submission, mock_client, 50) -@pytest.mark.parametrize( - "playlist_track_names", - [ - {"Arctic Oscillations - Fanu"}, - {"Not a Match - Some Artist"}, - ], -) -def test_track_name_too_similar(playlist_track_names, caplog): - """Test for the _track_name_too_similar function.""" - caplog.set_level("WARNING") - track = "Arctic Oscillations - Fanu" - ret = _track_name_too_similar(track, playlist_track_names) - if any("Not a Match" in x for x in playlist_track_names): - assert not ret + assert isinstance(ret, tuple) + if "spotify.com/track/" in url: + assert ret == (url, title) else: - assert ret - assert "too similar" in caplog.text + assert ret[0] == "track123" def test_write_playlist_ids(): @@ -266,47 +258,39 @@ def test_write_playlist_ids(): ids_path.unlink() +@mock.patch("djtools.spotify.helpers.search_track_fuzzy") @mock.patch("djtools.spotify.helpers.Client") -def test_filter_results_with_matches(mock_client): +def test_filter_results_with_matches(mock_client, mock_search): """Test filter_results returns best match.""" - results = { - "tracks": { - "items": [ - { - "id": "track1", - "name": "Test Song", - "artists": [{"name": "Test Artist"}], - "uri": "spotify:track:track1", - } - ], - "next": None, - } + mock_track = mock.MagicMock() + mock_track.model_dump.return_value = { + "id": "track1", + "name": "Test Song", + "artists": [{"name": "Test Artist"}], } + mock_result = mock.MagicMock() + mock_result.track = mock_track + mock_result.score = 180.0 + mock_search.return_value = mock_result + + results = {"tracks": {"items": []}} # unused but kept for API compat + track, score = filter_results( mock_client, results, 70.0, "Test Song", "Test Artist" ) assert track.get("id") == "track1" - assert score > 0 + assert score == 180.0 # noqa: PLR2004 +@mock.patch("djtools.spotify.helpers.search_track_fuzzy") @mock.patch("djtools.spotify.helpers.Client") -def test_filter_results_no_matches(mock_client): +def test_filter_results_no_matches(mock_client, mock_search): """Test filter_results returns empty when no matches.""" - results = { - "tracks": { - "items": [ - { - "id": "track1", - "name": "Completely Different", - "artists": [{"name": "Unknown"}], - "uri": "spotify:track:track1", - } - ], - "next": None, - } - } + mock_search.return_value = None + + results = {"tracks": {"items": []}} track, score = filter_results( mock_client, results, 90.0, "Test Song", "Test Artist" @@ -316,514 +300,170 @@ def test_filter_results_no_matches(mock_client): assert score == 0.0 -@mock.patch("djtools.spotify.helpers.Client") -def test_filter_results_with_pagination(mock_client): - """Test filter_results handles pagination.""" - results = { - "tracks": { - "items": [ - { - "id": "track1", - "name": "First Track", - "artists": [{"name": "Artist"}], - "uri": "spotify:track:track1", - } - ], - "next": "https://api.spotify.com/next", - } - } - mock_client.next.return_value = { - "tracks": { - "items": [ - { - "id": "track2", - "name": "Test Song", - "artists": [{"name": "Test Artist"}], - "uri": "spotify:track:track2", - } - ], - "next": None, - } - } - - track, score = filter_results( - mock_client, results, 70.0, "Test Song", "Test Artist" - ) - - assert track.get("id") == "track2" - assert score > 0 - mock_client.next.assert_called() - - -@mock.patch("djtools.spotify.helpers.Client") -def test_filter_results_pagination_exception(mock_client, caplog): - """Test filter_results handles pagination errors.""" - caplog.set_level("WARNING") - results = { - "tracks": { - "items": [ - { - "id": "track1", - "name": "Test Song", - "artists": [{"name": "Test Artist"}], - "uri": "spotify:track:track1", - } - ], - "next": "https://api.spotify.com/next", - } - } - mock_client.next.side_effect = Exception("API Error") - - track, _score = filter_results( - mock_client, results, 70.0, "Test Song", "Test Artist" - ) - - assert track.get("id") == "track1" - assert "Failed to get next tracks" in caplog.text - - -@mock.patch("djtools.spotify.helpers.Client") -def test_filter_results_pagination_empty_response(mock_client): - """Test filter_results handles empty pagination response.""" - results = { - "tracks": { - "items": [ - { - "id": "track1", - "name": "Test Song", - "artists": [{"name": "Test Artist"}], - "uri": "spotify:track:track1", - } - ], - "next": "https://api.spotify.com/next", - } - } - mock_client.next.return_value = None - - track, _ = filter_results( - mock_client, results, 70.0, "Test Song", "Test Artist" - ) - - assert track.get("id") == "track1" - - -@mock.patch("djtools.spotify.helpers.Client") -def test_filter_results_pagination_items_key(mock_client): - """Test filter_results handles 'items' key in pagination response.""" - results = { - "tracks": { - "items": [ - { - "id": "track1", - "name": "First", - "artists": [{"name": "Artist"}], - "uri": "spotify:track:track1", - } - ], - "next": "https://api.spotify.com/next", - } - } - # Simulating response where items is at top level instead of tracks.items - mock_client.next.return_value = { - "items": [ - { - "id": "track2", - "name": "Test Song", - "artists": [{"name": "Test Artist"}], - "uri": "spotify:track:track2", - } - ], - "next": None, - } - - _, score = filter_results( - mock_client, results, 70.0, "Test Song", "Test Artist" - ) - - assert score > 0 - - -def test_filter_tracks(): - """Test _filter_tracks function.""" - tracks = [ - { - "id": "track1", - "name": "Test Song", - "artists": [{"name": "Test Artist"}], - "uri": "spotify:track:track1", - }, - { - "id": "track2", - "name": "Another Track", - "artists": [{"name": "Other Artist"}], - "uri": "spotify:track:track2", - }, - ] - - results = _filter_tracks(tracks, 70.0, "Test Song", "Test Artist") - assert len(results) == 1 - assert results[0][0]["id"] == "track1" - assert results[0][1] > 0 - - -def test_filter_tracks_no_matches(): - """Test _filter_tracks returns empty when no matches.""" - tracks = [ - { - "id": "track1", - "name": "Completely Different", - "artists": [{"name": "Unknown"}], - "uri": "spotify:track:track1", - } - ] - - results = _filter_tracks(tracks, 90.0, "Test Song", "Test Artist") - assert results == [] - +def test_log_update_result_added_tracks(caplog): + """Test _log_update_result logs added tracks.""" + caplog.set_level("INFO") -@mock.patch("djtools.spotify.helpers.search_track_fuzzy") -@mock.patch("djtools.spotify.helpers.Client") -def test_fuzzy_match_success(mock_client, mock_search): - """Test _fuzzy_match returns track when found.""" mock_track = mock.MagicMock() - mock_track.id = "track123" - mock_track.name = "Test Song" - mock_artist = mock.MagicMock() - mock_artist.name = "Test Artist" - mock_track.artists = [mock_artist] + mock_track.display_name = "New Track - Artist" - mock_result = mock.MagicMock() - mock_result.track = mock_track - mock_search.return_value = mock_result + result = mock.MagicMock() + result.tracks_added = [mock_track] + result.tracks_removed = [] + result.skipped_existing = [] + result.skipped_duplicates = [] - result = _fuzzy_match(mock_client, "Test Song - Test Artist", 70.0) + _log_update_result(result, 0) - assert result is not None - assert result[0] == "track123" - assert "Test Song" in result[1] - - -@mock.patch("djtools.spotify.helpers.search_track_fuzzy") -@mock.patch("djtools.spotify.helpers.Client") -def test_fuzzy_match_no_match(mock_client, mock_search): - """Test _fuzzy_match returns None when no match found.""" - mock_search.return_value = None + assert "1 new tracks added" in caplog.text - result = _fuzzy_match(mock_client, "Test Song - Test Artist", 70.0) - assert result is None +def test_log_update_result_removed_tracks(caplog): + """Test _log_update_result logs removed tracks.""" + caplog.set_level("INFO") + mock_track = mock.MagicMock() + mock_track.display_name = "Old Track - Artist" -@mock.patch("djtools.spotify.helpers.Client") -def test_fuzzy_match_unparseable_title(mock_client): - """Test _fuzzy_match returns None for unparseable titles.""" - result = _fuzzy_match(mock_client, "No delimiter here", 70.0) - assert result is None + result = mock.MagicMock() + result.tracks_added = [] + result.tracks_removed = [mock_track] + result.skipped_existing = [] + result.skipped_duplicates = [] + _log_update_result(result, 0) -@mock.patch("djtools.spotify.helpers.search_track_fuzzy") -@mock.patch("djtools.spotify.helpers.Client") -def test_fuzzy_match_exception(mock_client, mock_search, caplog): - """Test _fuzzy_match handles exceptions gracefully.""" - caplog.set_level("ERROR") - mock_search.side_effect = Exception("API Error") - - result = _fuzzy_match(mock_client, "Test Song - Test Artist", 70.0) + assert "1 old tracks removed" in caplog.text - assert result is None - assert "Error searching" in caplog.text +def test_log_update_result_skipped_existing(caplog): + """Test _log_update_result logs skipped existing tracks.""" + caplog.set_level("WARNING") -@mock.patch("djtools.spotify.helpers.Client") -def test_update_existing_playlist_add_tracks(mock_client, caplog): - """Test _update_existing_playlist adds new tracks.""" - caplog.set_level("INFO") - playlist_data = { - "id": "playlist123", - "name": "Test Playlist", - "tracks": { - "items": [ - { - "track": { - "id": "existing1", - "name": "Existing Track", - "artists": [{"name": "Artist"}], - "uri": "spotify:track:existing1", - } - } - ], - "next": None, - }, - } - mock_client.playlist.return_value = playlist_data + mock_track = mock.MagicMock() + mock_track.display_name = "Existing Track - Artist" - new_tracks = [("new_track_id", "New Track - New Artist")] + result = mock.MagicMock() + result.tracks_added = [] + result.tracks_removed = [] + result.skipped_existing = [mock_track] + result.skipped_duplicates = [] - result = _update_existing_playlist( - mock_client, "playlist123", new_tracks, 50, 0 - ) + _log_update_result(result, 0) - assert result["id"] == "playlist123" - mock_client.playlist_add_items.assert_called_once() - assert "1 new tracks added" in caplog.text + assert "already in the playlist" in caplog.text -@mock.patch("djtools.spotify.helpers.Client") -def test_update_existing_playlist_duplicate_id(mock_client, caplog): - """Test _update_existing_playlist skips duplicate track IDs.""" +def test_log_update_result_skipped_duplicates(caplog): + """Test _log_update_result logs skipped duplicate tracks.""" caplog.set_level("WARNING") - playlist_data = { - "id": "playlist123", - "name": "Test Playlist", - "tracks": { - "items": [ - { - "track": { - "id": "existing1", - "name": "Existing Track", - "artists": [{"name": "Artist"}], - "uri": "spotify:track:existing1", - } - } - ], - "next": None, - }, - } - mock_client.playlist.return_value = playlist_data - - new_tracks = [("existing1", "Existing Track - Artist")] - - _update_existing_playlist(mock_client, "playlist123", new_tracks, 50, 0) - - mock_client.playlist_add_items.assert_not_called() - assert "already in the playlist" in caplog.text + mock_track = mock.MagicMock() + mock_track.display_name = "Similar Track - Artist" -@mock.patch("djtools.spotify.helpers.Client") -def test_update_existing_playlist_resolves_url(mock_client, caplog): - """Test _update_existing_playlist resolves Spotify URLs to track IDs.""" - caplog.set_level("INFO") - playlist_data = { - "id": "playlist123", - "name": "Test Playlist", - "tracks": { - "items": [], - "next": None, - }, - } - mock_client.playlist.return_value = playlist_data - mock_client.track.return_value = { - "id": "resolved_id", - "name": "Resolved Track", - "artists": [{"name": "Artist"}], - } + result = mock.MagicMock() + result.tracks_added = [] + result.tracks_removed = [] + result.skipped_existing = [] + result.skipped_duplicates = [mock_track] - new_tracks = [ - ( - "https://open.spotify.com/track/abc123", - "Some Track - Some Artist", - ) - ] + _log_update_result(result, 0) - _update_existing_playlist(mock_client, "playlist123", new_tracks, 50, 0) + assert "too similar to existing" in caplog.text - mock_client.track.assert_called_once() - mock_client.playlist_add_items.assert_called_once() - -@mock.patch("djtools.spotify.helpers.Client") -def test_update_existing_playlist_removes_old_tracks(mock_client, caplog): - """Test _update_existing_playlist removes old tracks when limit exceeded.""" +def test_log_update_result_no_changes(caplog): + """Test _log_update_result logs when no changes.""" caplog.set_level("INFO") - playlist_data = { - "id": "playlist123", - "name": "Test Playlist", - "tracks": { - "items": [ - { - "track": { - "id": "old1", - "name": "Old Track", - "artists": [{"name": "Artist"}], - "uri": "spotify:track:old1", - } - } - ], - "next": None, - }, - } - mock_client.playlist.return_value = playlist_data - new_tracks = [("new1", "New Track - Artist")] + result = mock.MagicMock() + result.tracks_added = [] + result.tracks_removed = [] + result.skipped_existing = [] + result.skipped_duplicates = [] - _update_existing_playlist(mock_client, "playlist123", new_tracks, 1, 0) + _log_update_result(result, 0) - mock_client.playlist_remove_specific_occurrences_of_items.assert_called() - mock_client.playlist_add_items.assert_called_once() - assert "1 old tracks removed" in caplog.text + assert "No tracks added or removed" in caplog.text +@mock.patch("djtools.spotify.helpers.resolve_track_from_url") @mock.patch("djtools.spotify.helpers.Client") -def test_update_existing_playlist_verbose_logging(mock_client, caplog): - """Test _update_existing_playlist logs track details when verbose.""" - caplog.set_level("INFO") - playlist_data = { - "id": "playlist123", - "name": "Test Playlist", - "tracks": { - "items": [], - "next": None, - }, - } - mock_client.playlist.return_value = playlist_data +def test_resolve_tracks_with_urls(mock_client, mock_resolve): + """Test _resolve_tracks resolves Spotify URLs.""" + mock_pt = mock.MagicMock() + mock_pt.id = "resolved_id" + mock_resolve.return_value = mock_pt - new_tracks = [("new1", "New Track - Artist")] + tracks = [ + ("https://open.spotify.com/track/abc123", "Track - Artist"), + ] - _update_existing_playlist(mock_client, "playlist123", new_tracks, 50, 1) + result = _resolve_tracks(mock_client, tracks) - assert "New Track - Artist" in caplog.text + assert len(result) == 1 + mock_resolve.assert_called_once() @mock.patch("djtools.spotify.helpers.Client") -def test_update_existing_playlist_no_changes(mock_client, caplog): - """Test _update_existing_playlist logs when no changes made.""" - caplog.set_level("INFO") - playlist_data = { - "id": "playlist123", - "name": "Test Playlist", - "tracks": { - "items": [ - { - "track": { - "id": "existing1", - "name": "Existing Track", - "artists": [{"name": "Artist"}], - "uri": "spotify:track:existing1", - } - } - ], - "next": None, - }, - } - mock_client.playlist.return_value = playlist_data - - new_tracks = [("existing1", "Existing Track - Artist")] +def test_resolve_tracks_with_ids(mock_client): + """Test _resolve_tracks creates PlaylistTrack from IDs.""" + tracks = [ + ("track_id_123", "Track Name - Artist Name"), + ] - _update_existing_playlist(mock_client, "playlist123", new_tracks, 50, 0) + result = _resolve_tracks(mock_client, tracks) - assert "No tracks added or removed" in caplog.text + assert len(result) == 1 + assert result[0].id == "track_id_123" + assert result[0].name == "Track Name" + assert result[0].artists == "Artist Name" +@mock.patch("djtools.spotify.helpers.search_track_fuzzy") @mock.patch("djtools.spotify.helpers.Client") -def test_update_existing_playlist_pagination(mock_client): - """Test _update_existing_playlist handles pagination.""" - page1 = { - "id": "playlist123", - "name": "Test Playlist", - "tracks": { - "items": [ - { - "track": { - "id": "track1", - "name": "Track 1", - "artists": [{"name": "Artist"}], - "uri": "spotify:track:track1", - } - } - ], - "next": "https://api.spotify.com/next", - }, - } - page2 = { - "tracks": { - "items": [ - { - "track": { - "id": "track2", - "name": "Track 2", - "artists": [{"name": "Artist"}], - "uri": "spotify:track:track2", - } - } - ], - "next": None, - } - } - mock_client.playlist.return_value = page1 - mock_client.next.return_value = page2 - - new_tracks = [("new_track", "New Track - Artist")] +@mock.patch("djtools.spotify.helpers.praw.models.Submission") +def test_process_unparseable_title(mock_submission, mock_client, mock_search): + """Test _process returns None for unparseable titles.""" + mock_submission.url = "https://some-url.com" + mock_submission.title = "No delimiter here" - _update_existing_playlist(mock_client, "playlist123", new_tracks, 50, 0) + ret = _process(mock_submission, mock_client, 50) - mock_client.next.assert_called() - mock_client.playlist_add_items.assert_called_once() + assert ret is None + mock_search.assert_not_called() +@mock.patch("djtools.spotify.helpers.search_track_fuzzy") @mock.patch("djtools.spotify.helpers.Client") -def test_update_existing_playlist_pagination_items_key(mock_client): - """Test _update_existing_playlist handles pagination with items key.""" - page1 = { - "id": "playlist123", - "name": "Test Playlist", - "tracks": { - "items": [ - { - "track": { - "id": "track1", - "name": "Track 1", - "artists": [{"name": "Artist"}], - "uri": "spotify:track:track1", - } - } - ], - "next": "https://api.spotify.com/next", - }, - } - # Response with items at top level instead of tracks.items - page2 = { - "items": [ - { - "track": { - "id": "track2", - "name": "Track 2", - "artists": [{"name": "Artist"}], - "uri": "spotify:track:track2", - } - } - ], - "next": None, - } - mock_client.playlist.return_value = page1 - mock_client.next.return_value = page2 - - new_tracks = [("new_track", "New Track - Artist")] +@mock.patch("djtools.spotify.helpers.praw.models.Submission") +def test_process_search_exception( + mock_submission, mock_client, mock_search, caplog +): + """Test _process handles search exceptions gracefully.""" + caplog.set_level("ERROR") + mock_submission.url = "https://some-url.com" + mock_submission.title = "Track Name - Artist" + mock_search.side_effect = Exception("API Error") - _update_existing_playlist(mock_client, "playlist123", new_tracks, 50, 0) + ret = _process(mock_submission, mock_client, 50) - mock_client.next.assert_called() + assert ret is None + assert "Error searching" in caplog.text +@mock.patch("djtools.spotify.helpers.search_track_fuzzy") @mock.patch("djtools.spotify.helpers.Client") -def test_update_existing_playlist_pagination_error(mock_client, caplog): - """Test _update_existing_playlist handles pagination errors.""" - caplog.set_level("ERROR") - page1 = { - "id": "playlist123", - "name": "Test Playlist", - "tracks": { - "items": [], - "next": "https://api.spotify.com/next", - }, - } - mock_client.playlist.return_value = page1 - mock_client.next.side_effect = Exception("API Error") - - new_tracks = [("new_track", "New Track - Artist")] +@mock.patch("djtools.spotify.helpers.praw.models.Submission") +def test_process_no_match(mock_submission, mock_client, mock_search): + """Test _process returns None when no match found.""" + mock_submission.url = "https://some-url.com" + mock_submission.title = "Track Name - Artist" + mock_search.return_value = None - _update_existing_playlist(mock_client, "playlist123", new_tracks, 50, 0) + ret = _process(mock_submission, mock_client, 50) - assert "Failed to get tracks" in caplog.text + assert ret is None class MockTqdm: @@ -841,7 +481,7 @@ def __enter__(self): def __exit__(self, *args): pass - def update(self, n=1): + def update(self, n=1): # pylint: disable=unused-argument """Mock update method.""" diff --git a/uv.lock b/uv.lock index 582c1f6..50358ae 100644 --- a/uv.lock +++ b/uv.lock @@ -672,7 +672,7 @@ wheels = [ [[package]] name = "djtools" -version = "2.9.1rc6" +version = "2.9.1rc7" source = { editable = "." } dependencies = [ { name = "asyncpraw" }, @@ -729,7 +729,7 @@ requires-dist = [ { name = "python-levenshtein", marker = "extra == 'accelerated'" }, { name = "pyyaml" }, { name = "requests" }, - { name = "spotify-tools", specifier = ">=0.0.1" }, + { name = "spotify-tools", specifier = ">=0.0.2" }, { name = "tqdm" }, { name = "youtube-dl" }, ] @@ -2540,7 +2540,7 @@ wheels = [ [[package]] name = "spotify-tools" -version = "0.0.1" +version = "0.0.2" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "fuzzywuzzy", extra = ["speedup"] }, @@ -2549,9 +2549,9 @@ dependencies = [ { name = "spotipy" }, { name = "typer" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/ae/53/1b5e3e73cd0a4adb4f95b9e4cdd344cac4f1edd53e5c0be7d7ba085d3691/spotify_tools-0.0.1.tar.gz", hash = "sha256:4b198b605d960a4d1fc911d8d2c940ad546fc8269e153e8000c349e45e60219a", size = 121112, upload-time = "2026-01-05T20:42:05.494Z" } +sdist = { url = "https://files.pythonhosted.org/packages/2f/f7/ad6fd464962b701d60c4faf122a0a9a1ec80700c99e0fc7b2f6ad82d58f6/spotify_tools-0.0.2.tar.gz", hash = "sha256:821eee249fc1b4d0dc26d7fe97cbc773b547e5fe1bbc98bcedcd26e4e1e8307b", size = 87203, upload-time = "2026-01-06T15:59:51.561Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/71/02/70c24f699a05b184b250b70b612c8c8c9c2b9a0a7f8c069f8144f938a2e1/spotify_tools-0.0.1-py3-none-any.whl", hash = "sha256:241dd69f8478f53b39b0f7f386197a04926843552f0a19d0d4c9048695334712", size = 26107, upload-time = "2026-01-05T20:42:03.893Z" }, + { url = "https://files.pythonhosted.org/packages/ed/99/d2ef3efdad1a563f95804fc048fa64b1e455fcbe46de993173b86c1208d1/spotify_tools-0.0.2-py3-none-any.whl", hash = "sha256:bed7c65f7c0b940a3ef6a1e5392d9616ecd8473e3bff01cd4c94f961ff71e7a4", size = 25490, upload-time = "2026-01-06T15:59:50.072Z" }, ] [[package]]