diff --git a/backend/app/youtube/client.py b/backend/app/youtube/client.py index 9d22961..4bea310 100644 --- a/backend/app/youtube/client.py +++ b/backend/app/youtube/client.py @@ -213,6 +213,93 @@ class YouTubeClient: f"DELETE subscriptions -> {resp.status_code}: {resp.text[:300]}" ) + def iter_playlist_items_with_ids(self, playlist_id: str) -> Iterator[dict]: + """Yield {item_id, video_id} for each item of one of the user's playlists, in + playlist order. `item_id` is the playlistItem resource id, needed to delete or + reposition the item via the write API. OAuth (private playlists work).""" + page_token = None + while True: + params = { + "part": "contentDetails", + "playlistId": playlist_id, + "maxResults": 50, + } + if page_token: + params["pageToken"] = page_token + data = self._get("playlistItems", params, cost=1, allow_key=False) + for item in data.get("items", []): + vid = item.get("contentDetails", {}).get("videoId") + if vid: + yield {"item_id": item.get("id"), "video_id": vid} + page_token = data.get("nextPageToken") + if not page_token: + break + + # --- write endpoints (OAuth only, never the API key; each costs 50 quota units) --- + def _write(self, method: str, path: str, *, params: dict, json: dict | None = None) -> dict: + headers = {"Authorization": f"Bearer {self._access_token()}"} + resp = self._http.request( + method, f"{API_BASE}/{path}", params=params, json=json, headers=headers + ) + quota.record_usage(self.db, 50) + if resp.status_code not in (200, 204): + log.warning("YouTube %s %s -> %s: %s", method, path, resp.status_code, resp.text[:200]) + raise YouTubeError(f"{method} {path} -> {resp.status_code}: {resp.text[:300]}") + return resp.json() if resp.content else {} + + def create_playlist(self, title: str, description: str = "", privacy: str = "private") -> str: + """Create a new YouTube playlist; returns its id. 50 units.""" + data = self._write( + "POST", + "playlists", + params={"part": "snippet,status"}, + json={ + "snippet": {"title": title[:150], "description": description[:5000]}, + "status": {"privacyStatus": privacy}, + }, + ) + return data["id"] + + def add_playlist_item(self, playlist_id: str, video_id: str, position: int | None = None) -> str: + """Append (or insert at `position`) a video into a playlist; returns the new + playlistItem id. 50 units.""" + snippet = { + "playlistId": playlist_id, + "resourceId": {"kind": "youtube#video", "videoId": video_id}, + } + if position is not None: + snippet["position"] = position + data = self._write( + "POST", "playlistItems", params={"part": "snippet"}, json={"snippet": snippet} + ) + return data["id"] + + def move_playlist_item( + self, item_id: str, playlist_id: str, video_id: str, position: int + ) -> None: + """Reposition an existing playlistItem to absolute 0-based `position`. 50 units.""" + self._write( + "PUT", + "playlistItems", + params={"part": "snippet"}, + json={ + "id": item_id, + "snippet": { + "playlistId": playlist_id, + "position": position, + "resourceId": {"kind": "youtube#video", "videoId": video_id}, + }, + }, + ) + + def delete_playlist_item(self, item_id: str) -> None: + """Remove a playlistItem from its playlist. 50 units.""" + self._write("DELETE", "playlistItems", params={"id": item_id}) + + def delete_playlist(self, playlist_id: str) -> None: + """Delete a whole playlist on YouTube. 50 units.""" + self._write("DELETE", "playlists", params={"id": playlist_id}) + def get_videos(self, video_ids: list[str]) -> list[dict]: items: list[dict] = [] for batch in _chunks(video_ids, 50):