How to Automate YouTube Playlist Management using the API

1#!/usr/bin/env python3
2"""Auto-curates a rolling 'Latest Videos' playlist on YouTube.
3
4Reads from a source playlist (or channel uploads), adds new videos to a
5target playlist, and trims it to MAX_SIZE by removing oldest items.
6Tracks quota usage to avoid exhausting the 10,000 units/day limit.
7"""
8import os
9import json
10import logging
11from google.oauth2.credentials import Credentials
12from google.auth.transport.requests import Request
13from googleapiclient.discovery import build
14from googleapiclient.errors import HttpError
15
16logging.basicConfig(level=logging.INFO, format='%(asctime)s %(message)s')
17
18SCOPES = ['https://www.googleapis.com/auth/youtube']
19CREDS_FILE = os.environ.get('YOUTUBE_CREDS_FILE', 'credentials.json')
20
21SOURCE_PLAYLIST_ID = os.environ['YOUTUBE_SOURCE_PLAYLIST']  # Uploads playlist or any playlist
22TARGET_PLAYLIST_ID = os.environ['YOUTUBE_TARGET_PLAYLIST']  # The curated playlist
23MAX_PLAYLIST_SIZE = 20
24SEEN_VIDEOS_FILE = 'seen_videos.json'
25
26QUOTA_BUDGET = 5000  # Use at most half the daily quota in one run
27quota_used = 0
28
29def load_seen_videos():
30    if os.path.exists(SEEN_VIDEOS_FILE):
31        with open(SEEN_VIDEOS_FILE) as f:
32            return set(json.load(f))
33    return set()
34
35def save_seen_videos(seen):
36    with open(SEEN_VIDEOS_FILE, 'w') as f:
37        json.dump(list(seen), f)
38
39def get_youtube():
40    creds = Credentials.from_authorized_user_file(CREDS_FILE, SCOPES)
41    if creds.expired and creds.refresh_token:
42        creds.refresh(Request())
43        with open(CREDS_FILE, 'w') as f:
44            f.write(creds.to_json())
45    return build('youtube', 'v3', credentials=creds)
46
47def list_playlist_items(yt, playlist_id):
48    global quota_used
49    items = []
50    page_token = None
51    while True:
52        resp = yt.playlistItems().list(
53            part='snippet,contentDetails', playlistId=playlist_id,
54            maxResults=50, pageToken=page_token
55        ).execute()
56        quota_used += 1  # List costs 1 unit per page
57        items.extend(resp['items'])
58        page_token = resp.get('nextPageToken')
59        if not page_token:
60            break
61    return items
62
63def add_video(yt, playlist_id, video_id):
64    global quota_used
65    if quota_used + 50 > QUOTA_BUDGET:
66        raise Exception(f'Quota budget ({QUOTA_BUDGET}) would be exceeded')
67    resp = yt.playlistItems().insert(
68        part='snippet',
69        body={'snippet': {'playlistId': playlist_id, 'resourceId': {'kind': 'youtube#video', 'videoId': video_id}}}
70    ).execute()
71    quota_used += 50
72    return resp['id']
73
74def remove_item(yt, item_id):
75    global quota_used
76    if quota_used + 50 > QUOTA_BUDGET:
77        raise Exception(f'Quota budget ({QUOTA_BUDGET}) would be exceeded')
78    yt.playlistItems().delete(id=item_id).execute()
79    quota_used += 50
80
81def main():
82    yt = get_youtube()
83    seen = load_seen_videos()
84    
85    # Get source videos
86    source_items = list_playlist_items(yt, SOURCE_PLAYLIST_ID)
87    source_video_ids = [i['snippet']['resourceId']['videoId'] for i in source_items]
88    new_videos = [v for v in source_video_ids if v not in seen]
89    logging.info(f'Found {len(new_videos)} new videos to add')
90    
91    # Get current target playlist state
92    target_items = list_playlist_items(yt, TARGET_PLAYLIST_ID)
93    logging.info(f'Target playlist has {len(target_items)} items')
94    
95    # Add new videos to target (newest first, position 0)
96    added = []
97    for vid in new_videos[:10]:  # Add at most 10 per run
98        try:
99            item_id = add_video(yt, TARGET_PLAYLIST_ID, vid)
100            added.append(vid)
101            seen.add(vid)
102            logging.info(f'Added {vid} (quota: {quota_used})')
103        except HttpError as e:
104            logging.error(f'Failed to add {vid}: {e}')
105    
106    # Re-fetch and trim to MAX_PLAYLIST_SIZE
107    if added:
108        current_items = list_playlist_items(yt, TARGET_PLAYLIST_ID)
109        while len(current_items) > MAX_PLAYLIST_SIZE:
110            oldest = current_items[-1]  # Last item is oldest
111            remove_item(yt, oldest['id'])
112            logging.info(f'Removed old item {oldest["id"]} (quota: {quota_used})')
113            current_items.pop()
114    
115    save_seen_videos(seen)
116    logging.info(f'Done. Added {len(added)} videos. Total quota used: {quota_used}')
117
118if __name__ == '__main__':
119    main()