YouTube API: 再生時間が条件に合う動画を選んで再生リストを作成（1/2）
YouTubeで再生時間の条件で動画の再生リストを作りたくなることありませんか？今回YouTube Data APIを使って再生時間が10分以上、15分以下の動画のみを含む再生リスト作成するPythonコードを書いたので共有します。長くなりそうなので、2回に分けて投稿します。
背景：YouTubeのフィルタが微妙
運動不足解消のため、毎朝妻と一緒にB-lifeというYouTubeのヨガレッスンやっています。B-lifeのチャンネルには質の高いレッスン動画が多くアップロードされていて嬉しいのですが、毎朝やっているとその日やる動画を選ぶのも結構迷います。このとき「今日は時間がないので5分のレッスンが良い」「今日は15分」というような状況が頻繁に発生するので、再生時間の条件で再生リストが作れると便利です。
ただ残念ながらYouTubeの検索フィルタでは細かい再生時間の絞り込みはできなさそうです。
YouTube Data APIを使ったらできるだろうということでやってみました。
実現までのステップ
YouTube Data APIを使った今回の実現ステップをまとめると次の3つです。
特定のチャンネル（B-life）の中にある動画ID一覧を取得する
動画の再生時間を取得し、再生時間の条件に合う動画を選ぶ
自分のYouTubeアカウント上に該当する動画の再生リストを作成する
1と2に関しては、認証も必要なくAPIキーさえ用意すれば気軽に行えます。3にはOAuth認証が必要になります。本投稿では、第1回として上記1と2の手順を一つひとつ見ていきたいと思います。3は次の投稿で説明します。
APIキーやOAuthクライアントキーについては以下を参考にしてください。
0. 準備
YouTubeチャンネルのIDを確認
APIで動画やチャンネルを指定するためには、それらのIDが必要です。準備として対象とするYouTubeチャンネルのIDを確認します。チャンネルIDはURLから直接分かります。YouTubeチャンネルのURLはhttps://www.youtube.com/channel/<チャンネルID>となっているので、ブラウザでアクセスして確認します。
ちなみにB-lifeのチャンネルIDはUCd0pUnH7i5CM-Y8xRe7cZVgでした。
B-lifeのYouTubeチャンネルのホーム画面
赤枠がチャンネルIDです。
なおB-lifeは初心者向けの動画も多く、Mariko先生のレクチャが大変分かりやすいのでおすすめです
ライブラリをインストール
pipでPythonのGoogle API Client、をインストールしておきましょう。あとpandasも使いますのでなければインストールしましょう。
pip install --upgrade google-api-python-client pip install pandas
1. チャンネルにある動画ID一覧を取得
さっそくYouTubeチャンネルの中にある動画ID一覧を取得していきます。
APIのビルドと初期化
まずAPIの初期化です。
Google APIを使うためには、APIキーが必要です。Google Cloud Consoleから作成して、下記のYOUTUBE_API_KEY = 'your-api-key-here'を作成したAPIキーに書き換えてください
from googleapiclient.discovery import build # 利用するAPIサービス YOUTUBE_API_SERVICE_NAME = 'youtube' YOUTUBE_API_VERSION = 'v3' # APIキー YOUTUBE_API_KEY = 'your-api-key-here' # API のビルドと初期化 youtube = build(YOUTUBE_API_SERVICE_NAME, YOUTUBE_API_VERSION, developerKey=YOUTUBE_API_KEY)
動画情報一覧を取得
次にsearch()APIを使って動画ID一覧を取得するのですが、それを関数get_video_list_in_channel()で定義しています。 get_video_list_in_channel()はチャンネルIDを入力とし、動画情報を出力します。仕様としては、指定したチャンネルIDの中で公開日時が新しいものから50件ずつ最大2回リクエストされます。なので最大50 * m_re_cntの動画情報が取得されます。
Google APIでは1日あたりのAPI使用量（クォータ）が限られているので、沢山リクエストすると一発で上限を超えてしまいます。今回実験ということで引数max_req_cnt=2と小さめに指定しています。必要に応じて調整してください。
from datetime import datetime, timedelta import pandas as pd def get_video_list_in_channel(youtube, channel_id, max_req_cnt=2): '''特定のチャンネルの動画情報を取得し、必要な動画情報を返す 公開時刻が新しい順に50ずつリクエスト デフォルトでは最大2リクエストで終了 ''' n_requested = 50 earliest_publishedtime =\ datetime.now(tz=timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ') req_cnt = 0 result = [] while True: response = youtube.search().list(part='snippet', channelId=channel_id, order='date', type='video', publishedBefore=earliest_publishedtime, maxResults=n_requested).execute() req_cnt += 1 video_info = fetch_video_info(response) result.append(video_info) # 取得動画の最も遅い公開日時の1秒前以前を次の動画一覧の取得条件とする last_publishedtime = video_info['publishTime'].min() last_publishedtime_next =\ datetime.strptime(last_publishedtime, '%Y-%m-%dT%H:%M:%SZ') - timedelta(seconds=1) earliest_publishedtime =\ last_publishedtime_next.strftime('%Y-%m-%dT%H:%M:%SZ') if req_cnt > max_req_cnt: # リクエスト回数がmax_req_cntを超えたらループを抜ける print('Result count exceeded max count {}.'.format(max_req_cnt)) break if len(response['items']) < n_requested: # リクエストした動画数より少ない数が返った場合はループを抜ける print('Number of results are less than {}.'.format(n_requested)) break if len(result) > 1: df_video_list = pd.concat(result, axis=0).reset_index(drop=True) else: df_video_list = result[0] return df_video_list def fetch_video_info(response, as_df=True): '''APIのレスポンスから必要な動画情報を抜き出す''' info_list = [] for item in response['items']: info = {} info['title'] = item['snippet']['title'] info['kind'] = item['id']['kind'] info['videoId'] = item['id']['videoId'] info['description'] = item['snippet']['description'] info['publishTime'] = item['snippet']['publishTime'] info['channelTitle'] = item['snippet']['channelTitle'] info['thumbnails_url'] = item['snippet']['thumbnails']['default']['url'] info_list.append(info) if as_df: return pd.DataFrame(info_list) else: info_list
実行部分のコードは以下です。df_video_listに取得した情報が格納されています。df_video_listのvideoId列が動画IDになります。動画ID一覧が取得できました。
# 動画一覧を取得したいチャンネルID channel_id = 'UCd0pUnH7i5CM-Y8xRe7cZVg' # 動画一覧を取得 df_video_list = get_video_list_in_channel(youtube, channel_id) # 出力 df_video_list.head()
2. 動画の再生時間を取得し、フィルタをかける
上で取得した動画情報一覧に動画の再生時間は含まれていませんので、別途videoIdをキーにしてAPI経由で取得する必要があります。再生時間を取得した後はpandasで条件に合ったものを絞り込むだけです。
再生時間を取得
再生時間を取得するにはvideoAPIに動画IDを渡して詳細情報を取得します。ここでは、下表のような機能のラッパー関数をいくつか定義しています。
| 関数 | 機能概要 |
|---|---|
| get_contents_detail(), get_contents_detail_core() | 動画の詳細情報取得 |
| get_duration(), get_basicinfo(), | レスポンスjsonのパース |
| pt2sec() | 再生時間のフォーマット変換 |
1回のリクエストで50動画IDまでしか問い合わせできないみたいなので、get_contents_detail()では動画ID一覧を50件ずつに分割してから処理しています。
import re import numpy as np def get_contents_detail_core(youtube, videoids): '''動画の詳細情報を取得''' part = ['snippet', 'contentDetails'] response = youtube.videos().list(part=part, id=videoids).execute() results = [] for item in response['items']: info = get_basicinfo(item) info['duration'] = get_duration(item) results.append(info) return pd.DataFrame(results) def get_contents_detail(youtube, videoids): '''必要に応じて50件ずつにIDを分割し、詳細情報を取得''' n_req_pre_once = 50 # IDの数が多い場合は50件ずつ動画IDのリストを作成 if len(videoids) > n_req_pre_once: videoids_list = np.array_split(videoids, len(videoids) // n_req_pre_once + 1) else: videoids_list = [videoids,] # 50件ずつ動画IDのリストを渡し、動画の詳細情報を取得 details_list = [] for vids in videoids_list: df_video_details_part =\ get_contents_detail_core(youtube, vids.tolist()) details_list.append(df_video_details_part) df_video_details =\ pd.concat(details_list, axis=0).reset_index(drop=True) return df_video_details def get_duration(item): '''動画時間を抜き出す（ISO表記を秒に変換）''' content_details = item['contentDetails'] pt_time = content_details['duration'] return pt2sec(pt_time) def get_basicinfo(item): '''動画の基本情報の抜き出し''' basicinfo = dict(id=item['id']) # snippets keys = ('title', 'description', 'channelTitle') snippets = {k: item['snippet'][k] for k in keys} basicinfo.update(snippets) return basicinfo def pt2sec(pt_time): '''ISO表記の動画時間を秒に変換 ''' pttn_time = re.compile(r'PT(\d+H)?(\d+M)?(\d+S)?') keys = ['hours', 'minutes', 'seconds'] m = pttn_time.search(pt_time) if m: kwargs = {k: 0 if v is None else int(v[:-1]) for k, v in zip(keys, m.groups())} return timedelta(**kwargs).total_seconds() else: msg = '{} is not valid ISO time format.'.format(pt_time) raise ValueError(msg)
再生時間取得の実行部分です。df_video_detailsが結果で、durationカラムに再生時間が入っています。再生時間の単位は秒です。
# 動画ID videoids = df_video_list['videoId'].values # 動画の詳細情報を取得 df_video_details = get_contents_detail(youtube, videoids) df_video_details.head()
再生時間でフィルタをかける
ここではpandas.between()を使って10分以上15分以下の動画に絞り込みこみました。
# 再生時間が10分以上、15分以下の動画に絞り込む lower_duration = 10 * 60 # 10分以上 upper_duration = 15 * 60 # 15分以下 is_matched = df_video_details['duration'].between(lower_duration, upper_duration) df_video_playlist = df_video_details.loc[is_matched, :] # 出力 df_video_playlist.head()
これで10分以上、15分以下の動画IDのリストができました。めでたし、めでした。次回はこれら動画を含む再生リストをAPI経由で生成したいと思います。 コードはGithubに挙げています。
久々にGoogle APIを触りました。サービスも年々増えているみたいで、やりさえすれば色々できそうです。やりさえすれば。。。日々精進
