YouTube Data API v3 を利用して Google Drive の動画を自動で YouTube にアップロードしてみた

はじめに

YouTube Data API v3（以下、YouTube API） を利用して Google Drive の動画を自動で YouTube にアップロードする方法を構築してみました。経緯としては、Google Drive に友人たちと遊んだ動画を保存していたのですが、Google Drive の保存容量が無料版では 15GB と少なく、容量が逼迫してきたため、YouTube に動画をアップロードしてしまおうと考えました。
しかしながら、YouTube API のクォータ制限（1 日あたりのリクエスト数やアップロード数などの制限）は 1 日当たり 10,000 単位となっています。ビデオのアップロードは、1,600 単位とかなり高いクォータを消費します。したがって、1 日に行えるアップロードは数回に限られます。クォータ制限がリセットされるのは、太平洋時間（PT）の午前 0 時にリセットされます。日本時間では、午後 4 時～5 時の間です。クォータ制限を考慮した応用実装のソースコードも載せてあるので参考にしてみて下さい。
クォータ制限の問題を解決するには、クォータの増加申請が必要になってきます。こちらに関しては審査が必要で、すぐには反映されません（1ヶ月以上やり取りしています）。2023年6月下旬に申請が通ったので、申請方法や申請内容、やり取りについて以下記事にまとめました。

クォータ制限が 10,000 単位から 1,000,000 単位になったので 1 日あたり、600 本くらい動画アップロードできるじゃんと浮かれていましたが、落とし罠がありました。それはまた後ほど。

環境

• Raspberry Pi 4 Model B
  • CentOS Stream 8
    • Python 3.11.0
      • google 3.0.0
      • google-api-core 2.11.0
      • google-api-python-client 2.86.0
      • google-auth 2.18.0
      • google-auth-httplib2 0.1.0
      • google-auth-oauthlib 1.0.0
      • googleapis-common-protos 1.59.0

導入

Google Cloud Platform の設定

まずは、Google Cloud Platformアクセスして下さい。その後、Google Drive にある動画を YouTube に上げたいアカウントに切り替え利用規約に同意して続行して下さい。

左上の[プロジェクトの選択]を押下し、[新しいプロジェクト]を押下して下さい。

適当な名前でプロジェクトが作られるので、名前を変更したい方は変更して[作成]ボタンを押下して下さい。

プロジェクトが作成できたら作成したプロジェクトを選択し、[APIの概要に移動]を押下して下さい。

API とサービスに移動できたら[ライブラリ]を押下して、YouTube Data API と Google Drive API を有効化していきます。

API ライブラリをスクロールしていくと、YouTube Data API と Google Drive APIが見つかるので、それぞれ有効化して下さい。

追加できたら、左上のメニューから[Cloud の概要]→[ダッシュボード]を選択して下さい。その後、[APIの概要に移動]を押下して下さい。

API とサービスに移動できたら[OAuth 同意画面]を押下して、[外部]を選択し、[作成]ボタンを押下して下さい。

アプリ情報の入力が求められるので適当に名前を入力し、[ユーザーサポートメール]は選択肢から選び、[デベロッパーの連絡先情報]にもメールアドレスを入力して下さい。[保存して次へ]を押下するとテストユーザー画面に遷移します。


[+ ADD USERS]からテストユーザーを追加して下さい。Google Drive のアカウントでいいと思います。最後に[保存して次へ]を押下して下さい。

続いて、OAuth クライアントを作っていきます。左メニューの[認証情報]から[認証情報を作成]→[OAuth クライアントID]を選択して下さい。

アプリケーションの種類を選択する画面に遷移するので、アプリケーションとして[デスクトップアプリ]を選択して下さい。

OAuth クライアントが作成できたら、[JSON をダウンロード]を押下してダウンロードし、JSON ファイルを client_secrets.json というファイル名で保存して下さい。保存先は Python コードを実行する先に保存して下さい。

以上で、Google Cloud Platform の設定は完了です。

認証/トークンの取得

ここからは Python を用いトークンの取得を行っていきます。まず始めに、必要なモジュールをインストールしていきましょう。

pip install

pip install google google-api-python-client google-api-core google_auth_oauthlib

続いて認証です。最初は flow.run_local_server という関数を使おうとしたのですが、上手く行かなかったのでここでは、flow.run_console 関数を使用していきます。以下の Python のコードを実行して下さい。但し、バージョンによってはエラーが発生するはずです（執筆現在（2023/07/08）ではエラーが出ます）。

auth.py

#!/usr/bin/env python3
import os
import google.auth
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from google_auth_oauthlib.flow import InstalledAppFlow
from google.oauth2.credentials import Credentials
import google_auth_oauthlib.flow

def authenticate():
    print(google_auth_oauthlib.flow.__file__)
    SCOPES = ['https://www.googleapis.com/auth/drive.readonly', 'https://www.googleapis.com/auth/youtube.force-ssl', 'https://www.googleapis.com/auth/youtube.upload']

    CLIENT_SECRET_FILE = 'client_secrets.json'
    TOKEN_PATH = 'token.json'

    creds = None
    if os.path.exists(TOKEN_PATH):
        creds = Credentials.from_authorized_user_file(TOKEN_PATH, SCOPES)

    if not creds or not creds.valid:
        if creds and creds.expired and creds.refresh_token:
            creds.refresh(google.auth.transport.requests.Request())
        else:
            flow = InstalledAppFlow.from_client_secrets_file(CLIENT_SECRET_FILE, SCOPES)
            creds = flow.run_console()
            with open(TOKEN_PATH, 'w') as token:
                token.write(creds.to_json())

    return creds

