feat(playlists): YouTube write API client methods
Add OAuth-only write helpers to YouTubeClient (each 50 quota units): create_playlist, add_playlist_item, move_playlist_item, delete_playlist_item, delete_playlist, plus iter_playlist_items_with_ids for diffing. A shared _write helper records quota and raises YouTubeError on non-2xx.
This commit is contained in:
parent
35831dc3ce
commit
b7bf6cc6f9
1 changed files with 87 additions and 0 deletions
|
|
@ -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):
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue