PythonでYouTube Liveのアーカイブからチャット（コメント）を取得する
※こちらに修正して高速化した、コードの改訂版を載せています。基本的な構造は変わっていないので解説は本記事を参照してもらって大丈夫ですが、最終的なコードはリンク先を用いてください。
watagassy.hatenablog.com
YouTube Liveのチャット（コメント）は、ライブ中にリアルタイムで取得することはYouTube Data APIを用いることで可能なのですが、アーカイブから取得するという操作はAPIにはありません。
それでもチャットリプレイ機能で過去のコメントが表示されるので、クソ～～～どっかからどうにかして取得できるやろ～～～できんかな～～～と思っていたらこんな記事を見つけました。
この記事の最後の方にこんなことが書かれていました。
放送終了後のチャットを取得できるかについて
まさにやりたいことはここから..
APIで取得できない.. リプレイの場合, webで確認してみると get_live_chat_replay?continuation=xxxxxxxxxxx.jsonで一定間隔でチャットを取得してることがわかった.
ここでcontinuationを知る方法について調査なう. videoIdから一番初めのcontinuationの値を知ることができたらあとは芋づる式に取得することができる. get_live_chat_replay?continuation=xxxxxxxxxxx.jsonには次に取得するcontinuationの値が含まれているから...
これをヒントにあれこれやっていたらアーカイブからチャットを取得できたので、今回それについて書いていこうと思います。
実装はPython3.6で行っています。
ちなみに先に言っておきますと、この方式で全コメント取得するのはめちゃくちゃ遅いです。ご了承ください。
まずはソースコードは以下です。
from bs4 import BeautifulSoup from selenium import webdriver from selenium.webdriver.chrome.options import Options import requests target_url = youtube_url options = Options() options.add_argument('--headless') options.add_argument('--disable-gpu') options.add_argument('--disable-desktop-notifications') options.add_argument("--disable-extensions") options.add_argument('--blink-settings=imagesEnabled=false') dict_str = "" next_url = "" comment_data = [] # まず動画ページにrequestsを実行しhtmlソースを手に入れてlive_chat_replayの先頭のurlを入手 html = requests.get(target_url) soup = BeautifulSoup(html.text, "html.parser") for iframe in soup.find_all("iframe"): if("live_chat_replay" in iframe["src"]): next_url= iframe["src"] while(1): try: # chromedriverを開きnext_urlのソースを入手 driver = webdriver.Chrome(chrome_options=options, executable_path=r"C:\chromedriver_win32\chromedriver.exe") driver.get(next_url) soup = BeautifulSoup(driver.page_source,"lxml") # 必ずquitすること。忘れるとgoogle chromeが開かれまくって大変なことになる driver.quit() # 次に飛ぶurlのデータがある部分をfind_allで探してsplitで整形 for scrp in soup.find_all("script"): if "window[\"ytInitialData\"]" in scrp.text: dict_str = scrp.text.split(" = ")[1] # javascript表記なので更に整形. falseとtrueの表記を直す dict_str = dict_str.replace("false","False") dict_str = dict_str.replace("true","True") # 辞書形式と認識すると簡単にデータを取得できるが, 末尾に邪魔なのがあるので消しておく（「空白2つ + \n + ;」を消す） dict_str = dict_str.rstrip(" \n;") # 辞書形式に変換 dics = eval(dict_str) # "https://www.youtube.com/live_chat_replay?continuation=" + continue_url が次のlive_chat_replayのurl continue_url = dics["continuationContents"]["liveChatContinuation"]["continuations"][0]["liveChatReplayContinuationData"]["continuation"] next_url = "https://www.youtube.com/live_chat_replay?continuation=" + continue_url # dics["continuationContents"]["liveChatContinuation"]["actions"]がコメントデータのリスト。先頭はノイズデータなので[1:]で保存 for samp in dics["continuationContents"]["liveChatContinuation"]["actions"][1:]: comment_data.append(str(samp)+"\n") # next_urlが入手できなくなったら終わり except: break # comment_data.txt にコメントデータを書き込む with open("comment_data.txt", mode='w', encoding="utf-8") as f: f.writelines(comment_data)
「youtube_url」にチャットを取得したい動画のURL
（"https://www.youtube.com/watch?v=xxxxxxxx"の形式）を入れてこのプログラムを実行すると、comment_data.txtというファイルに1行ずつチャットに関するデータが保存されます。
先にこのプログラムが何をやっているか大雑把に説明すると、まずrequestsを用いて、youtube_urlの動画のhtmlからチャットのデータが存在するURL
（"https://www.youtube.com/live_chat_replay?continuation=xxxxxxxx"という名前になっています）を探して飛びます。
そこでチャットがjavascriptで動的に生成されているので、seleniumを用いてスクレイピングします。
すると先頭40件くらいのチャットデータと次のチャットデータが存在するURLが取得できるので、そこに飛んで同様の操作をします。
これを繰り返してtxtファイルにチャットデータを保存する、というのが一連の流れです。
説明が必要そうな部分に分けて先頭から解説していきます。
from bs4 import BeautifulSoup from selenium import webdriver from selenium.webdriver.chrome.options import Options import requests target_url = youtube_url options = Options() options.add_argument('--headless') options.add_argument('--disable-gpu') options.add_argument('--disable-desktop-notifications') options.add_argument("--disable-extensions") options.add_argument('--blink-settings=imagesEnabled=false')
今回スクレイピングでデータを取得するので、BeautifulSoupとselenium、requestsを用います。導入していなければインストールしておいてください。
また今回seleniumでスクレイピングする際にchromedriverを用いるので、こちらからダウンロードしておいてください。
その上でchromedriverを任意のファイルに置いて環境変数を設定してパスを通しておいてください。
Cドライブ直下に置いてパスを通せばそのまま上記のコードを動かすことができますが、別の場所においている場合は、パスをそちらに設定して以下のコードの「executable_path=」以降をchromedriverが置いている場所に書き換えてください。
driver = webdriver.Chrome(chrome_options=options, executable_path=r"C:\chromedriver_win32\chromedriver.exe")
options.add_argumentとかいうのがいっぱいありますが、これはchromedriverでスクレイピングする際のオプションです。それぞれの内訳がどういうものかについてはこちらが詳しいので参照してください。
vaaaaaanquish.hatenablog.com
次です。
html = requests.get(target_url) soup = BeautifulSoup(html.text, "html.parser") for iframe in soup.find_all("iframe"): if("live_chat_replay" in iframe["src"]): next_url= iframe["src"]
次は動画のURLのhtmlを取得し、その中からfind_allで"iframe"タグを持つ部分を探します。
その中に
iframe["src"]="https://www.youtube.com/live_chat_replay?continuation=xxxxxxxx"であるデータが存在するので、それを探し出してnext_urlに代入します。
次からがいちばん重要な部分ですね。
# chromedriverを開きnext_urlのソースを入手 driver = webdriver.Chrome(chrome_options=options, executable_path=r"C:\chromedriver_win32\chromedriver.exe") driver.get(next_url) soup = BeautifulSoup(driver.page_source,"lxml") # 必ずquitすること。忘れるとgoogle chromeが開かれまくって大変なことになる driver.quit()
chromedriverを用いて、さきほど入手したnext_urlのhtmlを入手します。
# 次に飛ぶurlのデータがある部分をfind_allで探してsplitで整形 for scrp in soup.find_all("script"): if "window[\"ytInitialData\"]" in scrp.text: dict_str = scrp.text.split(" = ")[1] # javascript表記なので更に整形. falseとtrueの表記を直す dict_str = dict_str.replace("false","False") dict_str = dict_str.replace("true","True") # 辞書形式と認識すると簡単にデータを取得できるが, 末尾に邪魔なのがあるので消しておく（「空白2つ + \n + ;」を消す） dict_str = dict_str.rstrip(" \n;") # 辞書形式に変換 dics = eval(dict_str)
html中の"script"タグを持つ部分で、"window["ytInitialData"] = xxxxx"となっている部分のxxxxにチャットデータが存在するので、まず"window["ytInitialData"]"を探して、"="でsplitしxxxxの部分を取得します。
xxxxの部分はjson形式のデータになっていますが、javascriptのソースを取得しているのでそのままでは扱えません。ですのでいくつか修正を行ってevalで辞書形式にして扱いやすくします。
# "https://www.youtube.com/live_chat_replay?continuation=" + continue_url が次のlive_chat_replayのurl continue_url = dics["continuationContents"]["liveChatContinuation"]["continuations"][0]["liveChatReplayContinuationData"]["continuation"] next_url = "https://www.youtube.com/live_chat_replay?continuation=" + continue_url # dics["continuationContents"]["liveChatContinuation"]["actions"]がコメントデータのリスト。先頭はノイズデータなので[1:]で保存 for samp in dics["continuationContents"]["liveChatContinuation"]["actions"][1:]: comment_data.append(str(samp)+"\n")
dics["continuationContents"]["liveChatContinuation"]["continuations"][0]["liveChatReplayContinuationData"]["continuation"]が次に飛ぶURLのデータなので取得。
先頭に"https://www.youtube.com/live_chat_replay?continuation="をつけると次に飛ぶURLになるのでnext_urlに代入します。
コメントデータはdics["continuationContents"]["liveChatContinuation"]["actions"]にリストで入っているので、forを用いて1行ずつ取得します。
ただし注意が必要なのは、dics["continuationContents"]["liveChatContinuation"]["actions"]の各要素はコメントだけでなく、コメントに関するコメント以外のデータも含まれています。
ですのでコメントのみを取得したい場合は、comment_data.append(str(samp)+"\n")を以下のように書き換えてください。
comment_data.append(str(samp["replayChatItemAction"]["actions"][0]["addChatItemAction"]["item"]["liveChatTextMessageRenderer"]["message"]["simpleText"])+"\n")
クソめんどくさいですね…。正直このあたりのデータ構造を確認するのが一番大変でした。
他に「チャットがリプレイ欄に表示された時間」を取得したいときは以下のようにしてください。
comment_data.append(str(samp["replayChatItemAction"]["actions"][0]["addChatItemAction"]["item"]["liveChatTextMessageRenderer"]["timestampText"]["simpleText"])+"\n")
最後はnext_urlを取得できなくなれば（実際はcontinue_urlを取得できなくなれば、ですが）全てのデータを取得し終えたということなのですが、処理を書くのがめんどくさいのでexceptで書いています。
# next_urlが入手できなくなったら終わり except: break
以上がyoutube liveのアーカイブからチャットを取得する方法です。
一応この方式で遅くなる原因ですが、コメントを取得するために何度も別のURLを開き続けなければいけないので、その分だけchromedriverでchromeを開くことになるからです。
しかも一度に取得できるコメントの最大数が40件くらいなので、何千件とコメントがある動画で実行したときにはもうはちゃめちゃ遅いです。なにか解決方法ありませんかね？
まあ書いてみればなんのことはないという感じですが、プログラムを書く時間よりもhtmlファイルとずっとにらめっこする時間のほうが長くて文字にならない苦労がめちゃくちゃありました…。
とりあえず記事にすることができてホッと一息です。
