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

# はじめに
初めまして。Qiita初投稿のHagataです。

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

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


https://github.com/hagatasdelus/ClipSpotter_for_personal


本記事では、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. 必要な環境変数を設定（詳細は後述）

https://discord.com/developers/applications

https://dev.twitch.tv/console

## 環境変数
環境変数はプロジェクトのルートディレクトリに`.env`ファイルを作成し、以下のように環境変数を設定します：
```plaintext:.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で行い、クリップ情報の取得ロジックを実装しました。以下は主要な部分のコードサンプルです。
```python: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の記述内容を一部抜粋します。
```python: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）

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