DiscordチャットボットをPythonでつくる (6) - 「クイズ」機能を追加する
今回は、ユーザの返信を待機する方法を紹介し、
それを利用してチャットボットに新たに「クイズ」機能を追加します。
前回までのまとめ
前回は、ユーザメッセージに返信する方法を紹介しました。
そして、その仕組を利用した「おみくじ」機能を追加しました。
import discord
import random
class MyClient(discord.Client):
async def on_ready(self):
"""Discord接続時に実行される関数."""
print(f'{self.user}として接続しました。')
async def on_message(self, message: discord.Message):
"""メッセージ受信時に実行される関数."""
# ボット自身のメッセージは無視する
if message.author == self.user:
return
print(f'{message.author}よりメッセージを受信しました: {message.content}')
if message.content == '!omikuji':
choice = random.choice(['大吉', '吉', '小吉', '凶', '大凶'])
await message.channel.send(f"あなたの今日の運勢は **{choice}** です!")
with open('.discord_token') as f:
token = f.read().strip()
intents = discord.Intents.default()
intents.message_content = True
client = MyClient(intents=intents)
client.run(token)
ユーザからの返信を待機する
前回は、ユーザからのメッセージに対しチャットボットが応答する、
1往復のコミュニケーションができるようになりました。
それでは、さらにチャットボットの応答に対するユーザの返信を受け取りたい場合はどうしたら良いでしょう?
これはdiscord.Clientのwait_forメソッドで実現できます。
wait_forでユーザの返信を待機する
メッセージを待機するには以下のようにします(参照: ドキュメント):
reply_message = await client.wait_for('message', check=<メッセージ判定関数>, timeout=<タイムアウト秒数>)
<メッセージ判定関数>の部分に、「どのメッセージを返答と判定するか」を決める関数を指定します。<タイムアウト秒数>の部分は、返答を待機する最大秒数を指定します。- 時間内に返答が帰ってきた場合、
reply_message変数にその返答が代入されます。 - タイムアウトの場合、
asyncio.TimeoutErrorエラーを出力します。
「クイズ」機能を追加する。
それでは、早速wait_forを使用してクイズ機能を追加しましょう。
仕様を決める
クイズ機能は以下の仕様とします。
!quizと入力すると発動する。- 用意しておいた3択のクイズ集の中からランダムに1問出題する。
- ユーザの応答を最大30秒待つ。
- 応答に応じて「正解」または「不正解」を返答する。
実装する
では実装していきましょう。
3択クイズ集を用意する
まずは使用するクイズを用意します。
今回は、各問題を以下のような形式としたJSONファイルとして用意することにします。
{
"question": "Pythonの`enumerate`関数は何を返しますか？",
"choices": ["インデックスのリスト", "要素のリスト", "インデックスと要素のタプルのリスト"],
"answer_num": 3
},
この解説では、ChatGPTに作成してもらった次のPython問題集を使用します: サンプルデータ
クイズデータをどこで読み込むか
JSONファイルに記録したクイズデータですが、!quizコマンド実行時に毎回読み込むのでは動作が遅くなります。
そのため、ボット起動時に読み込んでおくのが良いでしょう。
以下のようにMyClientの初期化時にクイズデータを読み込むことにします。
import json
...
class MyClient(discord.Client):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
quiz_file = 'python_problems.json'
with open(quiz_file) as f:
self.quiz_list = json.load(f)
メッセージ判定関数
メッセージ判定関数ですが、今回3択であるため、数値であればOKと判定することにします。
また、!quizと入力したユーザによるメッセージである必要があります。
よって判定関数 quiz_reply_check は以下のようになります。
class MyClient(discord.Client):
...
async def on_message(self, message: discord.Message):
...
if message.content == '!quiz':
def quiz_reply_check(m: discord.Message):
"""クイズの返答を判定する関数."""
return m.author == message.author and m.content.isdigit()
クイズを出題する
クイズ出題部分です。クイズデータはself.quiz_listに入れてあるのでここからランダムに選択して出題します。
class MyClient(discord.Client):
...
async def on_message(self, message: discord.Message):
...
if message.content == '!quiz':
...
quiz = random.choice(self.quiz_list)
quiz_str = (
f'Q: {quiz["question"]}\n\n' +
'\n'.join([f'[{i+1}] {c}' for (i, c) in enumerate(quiz["choices"])])
)
await message.channel.send(quiz_str)
解答を待つ
解答を待つには wait_for メソッドを使用しますが、
この関数は時間切れの場合に asyncio.TimeoutError エラーを出力します。
そのため、try構文を使用してエラー処理を行いましょう。
import asyncio
...
class MyClient(discord.Client):
...
async def on_message(self, message: discord.Message):
...
if message.content == '!quiz':
...
await message.channel.send(quiz_str)
try:
reply_message = await self.wait_for('message', check=quiz_reply_check, timeout=30)
...
except asyncio.TimeoutError:
...
解答に応じて返答する
最後に、解答に応じてメッセージを返します。
class MyClient(discord.Client):
...
async def on_message(self, message: discord.Message):
...
if message.content == '!quiz':
...
await message.channel.send(quiz_str)
try:
reply_message = await self.wait_for('message', check=quiz_reply_check, timeout=30)
if reply_message.content == str(quiz['answer_num']):
result = "😀 正解!"
else:
result = f"😢 残念、正解は {quiz['answer_num']}でした。"
except asyncio.TimeoutError:
result = f"😢 時間切れです。正解は {quiz['answer_num']}でした。"
await message.channel.send(result)
実行する
それでは実行してみましょう。
!quiz コマンドを実行するとランダムに問題が出力されます。
正解の場合、不正解の場合、タイムアウトの場合、数値でない文字を入力した場合等、
色々なパターンを試してみてください。
まとめ
今回は、返答を待機する wait_for 関数を紹介し、それを利用してクイズ機能を新たに追加しました。
しかし、クイズ機能はこれで完成ではありません！
…ですが続きを進める前に、on_message関数が少しごちゃごちゃしてきたので、
次回はリファクタリングを行いたいと思います。
今回のコード
import asyncio
import json
import random
import discord
class MyClient(discord.Client):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
quiz_file = 'python_problems.json'
with open(quiz_file) as f:
self.quiz_list = json.load(f)
async def on_ready(self):
"""Discord接続時に実行される関数."""
print(f'{self.user}として接続しました。')
async def on_message(self, message: discord.Message):
"""メッセージ受信時に実行される関数."""
# ボット自身のメッセージは無視する
if message.author == self.user:
return
print(f'{message.author}よりメッセージを受信しました: {message.content}')
if message.content == '!omikuji':
choice = random.choice(['大吉', '吉', '小吉', '凶', '大凶'])
await message.channel.send(f"あなたの今日の運勢は **{choice}** です!")
if message.content == '!quiz':
def quiz_reply_check(m: discord.Message):
"""クイズの返答を判定する関数."""
return m.author == message.author and m.content.isdigit()
quiz = random.choice(self.quiz_list)
quiz_str = (
f'Q: {quiz["question"]}\n\n' +
'\n'.join(
[f'[{i+1}] {c}' for (i, c) in enumerate(quiz["choices"])]
)
)
await message.channel.send(quiz_str)
try:
reply_message = await self.wait_for(
'message', check=quiz_reply_check, timeout=30
)
if reply_message.content == str(quiz['answer_num']):
result = "😀 正解!"
else:
result = f"😢 残念、正解は {quiz['answer_num']}でした。"
except asyncio.TimeoutError:
result = f"😢 時間切れです。正解は {quiz['answer_num']}でした。"
await message.channel.send(result)
with open('.discord_token') as f:
token = f.read().strip()
intents = discord.Intents.default()
intents.message_content = True
client = MyClient(intents=intents)
client.run(token)
