LoginSignup
1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@HaGaTa_

Twitch APIを用いてクリップ表示Botを作成しました

Last updated at Posted at 2024-09-23

はじめに

初めまして。Qiita初投稿のHagataです。

Twitchには視聴者が面白かった場面や盛り上がった場面など、共有したい瞬間を10秒から60秒の動画（クリップ）として切り抜き、共有できるという機能があります。

Twitchを利用している中でTwitch APIを用いて、その機能により作成されたクリップをDiscord上でも表示できたらな〜という考えが過り、クリップ機能により作成された特定のストリーマーorゲームタイトルのクリップをDiscord上で表示するBotを作成しました。

本記事では、discord.pyとTwitch APIを用いたクリップ表示Botを紹介するという形で練習がてら記事を書きます。お手柔らかに。

背景

Twitchを利用している中でTwitch APIを用いて、その機能により作成されたクリップをDiscord上でも表示できたらな〜という考えが過りました。またDiscord内でTwitchの面白いクリップを簡単に共有したいというのがきっかけです。

環境

Python 3.11

使用技術等

Docker
Discord.py
FastAPI
SQLAlchemy
SQLite + aiosqlite

各自でホストする用として作成したためSQLiteを使っています。

機能（コマンド）

  • set：誰の、または何のゲームのクリップを取得するかを設定
  • set-streamer：表示名から誰のクリップを取得するかを設定
  • clip-range：現在から何日前までのクリップを取得するかを設定
  • display：現在の設定からクリップを指定個表示
  • status：ギルドの設定情報を表示
  • remove：設定情報を削除
  • help：上記コマンドの説明

準備

  1. Discord Developer PortalでBotを作成
  2. Twitch Developerでアプリケーションを作成
  3. 必要な環境変数を設定（詳細は後述）

環境変数

環境変数はプロジェクトのルートディレクトリに.envファイルを作成し、以下のように環境変数を設定します：

.env
DISCORD_ACCESS_TOKEN=  # 作成したBotのアクセストークン
TWITCH_CLIENT_ID=      # TwitchアプリケーションのクライアントID
TWITCH_CLIENT_SECRET=  # Twitchアプリケーションのクライアントシークレット

ここで定義した環境変数はdotenvを用いて読み込んでいます。

設計

ソフトウェア設計モデルを参考に、以下のようなディレクトリ構造にしました。

├── clipspotter
│   ├── __init__.py
│   ├── app.py
│   ├── api
│   │   ├── __init__.py
│   │   └── twitch.py
│   ├── cogs
│   │   ├── clip_range.py
│   │   ├── cshelp.py
│   │   ├── display.py
│   │   ├── remove.py
│   │   ├── set.py
│   │   ├── set_streamer.py
│   │   └── status.py
│   ├── config
│   │   ├── __init__.py
│   │   ├── config.py
│   │   └── constants.py
│   ├── models
│   │   ├── __init__.py
│   │   ├── base_model.py
│   │   ├── category.py
│   │   ├── discord_model.py
│   │   ├── twitch_model.py
│   │   └── visibility.py
│   └── utils
│       └── database.py
└── setup.py

主要な実装ポイント

Twitch APIとの連携

TwtichのAPI呼び出しをclipspotter/api/twitch.pyで行い、クリップ情報の取得ロジックを実装しました。以下は主要な部分のコードサンプルです。

