ラズパイでTwitch APIを利用しよう

Japanese

はじめに

Twitchで配信を視聴するとチャンネルポイントが各ユーザーに付与され、そのポイントは配信者が定めた使用方法で使用することができます。

例えば、

  • 文字を強調する
  • 質問、相談をする
  • 一緒にゲームを1プレイする
  • スタンプのアンロック

など様々な使用方法があります。
そのポイントを荒稼ぎできないかなという邪な考えから始まりました。

プログラムで特定の配信者が配信開始するのを確認し、視聴開始する。
その後、ポイントを貰い配信が終了すると自動で切断を繰り返すことができれば可能では?
と思って作成してみた。

結果、ブラウザとアプリを使用しログインしている状態でないとポイントは付与されないので夢は儚く散ってしまったのですが、、、

Twitch Developersへアクセス

TwitchAPIを使用するにはTwitch Developersへのアクセスが不可欠です。
そこでIDなどを取得しプログラムに埋め込み使用します。

Twitch
Twitch is the world's leading video platform and community for gamers.

クライアントIDとクライアントシークレット

クライアントID

画面が表示されたら[アプリケーションを登録]を選択します。

名前:任意(自分がわかればOK)
OAuthのリダイレクトURL:http://localhost
カテゴリー:Other

以上を入力して作成。

作成後、アプリケーションに表示されるので[管理]を選択

そこでクライアントIDが表示されるのでメモする。

クライアントシークレット

ここからラズパイでの操作になります。

ラズパイのCMDプロンプトで以下のコードを入力しクライアントシークレットを取得します。
クライアントシークレットは以下の画像の[新しい秘密]を選択すると表示することができます。


curl -X POST "https://id.twitch.tv/oauth2/token?client_id=[クライアントID]&client_secret=[クライアントシークレット]&grant_type=client_credentials"

すると以下のような結果が返ってくるので”access_token”をメモする。

{"access_token":"xxxxxxxxxxxxx","expires_in"~~~

ラズパイの準備

Pythonライブラリ


pip install requests

動画プレイヤー(mpv)


sudo apt update
sudo apt install mpv

コード

上記の準備が完了したらいよいよPythonのプログラムを作っていきます。


import requests
import subprocess
import time
import signal
import os
# ===========================================
CLIENT_ID = 'Client ID'
ACCESS_TOKEN = 'アクセストークン'
STREAMER_NAME = '視聴したいtwitchユーザー名'
# ===========================================
HEADERS = {
    'Client-ID': CLIENT_ID,
    'Authorization': f'Bearer {ACCESS_TOKEN}'
}
# mpvのプロセス保持用
mpv_process = None
def is_stream_live(username):
    url = f'https://api.twitch.tv/helix/streams?user_login={username}'
    response = requests.get(url, headers=HEADERS)
    data = response.json()
    return bool(data['data'])
def start_stream(username):
    global mpv_process
    print(f"{username} is live! Starting stream...")
    # mpvを非同期起動してプロセスを保存
    mpv_process = subprocess.Popen(['mpv', f'https://twitch.tv/{username}'])
def stop_stream():
    global mpv_process
    if mpv_process is not None:
        print("Stream ended. Stopping mpv...")
        mpv_process.terminate()
        try:
            mpv_process.wait(timeout=5)
        except subprocess.TimeoutExpired:
            mpv_process.kill()  # 強制終了
        mpv_process = None
def main():
    already_live = False
    while True:
        try:
            live = is_stream_live(STREAMER_NAME)
            if live and not already_live:
                start_stream(STREAMER_NAME)
                already_live = True
            elif not live and already_live:
                stop_stream()
                already_live = False
        except Exception as e:
            print("Error:", e)
        time.sleep(60)
if __name__ == '__main__':
    main()

上記のコードを使用し実行すれば視聴可能です。

このようにAPI経由で配信が視聴可能になります。
ラズパイなので配信自体若干かくついているのですが問題ないです。

このプログラムをcrontabなどで自動実行すると自分が見れない時でもポイントが楽に稼げると思ったのですが、API経由からの視聴はポイントは貰えないみたいです。残念

タイトルとURLをコピーしました