現在、ホロライブ+イノナカミュージックには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に自動的にデプロイしてくれるのでめちゃくちゃ便利だなーと思っています。

ホロライブメンバーの配信予定を取得して配信開始時刻に通知するDiscord botを作った

仕組み

YouTube APIの取得

ソースコード

実行環境