注意

　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)

YouTube Live チャット欄をAPIでリアルタイム取得

注意

YouTube APIを使い、リアルタイムで取得

概要

流れ

ソースコード