🎬 YouTube API for Creators - Complete Automation Guide #

# Install the Google API client library
pip install google-api-python-client google-auth google-auth-oauthlib google-auth-httplib2

{
  "web": {
    "client_id": "YOUR_CLIENT_ID",
    "client_secret": "YOUR_CLIENT_SECRET",
    "redirect_uris": [],
    "auth_uri": "https://accounts.google.com/o/oauth2/auth",
    "token_uri": "https://accounts.google.com/o/oauth2/token"
  }
}

YOUTUBE_UPLOAD_SCOPE = "https://www.googleapis.com/auth/youtube.upload"
YOUTUBE_READONLY_SCOPE = "https://www.googleapis.com/auth/youtube.readonly"
YOUTUBE_FORCE_SSL_SCOPE = "https://www.googleapis.com/auth/youtube.force-ssl"

#!/usr/bin/env python3
import os
import sys
import httplib2
import random
import time
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from googleapiclient.http import MediaFileUpload
from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request
import pickle

# Configuration
CLIENT_SECRETS_FILE = "client_secrets.json"
SCOPES = ['https://www.googleapis.com/auth/youtube.upload']
API_SERVICE_NAME = 'youtube'
API_VERSION = 'v3'

def get_authenticated_service():
    credentials = None
    
    # Load existing credentials
    if os.path.exists('token.pickle'):
        with open('token.pickle', 'rb') as token:
            credentials = pickle.load(token)
    
    # Refresh or get new credentials
    if not credentials or not credentials.valid:
        if credentials and credentials.expired and credentials.refresh_token:
            credentials.refresh(Request())
        else:
            flow = InstalledAppFlow.from_client_secrets_file(
                CLIENT_SECRETS_FILE, SCOPES)
            credentials = flow.run_local_server(port=0)
        
        # Save credentials for future use
        with open('token.pickle', 'wb') as token:
            pickle.dump(credentials, token)
    
    return build(API_SERVICE_NAME, API_VERSION, credentials=credentials)

def upload_video(youtube, options):
    tags = None
    if options.get('keywords'):
        tags = options['keywords'].split(',')
    
    body = {
        'snippet': {
            'title': options['title'],
            'description': options['description'],
            'tags': tags,
            'categoryId': options.get('category', '22')  # Default: People & Blogs
        },
        'status': {
            'privacyStatus': options.get('privacy', 'private'),
            'selfDeclaredMadeForKids': False
        }
    }
    
    # Create upload request
    insert_request = youtube.videos().insert(
        part=','.join(body.keys()),
        body=body,
        media_body=MediaFileUpload(
            options['file'], 
            chunksize=-1, 
            resumable=True
        )
    )
    
    return resumable_upload(insert_request)

def resumable_upload(insert_request):
    response = None
    error = None
    retry = 0
    
    while response is None:
        try:
            print("Uploading file...")
            status, response = insert_request.next_chunk()
            
            if response is not None:
                if 'id' in response:
                    print(f"Video ID '{response['id']}' uploaded successfully!")
                    return response['id']
                else:
                    raise Exception(f"Upload failed: {response}")
                    
        except HttpError as e:
            if e.resp.status in [500, 502, 503, 504]:
                error = f"Retriable HTTP error {e.resp.status}: {e.content}"
            else:
                raise
        except Exception as e:
            error = f"Retriable error: {e}"
        
        if error is not None:
            print(error)
            retry += 1
            if retry > 3:
                raise Exception("Max retries exceeded")
            
            max_sleep = 2 ** retry
            sleep_seconds = random.random() * max_sleep
            print(f"Sleeping {sleep_seconds:.2f} seconds...")
            time.sleep(sleep_seconds)

# Usage example
if __name__ == '__main__':
    youtube = get_authenticated_service()
    
    upload_options = {
        'file': '/path/to/your/video.mp4',
        'title': 'My Awesome Video',
        'description': 'This is an automated upload using YouTube API',
        'keywords': 'automation,youtube,api',
        'category': '22',
        'privacy': 'private'  # Options: private, public, unlisted
    }
    
    try:
        video_id = upload_video(youtube, upload_options)
        print(f"Upload complete! Video ID: {video_id}")
    except HttpError as e:
        print(f"HTTP error {e.resp.status}: {e.content}")

