discordのボットをAWS Lambdaで(wslなしのWindows環境で)作る with Python

## 概要

- Windows (**wslなし**) で、discord のスラッシュコマンドのボットを AWS Lambda を用いて作成したよ

wsl なしの謎縛りプレイですが、Python 3.8 なら簡単に実現できることがわかったので共有します。
→ 注: Python 3.8は2024/10でサポート終了らしい...
→ 注2: 3.12でも簡単にできそう！

本記事は、下記リポジトリの部分的な解説も兼ねています。

https://github.com/Take-Me1010/DMP-bot-lambda/tree/main

## 環境

- Windows 10
- Python 3.8.0
    - 3.8じゃないと、windowsだけでは難しそうです
- pipenv 2023.12.1
    - 別にPipenvでなくても、環境変数が読み込めて、かつPythonのバージョン管理ができれば何でもよいです
- PowerShell 7.4.2

## 本記事の新規性

- wsl なしで windows 環境のみでディスコボット作ったよ
- event のボディの型を作ったよ

## 注意点

ここでは、ギルド (特定のサーバー) 限定のスラッシュコマンドを実装します。
Python の文法等は解説しません。
また、最初に述べたように、[身内サーバーで運用しているBOT](https://github.com/Take-Me1010/DMP-bot-lambda/tree/main)の部分的な解説を含んでいます。

## 作り方

ボットの作成と、使いたいサーバー (開発用サーバー等) への招待を済ませましょう。

https://qiita.com/1ntegrale9/items/cb285053f2fa5d0cccdf

また、AWSへの登録を済ませておいてください。

ボット作成の流れ
1. トークン等の取得
1. AWS Lambda のハンドラー作成
1. コマンドの登録
1. アップロード用の zip 作成
1. API Gateway の追加とデプロイ
1. API Gateway の URL をポータルへコピペ
1. 完成！

### まずはディレクトリとかの整理

```ps1
# 3.8じゃないとだめです。それ以外のバージョンを使うにはwslを要します。
pipenv --python 3.8
# 実はインストールしなくても大丈夫です (後述)
pipenv install pynacl
# コマンドの登録に利用するスクリプトで使います
pipenv install --dev requests

mkdir app
```

### トークンの設定

適当なディレクトリ (以下 bot_project_dir) で .env ファイルを作っておきます。

```.env
APP_ID = "YOUR_APP_ID"
SERVER_ID = "YOUR_SERVER_ID"
BOT_TOKEN = "YOUR_BOT_TOKEN"
```

`APP_ID` は、dev portal で取得できるはず。
`SERVER_ID` は、ディスコの開発者モードをクライアントアプリで有効化して、サーバーで右クリックすると見れるはず。

### AWS Lambda のハンドラー作成

AWS Lambda では、リクエストを処理するハンドラー関数を定義して指定する必要があります。
このセクションでは、下記のハンドラーを説明します。
詳細に興味がない場合は、下記のコードをapp/lambda_function.pyに貼り付けてください。
[discord_types.py](https://github.com/Take-Me1010/DMP-bot-lambda/blob/main/app/discord_types.py), [lambda_types.py](https://github.com/Take-Me1010/DMP-bot-lambda/blob/main/app/lambda_types.py) をダウンロード(またはコピペ)して、appフォルダにいれてください。

```python: app/lambda_function.py
import json

from nacl.signing import VerifyKey
from nacl.exceptions import BadSignatureError

from discord_types import (
    DiscordCommandRequestBody,
    DiscordCommandResponseBody,
    DiscordResponseInteractionType,
)
from lambda_types import (
    MyApiGatewayEvent,
    LambdaContext,
    MyApiGatewayResponse,
)

PUBLIC_KEY = "YOUR_PUBLIC_KEY"

def toResponse(
    body: DiscordCommandResponseBody, statusCode=200
) -> MyApiGatewayResponse:
    return {
        "statusCode": statusCode,
        "headers": {
            "Content-Type": "application/json",
        },
        "body": json.dumps(body),
    }

def discord_response(body: DiscordCommandRequestBody) -> MyApiGatewayResponse:
    name = body["data"]["name"]

    res: DiscordCommandResponseBody
    if name == "hello":
        res = {
            "type": DiscordResponseInteractionType.CHANNEL_MESSAGE_WITH_SOURCE,
            "data": {"content": "World!"},
        }
    else:
        res = {
            "type": DiscordResponseInteractionType.CHANNEL_MESSAGE_WITH_SOURCE,
            "data": {"content": "Command not found."},
        }
    return toResponse(res)

def verify_request(event: MyApiGatewayEvent):
    """
    Verify the request signature.
    """
    signature = event["headers"]["x-signature-ed25519"]
    timestamp = event["headers"]["x-signature-timestamp"]

    verify_key = VerifyKey(bytes.fromhex(PUBLIC_KEY))

    message = timestamp + event["body"]

    try:
        verify_key.verify(message.encode(), signature=bytes.fromhex(signature))
    except BadSignatureError:
        return False

    return True
    
def lambda_handler(
    event: MyApiGatewayEvent, context: LambdaContext
) -> MyApiGatewayResponse:
    """
    Entry point for AWS Lambda.
    """
    if not verify_request(event):
        return {
            "statusCode": 401,
            "headers": {
                "Content-Type": "application/json",
            },
            "body": json.dumps("invalid request signature"),
        }

    body: DiscordCommandRequestBody = json.loads(event["body"])
    t = body["type"]
    if t == 1:
        # handle ping
        return {
            "statusCode": 200,
            "headers": {
                "Content-Type": "application/json",
            },
            "body": json.dumps({"type": DiscordResponseInteractionType.PONG}),
        }
    elif t == 2:  # handle application command
        return discord_response(body)

    return {
        "statusCode": 400,
        "headers": {
            "Content-Type": "application/json",
        },
        "body": json.dumps("invalid request type"),
    }
```

---
以下、解説。

基本的には、ハンドラーは下記のような形になります。

```python
def lambda_handler(event, context):
    return {
        "statusCode": 400,
        "headers": {
            "Content-Type": "application/json",
        },
        "body": json.dumps("invalid request type"),
    }
```

上記のように、`event`: リクエスト内容等を含むオブジェクトと`context`: AWS のサービスに関連する情報等を含むオブジェクトの二つを受け取って、http通信のレスポンスを返すものになっています。

`context` については、次の参考リンクで型情報がわかります。

https://github.com/aws/aws-lambda-python-runtime-interface-client/blob/main/awslambdaric/lambda_context.py

`event` については、`MyApiGatewayEvent` のような辞書のようです。

```python
# event の型
class MyApiGatewayEvent(tp.TypedDict):
    headers: tp.Dict[str, str]
    body: str

# レスポンスの型
class MyApiGatewayResponse(tp.TypedDict):
    statusCode: int
    headers: tp.Dict[str, str]  # "Content-Type: application/json" は必須っぽいです。
    body: str
```

`json.loads(event["body"])` は、次のような型 (辞書) になっていることがわかりました (公式ドキュメントと睨めっこ)。

```python
import typing as tp

class DiscordCommandRequestBodyDataOption(tp.TypedDict):
    type: DiscordApplicationCommandOptionType
    name: str
    value: tp.Any  # NOTE: type depends on `type`. e.g. str, int, float


class DiscordCommandRequestBodyData(tp.TypedDict):
    id: tp.Any
    name: str
    options: tp.List[DiscordCommandRequestBodyDataOption]


class DiscordCommandRequestBody(tp.TypedDict):
    type: tp.Literal[
        DiscordRequestInteractionType.PING,
        DiscordRequestInteractionType.APPLICATION_COMMAND,
    ]

    # only for `APPLICATION_COMMAND`
    data: DiscordCommandRequestBodyData
```

なお、上記の型は最低限のキーしか書いていませんので、複雑なコマンドの際には他にも色々必要見ないです。下記の公式リファレンスをご覧ください。

https://discord.com/developers/docs/interactions/application-commands#slash-commands

上記の型情報を踏まえると、ハンドラーは次のよう定義すればよいことになります。

```python
def lambda_handler(
    event: MyApiGatewayEvent, context: LambdaContext
) -> MyApiGatewayResponse:
    """
    Entry point for AWS Lambda.
    """
    if not verify_request(event):
        return {
            "statusCode": 401,
            "headers": {
                "Content-Type": "application/json",
            },
            "body": json.dumps("invalid request signature"),
        }

    body: DiscordCommandRequestBody = json.loads(event["body"])
    ...
```

では、後はコマンドに応じた処理を行うだけ...ではなくて、discordから飛んでくる ping に応答できる必要があります。

どうやら nacl とやらで認証を行えるようです:

https://github.com/ker0olos/aws-lambda-discord-bot

ここで windows 縛りの弊害なのですが、上記のチュートリアル通りに行うと動きません。下記のようにLambda のログで怒られます。

```log
[ERROR] Runtime.ImportModuleError: Unable to import module 'lambda_function': No module named 'nacl._sodium'
```

これは nacl がプラットフォームに依存したビルドを行っているからであり、windows 環境でビルドしたものと、Lambda のランタイム上の OS (Linux系) と互換性がないからです。

https://www.reddit.com/r/aws/comments/p1z95f/help_with_python_lambda_pynacl_module_issue/

最も簡単な解決策は WSL を利用することですが、ここでは Layers を利用することでこれを解決できます。
したがってここでは `pip install` **しません**。

ということで、nacl をインストールせずに、認証用の関数は、[参考記事](https://github.com/ker0olos/aws-lambda-discord-bot)よりほぼコピペして、次のように定義しましょう。

```python
def verify_request(event: MyApiGatewayEvent):
    """
    Verify the request signature.
    """
    signature = event["headers"]["x-signature-ed25519"]
    timestamp = event["headers"]["x-signature-timestamp"]

    verify_key = VerifyKey(bytes.fromhex(PUBLIC_KEY))

    message = timestamp + event["body"]

    try:
        verify_key.verify(message.encode(), signature=bytes.fromhex(signature))
    except BadSignatureError:
        return False

    return True
```

認証用関数を用いて、Discord が飛ばしてくる ping に対して
- 認証ができていないなら`401`を返す
- 認証ができているなら`200`を返す

ようなハンドラーを実装します。

```python
def lambda_handler(
    event: MyApiGatewayEvent, context: LambdaContext
) -> MyApiGatewayResponse:
    """
    Entry point for AWS Lambda.
    """
    if not verify_request(event):
        return {
            "statusCode": 401,
            "headers": {
                "Content-Type": "application/json",
            },
            "body": json.dumps("invalid request signature"),
        }

    body: DiscordCommandRequestBody = json.loads(event["body"])
    t = body["type"]  # 1 は ping, 2 はコマンド と決まっています。
    if t == 1:
        # handle ping
        return {
            "statusCode": 200,
            "headers": {
                "Content-Type": "application/json",
            },
            "body": json.dumps({"type": DiscordResponseInteractionType.PONG}),
        }
    elif t == 2:  # handle application command
        ...
```

これで後は、コマンドに応じた処理を書くだけです。
ハンドラーの最後に、受け取ったボディを処理してレスポンスを返す関数を定義します。

```python
def discord_response(body: DiscordCommandRequestBody) -> MyApiGatewayResponse:
    name = body["data"]["name"]

    res: DiscordCommandResponseBody
    # hello コマンドをとりあえず定義
    if name == "hello":
        res = {
            "type": DiscordResponseInteractionType.CHANNEL_MESSAGE_WITH_SOURCE,
            "data": {"content": "World!"},
        }
    else:
        res = {
            "type": DiscordResponseInteractionType.CHANNEL_MESSAGE_WITH_SOURCE,
            "data": {"content": "Command not found."},
        }
    return toResponse(res)
```

### コマンドの登録

スラッシュコマンドの登録は、リクエストをディスコのAPIに投げることで行います。


```python: register.py
import os

import requests


APP_ID = os.environ["APP_ID"]
SERVER_ID = os.environ["SERVER_ID"]
BOT_TOKEN = os.environ["BOT_TOKEN"]

url = (
    f"https://discord.com/api/v10/applications/{APP_ID}/guilds/{SERVER_ID}/commands"
)

payload = [
    {
        "name": "hello",
        "description": "return world",
        "options": []  # 引数の設定等。今回は引数なしなので空の配列で。 See https://discord.com/developers/docs/interactions/application-commands#application-command-object-application-command-option-structure
    }
]

response = requests.put(
    url, headers={"Authorization": f"Bot {BOT_TOKEN}"}, json=payload
)

print(response.json())
```

このスクリプトを実行。
```
pipenv run python -u register.py
```

成功だと200が返ってくる。
サーバーコマンドなので、ほぼ即時にコマンドが登録されます。

### アップロード用の zip 作成

Lambda では、依存関係も全てまとめてアップロードする必要があります。
今回、pynaclに依存していますが、これは後ほどLambdaの Layer で解決するので、単に zip を作るだけです。

```python
Compress-Archive ./app/*.py -DestinationPath app.zip -Force
```

出来上がったzipファイルを、AWS Lambda でアップロードしてください。

参考 (公式):

https://docs.aws.amazon.com/lambda/latest/dg/python-package.html

### API Gateway の追加とデプロイ

トリガーとして、API Gateway を追加します。
設定はすべてデフォルト値でおっけー。

Lambda の Layers に arn 値を指定で、下記の値を入力・追加してください。
```
arn:aws:lambda:ap-northeast-1:770693421928:layer:Klayers-p38-PyNaCl:2
```

これを追加すると、**PyNaCl がインストールされた環境が用意されます**。

これが Python 3.8 しか使えない所以で、先人が作成した Layers は下記のリポジトリで見ることができます。

https://github.com/keithrozario/Klayers/tree/master

PyNaCl がインストールされた Python の最新バージョンが 3.8 だったのです。~~本当は 3.12 使いたかった...~~
→ 自分でLayerを簡単に作成できるので、Python 3.12にすんなりと移行できました！

https://medium.com/@geoff.ford_33546/creating-a-pynacl-lambda-layer-c3f2e1b6ff11

最後に、urlの値をコピーして、discord dev portal の INTERACTIONS ENDPOINT URL の欄に貼り付けます。
登録の際に早速 ping が飛ぶようで、ここで蹴られると実装が間違っていることになります。
Cloud Watch等でログを確認してみてください。

## まとめ

本記事では、wslを使わずにwindowsだけでAWSのサーバーレス構成のディスコボットを作成するtipsを紹介しました。
また、Python 用の型情報を考察・定義しました。
型ガチガチPythonで、wslなし縛りwindows環境下で、ディスコボットを開発する非常に稀な人種に役に立つことを願います。