twitch.py
class TwitchAPI:
    base_url = "https://api.twitch.tv/helix/"

    def __init__(self):
        self.client_id = CLIENT_ID
        self.client_secret = CLIENT_SECRET
        self.access_token = self._get_access_token()

    # アクセストークンの取得
    def _get_access_token(self) -> str:
        url = "https://id.twitch.tv/oauth2/token"
        params = {
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "grant_type": "client_credentials",
        }
        # POSTリクエストを送信し、レスポンスを取得
        response = requests.post(url, params=params, timeout=REQUEST_TIMEOUT)
        response.raise_for_status()
        # レスポンスのJSONからアクセストークンを抽出して返す
        return response.json()["access_token"]
        
    # リクエストヘッダーの生成
    def _get_headers(self):
        # Client-IDとBearerトークンを含むヘッダー辞書を作成して返す
        return {
            "Client-ID": self.client_id,
            "Authorization": f"Bearer {self.access_token}",
        }
        
    # APIリクエストの実行とレスポンスの取得
    def _get_response(self, url: str, query_params: dict[str, Any] | None) -> list[dict[str, Any]] | None:
        # 指定されたURLに対してGETリクエストを送信
        # _get_headers()で生成したヘッダーと指定されたクエリパラメータを使用
        response = requests.get(url, headers=self._get_headers(), params=query_params, timeout=REQUEST_TIMEOUT)
        response.raise_for_status()
        # レスポンスのJSONから'data'キーの値を抽出して返す
        return response.json().get("data")

    # クリップの取得
    def get_clips(self, category: Category, set_id: str, days_ago: int, first: int) -> list[dict[str, Any]] | None:
        # クリップ取得用のエンドポイントURLを構築
        url = self.base_url + "clips"
        # リクエストパラメータを設定（取得数、開始日時、終了日時）
        params = {
            "first": first,
            "started_at": (datetime.now(tz=timezone.utc) - timedelta(days_ago)).strftime("%Y-%m-%dT%H:%M:%SZ"),
            "ended_at": datetime.now(tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
        }
        # カテゴリに応じてbroadcaster_idまたはgame_idパラメータを設定
        cat_id_name = "broadcaster_id" if category == Category.STREAMER else "game_id"
        params[cat_id_name] = set_id
        # _get_response()を呼び出してクリップデータを取得し返す
        return self._get_response(url, params)

    # ストリーマー情報の取得
    def _get_streamer_keys(self, name: str) -> tuple[str | None, str | None, str | None]:
        url = self.base_url + "users"
        # ユーザー名をクエリパラメータとして設定
        query_params = {"login": name}
        data = self._get_response(url, query_params)
        if not data:
            return None, None, None
        # データが存在する場合、id、login、display_nameを抽出してタプルとして返す
        return data[0].get("id"), data[0].get("login"), data[0].get("display_name")

    # ゲーム情報の取得
    def _get_game_keys(self, name: str) -> tuple[str | None, str | None]:
        url = self.base_url + "games"
        # ゲーム名をクエリパラメータとして設定
        query_params = {"name": name}
        data = self._get_response(url, query_params)
        if not data:
            return None, None
        # データが存在する場合、idとnameを抽出してタプルとして返す
        return data[0].get("id"), data[0].get("name")

BotとDBの起動

clipspotter/app.pyでは、コグの読み込みとデータベース起動の処理を記述します。
以下にtwitch.pyの記述内容を一部抜粋します。

app.py
# commands.Bot インスタンスを作成し、コマンドプレフィックスや権限を指定
intents = discord.Intents.default()
intents.guilds = True
intents.messages = True
intents.message_content = True
bot = commands.Bot(command_prefix="cs$", case_insensitive=True, intents=intents)

# SQLAlchemyのORMを使用するための基本クラス
Base = declarative_base()


# FastAPIアプリケーションを作成
def create_app():
    # アプリケーションの起動時に実行される処理を定義
    @asynccontextmanager
    async def lifespan(app: FastAPI) -> AsyncGenerator[None, Any]:
        await init_models()
        yield

    app = FastAPI(lifespan=lifespan)

    # Botを非同期で起動するためのタスクを生成
    loop = asyncio.get_event_loop()
    loop.create_task(main())

    return app


# データベースモデルの初期化
async def init_models() -> None:
    async with engine.begin() as conn:
        # Base.metadata.create_all()を実行し、定義されたモデルに基づいてテーブルを作成
        await conn.run_sync(Base.metadata.create_all)


# 拡張機能（Cog）の読み込み
async def load_extensions():
    initial_extensions = [
        "clipspotter.cogs.set",
        "clipspotter.cogs.display",
        "clipspotter.cogs.clip_range",
        "clipspotter.cogs.status",
        "clipspotter.cogs.remove",
        "clipspotter.cogs.set_streamer",
        "clipspotter.cogs.cshelp",
    ]
    # リスト内の各Cogに対してbot.load_extension()を呼び出し、拡張機能を読み込む
    for cog in initial_extensions:
        await bot.load_extension(cog)


# メイン処理（Botの起動）
async def main():
    async with bot:
        # 拡張機能を読み込み、Botを起動
        await load_extensions()
        await bot.start(ACCESS_TOKEN)

ゲーム名の類似度

DBからゲーム情報を取り出す際、rapidfuzzとlevenshteinライブラリを使用して、ゲームの類似度を計算しています。これにより、ユーザーが正確なゲーム名を覚えていなくても、近い名前で検索できるようにしました。

ユーザー名と表示名（set使用時）

Twitchアカウントには、名前に関して2つの概念があります：ユーザー名（Username）と表示名（Display Name）です。

  1. ユーザー名：英数字と"_"のみ使用可能
  2. 表示名：ユーザー名で使える文字に加え、日本語や韓国語、中国語などのマルチバイトのUnicode文字も使用可能

Twitch APIには多数のエンドポイントがありますが、これらのエンドポイントの多くはユーザー名を使用するエンドポイントであり、表示名を使用するエンドポイントは存在しません。
しかし、日本語を含む非ASCII文字を使用するユーザーにとっては、ローカライズされた表示名を使えないことは不便です。そこで本Botでは以下のようにストリーマー情報を設定できるようにしました。

  1. 初回登録時：ユーザー名での登録が必要（set）
  2. 2回目以降：表示名でのコマンド実行が可能（set-streamer）

さいごに

最後まで読んでいただきありがとうございました！
クリップを通して新たなストリーマー、ゲームを見つけられる一助となれば幸いです！

1
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?