# ✅ DO: Store credentials securely
import os
from google.oauth2.credentials import Credentials

# Use environment variables
CLIENT_ID = os.environ.get('YOUTUBE_CLIENT_ID')
CLIENT_SECRET = os.environ.get('YOUTUBE_CLIENT_SECRET')

# ❌ DON'T: Hardcode credentials
CLIENT_ID = "your_actual_client_id"  # Never do this!

def refresh_credentials(credentials):
    """Safely refresh expired credentials"""
    if credentials.expired and credentials.refresh_token:
        try:
            credentials.refresh(Request())
            return credentials
        except Exception as e:
            print(f"Token refresh failed: {e}")
            # Re-authenticate user
            return get_new_credentials()
    return credentials

import time
from functools import wraps

def rate_limit(calls_per_second=1):
    """Decorator to limit API calls"""
    min_interval = 1.0 / calls_per_second
    last_called = [0.0]
    
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            elapsed = time.time() - last_called[0]
            left_to_wait = min_interval - elapsed
            if left_to_wait > 0:
                time.sleep(left_to_wait)
            ret = func(*args, **kwargs)
            last_called[0] = time.time()
            return ret
        return wrapper
    return decorator

@rate_limit(calls_per_second=0.5)  # Max 1 call every 2 seconds
def upload_with_rate_limit(youtube, options):
    return upload_video(youtube, options)

import json
import logging
from pathlib import Path

def batch_upload_videos(video_configs):
    """Upload multiple videos with comprehensive error handling"""
    youtube = get_authenticated_service()
    results = []
    
    for i, config in enumerate(video_configs):
        try:
            print(f"Uploading video {i+1}/{len(video_configs)}: {config['title']}")
            video_id = upload_video(youtube, config)
            
            results.append({
                'status': 'success',
                'video_id': video_id,
                'title': config['title'],
                'file': config['file']
            })
            
            # Rate limiting between uploads
            time.sleep(2)
            
        except Exception as e:
            logging.error(f"Failed to upload {config['file']}: {e}")
            results.append({
                'status': 'failed',
                'error': str(e),
                'title': config['title'],
                'file': config['file']
            })
    
    return results

# Usage
video_configs = [
    {
        'file': '/path/to/video1.mp4',
        'title': 'Video 1 Title',
        'description': 'Description for video 1',
        'keywords': 'tag1,tag2,tag3',
        'privacy': 'private'
    },
    # Add more video configurations...
]

results = batch_upload_videos(video_configs)

def upload_thumbnail(youtube, video_id, thumbnail_path):
    """Upload custom thumbnail for a video"""
    try:
        youtube.thumbnails().set(
            videoId=video_id,
            media_body=MediaFileUpload(thumbnail_path)
        ).execute()
        print(f"Thumbnail uploaded for video {video_id}")
    except HttpError as e:
        print(f"Thumbnail upload failed: {e}")

def update_video_metadata(youtube, video_id, updates):
    """Update video title, description, tags, etc."""
    try:
        # Get current video details
        video_response = youtube.videos().list(
            part='snippet',
            id=video_id
        ).execute()
        
        if not video_response['items']:
            raise Exception(f"Video {video_id} not found")
        
        snippet = video_response['items'][0]['snippet']
        
        # Apply updates
        for key, value in updates.items():
            if key in snippet:
                snippet[key] = value
        
        # Update video
        youtube.videos().update(
            part='snippet',
            body={
                'id': video_id,
                'snippet': snippet
            }
        ).execute()
        
        print(f"Video {video_id} updated successfully")
        
    except HttpError as e:
        print(f"Update failed: {e}")

import schedule
import time
from datetime import datetime

