python_scripts/uploader.py

#!/usr/bin/env python3
import os
import json
import argparse
from enum import Enum
from datetime import datetime
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request
from googleapiclient.discovery import build
from googleapiclient.http import MediaFileUpload
from googleapiclient.errors import HttpError

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 load_credentials():
    """Load credentials from secrets.json"""
    try:
        with open('secrets.json', 'r') as f:
            creds_data = json.load(f)
            
        credentials = Credentials(
            token=creds_data['token'],
            refresh_token=creds_data['refresh_token'],
            token_uri=creds_data['token_uri'],
            client_id=creds_data['client_id'],
            client_secret=creds_data['client_secret'],
            scopes=creds_data['scopes']
        )
        
        # Refresh token if expired
        if credentials.expired:
            credentials.refresh(Request())
            
        return credentials
    except FileNotFoundError:
        print("Error: secrets.json not found! Please run setup.py first.")
        return None
    except Exception as e:
        print(f"Error loading credentials: {str(e)}")
        return None

def upload_video(file_path: str, title: str, category: CategoryId, description: str = "", privacyStatus: str = 'private', tags: list = None, release_time: datetime = 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)
    """
    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 False

        # Load credentials
        credentials = load_credentials()
        if not credentials:
            return False

        # Create YouTube API client
        youtube = build('youtube', 'v3', credentials=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')
            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 True

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

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()