【Python】TwitchのアーカイブからAPI経由で全コメントを取得する方法

はじめに・注意事項

この記事の内容はTwitch公式で推奨されているやり方ではありません。
試す場合は自己責任でお願いします。

概要

アーカイブコメントを集計・解析するツールを制作する際、Twitchアーカイブからコメントを取得するのに意外と時間がかかったので記録。

技術スタック

• OS：Windows11
• 言語：Python3
• IDE：VSCode

Twitchのコメント数を取得する

一番最初にたどり着いた記事がこちら↓

しかし、https://api.twitch.tv/v5/というエンドポイントは2022年2月末をもって廃止となったとのこと。（以下Twitch公式ブログ）

2024年8月時点ではhttps://api.twitch.tv/helix/のエンドポイントが使えるらしい。
APIドキュメント：Twitch API

PythonでTwitch APIを扱う場合はtwitchAPIというライブラリを使用できるとのこと。

しかし、現在のAPIではアーカイブの情報を取得するようなエンドポイントはありませんでした。
ただ、https://www.twitchchatdownloader.com/ のようなサービスがあるので何かの方法で取得できるはず。

ということでいろいろ調べてみることに。

Twitchを検証ツールで調べてみる。

Twitchのアーカイブを再生しながらChromeの検証ツールで調べてみた。

---

Headersの中身を見てみる。

ふむ。どうやらhttps://gql.twitch.tv/gqlというAPIが存在しているらしい。

調べてみたら以下のGithubがヒット。

読んでみると、

通常のClient-IDでリクエストを送るとはじき返されるけど、kimne78kx3ncx6brgo4mv6wki5h1koだったら通るぞ。

みたいなことが書いてある。
確かにRequest Headersの中身をよく見ると同じのがあった。

以上から、Request Headerの中身は Client-ID:kimne78kx3ncx6brgo4mv6wki5h1koでよさそう。

---

GraphQLのため、POSTリクエストに応じてResponseが返ってくる。
bodyに何かを書いて情報を送る必要があるため、Request Payloadを見てみる。

[
    {
        "operationName": "VideoCommentsByOffsetOrCursor",
        "variables": {
            "videoID": "2219755462",
            "contentOffsetSeconds": 0
        },
        "extensions": {
            "persistedQuery": {
                "version": 1,
                "sha256Hash": "b70a3591ff0f4e0313d126c6a1502d79a1c02baebb288227c582044aa76adf6a"
            }
        }
    }
]

• operationName：文字列から予測するに、経過時間またはCursorに応じてアーカイブのコメントを取得する操作をするためのキー
• variables - contentOffsetSeconds：動画開始からの経過時間
• variables - videoID：アーカイブ動画のID（URLから取得可能）
• extentions - persistedQuery：GraphQL APIを利用するとき、クエリに対応するIDを予め用意し、GraphQLサーバー>の前段でIDとクエリを交換することでリクエストパラメータを小さく収めよう、みたいなやつ。 コピペでよさそう。正直よくわかってない

POSTMANで検証

うまいこと返ってきた。
だけどどう見ても取得できてるコメントが少ない。

Responseの中を見てみると以下を発見。

"pageInfo": {
    "hasNextPage": true,
    "hasPreviousPage": false,
    "__typename": "PageInfo"
}

"hasNextPage": trueということは次のページがありそう。

またここで取得されたコメントは
"eyJpZCI6ImI4ZDFiNTI5LWNmNjEtNGRjMy1hY2M5LTAyNDI1MjMyNjJlZSIsImhrIjoiYnJvYWRjYXN0OjQxNTA2NjU2MTM1Iiwic2siOiJBQUFBVjY5WV93QVg2Z3JJTGY0WUFBIn0"
という同一のcursorを含んでいる構造だとわかった。

ということは、"hasNextPage"がFalseになるまでリクエストを送り続けるとすべてのコメントが取得できそう。

Pythonスクリプトで検証

import requests
import time
import json
import pprint

video_id = "2219755462"
start = 0
cursor = ""

api_url = "https://gql.twitch.tv/gql"

first_data = json.dumps([
    {
        "operationName": "VideoCommentsByOffsetOrCursor",
        "variables": {
            "videoID": video_id,
            "contentOffsetSeconds": 0
        },
        "extensions": {
            "persistedQuery": {
                "version": 1,
                "sha256Hash": "b70a3591ff0f4e0313d126c6a1502d79a1c02baebb288227c582044aa76adf6a"
            }
        }
    }
])

def get_json_data(video_id, cursor):
    loop_data = json.dumps([
        {
            "operationName": "VideoCommentsByOffsetOrCursor",
            "variables": {
                "videoID": video_id,
                "cursor": cursor
            },
            "extensions": {
                "persistedQuery": {
                    "version": 1,
                    "sha256Hash": "b70a3591ff0f4e0313d126c6a1502d79a1c02baebb288227c582044aa76adf6a"
                }
            }
        }
    ])
    return(loop_data)

#1回目のセッションスタート
session = requests.Session()
session.headers = { 'Client-ID': 'kd1unb4b3q4t58fwlpcbzcbnm76a8fp', 'content-type': 'application/json' }

response = session.post(
    api_url,
    first_data,
    timeout=10
)
print("接続に成功しました\n")
response.raise_for_status()
data = response.json()
#pprint.pprint(data)

for comment in data[0]['data']['video']['comments']['edges']:
    print(comment['node']['message']['fragments'][0]['text'], "\n")

cursor = None
if data[0]['data']['video']['comments']['pageInfo']['hasNextPage']:
    cursor = data[0]['data']['video']['comments']['edges'][-1]['cursor']
    print("\n", cursor)
    time.sleep(0.1)


# session loop
while cursor:
    response = session.post(
        api_url,
        get_json_data(video_id, cursor),
        timeout=10
    )
    response.raise_for_status()
    data = response.json()

    for comment in data[0]['data']['video']['comments']['edges']:
        print(comment['node']['message']['fragments'][0]['text'])

    if data[0]['data']['video']['comments']['pageInfo']['hasNextPage']:
        cursor = data[0]['data']['video']['comments']['edges'][-1]['cursor']
        print("\n", cursor)
        time.sleep(0.1)
    else:
        cursor = None

video_idを好きなアーカイブのidに指定するとコメントを全て取得、表示する。

細かいところは省略するが、送信するデータのvariablesはcontentOffsetSecondsだけでなくcursorも指定でき、その仕組みを利用して全ページのコメントを取得している。

これができたら、開始時間と終了時間を指定して特定の時間のコメントを取得したり、一番コメントした回数多い人を割り出せたりできる。

requestsモジュールのSessionインスタンスを利用してログイン状態を保持したままでないと、2回目以降のセッションで弾かれてしまうので注意。

あとがき

今回の記事のように、通信の中身を見て色々試してみるということがなかったのでAPIに対する理解が深まった。
と、同時に今までAPIを設計する時に「レスポンスが返ってくればいいや」くらいの感覚だったので、セキュリティ要件とかセッション管理とかちゃんとするべきだなと感じた。

Twitchは公式のAPIでこれができるようにしてくれ

宣伝

Web作ってたり、ライティングしてたり、ツール開発してたりします。
ポートフォリオサイト↓

スプレッドシートとコマンド上で動くツール作るのが好きなので、依頼がある場合上記からお問合せください。

超不定期でマンツーマンプログラミング講師もしてます。
興味のある方はお声がけください。