python_scripts/package/youtube/uploader.py

#!/usr/bin/env python3
import os
import argparse
from enum import Enum
from datetime import datetime, timezone

from googleapiclient.discovery import build
from googleapiclient.http import MediaFileUpload
from googleapiclient.errors import HttpError

import package.youtube.config as config

class CategoryId(Enum):
    """Official YouTube Category IDs for API Uploads."""

    FILM_AND_ANIMATION = "1"
    AUTOS_AND_VEHICLES = "2"
    MUSIC = "10"
    PETS_AND_ANIMALS = "15"
    SPORTS = "17"
    TRAVEL_AND_EVENTS = "19"
    GAMING = "20"
    PEOPLE_AND_BLOGS = "22"
    COMEDY = "23"
    ENTERTAINMENT = "24"
    NEWS_AND_POLITICS = "25"
    HOWTO_AND_STYLE = "26"
    EDUCATION = "27"
    SCIENCE_AND_TECHNOLOGY = "28"
    NONPROFITS_AND_ACTIVISM = "29"

def upload_video(file_path: str, title: str, category: CategoryId, description: str = "", privacyStatus: str = 'private', tags: list = None, release_time: datetime = None) -> str | None:
    """
    Upload a video to YouTube
    
    Args:
        file_path (str): Path to the video file
        title (str): Title of the Video
        categoryId (str): categoryId of the video
        description (str): Video description
        privacyStatus (str): privacyStatus of the video
        tags (str): tags to be use on the video
        release_time (datetime): Optional timezone-aware UTC datetime object for timed release

        schedule_date = datetime.now(timezone.utc) + timedelta(days=2)
    Returns:
        a String(str) of with the youtube video id or None if somthing fails
    """
    if tags is None:
        tags = []

    try:
        # Check if file exists
        if not os.path.exists(file_path):
            print(f"❌ Error: File not found: {file_path}")
            return None

        # Load credentials
        if not config.CREDENTIALS:
            print("❌ Error: YouTube API credentials not configured.")
            return None

        # Create YouTube API client
        youtube = build('youtube', 'v3', credentials=config.CREDENTIALS)

        # Configure the status object dynamically
        status_body = {
            'selfDeclaredMadeForKids': False
        }

        if release_time is not None:
            if release_time.tzinfo is None:
                release_time = release_time.replace(tzinfo=timezone.utc)

            # If release_time is passed, YouTube forces privacyStatus to 'private'
            status_body['privacyStatus'] = 'private'
            #status_body['publishAt'] = release_time.strftime('%Y-%m-%dT%H:%M:%S.000Z')
            status_body['publishAt'] = release_time.isoformat() 
            print(f"Configuring timed release for: {status_body['publishAt']}")
        else:
            # Standard immediate upload
            status_body['privacyStatus'] = privacyStatus
            print(f"Configuring immediate upload with status: {privacyStatus}")

        # Prepare the video upload request
        body = {
            'snippet': {
                'title': title,
                'description': description,
                'tags': tags,
                'categoryId': category.value
            },
            'status': status_body
        }

        # Create media file upload
        media = MediaFileUpload(
            file_path,
            chunksize=1024*1024,
            resumable=True
        )

        # Create the video insert request
        insert_request = youtube.videos().insert(
            part=','.join(body.keys()),
            body=body,
            media_body=media
        )

        print(f"Starting upload for '{title}'...")
        response = None
        while response is None:
            status, response = insert_request.next_chunk()
            if status:
                print(f"Uploaded {int(status.progress() * 100)}%")

        print(f"\nUpload Complete!")
        print(f"Video ID: {response['id']}")
        print(f"Title: {response['snippet']['title']}")
        print(f"URL: https://youtu.be/{response['id']}")

        # Output confirmation based on what was chosen
        if 'publishAt' in response['status']:
            print(f"Scheduled Release Time: {response['status']['publishAt']}")
        else:
            print(f"Current Privacy Status: {response['status']['privacyStatus']}")
            
        return response['id']

    except HttpError as e:
        print(f"An HTTP error occurred: {str(e)}")
        return None
    except Exception as e:
        print(f"An error occurred: {str(e)}")
        return None

def main():
    parser = argparse.ArgumentParser(description='Upload a video to YouTube')
    parser.add_argument('--file', required=True, help='Path to the video file')
    parser.add_argument('--title', required=True, help='Title of the video')
    parser.add_argument('--category', default='PEOPLE_AND_BLOGS', choices=[c.name for c in CategoryId], help='Video category genre')
    parser.add_argument('--description', default='', help='Video description text')
    parser.add_argument('--privacy', default='private', choices=['public', 'private', 'unlisted'], help='Video privacy settings')
    
    # ADDED: Feature parsing to easily pass tags from the CLI split by commas
    parser.add_argument('--tags', default='', help='Comma-separated tags list (e.g. "python,coding,api")')
    
    # ADDED: Option to provide a scheduled upload timestamp natively from CLI
    parser.add_argument('--schedule', default=None, help='UTC Release date/time in ISO format: YYYY-MM-DDTHH:MM:SS (e.g. 2026-08-15T14:30:00)')
    
    args = parser.parse_args()
    
    chosen_category = CategoryId[args.category]
    parsed_tags = [t.strip() for t in args.tags.split(',')] if args.tags else []
    
    # ADDED: Parse schedule string into datetime object dynamically
    release_datetime = None
    if args.schedule:
        try:
            # Assumes format matches CLI help instruction text
            release_datetime = datetime.strptime(args.schedule, '%Y-%m-%dT%H:%M:%S').replace(tzinfo=timezone.utc)
        except ValueError:
            print("Error: Schedule date must format strictly as YYYY-MM-DDTHH:MM:SS")
            return

    upload_video(
        file_path=args.file, 
        title=args.title, 
        category=chosen_category, 
        description=args.description,
        privacyStatus=args.privacy,
        tags=parsed_tags,
        release_time=release_datetime
    )

if __name__ == "__main__":
    main()