Uploader.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#!/usr/bin/python\n",
    "\n",
    "import http.client as httplib\n",
    "import httplib2\n",
    "import os\n",
    "import random\n",
    "import sys\n",
    "import time\n",
    "import argparse\n",
    "import glob\n",
    "import pickle\n",
    "\n",
    "from googleapiclient.discovery import build\n",
    "from googleapiclient.errors import HttpError\n",
    "from googleapiclient.http import MediaFileUpload\n",
    "from google.auth.transport.requests import Request\n",
    "from google_auth_oauthlib.flow import InstalledAppFlow\n",
    "\n",
    "# Retry configuration\n",
    "httplib2.RETRIES = 1\n",
    "MAX_RETRIES = 10\n",
    "RETRIABLE_EXCEPTIONS = (\n",
    "    httplib2.HttpLib2Error,\n",
    "    IOError,\n",
    "    httplib.NotConnected,\n",
    "    httplib.IncompleteRead,\n",
    "    httplib.ImproperConnectionState,\n",
    "    httplib.CannotSendRequest,\n",
    "    httplib.CannotSendHeader,\n",
    "    httplib.ResponseNotReady,\n",
    "    httplib.BadStatusLine,\n",
    ")\n",
    "RETRIABLE_STATUS_CODES = [500, 502, 503, 504]\n",
    "\n",
    "# OAuth 2.0 configuration\n",
    "CLIENT_SECRETS_FILE = \"client_secrets.json\"\n",
    "YOUTUBE_UPLOAD_SCOPE = \"https://www.googleapis.com/auth/youtube.upload\"\n",
    "YOUTUBE_API_SERVICE_NAME = \"youtube\"\n",
    "YOUTUBE_API_VERSION = \"v3\"\n",
    "MISSING_CLIENT_SECRETS_MESSAGE = \"\"\"\n",
    "WARNING: Please configure OAuth 2.0\n",
    "To make this sample run you will need to populate the client_secrets.json file\n",
    "found at:\n",
    "   %s\n",
    "with information from the API Console\n",
    "https://console.cloud.google.com/\n",
    "For more information about the client_secrets.json file format, please visit:\n",
    "https://developers.google.com/api-client-library/python/guide/aaa_client_secrets\n",
    "\"\"\"\n",
    "VALID_PRIVACY_STATUSES = (\"unlisted\", \"public\", \"private\")\n",
    "\n",
    "def get_authenticated_service():\n",
    "    flow = InstalledAppFlow.from_client_secrets_file(\n",
    "        CLIENT_SECRETS_FILE, scopes=[YOUTUBE_UPLOAD_SCOPE]\n",
    "    )\n",
    "\n",
    "    credentials = None\n",
    "\n",
    "    # Check if token file exists\n",
    "    if os.path.exists(\"token.pickle\"):\n",
    "        # If it does, load the credentials from it\n",
    "        try:\n",
    "            with open(\"token.pickle\", \"rb\") as token:\n",
    "                credentials = pickle.load(token)\n",
    "        except EOFError:\n",
    "            # If an EOFError was raised, the file is empty. Credentials will be None.\n",
    "            pass\n",
    "\n",
    "    # If there are no valid credentials available, prompt the user to log in\n",
    "    if not credentials or not credentials.valid:\n",
    "        if credentials and credentials.expired and credentials.refresh_token:\n",
    "            credentials.refresh(Request())\n",
    "        else:\n",
    "            credentials = flow.run_local_server(port=0)\n",
    "\n",
    "        # Save the credentials for the next run\n",
    "        with open(\"token.pickle\", \"wb\") as token:\n",
    "            pickle.dump(credentials, token)\n",
    "\n",
    "    return build(YOUTUBE_API_SERVICE_NAME, YOUTUBE_API_VERSION, credentials=credentials)\n",
    "\n",
    "def initialize_upload(youtube, options):\n",
    "    print(\"Initializing Upload\")\n",
    "    tags = None\n",
    "    if options.keywords:\n",
    "        tags = options.keywords.split(\",\")\n",
    "\n",
    "    body = dict(\n",
    "        snippet=dict(\n",
    "            title=options.title,\n",
    "            description=options.description,\n",
    "            tags=tags,\n",
    "            categoryId=options.category,\n",
    "        ),\n",
    "        status=dict(privacyStatus=options.privacyStatus),\n",
    "    )\n",
    "\n",
    "    # Call the API's videos.insert method to create and upload the video.\n",
    "    insert_request = youtube.videos().insert(\n",
    "        part=\",\".join(body.keys()),\n",
    "        body=body,\n",
    "        media_body=MediaFileUpload(options.file, chunksize=-1, resumable=True),\n",
    "    )\n",
    "\n",
    "    resumable_upload(insert_request)\n",
    "\n",
    "def resumable_upload(insert_request):\n",
    "    response = None\n",
    "    error = None\n",
    "    retry = 0\n",
    "    while response is None:\n",
    "        try:\n",
    "            print(\"Uploading file...\")\n",
    "            status, response = insert_request.next_chunk()\n",
    "            if response is not None:\n",
    "                if \"id\" in response:\n",
    "                    print(\"Video id '%s' was successfully uploaded.\" % response[\"id\"])\n",
    "                else:\n",
    "                    exit(\"The upload failed with an unexpected response: %s\" % response)\n",
    "        except HttpError as e:\n",
    "            if e.resp.status in RETRIABLE_STATUS_CODES:\n",
    "                error = \"A retriable HTTP error %d occurred:\\n%s\" % (\n",
    "                    e.resp.status,\n",
    "                    e.content,\n",
    "                )\n",
    "            else:\n",
    "                raise\n",
    "        except RETRIABLE_EXCEPTIONS as e:\n",
    "            error = \"A retriable error occurred: %s\" % e\n",
    "\n",
    "        if error is not None:\n",
    "            print(error)\n",
    "            retry += 1\n",
    "            if retry > MAX_RETRIES:\n",
    "                exit(\"No longer attempting to retry.\")\n",
    "\n",
    "            max_sleep = 2**retry\n",
    "            sleep_seconds = random.random() * max_sleep\n",
    "            print(\"Sleeping %f seconds and then retrying...\" % sleep_seconds)\n",
    "            time.sleep(sleep_seconds)\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    # Create an ArgumentParser object\n",
    "    argparser = argparse.ArgumentParser()\n",
    "\n",
    "    # Add the necessary arguments\n",
    "    # argparser.add_argument(\"--file\", required=True, help=\"Video file to upload\")\n",
    "    argparser.add_argument(\"--title\", help=\"Video title\", default=\"Test Title\")\n",
    "    argparser.add_argument(\n",
    "        \"--description\",\n",
    "        help=\"Video description\",\n",
    "        default=\"Uploading automatically using code\",\n",
    "    )\n",
    "    argparser.add_argument(\n",
    "        \"--category\",\n",
    "        default=\"28\",\n",
    "        help=\"Numeric video category. \"\n",
    "        \"See https://developers.google.com/youtube/v3/docs/videoCategories/list\",\n",
    "    )\n",
    "    argparser.add_argument(\n",
    "        \"--keywords\", help=\"Video keywords, comma separated\", default=\"\"\n",
    "    )\n",
    "    argparser.add_argument(\n",
    "        \"--privacyStatus\",\n",
    "        choices=VALID_PRIVACY_STATUSES,\n",
    "        default=VALID_PRIVACY_STATUSES[0],\n",
    "        help=\"Video privacy status.\",\n",
    "    )\n",
    "    args = argparser.parse_args()\n",
    "\n",
    "    youtube = get_authenticated_service()\n",
    "\n",
    "    # Get a list of all video files in the \"videos\" folder\n",
    "    video_files = (\n",
    "        glob.glob(\"videos/*.mp4\")\n",
    "        + glob.glob(\"videos/*.avi\")\n",
    "        + glob.glob(\"videos/*.mov\")\n",
    "        + glob.glob(\"videos/*.flv\")\n",
    "        + glob.glob(\"videos/*.wmv\")\n",
    "        + glob.glob(\"videos/*.mkv\")\n",
    "        + glob.glob(\"videos/*.webm\")\n",
    "        + glob.glob(\"videos/*.mpeg\")\n",
    "        + glob.glob(\"videos/*.mpg\")\n",
    "        + glob.glob(\"videos/*.m4v\")\n",
    "        + glob.glob(\"videos/*.3gp\")\n",
    "    )\n",
    "\n",
    "    # Loop over the list of video files and upload each one\n",
    "    for video_file in video_files:\n",
    "        if not os.path.exists(video_file):\n",
    "            print(f\"The file {video_file} does not exist.\")\n",
    "            continue\n",
    "\n",
    "        args.file = video_file\n",
    "        # If no title argument is specified, use the filename as the title\n",
    "        if args.title == \"Test Title\":\n",
    "            args.title = os.path.basename(video_file)\n",
    "        print(f\"Uploading {video_file}...\")\n",
    "        try:\n",
    "            initialize_upload(youtube, args)\n",
    "        except HttpError as e:\n",
    "            print(\"An HTTP error %d occurred:\\n%s\" % (e.resp.status, e.content))\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}