def main():
    creds = authenticate()
    youtube = build('youtube', 'v3', credentials=creds)

if __name__ == '__main__':
    main()

これをそのまま実行すると以下のようなエラーが出ると思います。

error

Traceback (most recent call last):
  File "/home/web/youtube/./auth3.py", line 38, in <module>
    main()
  File "/home/web/youtube/./auth3.py", line 33, in main
    creds = authenticate()
            ^^^^^^^^^^^^^^
  File "/home/web/youtube/./auth3.py", line 26, in authenticate
    creds = flow.run_console()
            ^^^^^^^^^^^^^^^^
AttributeError: 'InstalledAppFlow' object has no attribute 'run_console'

これは flow.py 中で run_console という関数が定義されていないものによるエラーになります（バグ報告しておきました）。インストールしたモジュールのバージョンが古く、run_console 関数が定義されていないか、そもそも定義されていないかが考えられます。結論としては、モジュールである google_auth_oauthlib のバグであり、実際に google_auth_oauthlib.flow の現物のソースコードを見てみると、run_console 関数が定義されていることが分かります（なんでこんなに改行入れてるんでしょうね…）。つまり、インストールしたモジュールには run_console 関数がないので、追加してあげればいいわけです。じゃあどこに追加すればいいのかという話になるので、以下のコードを実行して、 flow.py がどこにあるのか確認してあげましょう。

chk.py

#!/usr/bin/env python3
import google_auth_oauthlib.flow

print(google_auth_oauthlib.flow.__file__)

一般ユーザの環境であれば、以下のように表示されるでしょう。

Which flow.py

/home/web/.local/lib/python3.11/site-packages/google_auth_oauthlib/flow.py

root環境で実行している場合以下のように表示されるでしょう。

Which flow.py

/usr/local/lib/python3.11/site-packages/google_auth_oauthlib/flow.py

環境に合わせて flow.py のソースコードを以下のように編集して下さい。

flow.py

     #～省略～
    .. _Installed Application Authorization Flow:
        https://github.com/googleapis/google-api-python-client/blob/main/docs/oauth-installed.md
    #""" 見やすくするためコメントにしてます。

# add
    _OOB_REDIRECT_URI = "urn:ietf:wg:oauth:2.0:oob"
# end

    _DEFAULT_AUTH_PROMPT_MESSAGE = (
        "Please visit this URL to authorize this application: {url}"
    )
    """str: The message to display when prompting the user for
    authorization."""
    _DEFAULT_AUTH_CODE_MESSAGE = "Enter the authorization code: "
    """str: The message to display when prompting the user for the
    authorization code. Used only by the console strategy."""

    _DEFAULT_WEB_SUCCESS_MESSAGE = (
        "The authentication flow has completed. You may close this window."
    )

# add
    def run_console(
        self,
        authorization_prompt_message=_DEFAULT_AUTH_PROMPT_MESSAGE,
        authorization_code_message=_DEFAULT_AUTH_CODE_MESSAGE,
        **kwargs
    ):
        """Run the flow using the console strategy.

        The console strategy instructs the user to open the authorization URL
        in their browser. Once the authorization is complete the authorization
        server will give the user a code. The user then must copy & paste this
        code into the application. The code is then exchanged for a token.

        Args:
            authorization_prompt_message (str): The message to display to tell
                the user to navigate to the authorization URL.
            authorization_code_message (str): The message to display when
                prompting the user for the authorization code.
            kwargs: Additional keyword arguments passed through to
                :meth:`authorization_url`.

        Returns:
            google.oauth2.credentials.Credentials: The OAuth 2.0 credentials
                for the user.
        """
        kwargs.setdefault("prompt", "consent")

        self.redirect_uri = self._OOB_REDIRECT_URI

        auth_url, _ = self.authorization_url(**kwargs)

        print(authorization_prompt_message.format(url=auth_url))

        code = input(authorization_code_message)

        self.fetch_token(code=code)

        return self.credentials
