現在、ホロライブ+イノナカミュージックには28人のメンバーがおり、それぞれのメンバーの配信予定を把握することが難しいです。一応公式でホロジュールと呼ばれる各メンバーのスケジュールはありますが、それに乗らないような3時間前くらいに突如として現れるようなスケジュールまで追うのはなかなか難しいのではないでしょうか。
そこで、今回Discordに配信予定や配信時刻になったら通知が来るようなbotをミカグラくん(@hungry_and_fool)と作りました。
見た目
仕組み
YouTube APIの取得
これがめちゃくちゃ参考になりました。ここを見て、Google Cloud Platformにてプロジェクトを作成し、YouTube Data API v3を選択し、アクセストークンを取得しました。
今回使ったものは
https://www.googleapis.com/youtube/v3/search
https://www.googleapis.com/youtube/v3/videos
の2つです。上を使って配信予定のものを取得し、下を使ってそれぞれの配信開始時刻を取得しています。
ただ、YouTube APIには制約があって、1トークンにつき1万ユニット/dayしか使えません。searchのAPIを1回叩くのに100を消費し、それがメンバーだけかかるため、到底1つのトークンでは追いきることができません。なので複数プロジェクトを立ち上げ、それぞれでトークンを取得し、回すようにしています。
ソースコード
import time
import requests
import json
import copy
from datetime import datetime, timedelta, timezone
Hololive = {
"UCp6993wxpyDPHUpavwDFqgg": [
"ときのそら",
"https://yt3.ggpht.com/a/AATXAJzGvZJuJ92qM5WcfBcDZqPFSj_CGIEYp9VFmA=s288-c-k-c0xffffffff-no-rj-mo"
],
...
"UC1uv2Oq6kNxgATlCiez59hw": [
"常闇トワ",
"https://yt3.ggpht.com/a/AATXAJxqyp7DhLSSrSYRc5HaLcq5QvJvRp3jDnxTeA=s288-c-k-c0xffffffff-no-rj-mo"
],
"UCa9Y57gfeY0Zro_noHRVrnw": [
"姫森ルーナ",
"https://yt3.ggpht.com/a/AATXAJzzirDjRJkofWVeoE6gVjodJ0VXaJhy4b_CLg=s288-c-k-c0xffffffff-no-rj-mo"
],
} #配信者のチャンネルID, 配信者名, アイコン画像のURLのリスト
webhook_url_Hololive = '配信開始チャンネル用のwebhookリンク' #ホロライブ配信開始
webhook_url_Hololive_yotei = '配信開始予定用のwebhookリンク' #ホロライブ配信予定
broadcast_data = {} #配信予定のデータを格納
YOUTUBE_API_KEY = [複数のAPI(str型)をリストで管理]
def dataformat_for_python(at_time): #datetime型への変換
at_year = int(at_time[0:4])
at_month = int(at_time[5:7])
at_day = int(at_time[8:10])
at_hour = int(at_time[11:13])
at_minute = int(at_time[14:16])
at_second = int(at_time[17:19])
return datetime(at_year, at_month, at_day, at_hour, at_minute, at_second)
def replace_JST(s):
a = s.split("-")
u = a[2].split(" ")
t = u[1].split(":")
time = [int(a[0]), int(a[1]), int(u[0]), int(t[0]), int(t[1]), int(t[2])]
if(time[3] >= 15):
time[2] += 1
time[3] = time[3] + 9 - 24
else:
time[3] += 9
return (str(time[0]) + "/" + str(time[1]).zfill(2) + "/" + str(time[2]).zfill(2) + " " + str(time[3]).zfill(2) + "-" + str(time[4]).zfill(2) + "-" + str(time[5]).zfill(2))
def post_to_discord(userId, videoId):
haishin_url = "https://www.youtube.com/watch?v=" + videoId #配信URL
content = "配信中！\n" + haishin_url #Discordに投稿される文章
main_content = {
"username": Hololive[userId][0], #配信者名
"avatar_url": Hololive[userId][1], #アイコン
"content": content #文章
}
requests.post(webhook_url_Hololive, main_content) #Discordに送信
broadcast_data.pop(videoId)
def get_information():
tmp = copy.copy(broadcast_data)
api_now = 0 #現在どのYouTube APIを使っているかを記録
for idol in Hololive:
api_link = "https://www.googleapis.com/youtube/v3/search?part=snippet&channelId=" + idol + "&key=" + YOUTUBE_API_KEY[api_now] + "&eventType=upcoming&type=video"
api_now = (api_now + 1) % len(YOUTUBE_API_KEY) #apiを1つずらす
aaa = requests.get(api_link)
v_data = json.loads(aaa.text)
try:
for item in v_data['items']:#各配信予定動画データに関して
broadcast_data[item['id']['videoId']] = {'channelId':item['snippet']['channelId']} #channelIDを格納
for video in broadcast_data:
try:
a = broadcast_data[video]['starttime'] #既にbroadcast_dataにstarttimeがあるかチェック
except KeyError:#なかったら
aaaa = requests.get("https://www.googleapis.com/youtube/v3/videos?part=liveStreamingDetails&id=" + video + "&key=" + YOUTUBE_API_KEY[api_now])
api_now = (api_now + 1) % len(YOUTUBE_API_KEY) #apiを1つずらす
vd = json.loads(aaaa.text)
print(vd)
broadcast_data[video]['starttime'] = vd['items'][0]['liveStreamingDetails']['scheduledStartTime']
except KeyError: #配信予定がなくて403が出た
continue
for vi in broadcast_data:
if(not(vi in tmp)):
print(broadcast_data[vi])
try:
post_broadcast_schedule(broadcast_data[vi]['channelId'], vi, broadcast_data[vi]['starttime'])
except KeyError:
continue
def check_schedule(now_time, broadcast_data):
for bd in list(broadcast_data):
try:
# RFC 3339形式 => datetime
sd_time = datetime.strptime(broadcast_data[bd]['starttime'], '%Y-%m-%dT%H:%M:%SZ') #配信スタート時間をdatetime型で保管
sd_time += timedelta(hours=9)
if(now_time >= sd_time):#今の方が配信開始時刻よりも後だったら
post_to_discord(broadcast_data[bd]['channelId'], bd) #ツイート
except KeyError:
continue
def post_broadcast_schedule(userId, videoId, starttime):
st = starttime.replace('T', ' ')
sst = st.replace('Z', '')
ssst = replace_JST(sst)
haishin_url = "https://www.youtube.com/watch?v=" + videoId #配信URL
content = ssst + "に配信予定！\n" + haishin_url #Discordに投稿される文章
main_content = {
"username": Hololive[userId][0], #配信者名
"avatar_url": Hololive[userId][1], #アイコン
"content": content #文章
}
requests.post(webhook_url_Hololive_yotei, main_content) #Discordに送信
while True:
now_time = datetime.now() + timedelta(hours=9)
if(((now_time.year > 2020) or ((now_time.year == 2020) and (now_time.month >= 6) and (now_time.day >= 22))) and (now_time.minute == 0) and (now_time.hour % 2 == 0)):
get_information()
check_schedule(now_time, broadcast_data)
time.sleep(60)
1つずつ解説していきます。
Hololive = {
"UCp6993wxpyDPHUpavwDFqgg": [
"ときのそら",
"https://yt3.ggpht.com/a/AATXAJzGvZJuJ92qM5WcfBcDZqPFSj_CGIEYp9VFmA=s288-c-k-c0xffffffff-no-rj-mo"
],
...
"UC1uv2Oq6kNxgATlCiez59hw": [
"常闇トワ",
"https://yt3.ggpht.com/a/AATXAJxqyp7DhLSSrSYRc5HaLcq5QvJvRp3jDnxTeA=s288-c-k-c0xffffffff-no-rj-mo"
],
"UCa9Y57gfeY0Zro_noHRVrnw": [
"姫森ルーナ",
"https://yt3.ggpht.com/a/AATXAJzzirDjRJkofWVeoE6gVjodJ0VXaJhy4b_CLg=s288-c-k-c0xffffffff-no-rj-mo"
],
} #配信者のチャンネルID, 配信者名, アイコン画像のURLのリスト
webhook_url_Hololive = '配信開始チャンネル用のwebhookリンク' #ホロライブ配信開始
webhook_url_Hololive_yotei = '配信開始予定用のwebhookリンク' #ホロライブ配信予定
broadcast_data = {} #配信予定のデータを格納
YOUTUBE_API_KEY = [複数のAPI(str型)をリストで管理]
broadcast_dataという辞書型に配信のvideoIdをキーとして、そこに開始時刻、チャンネルIDを格納していきます。
def replace_JST(s):
a = s.split("-")
u = a[2].split(" ")
t = u[1].split(":")
time = [int(a[0]), int(a[1]), int(u[0]), int(t[0]), int(t[1]), int(t[2])]
if(time[3] >= 15):
time[2] += 1
time[3] = time[3] + 9 - 24
else:
time[3] += 9
return (str(time[0]) + "/" + str(time[1]).zfill(2) + "/" + str(time[2]).zfill(2) + " " + str(time[3]).zfill(2) + "-" + str(time[4]).zfill(2) + "-" + str(time[5]).zfill(2))
YYYY-MM-DD hh:mm :ssというフォーマットに変換をしていますがここで取得している時間が9時間の時差があります。なので時差を考えて変換しています。
def post_to_discord(userId, videoId):
haishin_url = "https://www.youtube.com/watch?v=" + videoId #配信URL
content = "配信中！\n" + haishin_url #Discordに投稿される文章
main_content = {
"username": Hololive[userId][0], #配信者名
"avatar_url": Hololive[userId][1], #アイコン
"content": content #文章
}
requests.post(webhook_url_Hololive, main_content) #Discordに送信
broadcast_data.pop(videoId)
配信開始時にDiscordにポストする関数、webhookを用いて指定したリンクにPostしています。また、メモリがあふれることを回避するために一度POSTしたデータはbroadcast_dataから削除します。
def get_information():
tmp = copy.copy(broadcast_data)
api_now = 0 #現在どのYouTube APIを使っているかを記録
for idol in Hololive:
api_link = "https://www.googleapis.com/youtube/v3/search?part=snippet&channelId=" + idol + "&key=" + YOUTUBE_API_KEY[api_now] + "&eventType=upcoming&type=video"
api_now = (api_now + 1) % len(YOUTUBE_API_KEY) #apiを1つずらす
aaa = requests.get(api_link)
v_data = json.loads(aaa.text)
try:
for item in v_data['items']:#各配信予定動画データに関して
broadcast_data[item['id']['videoId']] = {'channelId':item['snippet']['channelId']} #channelIDを格納
for video in broadcast_data:
try:
a = broadcast_data[video]['starttime'] #既にbroadcast_dataにstarttimeがあるかチェック
except KeyError:#なかったら
aaaa = requests.get("https://www.googleapis.com/youtube/v3/videos?part=liveStreamingDetails&id=" + video + "&key=" + YOUTUBE_API_KEY[api_now])
api_now = (api_now + 1) % len(YOUTUBE_API_KEY) #apiを1つずらす
vd = json.loads(aaaa.text)
print(vd)
broadcast_data[video]['starttime'] = vd['items'][0]['liveStreamingDetails']['scheduledStartTime']
except KeyError: #配信予定がなくて403が出た
continue
for vi in broadcast_data:
if(not(vi in tmp)):
print(broadcast_data[vi])
try:
post_broadcast_schedule(broadcast_data[vi]['channelId'], vi, broadcast_data[vi]['starttime'])
except KeyError:
continue
配信予定を取得するものです。28人のホロライブメンバーごとに、searchにおいてpart=snippet,eventType=upcoming,type=video,channelId=(各メンバーのチャンネルID), key=APIとしてAPIを叩きます。これにより配信予定を取得できます。また、取得していた各データにおいてvideosをpart=liveStreamingDetails, id=(データ中のvideoId), key=APIとしてAPIを叩き、配信開始予定時刻を取得します。これらの動作を1度終えたあとはAPIリストの次のものを使うようにすることを忘れないようにします。
一連の動作によってbroadcast_dataに各配信予定を格納できます。また、tmpに動作を行う前の辞書を保存しておき、それと動作後を比較することにより、今回の動作によって取得したデータがわかります。これら1つ1つにpost_broadcast_schedule関数を適用し、配信予定をDiscordにポストしています。
def check_schedule(now_time, broadcast_data):
for bd in list(broadcast_data):
try:
sd_time = dataformat_for_python(broadcast_data[bd]['starttime']) #配信スタート時間をdatetime型で保管
sd_time += timedelta(hours=9)
if(now_time >= sd_time):#今の方が配信開始時刻よりも後だったら
post_to_discord(broadcast_data[bd]['channelId'], bd) #ツイート
except KeyError:
continue
broadcast_dataの全要素の開始時刻と現在時刻を比較。現在時刻の方が時間が経過していたらpost_to_discordを実行しDiscordに配信開始をポストしています。
while True:
now_time = datetime.now() + timedelta(hours=9)
if((now_time.minute == 0) and (now_time.hour % 2 == 0)):
get_information()
check_schedule(now_time, broadcast_data)
time.sleep(60)
メインの実行部分です。現在時刻を取得し，2時間ごとにget_information()を実行、また1分ごとにcheck_schedule()を実行しています。
#実行環境
今回はHerokuとGitHubを使っています。やり方はhttps://qiita.com/1ntegrale9/items/aa4b373e8895273875a8 を見るとわかりやすいと思います。GitHubの中身を書き換えるとHerokuに自動的にデプロイしてくれるのでめちゃくちゃ便利だなーと思っています。