def scheduled_upload_job():
    """Job function for scheduled uploads"""
    # Check for new videos in upload queue
    queue_file = 'upload_queue.json'
    
    if os.path.exists(queue_file):
        with open(queue_file, 'r') as f:
            queue = json.load(f)
        
        if queue:
            video_config = queue.pop(0)
            try:
                youtube = get_authenticated_service()
                video_id = upload_video(youtube, video_config)
                print(f"Scheduled upload complete: {video_id}")
                
                # Update queue file
                with open(queue_file, 'w') as f:
                    json.dump(queue, f)
                    
            except Exception as e:
                print(f"Scheduled upload failed: {e}")
                # Re-add to queue for retry
                queue.insert(0, video_config)
                with open(queue_file, 'w') as f:
                    json.dump(queue, f)

# Schedule uploads
schedule.every().day.at("10:00").do(scheduled_upload_job)
schedule.every().day.at("14:00").do(scheduled_upload_job)

# Keep the scheduler running
while True:
    schedule.run_pending()
    time.sleep(60)

class YouTubeUploader:
    def __init__(self, credentials_file):
        self.credentials_file = credentials_file
        self.youtube = None
        
    def authenticate(self):
        """Initialize YouTube service"""
        self.youtube = get_authenticated_service()
        
    def process_video_folder(self, folder_path, template_config):
        """Process all videos in a folder"""
        folder = Path(folder_path)
        video_files = list(folder.glob('*.mp4')) + list(folder.glob('*.mov'))
        
        for video_file in video_files:
            config = template_config.copy()
            config['file'] = str(video_file)
            config['title'] = f"{template_config['title']} - {video_file.stem}"
            
            try:
                video_id = upload_video(self.youtube, config)
                print(f"Uploaded: {video_file.name} -> {video_id}")
                
                # Move processed file
                processed_folder = folder / 'processed'
                processed_folder.mkdir(exist_ok=True)
                video_file.rename(processed_folder / video_file.name)
                
            except Exception as e:
                print(f"Failed to upload {video_file.name}: {e}")

def handle_quota_exceeded(func):
    """Decorator to handle quota exceeded errors"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except HttpError as e:
            if e.resp.status == 403 and 'quotaExceeded' in str(e):
                print("Quota exceeded. Waiting until tomorrow...")
                # Implement exponential backoff or queue for next day
                return None
            raise
    return wrapper

def auto_refresh_token(func):
    """Auto-refresh expired tokens"""
    @wraps(func)
    def wrapper(self, *args, **kwargs):
        try:
            return func(self, *args, **kwargs)
        except HttpError as e:
            if e.resp.status == 401:
                print("Token expired, refreshing...")
                self.authenticate()
                return func(self, *args, **kwargs)
            raise
    return wrapper

def upload_large_video(youtube, options, chunk_size=1024*1024):
    """Handle large video uploads with chunking"""
    media = MediaFileUpload(
        options['file'],
        chunksize=chunk_size,
        resumable=True
    )
    
    # Implement progress tracking
    def progress_callback(request, response):
        if response:
            print(f"Upload progress: {response.get('progress', 0):.1%}")
    
    # Add progress callback to media upload
    media.stream().progress_callback = progress_callback
    
    # Continue with normal upload process
    return upload_video(youtube, options)

# .env file
YOUTUBE_CLIENT_ID=your_client_id_here
YOUTUBE_CLIENT_SECRET=your_client_secret_here
YOUTUBE_REFRESH_TOKEN=your_refresh_token_here
UPLOAD_FOLDER=/path/to/videos
DEFAULT_PRIVACY=private
DEFAULT_CATEGORY=22

import os
from dataclasses import dataclass
from typing import Optional

@dataclass
class YouTubeConfig:
    client_id: str
    client_secret: str
    refresh_token: Optional[str] = None
    upload_folder: str = './uploads'
    default_privacy: str = 'private'
    default_category: str = '22'
    max_retries: int = 3
    chunk_size: int = -1
    
    @classmethod
    def from_env(cls):
        return cls(
            client_id=os.environ['YOUTUBE_CLIENT_ID'],
            client_secret=os.environ['YOUTUBE_CLIENT_SECRET'],
            refresh_token=os.environ.get('YOUTUBE_REFRESH_TOKEN'),
            upload_folder=os.environ.get('UPLOAD_FOLDER', './uploads'),
            default_privacy=os.environ.get('DEFAULT_PRIVACY', 'private'),
            default_category=os.environ.get('DEFAULT_CATEGORY', '22'),
        )