はじめに
私はVTuber限界オタクです。
12月、両国国技館にて「Virtual to Live」が開催され、ニコ生で配信もされました。
現地に行けず、さらにはリアルタイムで見ることすらできなかった負け組ですが、円盤が出るまで待つこともできませんでした。
そんなわけで、ネットチケットを購入してタイムシフトを見たわけです。
まぁ、案の定死にまして。
タイムシフトが見れなくなるまでに保存したいと考えた私は、Pythonを片手にニコ生の分析を始めました。
ちなみに、僕の強さは「websocketってなに?」「hls?」「selenium?」程度です。
ニコ生の中身
※注意: 2019年12月17日現在の情報です。今後ニコ生の仕様が変わる可能性があります。
分析は、ChromeのDevToolsを活用しました。
Chrome上でF12
を押すことで表示することができます。
まずは適当な配信を開きましょう。
配信を開いてからF12を押した場合は、一度リロードしてください。
HLS
DevToolsのNetwork
タブには、そのウェブサイトでの通信ログを見ることができます。(Disable cache
にチェックを入れるとページが移ってもログが消えません)
動画の配信ということで、Size
が一番大きい通信を探しました。
すると、
https://{???}.dmc.nico/hlsarchive/ht2_nicolive/nicolive-hamster-{配信ID}_main_{十六進数}/4/ts/{数値}.ts?
というようなURLから大きなデータをダウンロードしていました。
{数値}
は順番に増えていっているように見えます。
そして、似たURLの
https://{???}.dmc.nico/hlsarchive/ht2_nicolive/nicolive-hamster-{配信ID}_main_{十六進数}/4/ts/playlist.m3u8?
のレスポンスで
#EXTM3U
#EXT-X-VERSION:3
#EXT-X-TARGETDURATION:5
(後略)
のようなデータが送られていました。
.m3u8
と.ts
で調べたところ、どうやらHLS (HTTP Live Streaming)
というプロトコルで使われるファイルのようです。
このHLS
はffmpeg
を用いて簡単にDLできるようなので、あとはURLを取得するだけですね。
URL取得
このURLはどこで得たのでしょうか。
DevToolsのNetwork
タブでCtrl+F
を押しましょう。
そこに、先ほど見つけたhttps://{???}.dmc.nico/hlsarchive/ht2_nicolive/nicolive-hamster-{配信ID}_main_{十六進数}
を入力して検索してみましょう。
どうやら出てくるのは「URLに対する通信」で、「URLを得た通信」は見当たりません。
「どういうことだ?」といろいろ試した結果、websocket内の通信は検索できないようでした。
なので、wss:
で検索をかけます。
すると、3つのwebsocketを見つけることができます。
4012/
はバイナリデータを送りあっているようです。とりあえず放置します。
websocket
では、サーバーからJSON形式でchatの情報を受け取っています。これも放置でいいでしょう。
残ったwebsocket(ここではtimeshift
)は、軽く見ただけではよくわかりません。とりあえず目当てのURLを検索してみましょう。
すると、ヒットする通信が存在しました。
この通信では、どうやらJSONを受け取っているようです。
このJSONの中のuri
に、お目当てのURLを発見しました。
Are you still watching?
まぁ、当たり前と言ってしまえばそうなのですが、サーバーが「コイツもう見てないな」と判断するとURLが無効になるようです。
そのため、「URLを取得して放置」ということができません。
じゃ、どうしましょうか…と考えて思いついたのは、
「ニコニコに任せる」という方法でした。
Seleniumで配信のURLを開きっぱなしにしておけば、勝手にクライアントが「まだ見てますよ~」というのをサーバーに送ってくれます。
その間にffmpeg
がDLするという、簡単なお話です。
ただ、この方法はDLするだけなら問題ないのですが、「独自のアプリで再生したい!」とかになると使えないと思うので、その場合は自力で「まだ見てますよ~」と送る必要があります。
また、ニコニコは同時に一つのウィンドウでしか配信を見ることができません(ニコニコ詳しくないので、詳しくはよく知りませんが)。
DL中に配信を開くとDLが止まってしまうのでお気を付けください。
Pythonでスクリプトを書く
コード
python dl.py {live_id} {id} {pass}
で実行できます。
from selenium import webdriver
import chromedriver_binary
import json
import time
import sys
import subprocess
lid = sys.argv[1]
id = sys.argv[2]
pa = sys.argv[3]
options = webdriver.ChromeOptions()
options.add_argument('--headless')
caps = webdriver.DesiredCapabilities.CHROME
caps['goog:loggingPrefs'] = {'performance': 'ALL'}
driver = webdriver.Chrome(
options=options, desired_capabilities=caps, service_log_path='NUL')
driver.get('https://account.nicovideo.jp/login')
fid = driver.find_element_by_xpath('//*[@id="input__mailtel"]')
fpa = driver.find_element_by_xpath('//*[@id="input__password"]')
fid.clear()
fid.send_keys(id)
fpa.clear()
fpa.send_keys(pa)
fpa.submit()
driver.get('https://live2.nicovideo.jp/watch/' + lid)
# setting_button = driver.find_element_by_xpath(
# '/html/body/div/div/div[4]/div[3]/div/div/div[1]/div[3]/div[1]/div[2]/div[3]/button[4]').click()
# time.sleep(1)
# driver.find_element_by_xpath(
# '/html/body/div/div/div[4]/div[3]/div/div/div[1]/div[3]/div[1]/div[2]/div[4]/div/div/div[2]').click()
# time.sleep(1)
# driver.find_element_by_xpath(
# '/html/body/div/div/div[4]/div[3]/div/div/div[1]/div[3]/div[1]/div[2]/div[4]/section[2]/ul/div[2]').click()
# time.sleep(3)
# driver.find_element_by_xpath(
# '/html/body/div/div/div[4]/div[3]/div/div/div[1]/div[3]/div[1]/div[2]/button').click()
time.sleep(3)
log = [json.loads(i['message']) for i in driver.get_log(
'performance') if json.loads(i['message'])['message']['method'] == 'Network.webSocketFrameReceived']
log = [json.loads(i['message']['params']['response']['payloadData'])
for i in log if i['message']['params']['response']['payloadData'][0] == '{']
log = [i['body'] for i in log if 'body' in i.keys()]
uri = ''
quality = 6
for i in log:
if 'command' in i.keys():
if i['command'] == 'currentstream':
if 0 != i['currentStream']['qualityTypes'].index(i['currentStream']['quality']) < quality:
uri = i['currentStream']['uri']
quality = i['currentStream']['qualityTypes'].index(
i['currentStream']['quality'])
if quality == 0:
break
subprocess.run(['ffmpeg', '-i', uri, '-c', 'copy', 'output.mp4'])
driver.quit()
解説
chromeの設定
options = webdriver.ChromeOptions()
options.add_argument('--headless')
caps = webdriver.DesiredCapabilities.CHROME
caps['goog:loggingPrefs'] = {'performance': 'ALL'}
driver = webdriver.Chrome(
options=options, desired_capabilities=caps, service_log_path='NUL')
--headless
オプションを付けることで、ウィンドウを表示せずに実行することができます。
caps['goog:loggingPrefs'] = {'performance': 'ALL'}
は、通信ログを見るための設定です。
詳しいことはよくわかりません。
ログイン
driver.get('https://account.nicovideo.jp/login')
fid = driver.find_element_by_xpath('//*[@id="input__mailtel"]')
fpa = driver.find_element_by_xpath('//*[@id="input__password"]')
fid.clear()
fid.send_keys(id)
fpa.clear()
fpa.send_keys(pa)
fpa.submit()
ニコ生はログインしないと見れません。
そのため、ここで一度ログインします。
websocketの監視
driver.get('https://live2.nicovideo.jp/watch/' + lid)
# setting_button = driver.find_element_by_xpath(
# '/html/body/div/div/div[4]/div[3]/div/div/div[1]/div[3]/div[1]/div[2]/div[3]/button[4]').click()
# time.sleep(1)
# driver.find_element_by_xpath(
# '/html/body/div/div/div[4]/div[3]/div/div/div[1]/div[3]/div[1]/div[2]/div[4]/div/div/div[2]').click()
# time.sleep(1)
# driver.find_element_by_xpath(
# '/html/body/div/div/div[4]/div[3]/div/div/div[1]/div[3]/div[1]/div[2]/div[4]/section[2]/ul/div[2]').click()
# time.sleep(3)
# driver.find_element_by_xpath(
# '/html/body/div/div/div[4]/div[3]/div/div/div[1]/div[3]/div[1]/div[2]/button').click()
time.sleep(3)
log = [json.loads(i['message']) for i in driver.get_log(
'performance') if json.loads(i['message'])['message']['method'] == 'Network.webSocketFrameReceived']
log = [json.loads(i['message']['params']['response']['payloadData'])
for i in log if i['message']['params']['response']['payloadData'][0] == '{']
log = [i['body'] for i in log if 'body' in i.keys()]
uri = ''
quality = 6
for i in log:
if 'command' in i.keys():
if i['command'] == 'currentstream':
if i['currentStream']['qualityTypes'].index(i['currentStream']['quality']) < quality:
uri = i['currentStream']['uri']
quality = i['currentStream']['qualityTypes'].index(
i['currentStream']['quality'])
if quality == 0:
break
driver.get_log('performance')
でパフォーマンスログを取得できます。
if json.loads(i['message'])['message']['method'] == 'Network.webSocketFrameReceived'
でwebsocketの受信だけを抽出しています。
そこからさらに、「JSON形式であり、'body'を持つ」ものだけを残します。
uri = i['currentStream']['uri']
でURLを保存しています。
ここでは、通信の中から「一番クオリティが良いURL」を得るようにしています。
コメントアウト部分では、最高クオリティの動画を選択し動画をストップしています。
ここは今後のアップデートで変わる可能性がとても大きいため、コメントアウトしました。
seleniumの使い方を調べて、ご自身で記述することをおススメします。
保存と終了
subprocess.run(['ffmpeg', '-i', uri, '-c', 'copy', 'output.mp4'])
driver.quit()
subprocess.run(['ffmpeg', '-i', uri, '-c', 'copy', 'output.mp4'])
でffmpeg
を実行しています。
ffmpeg
が既にインストールされている必要があります。
'driver.quit()'でchromeを閉じます。
最後に
webサービスの分析をしてそれを世に公開しても、サービスの仕様変更により使えなくなることがよくあります。
今回は、「今の仕様」を共有することよりも「こうやれば仕様を分析できるんじゃね?」というのを共有したいと思い、このような記事になりました。
最初に申した通り、僕は何もわからない状態でそれを共有しています。
もし、「ここおかしくね?」と思いましたら、ご指摘ください。
12/22追記
また同じことをする機会があったので、12/22日現在動いているコードを載せておきます
from os import read
from selenium import webdriver
import chromedriver_binary
import json
import time
import sys
import subprocess
import getpass
id = input("email adress: ")
pa = getpass.getpass("password: ")
lid = input("live id: ")
options = webdriver.ChromeOptions()
# options.add_argument('--headless')
caps = webdriver.DesiredCapabilities.CHROME
caps['goog:loggingPrefs'] = {'performance': 'ALL'}
driver = webdriver.Chrome(options=options, desired_capabilities=caps, service_log_path='NUL')
# ログイン
driver.get('https://account.nicovideo.jp/login')
fid = driver.find_element_by_xpath('//*[@id="input__mailtel"]')
fpa = driver.find_element_by_xpath('//*[@id="input__password"]')
fid.clear()
fid.send_keys(id)
fpa.clear()
fpa.send_keys(pa)
fpa.submit()
driver.get('https://live2.nicovideo.jp/watch/' + lid)
time.sleep(5)
# JSONでやり取りしているWebSocketの通信を全取得
log = [json.loads(i['message']) for i in driver.get_log('performance') if json.loads(i['message'])['message']['method'] == 'Network.webSocketFrameReceived']
log = [json.loads(i['message']['params']['response']['payloadData']) for i in log if i['message']['params']['response']['payloadData'][0] == '{']
# 目的の通信からURIを取得
log = [i['data'] for i in log if 'data' in i.keys()]
uri = ''
quality = 6
for i in log:
if 'uri' in i.keys():
uri = i['uri']
break
# mp4形式で保存(ffmpegバイナリが必要)
subprocess.run(['ffmpeg', '-i', uri, '-c', 'copy', 'output.mp4'])
driver.quit()