#!/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(): # FIXED: Reconstructed arguments to provide all required inputs for upload_video 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') args = parser.parse_args() # Map the text input choice back to your CategoryId Enum object chosen_category = CategoryId[args.category] # FIXED: Properly pass all required positional and optional parameters upload_video( file_path=args.file, title=args.title, category=chosen_category, description=args.description, privacyStatus=args.privacy ) if __name__ == "__main__": main()