#end

    def run_local_server(
        self,

     #～省略～

ソースコードの編集が出来たら、auth.py ファイルを実行してみましょう。以下のように表示されるはずです。

auth result

[web@localhost youtube]$ ./auth.py 
Please visit this URL to authorize this application: https://accountstype=code&client_id=14671478899-d40upuv1r99m9t898of79fbl60qokuii.apps=urn%3Aietf%3Awg%3Aoauth%3A2.0%3Aoob&scope=https%3A%2F%2Fwww.googleaps%3A%2F%2Fwww.googleapis.com%2Fauth%2Fdrive.readonly&state=5eLsIcL1Nw&access_type=offline
Enter the authorization code: 

続いて、表示された URL にアクセスし、テストユーザとして許可したアカウントでログインして下さい。「このアプリは Google で確認されていません」と表示されますが、[続行]を選択して下さい。続いて、アクセス許可が求められるのですべて選択し、[続行]を押下して下さい。すると、以下のように認証コードが表示されるので、コピーボタンからコピーし、auth.py を実行しているコンソールに貼り付けて下さい。これで、ディレクトリ内に token.json というファイルが生成されトークンの取得が完了です。

本実装

ようやく、YouTube API を利用して Google Drive の動画を自動で YouTube にアップロードする準備が整いました。ソースコードは以下の通りです。ここで重要となってくる変数は、folder_id, categoryId, description, title, madeForKids, privacyStatus になります。変数については後ほど説明するとして、コードの流れを軽く説明すると、まず始めに、drive_service.files().list にて Google Drive のフォルダからファイル一覧を取得します。取得したファイル一覧に対してループを回し、googleapiclient.http.MediaIoBaseDownload にて Google Drive からファイルをダウンロードし、request.execute にて動画を YouTube にアップロードします。最後に os.remove で一次取得した動画ファイルを削除してループが終わります。

main.py

#!/usr/bin/env python3
import os
import io
import googleapiclient.discovery
import googleapiclient.errors
import googleapiclient.http
from google.oauth2.credentials import Credentials

# Google Driveから動画を取得し、YouTubeにアップロードする関数
def upload_videos(drive_service, youtube_service, folder_id):
    print("Start")

    # Google Driveから動画ファイルを取得
    query = f"'{folder_id}' in parents"
    results = drive_service.files().list(q=query, fields="files(id, name, mimeType)").execute()
    items = results.get('files', [])

    for item in items:
        if item['mimeType'].startswith('video'):  # 動画ファイルのみを対象とする
            # ファイルを一時的にダウンロード
            request = drive_service.files().get_media(fileId=item['id'])
            fh = io.FileIO(item['name'], 'wb')
            downloader = googleapiclient.http.MediaIoBaseDownload(fh, request)
            done = False
            while done is False:
                status, done = downloader.next_chunk()
            fh.close()
            print("Download Complete") 

            # YouTubeへのアップロード
            print("Upload Start")
            filename, file_extension = os.path.splitext(item['name'])  # 拡張子を取り除く
            request = youtube_service.videos().insert(
                part="snippet,status",
                body={
                    "snippet": {
                        "categoryId": "20",  # 指定したカテゴリID
                        "description": "Video uploaded from Google Drive",  # ビデオの説明
                        "title": filename,  # ビデオのタイトル
                    },
                    "status": {
                        "privacyStatus": "unlisted"  # 限定公開に設定
                    }
                },
                media_body=googleapiclient.http.MediaFileUpload(item['name'])
            )
            response = request.execute()
            print("Upload Complete")
            print(response)

            video_id = response['id']  # アップロードしたビデオのIDを取得
            print(f'Uploaded video ID: {video_id}')
    
            # アップロードした動画の情報を更新
            request = youtube_service.videos().update(
                part="status",
                body={
                    "id": video_id,  # 更新する動画のID
                    "status": {
                        "selfDeclaredMadeForKids": True,  # 自己申告した子供向け動画であるかどうか
                        "madeForKids": True,  # 子供向けに設定
                    }
                }
            )
            response = request.execute()
            print(f"Video Updated: {video_id}")

            # 一時ファイルを削除
            os.remove(item['name'])

def main():
    folder_id = '1T0r-OsJ4ZE_WdZaLZLfodqrzZRIG2lPd'  # ここにGoogle DriveのフォルダIDを指定
    creds = Credentials.from_authorized_user_file('token.json')
    drive_service = googleapiclient.discovery.build('drive', 'v3', credentials=creds)
    youtube_service = googleapiclient.discovery.build('youtube', 'v3', credentials=creds)
    upload_videos(drive_service, youtube_service, folder_id)

if __name__ == "__main__":
    main()

コードの説明が終わったので変数について説明していきましょう。

• folder_id
folder_id とは、Google Drive 上の特定のフォルダの ID を指す変数です。この ID を使用して、上記スクリプトはそのフォルダ内のすべてのファイルとサブフォルダにアクセスできます。具体例として、以下のように Google Drive にフォルダがあるとします。この時、temp フォルダの URL は以下の通りです（少し編集してあります）。この URL の末尾 1Acrab808xEwyUyntnfEJNhb76HVv7 が folder_id になります。
  https://drive.google.com/drive/u/1/folders/1Acrab808xEwyUyntnfEJNhb76HVv7
  
• categoryId
categoryId とは、YouTube のビデオカテゴリを指定するための ID です。例えば、音楽ビデオであればカテゴリ ID は 10、科学や技術のビデオであればカテゴリ ID は 28 となります。以下にカテゴリ ID のリストを載せておきます。参考にしてみて下さい。私の場合ゲームの動画をアップロードすることが目的なのでカテゴリ ID は 20 になります。

カテゴリ ID のリスト

カテゴリ ID のリスト

'1' : 映画とアニメ
'2' : 自動車と乗り物
'10' : 音楽
'15' : ペットと動物
'17' : スポーツ
'18' : 短編映画
'19' : 旅行とイベント
'20' : ゲーム
'21' : ビデオブログ
'22' : 人々とブログ
'23' : コメディ
'24' : エンターテイメント
'25' : ニュースと政治
'26' : ハウツーとスタイル
'27' : 教育
'28' : 科学とテクノロジー
'29' : 非営利活動と社会運動
'30' : 映画
'31' : アニメ/アニメーション
'32' : アクション/アドベンチャー
'33' : クラシック
'34' : コメディ
'35' : ドキュメンタリー
'36' : ドラマ
'37' : ファミリー
'38' : 外国映画
'39' : ホラー
'40' : SF/ファンタジー
'41' : スリラー
'42' : 短編映画
'43' : ショー
'44' : 予告編

• description
description とは、 YouTube にアップロードされるビデオの説明を指定するための変数です。ビデオに関する詳細な情報や、他のビデオへのリンクなどを含めることができます。上記のスクリプトでは Video uploaded from Google Drive と設定しています

• title
title とは、YouTube にアップロードされるビデオのタイトルを指定するための変数です。ビデオの内容を示す簡潔で分かりやすいタイトルを設定することが推奨されます。上記のスクリプトではタイトルとして、拡張子を除いたファイル名にしています。

• madeForKids
madeForKids とは、ビデオが子供向けに作成されたものかどうかを指定するための変数です。ビデオが子供向けである場合、YouTube は特定の機能を制限します（例えば、コメントの投稿やプッシュ通知の送信など）。

• privacyStatus
privacyStatus とは、YouTube にアップロードされるビデオのプライバシー設定を指定するための変数です。設定可能な値は public（公開）、private（非公開）、および unlisted（限定公開）の3つです。公開設定ではビデオがYouTubeの検索結果、ホームページ、チャンネルページなどで表示されます。非公開設定では、ビデオのURLを知っているユーザーだけがビデオを見ることができます。限定公開設定では、ビデオのURLを知っているユーザーだけがビデオを見ることができ、YouTubeの検索結果、ホームページ、チャンネルページ、おすすめビデオには表示されません。

ソースコードの説明が終わったので実行してみましょう。実行の様子は以下の通りです。

実行すると以下のようなエラーが出てくるかと思います。

error

  File "/home/web/youtube/./main.py", line 46, in upload_videos
    response = request.execute()
               ^^^^^^^^^^^^^^^^^
  File "/home/web/.local/lib/python3.11/site-packages/googleapiclient/_helpers.py", line 130, in positional_wrapper
    return wrapped(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/web/.local/lib/python3.11/site-packages/googleapiclient/http.py", line 938, in execute
    raise HttpError(resp, content, uri=self.uri)
googleapiclient.errors.HttpError: <HttpError 403 when requesting https://youtube.googleapis.com/upload/youtube/v3/videos?part=snippet%2Cstatus&alt=json&uploadType=multipart returned "The request cannot be completed because you have exceeded your <a href="/youtube/v3/getting-started#quota">quota</a>.". Details: "[{'message': 'The request cannot be completed because you have exceeded your <a href="/youtube/v3/getting-started#quota">quota</a>.', 'domain': 'youtube.quota', 'reason': 'quotaExceeded'}]">

このエラーメッセージは、YouTube API のクォータ（1日あたりのリクエスト数やアップロードサイズなどの制限）を超えてしまったことを示しています。冒頭でも説明しましたが、YouTube API のクォータ制限は 1 日当たり 10,000 単位となっています。ビデオのアップロードは、1,600 単位とかなり高いクォータを消費します。したがって、1 日に行えるアップロードは数回に限られます。今回アップロードできた動画は 6 本なので 1,600×6=9,600 となり、クォータ制限が 10,000 なのでアップロードできる動画は 1 日当たり 6 本までということになりますね。クォータ制限がリセットされるのは、太平洋時間（PT）の午前 0 時にリセットされます。日本時間では、午後 4 時～5 時の間です。
クォータ制限に引っかかってしまい、Google Drive の動画を全て YouTube にアップロードできないじゃないかという話になってくるのですが、応用実装としてクォータ制限を考慮したソースコードについて説明していきます。

応用実装

アップロードした動画をリストに保存し、次回のアップロード時に対象外とする

以下のソースコードは、アップロードした動画をリストに保存しておき、次回のアップロード時にはその動画をアップロード対象外とするように改良したコードになります。リストのファイル名は uploaded_videos.jsonになります。これで、大量に動画があっても1日おきに以下のコードを実行することで地道ではありますが（1 日当たり 6 本しかアップロードできませんからね…）、Google Driveの動画を自動で YouTube にアップロードすることができるようになります。

upload_list_create.py.py

#!/usr/bin/env python3
import os
import io
import json
import googleapiclient.discovery
import googleapiclient.errors
import googleapiclient.http
from google.oauth2.credentials import Credentials

# Google Driveから動画を取得し、YouTubeにアップロードする関数
def upload_videos(drive_service, youtube_service, folder_id):
    print("Start")

    # 既にアップロードした動画のリストの取得
    try:
        with open('uploaded_videos.json', 'r') as f:
            uploaded_videos = json.load(f)
    except FileNotFoundError:
        uploaded_videos = []

    # Google Driveから動画ファイルを取得
    query = f"'{folder_id}' in parents"
    results = drive_service.files().list(q=query, fields="files(id, name, mimeType)").execute()
    items = results.get('files', [])

    for item in items:
        if item['id'] in uploaded_videos:
            print(f"Skipping {item['name']} (already uploaded)")
            continue

        if item['mimeType'].startswith('video'):  # 動画ファイルのみを対象とする
            # ファイルを一時的にダウンロード
            request = drive_service.files().get_media(fileId=item['id'])
            fh = io.FileIO(item['name'], 'wb')
            downloader = googleapiclient.http.MediaIoBaseDownload(fh, request)
            done = False
            while done is False:
                status, done = downloader.next_chunk()
            fh.close()
            print("Download Complete") 

            print("Upload Start")
            # YouTubeへのアップロード
            filename, file_extension = os.path.splitext(item['name'])  # 拡張子を取り除く
            request = youtube_service.videos().insert(
                part="snippet,status",
                body={
                    "snippet": {
                        "categoryId": "20",  # 指定したカテゴリID
                        "description": "Video uploaded from Google Drive",  # ビデオの説明
                        "title": filename,  # 拡張子を取り除いたビデオのタイトル
                    },
                    "status": {
                        "privacyStatus": "unlisted"   # 限定公開に設定
                    }
                },
                media_body=googleapiclient.http.MediaFileUpload(item['name'])
            )
            response = request.execute()
            print("Upload Complete")
            print(response)

            video_id = response['id']  # アップロードしたビデオのIDを取得
            print(f'Uploaded video ID: {video_id}')
    
            # アップロードした動画の情報を更新
            request = youtube_service.videos().update(
                part="status",
                body={
                    "id": video_id,  # 更新する動画のID
                    "status": {
                        "selfDeclaredMadeForKids": True,  # 自己申告した子供向け動画であるかどうか
                        "madeForKids": True,  # 子供向けに設定
                    }
                }
            )
            response = request.execute()
            print(f"Video Updated: {video_id}")

            # アップロードした動画をリストに保存
            uploaded_videos.append(item['id'])
            with open('uploaded_videos.json', 'w') as f:
                json.dump(uploaded_videos, f)

            # 一時ファイルを削除
            os.remove(item['name'])

def main():
    folder_id = '1T0r-OsJ4ZE_WdZaLZLfodqrzZRIG2lPd'  # ここにGoogle DriveのフォルダIDを指定
    creds = Credentials.from_authorized_user_file('token.json')
    drive_service = googleapiclient.discovery.build('drive', 'v3', credentials=creds)
    youtube_service = googleapiclient.discovery.build('youtube', 'v3', credentials=creds)
    upload_videos(drive_service, youtube_service, folder_id)

if __name__ == "__main__":
    main()

実行の様子は、次の「再生リストを作成し、アップロードした動画を再生リストに自動で追加する」を参照下さい。

再生リストを作成し、アップロードした動画を再生リストに自動で追加する

以下のソースコードは、Google Drive のフォルダ名を再生リストの名前とし、再生リストを自動で作成して、フォルダ内の動画をアップロードした時に、自動で作成した再生リストに追加するように改良したコードになります。こちらでも既にアップロードした動画はアップロードの対象外とするようにしています。

make_playlist.py

#!/usr/bin/env python3
import os
import io
import json
import googleapiclient.discovery
import googleapiclient.errors
import googleapiclient.http
from google.oauth2.credentials import Credentials

# プレイリストを探し、無ければ新たに作成する
def get_or_create_playlist(youtube_service, title, description=None):
    request = youtube_service.playlists().list(
        part="snippet", 
        maxResults=50, 
        mine=True
    )

    while request is not None:
        response = request.execute()
        for item in response["items"]:
            if item["snippet"]["title"] == title:
                # 既存のプレイリストを見つけたので、そのIDを返す
                return item["id"]

        # 次のページのリクエストを作成する
        request = youtube_service.playlists().list_next(request, response)

    return create_playlist(youtube_service, title, description)

# プレイリストを作成する関数
def create_playlist(youtube_service, title, description=None):
    # YouTube にプレイリストを作成するリクエストを送る
    playlists_insert_response = youtube_service.playlists().insert(
        part="snippet,status", 
        body=dict(
            snippet=dict(
                title=title,  
                description=description
            ),
            status=dict(
                privacyStatus="unlisted"
            )
        )
    ).execute()

    # 作成したプレイリストのIDを返します
    return playlists_insert_response["id"]

# 動画をプレイリストに追加する関数
def add_video_to_playlist(youtube_service, playlist_id, video_id):
    # YouTube に動画をプレイリストに追加するリクエストを送る
    youtube_service.playlistItems().insert(
        part="snippet",
        body={
            'snippet': {
                'playlistId': playlist_id,
                'resourceId': {
                    'kind': 'youtube#video',
                    'videoId': video_id
                }
            }
        }
    ).execute()

# Google Drive から動画を取得し、YouTube にアップロードする関数
def upload_videos(drive_service, youtube_service, folder_id):
    print("Start")

    # 既にアップロードした動画のリストの取得
    try:
        with open('uploaded_videos.json', 'r') as f:
            uploaded_videos = json.load(f)
    except FileNotFoundError:
        uploaded_videos = []

    # Google Drive から動画ファイルを取得して YouTube にアップロードし、作成した再生リストに追加する
    query = f"'{folder_id}' in parents"
    results = drive_service.files().list(q=query, fields="files(id, name, mimeType)").execute()
    items = results.get('files', [])

    # Google Drive のフォルダパス名を取得
    folder = drive_service.files().get(fileId=folder_id).execute()
    folder_name = folder["name"]
    playlist_id = get_or_create_playlist(youtube_service, folder_name, "This is a playlist created from a Google Drive folder.")

    for item in items:
        if item['id'] in uploaded_videos:
            print(f"Skipping {item['name']} (already uploaded)")
            continue

        if item['mimeType'].startswith('video'):  # 動画ファイルのみを対象とする
            # ファイルを一時的にダウンロード
            request = drive_service.files().get_media(fileId=item['id'])
            fh = io.FileIO(item['name'], 'wb')
            downloader = googleapiclient.http.MediaIoBaseDownload(fh, request)
            done = False
            while done is False:
                status, done = downloader.next_chunk()
            fh.close()
            print("Download Complete") 

            # YouTubeへのアップロード
            print("Upload Start")
            filename, file_extension = os.path.splitext(item['name'])  # 拡張子を取り除く
            request = youtube_service.videos().insert(
                part="snippet,status",
                body={
                    "snippet": {
                        "categoryId": "20",  # 指定したカテゴリID
                        "description": "Video uploaded from Google Drive",  # ビデオの説明
                        "title": filename,  # ビデオのタイトル
                    },
                    "status": {
                        "privacyStatus": "unlisted"  # 限定公開に設定
                    }
                },
                media_body=googleapiclient.http.MediaFileUpload(item['name'])
            )
            response = request.execute()
            print("Upload Complete")
            print(response)

            video_id = response['id']  # アップロードしたビデオのIDを取得
            print(f'Uploaded video ID: {video_id}')
    
            # アップロードした動画の情報を更新
            request = youtube_service.videos().update(
                part="status",
                body={
                    "id": video_id,  # 更新する動画のID
                    "status": {
                        "selfDeclaredMadeForKids": True,  # 自己申告した子供向け動画であるかどうか
                        "madeForKids": True,  # 子供向けに設定
                    }
                }
            )
            response = request.execute()
            print(f"Video Updated: {video_id}")

            # アップロードした動画をリストに保存
            uploaded_videos.append(item['id'])
            with open('uploaded_videos.json', 'w') as f:
                json.dump(uploaded_videos, f)

            # プレイリスト作成と動画の追加
            add_video_to_playlist(youtube_service, playlist_id, video_id)

            # 一時ファイルの削除
            os.remove(item['name'])

def main():
    folder_id = '1T0r-OsJ4ZE_WdZaLZLfodqrzZRIG2lPd'  # ここにGoogle DriveのフォルダIDを指定
    creds = Credentials.from_authorized_user_file('token.json')
    drive_service = googleapiclient.discovery.build('drive', 'v3', credentials=creds)
    youtube_service = googleapiclient.discovery.build('youtube', 'v3', credentials=creds)
    upload_videos(drive_service, youtube_service, folder_id)

if __name__ == "__main__":
    main()

上記のコードでは、request = youtube_service.playlists().list にて再生リストを取得していますが、一度に取得できる再生リストの最大数が 50 個という制約があります。そのため、request = youtube_service.playlists().list_next(request, response) で次のページの再生リストを取得していますが、私の環境では再生リストを 50 個以上も作っていないので、実証はしていません。ご容赦下さい。
続いて、再生リストの変数について説明していきます。

• privacyStatus（def create_playlist 関数中）
  privacyStatus とは、YouTube で作成したプレイリストのプライバシー設定を指定するための変数です。設定可能な値は public（公開）、unlisted（限定公開）、および private（非公開）の3つです。公開設定ではプレイリストは一般公開され、誰でも検索したり視聴したりすることができます。再生リストのタイトルと説明も公開され、再生リストのビデオは他のユーザーに表示されます。限定公開設定では、プレイリストは公開されませんが、リンクを知っている人なら誰でも視聴することができます。検索結果には表示されず、チャンネルページでも表示されません。しかし、再生リストのURLが共有されていれば、そのURLを持っている人は再生リストにアクセスし視聴することができます。非公開設定では、プレイリストは完全に非公開で、作成者本人しか視聴することができません。他のユーザーには全く表示されず、URLを知っている人でも視聴することはできません。

実行の様子は以下の通りです。

クォータ制限解除 一括アップロード

以下のソースコードは、クォータ制限が解除され 1 日当たりのクォータが 10,000 単位から 1,000,000 単位まで増加したことを想定したコードになります。つまり、1 日当たりのアップロードできる動画の数は 1000000 ÷ 1600 = 625 となります。そこで、Google Drive にある動画をフォルダ単位の ID を指定するのではなく、親ディレクトリに存在する全てのフォルダに対して、フォルダ内の動画をアップロードし、自動で作成した再生リストに追加するようにしました。
この時、folder_id は親ディレクトリのフォルダIDを指定して下さい。

all_upload.py

#!/usr/bin/env python3
import os
import io
import json
import googleapiclient.discovery
import googleapiclient.errors
import googleapiclient.http
from google.oauth2.credentials import Credentials

# プレイリストを探し、無ければ新たに作成する
def get_or_create_playlist(youtube_service, title, description=None):
    request = youtube_service.playlists().list(
        part="snippet", 
        maxResults=50, 
        mine=True
    )

    while request is not None:
        response = request.execute()
        for item in response["items"]:
            if item["snippet"]["title"] == title:
                # 既存のプレイリストを見つけたので、そのIDを返す
                return item["id"]

        # 次のページのリクエストを作成する
        request = youtube_service.playlists().list_next(request, response)

    return create_playlist(youtube_service, title, description)

# プレイリストを作成する関数
def create_playlist(youtube_service, title, description=None):
    # YouTube にプレイリストを作成するリクエストを送る
    playlists_insert_response = youtube_service.playlists().insert(
        part="snippet,status", 
        body=dict(
            snippet=dict(
                title=title,  
                description=description
            ),
            status=dict(
                privacyStatus="unlisted"
            )
        )
    ).execute()

    # 作成したプレイリストのIDを返します
    return playlists_insert_response["id"]

# 動画をプレイリストに追加する関数
def add_video_to_playlist(youtube_service, playlist_id, video_id):
    # YouTubeに動画をプレイリストに追加するリクエストを送る
    youtube_service.playlistItems().insert(
        part="snippet",
        body={
            'snippet': {
                'playlistId': playlist_id,
                'resourceId': {
                    'kind': 'youtube#video',
                    'videoId': video_id
                }
            }
        }
    ).execute()

# Google Driveから動画を取得し、YouTubeにアップロードする関数
def upload_videos(drive_service, youtube_service, folder_id):
    print("Start")

    # 既にアップロードした動画のリストの取得
    try:
        with open('uploaded_videos.json', 'r') as f:
            uploaded_videos = json.load(f)
    except FileNotFoundError:
        uploaded_videos = []

    # Google Driveから動画ファイルを取得してYouTubeにアップロードし、作成した再生リストに追加する
    query = f"'{folder_id}' in parents"
    results = drive_service.files().list(q=query, fields="files(id, name, mimeType)").execute()
    items = results.get('files', [])


    for item in items:
        if item['mimeType'] == 'application/vnd.google-apps.folder':  # フォルダの場合
            # Google Driveから動画ファイルを取得してYouTubeにアップロードし、作成した再生リストに追加する
            query = f"'{item['id']}' in parents"
            results = drive_service.files().list(q=query, fields="files(id, name, mimeType)").execute()
            videos = results.get('files', [])
        
            # Google Drive のフォルダパス名を取得
            folder_name = item["name"]
            playlist_id = get_or_create_playlist(youtube_service, folder_name, "This is a playlist created from a Google Drive folder.")

            for video in videos:
                if video['id'] in uploaded_videos:
                    print(f"Skipping {video['name']} (already uploaded)")
                    continue

                if video['mimeType'].startswith('video'):  # 動画ファイルのみを対象とする
                    # ファイルを一時的にダウンロード
                    request = drive_service.files().get_media(fileId=video['id'])
                    fh = io.FileIO(video['name'], 'wb')
                    downloader = googleapiclient.http.MediaIoBaseDownload(fh, request)
                    done = False
                    while done is False:
                        status, done = downloader.next_chunk()
                    fh.close()
                    print("Download Complete") 

                    # YouTubeへのアップロード
                    print("Upload Start")
                    filename, file_extension = os.path.splitext(video['name'])  # 拡張子を取り除く
                    request = youtube_service.videos().insert(
                        part="snippet,status",
                        body={
                            "snippet": {
                                "categoryId": "20",  # 指定したカテゴリID
                                "description": "Video uploaded from Google Drive",  # ビデオの説明
                                "title": filename  # ビデオのタイトル
                            },
                            "status": {
                                "privacyStatus": "unlisted"  # 限定公開に設定
                            }
                        },
                        media_body=googleapiclient.http.MediaFileUpload(video['name'])
                    )
                    response = request.execute()
                    print("Upload Complete")
                    print(response)

                    video_id = response['id']  # アップロードしたビデオのIDを取得
                    print(f'Uploaded video ID: {video_id}')
    
                    # アップロードした動画の情報を更新
                    request = youtube_service.videos().update(
                        part="status",
                        body={
                            "id": video_id,  # 更新する動画のID
                            "status": {
                                "selfDeclaredMadeForKids": True,  # 自己申告した子供向け動画であるかどうか
                                "madeForKids": True,  # 子供向けに設定
                            }
                        }
                    )
                    response = request.execute()
                    print(f"Video Updated: {video_id}")

                    # アップロードした動画をリストに保存
                    uploaded_videos.append(video['id'])
                    with open('uploaded_videos.json', 'w') as f:
                        json.dump(uploaded_videos, f)

                    # プレイリスト作成と動画の追加
                    add_video_to_playlist(youtube_service, playlist_id, video_id)

                    # 一時ファイルの削除
                    os.remove(video['name'])

def main():
    folder_id = '12HG54TKzrwdQhaw7nJhx9hPL9RrKYwtJ'  # Google Driveの親フォルダのIDを指定
    creds = Credentials.from_authorized_user_file('token.json')
    drive_service = googleapiclient.discovery.build('drive', 'v3', credentials=creds)
    youtube_service = googleapiclient.discovery.build('youtube', 'v3', credentials=creds)
    upload_videos(drive_service, youtube_service, folder_id)

if __name__ == "__main__":
    main()

実行の様子は以下の通りです。

やったー、これで Google Drive の動画全部 YouTube にアップロードできるぞ、と思ったら思わぬ落とし罠がありました。そう、YouTube には 24 時間当たりのアップロード制限があるのです。制限については公式から明示されていませんが、おそらく動画の本数ではなく、総アップロード容量だと思います。89 本アップロードできたり、100 本弱アップロードできたりしました（検証はめんどくさいのでしません）。この制限に引っかかると以下のようなエラーが発生します。

error

  File "/home/web/.local/lib/python3.11/site-packages/googleapiclient/_helpers.py", line 130, in positional_wrapper
    return wrapped(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/web/.local/lib/python3.11/site-packages/googleapiclient/http.py", line 938, in execute
    raise HttpError(resp, content, uri=self.uri)
googleapiclient.errors.HttpError: <HttpError 400 when requesting https://youtube.googleapis.com/upload/youtube/v3/videos?part=snippet%2Cstatus&alt=json&uploadType=multipart returned "The user has exceeded the number of videos they may upload.". Details: "[{'message': 'The user has exceeded the number of videos they may upload.', 'domain': 'youtube.video', 'reason': 'uploadLimitExceeded'}]">

'youtube.video', 'reason': 'uploadLimitExceeded' とエラーが出ていますね。この制限が解除されるのは 24 時間後になります。

注意点

2023/08/21追記
ラズパイのメモリの容量が少ないと、大きなサイズの動画（大体1GB以上）をアップロードする時に、ジョブが停止してしまいます。回避策として分割アップロードというものがあるらしいのでそちらを利用しました。分割アップロードといっても、アップロードする時に分割しながらアップロードするだけであって、最終的には一つの動画としてアップロードされるのでご安心下さい。以下が修正前のコードと修正後のコードになります。

before.py

# YouTubeへのアップロード
print("Upload Start")
filename, file_extension = os.path.splitext(video['name'])  # 拡張子を取り除く
request = youtube_service.videos().insert(
    part="snippet,status",
    body={
        "snippet": {
            "categoryId": "20",  # 指定したカテゴリID
            "description": "Video uploaded from Google Drive",  # ビデオの説明
            "title": filename  # ビデオのタイトル
        },
        "status": {
            "privacyStatus": "unlisted"  # 限定公開に設定
        }
    },
    media_body=googleapiclient.http.MediaFileUpload(video['name'])
)
response = request.execute()
print("Upload Complete")
print(response)

after.py

# YouTubeへのアップロード
print("Upload Start")
filename, file_extension = os.path.splitext(video['name'])  # 拡張子を取り除く

media = googleapiclient.http.MediaFileUpload(video['name'], chunksize=-1, resumable=True)
request = youtube_service.videos().insert(
    part="snippet,status",
    body={
        "snippet": {
            "categoryId": "20",  # 指定したカテゴリID
            "description": "Video uploaded from Google Drive",  # ビデオの説明
            "title": filename  # ビデオのタイトル
        },
        "status": {
            "privacyStatus": "unlisted"  # 限定公開に設定
        }
    },
    media_body=media
)

response = None
while response is None:
    status, response = request.next_chunk()
    if status:
        print("Uploaded %d%%." % int(status.progress() * 100))

print("Upload Complete")
print(response)

クォータの消費量

最後にスクリプトの各 API 呼び出し時のクォータ消費量についてまとめます。

• Google Drive API:

  • ファイルのリスト取得（drive_service.files().list(q=query, fields="files(id, name, mimeType)").execute()）：リクエスト 1 回あたりのクォータ消費量は 10 です。（実際の数値は公開されていませんが、これは一般的な推定値です）
  • ファイルのダウンロード（drive_service.files().get_media(fileId=video['id'])）：リクエスト1 回あたりのクォータ消費量は 10 です。（これも一般的な推定値です）

• YouTube Data API v3:

  • プレイリストのリスト取得（youtube_service.playlists().list(part="snippet", maxResults=50, mine=True).execute()）：これは読み取り操作で、1 回のリクエストあたり 1 単位のクォータを消費します。
  • 新しいプレイリストの作成（youtube_service.playlists().insert(part="snippet,status", body=body_dict).execute()）：これは書き込み操作で、1 回のリクエストあたり 50 単位のクォータを消費します。
  • 動画のアップロード（youtube_service.videos().insert(part="snippet,status", body=body_dict, media_body=media_body).execute()）： これは書き込み操作で、1 回のリクエストあたり 1600 単位のクォータを消費します。
  • 動画の情報の更新（youtube_service.videos().update(part="status", body=body_dict).execute()）：これは書き込み操作で、1 回のリクエストあたり 50 単位のクォータを消費します。
  • プレイリストにアイテムを追加（youtube_service.playlistItems().insert(part="snippet", body=body_dict).execute()）：これは書き込み操作で、1 回のリクエストあたり 50 単位のクォータを消費します。

最後に

需要があるのかは分かりません（おそらくないでしょう）。YouTube Data API を利用して Google Drive の動画を自動でアップロードできたら面白いし、楽だなと思い実装してみました。しかし、クォータ制限があることを実装当初は知らず、全部自動でアップロードできるだろと、楽観的でいました（現実はそう甘くないですね…）。クォータ申請については、ChatGPT を用いて適当に生成した内容で申請したため、それが原因で審査が遅くなった可能性があります。おそらくきちんとクォータ申請を行えばすぐ審査が通るのではないでしょうか。申請方法については以下記事にまとめました。