diff --git a/src/youtube_kit/core/ops/playlists.py b/src/youtube_kit/core/ops/playlists.py index 84f7d20..7727b7a 100644 --- a/src/youtube_kit/core/ops/playlists.py +++ b/src/youtube_kit/core/ops/playlists.py @@ -33,6 +33,7 @@ from .channels import copied_from_recent_uploads, my_channel_id, owner_or _PARTS = "snippet,status,contentDetails" +_MAX_PLAYLIST_LENGTH = 5000 # YouTube's cap on a user playlist (uploads and Watch Later can be longer) @operation(quota=1, read_only=True, idempotent=True, paginated=True) @@ -200,6 +201,9 @@ def add_video_to_playlist(playlist: str, video: str, *, position: int | None = N def remove_from_playlist(playlist: str, video: str) -> PlaylistRemoval: """Remove a video from one of your playlists, every time it appears. The video itself is not affected. + YouTube can't look up entries of a deleted video, so for a deleted or unknown video the whole playlist is read: + 1 more unit per 50 entries. + Args: playlist: Playlist ID or URL. video: Video ID or URL. @@ -207,7 +211,8 @@ def remove_from_playlist(playlist: str, video: str) -> PlaylistRemoval: playlist_id, video_id = parse.playlist_id(playlist), parse.video_id(video) youtube = client.oauth() removed = sum( - delete_if_present(youtube.playlistItems().delete(id=item["id"])) for item in _occurrences(playlist_id, video_id) + delete_if_present(youtube.playlistItems().delete(id=item["id"])) + for item in _occurrences(playlist_id, video_id, scan_if_deleted=True) ) return PlaylistRemoval(playlist_id=playlist_id, video_id=video_id, removed=removed) @@ -280,14 +285,31 @@ def set_playlist_thumbnail(playlist: str, image: Path) -> PlaylistImage: ) -def _occurrences(playlist_id: str, video_id: str) -> list[dict]: +def _occurrences(playlist_id: str, video_id: str, *, scan_if_deleted: bool = False) -> list[dict]: """The playlist items holding a video; more than one if it was added repeatedly.""" - request = ( - client.oauth() - .playlistItems() - .list(part="id", playlistId=playlist_id, videoId=video_id, maxResults=API_PAGE_SIZE) + items = client.oauth().playlistItems() + try: + return ( + items.list(part="id", playlistId=playlist_id, videoId=video_id, maxResults=API_PAGE_SIZE) + .execute() + .get("items", []) + ) + except HttpError as e: + if not scan_if_deleted or http_reason(e) != "videoNotFound": + raise + # The videoId filter fails once a video is deleted, though the playlist keeps its entry: read the whole playlist. + page = collect( + lambda **page: items.list(part="contentDetails", playlistId=playlist_id, **page), + limit=_MAX_PLAYLIST_LENGTH, + page_token=None, + to_item=lambda item: item, ) - return request.execute().get("items", []) + if page.next_page_token: + raise YouTubeError( + f"Playlist {playlist_id} has more than {_MAX_PLAYLIST_LENGTH} entries, " + f"too many to search for video {video_id}. Remove it in YouTube Studio." + ) + return [item for item in page.items if item.get("contentDetails", {}).get("videoId") == video_id] def _item_snippet(playlist_id: str, video_id: str, position: int | None) -> dict: diff --git a/tests/ops/test_playlists.py b/tests/ops/test_playlists.py index 207a59e..520784d 100644 --- a/tests/ops/test_playlists.py +++ b/tests/ops/test_playlists.py @@ -135,6 +135,71 @@ def test_remove_from_playlist_is_idempotent(api): assert playlists.remove_from_playlist("PL1", "dQw4w9WgXcQ").removed == 0 +def _entries(*video_ids, token=None): + items = [{"id": f"item-{v}", "contentDetails": {"videoId": v}} for v in video_ids] + return {"items": items, "nextPageToken": token} if token else {"items": items} + + +VIDEO_NOT_FOUND = http_error(404, "videoNotFound", "Video not found.") + + +def test_remove_from_playlist_finds_a_deleted_video_by_reading_the_playlist(api): + # YouTube rejects the videoId filter once the video is deleted, though the playlist still holds it. + first_page = _entries("gKcKEGRfoAM", "7A22uuMVA6w", token="next") + first_page["items"].append({"id": "item-no-video", "contentDetails": {}}) + items = api.oauth.playlistItems() + items.list().execute.side_effect = [VIDEO_NOT_FOUND, first_page, _entries("7A22uuMVA6w")] + items.list.reset_mock() + + result = playlists.remove_from_playlist("PL1", "7A22uuMVA6w") + + assert result.removed == 2 + deleted = [call.kwargs["id"] for call in items.delete.call_args_list if call.kwargs] + assert deleted == ["item-7A22uuMVA6w", "item-7A22uuMVA6w"] + filtered, *scans = [call.kwargs for call in items.list.call_args_list] + assert filtered["videoId"] == "7A22uuMVA6w" + assert scans == [ + {"part": "contentDetails", "playlistId": "PL1", "maxResults": 50}, + {"part": "contentDetails", "playlistId": "PL1", "maxResults": 50, "pageToken": "next"}, + ] + + +def test_remove_from_playlist_refuses_a_scan_cut_off_at_the_playlist_cap(api, monkeypatch): + monkeypatch.setattr(playlists, "_MAX_PLAYLIST_LENGTH", 1) + api.oauth.playlistItems().list().execute.side_effect = [VIDEO_NOT_FOUND, _entries("gKcKEGRfoAM", token="next")] + + with pytest.raises(YouTubeError, match=r"more than 1 entries.*YouTube Studio"): + playlists.remove_from_playlist("PL1", "7A22uuMVA6w") + api.oauth.playlistItems().delete.assert_not_called() + + +def test_remove_from_playlist_of_a_video_in_no_entry_removes_nothing(api): + api.oauth.playlistItems().list().execute.side_effect = [VIDEO_NOT_FOUND, _entries("gKcKEGRfoAM")] + + assert playlists.remove_from_playlist("PL1", "7A22uuMVA6w").removed == 0 + api.oauth.playlistItems().delete.assert_not_called() + + +def test_remove_from_playlist_raises_other_errors_without_reading_the_playlist(api): + items = api.oauth.playlistItems() + items.list().execute.side_effect = http_error(404, "playlistNotFound", "Playlist not found.") + items.list.reset_mock() + + with pytest.raises(NotFound): + playlists.remove_from_playlist("PL1", "7A22uuMVA6w") + assert items.list.call_count == 1 + + +def test_move_in_playlist_does_not_read_the_playlist_for_a_deleted_video(api): + items = api.oauth.playlistItems() + items.list().execute.side_effect = VIDEO_NOT_FOUND + items.list.reset_mock() + + with pytest.raises(NotFound): + playlists.move_in_playlist("PL1", "7A22uuMVA6w", position=0) + assert items.list.call_count == 1 + + def test_move_in_playlist(api): api.oauth.playlistItems().list().execute.return_value = {"items": [{"id": "item1"}]} api.oauth.playlistItems().update().execute.return_value = playlist_item_resource(position=0) diff --git a/tests/snapshots/mcp/remove_from_playlist.json b/tests/snapshots/mcp/remove_from_playlist.json index 45e54d2..87d61cf 100644 --- a/tests/snapshots/mcp/remove_from_playlist.json +++ b/tests/snapshots/mcp/remove_from_playlist.json @@ -6,7 +6,7 @@ "read_only_hint": false, "title": null }, - "description": "Remove a video from one of your playlists, every time it appears. The video itself is not affected.\n\nQuota: 51 units.", + "description": "Remove a video from one of your playlists, every time it appears. The video itself is not affected.\n\nYouTube can't look up entries of a deleted video, so for a deleted or unknown video the whole playlist is read:\n1 more unit per 50 entries.\n\nQuota: 51 units.", "input_schema": { "additionalProperties": false, "properties": {