注意
YouTubeの利用規約
以下のコードは自己責任で使用すること。
YouTube APIを使い、リアルタイムで取得
概要
最近YouTube チャット欄でのなりすましや、スパチャした方の一欄に漏れがあるというような話があります。
その解決法の1つとして、YouTube APIを使用したチャット欄取得の方法を以下に書きました。
出力
[by ユーザー名1 チャンネルURL 1]
コメント1
[by ユーザー名2 チャンネルURL 2]
コメント2
[by ユーザー名3 チャンネルURL 3]
金額 from ユーザー名3: "コメント3" ← スパチャはこう見える
start : 2020-06-07T05:21:49.871000Z ← 取得したチャット群の1個目の時間 日本時間なら＋9時間
end : 2020-06-07T05:22:02.846000Z ← " 最後 の時間
- 公式のAPI
- YouTube APIキーの取得 (2020/03/25時点)にしたがって取得
- quotaと呼ばれるものを消費する。上限10000 quota/dayで、それ以上はincrement申請必須。
- quota calculatorに以下のコードで使っているAPIのquota消費量がないので、消費量がどのくらいか不明。
-
チャットID取得に 2 quota、チャット取得に 3.5 quota ?15秒ごとのチャット取得を240回（main関数内の変数slp_time = 15,iter_times = 240）してみたところ、消費量は842だった。
- 2020年7月末にquota消費量に変更がありました。現在はチャットID取得に 1 quota、チャット取得に 5 quota？
- quota消費量の確認は、
Google Cloud Platform > APIキー取得でつくったプロジェクト > 左上の「三」みたいなやつ > IAMと管理 > 割り当て > サービスをYouTube Data API v3に絞り込み
- 放送前後も取得可能
- 取得頻度が低すぎると取得漏れあり？
- 下のはスパチャを含む全チャットの記録をするコードになっている。用途に合わせて書き換えること。
- スパチャは
msg = '金額 from ユーザ名: "メッセージ"' - スパチャについては
supChat、ステッカーについてはsupSticに入る。#を消して、必要なとこだけ引っこ抜けば良い。どんな中身かはこちら
- スパチャは
流れ
- YouTube APIキーの取得 (2020/03/25時点)にしたがってAPI KEY取得
- YouTube Live のURLからvideo ID 取得
-
https://www.youtube.com/watch?v=***********の***********のこと
-
- video IDからchat ID取得
- このAPIを使う
{'key': YT_API_KEY, 'id': video_id, 'part': 'liveStreamingDetails'}
- chat IDを元に、チャット欄を繰り返し取得
- このAPIを使う
-
{'key': YT_API_KEY, 'liveChatId': chat_id, 'part': 'id,snippet,authorDetails', 'pageToken': pageToken}- 初回は
pageTokenをNoneにしておく - 2回目以降は、
pageTokenに前回のnextPageTokenを指定- これで差分が取得できる
- 初回は
- 時間指定してループするもよし、エラー吐くまで（ctrl+Cするまで）
while Trueするもよし
ソースコード
record_chat.py
import time
import requests
import json
#事前に取得したYouTube API key
YT_API_KEY = '***************************************'
def get_chat_id(yt_url):
'''
https://developers.google.com/youtube/v3/docs/videos/list?hl=ja
'''
video_id = yt_url.replace('https://www.youtube.com/watch?v=', '')
print('video_id : ', video_id)
url = 'https://www.googleapis.com/youtube/v3/videos'
params = {'key': YT_API_KEY, 'id': video_id, 'part': 'liveStreamingDetails'}
data = requests.get(url, params=params).json()
liveStreamingDetails = data['items'][0]['liveStreamingDetails']
if 'activeLiveChatId' in liveStreamingDetails.keys():
chat_id = liveStreamingDetails['activeLiveChatId']
print('get_chat_id done!')
else:
chat_id = None
print('NOT live')
return chat_id
def get_chat(chat_id, pageToken, log_file):
'''
https://developers.google.com/youtube/v3/live/docs/liveChatMessages/list
'''
url = 'https://www.googleapis.com/youtube/v3/liveChat/messages'
params = {'key': YT_API_KEY, 'liveChatId': chat_id, 'part': 'id,snippet,authorDetails'}
if type(pageToken) == str:
params['pageToken'] = pageToken
data = requests.get(url, params=params).json()
try:
for item in data['items']:
channelId = item['snippet']['authorChannelId']
msg = item['snippet']['displayMessage']
usr = item['authorDetails']['displayName']
#supChat = item['snippet']['superChatDetails']
#supStic = item['snippet']['superStickerDetails']
log_text = '[by {} https://www.youtube.com/channel/{}]\n {}'.format(usr, channelId, msg)
with open(log_file, 'a') as f:
print(log_text, file=f)
print(log_text)
print('start : ', data['items'][0]['snippet']['publishedAt'])
print('end : ', data['items'][-1]['snippet']['publishedAt'])
except:
pass
return data['nextPageToken']
def main(yt_url):
slp_time = 10 #sec
iter_times = 90 #回
take_time = slp_time / 60 * iter_times
print('{}分後 終了予定'.format(take_time))
print('work on {}'.format(yt_url))
log_file = yt_url.replace('https://www.youtube.com/watch?v=', '') + '.txt'
with open(log_file, 'a') as f:
print('{} のチャット欄を記録します。'.format(yt_url), file=f)
chat_id = get_chat_id(yt_url)
nextPageToken = None
for ii in range(iter_times):
#for jj in [0]:
try:
print('\n')
nextPageToken = get_chat(chat_id, nextPageToken, log_file)
time.sleep(slp_time)
except:
break
if __name__ == '__main__':
yt_url = input('Input YouTube URL > ')
main(yt_url)