DiscordチャットボットをPythonでつくる (7) - discord.pyの拡張機能を用いてリファクタリングする
シリーズ - DiscordチャットボットをPythonでつくる
コンテンツ
今回は、discord.pyの拡張ライブラリdiscord.ext.commandsを使用して、
前回少し複雑化していたon_message関数のリファクタリングを行います。
前回までのまとめ
前回までのコードは以下のとおりです。
!omikujiコマンドと!quizコマンドが追加され、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)
拡張ライブラリdiscord.ext.commandsを使用する
前回までのようにコマンドを複数追加していくと、次第に同じような処理も増え、
on_messageが次第に肥大化していきます。
そのため、discord.pyにはコマンド追加を簡潔かつ容易にする
拡張ライブラリ discord.ext.commands が用意されています。
(参照: APIリファレンス, APIリファレンス2)
それでは、このコマンド拡張機能を使ってコードのリファクタリングを行っていきましょう。
1. commands.Bot を継承する
はじめに、前回までは discord.Client を継承して独自のクライントMyClientを作成していましたが、
これを commands.Bot を継承するように変更します。
...
from discord.ext import commands
class MyClient(commands.Bot):
...
2. __init__関数の引数を変更する
command.Botでは、__init__関数の引数にcommand_prefixという値が追加されているため、この調整を行います。
この値はコマンドの先頭に付ける記号を表すので、デフォルトで!を入れておくことにします。
class MyClient(commands.Bot):
def __init__(self, command_prefix='!', *args, **kwargs):
super().__init__(command_prefix, *args, **kwargs)
...3. 各コマンドの定義を分離する
次に、各コマンドの定義をon_messageから分離して関数化します。
分離した関数は、commands.Context型のctxを第一引数とした関数にします。
先に結論を書くと、!omikujiコマンドの場合は以下のようになります。
@commands.command(name='omikuji')
async def omikuji_command(ctx: commands.Context):
"""おみくじを引く."""
choice = random.choice(['大吉', '吉', '小吉', '凶', '大凶'])
await ctx.channel.send(f"あなたの今日の運勢は **{choice}** です!")- 関数には
@commands.command()デコレータを付与します。nameでコマンド名を指定します。 - 関数の第一引数を
ctx: commands.Contextとします。
ctxは文脈(コンテキスト)を意味する変数であり、ctx.channel.sendで現在のチャンネルにメッセージを送信できます。
他の属性に関しては、APIリファレンスを参照してください。 - docstringの内容がコマンドヘルプに表示されます。
同様に!quizコマンドも分離すると以下のようになります。
@commands.command(name='quiz')
async def quiz_command(ctx: commands.Context):
"""3択クイズを出題する."""
def quiz_reply_check(m: discord.Message):
"""クイズの返答を判定する関数."""
return m.author == ctx.author and m.content.isdigit()
quiz = random.choice(ctx.bot.quiz_list)
quiz_str = (
f'Q: {quiz["question"]}\n\n' +
'\n'.join(
[f'[{i+1}] {c}' for (i, c) in enumerate(quiz["choices"])]
)
)
await ctx.channel.send(quiz_str)
try:
reply_message = await ctx.bot.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 ctx.channel.send(result)ctx.authorでメッセージの送信者が取得できます。- その他の
selfであった部分をctx.botに変更しています。
4. add_command でコマンドを追加する
分離したコマンドを以下のようにしてボットに追加します。
client = MyClient(intents=intents)
bot_commands = [omikuji_command, quiz_command]
for command in bot_commands:
client.add_command(command)
client.run(token)
5. on_message で継承元のon_messageを実行する
最後に、on_messageで継承元のcommands.Botのon_messageを実行するように、
以下のように変更します。
async def on_message(self, message: discord.Message):
"""メッセージ受信時に実行される関数."""
# ボット自身のメッセージは無視する
if message.author == self.user:
return
print(f'{message.author}よりメッセージを受信しました: {message.content}')
await super().on_message(message)6. 調整後のコードまとめ
上記調整後のコードは以下のようになりました。
import asyncio
import json
import random
import discord
from discord.ext import commands
class MyClient(commands.Bot):
def __init__(self, command_prefix='!', *args, **kwargs):
super().__init__(command_prefix, *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}')
await super().on_message(message)
@commands.command(name='omikuji')
async def omikuji_command(ctx: commands.Context):
"""おみくじを引く."""
choice = random.choice(['大吉', '吉', '小吉', '凶', '大凶'])
await ctx.channel.send(f"あなたの今日の運勢は **{choice}** です!")
@commands.command(name='quiz')
async def quiz_command(ctx: commands.Context):
"""3択クイズを出題する."""
def quiz_reply_check(m: discord.Message):
"""クイズの返答を判定する関数."""
return m.author == ctx.author and m.content.isdigit()
quiz = random.choice(ctx.bot.quiz_list)
quiz_str = (
f'Q: {quiz["question"]}\n\n' +
'\n'.join(
[f'[{i+1}] {c}' for (i, c) in enumerate(quiz["choices"])]
)
)
await ctx.channel.send(quiz_str)
try:
reply_message = await ctx.bot.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 ctx.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)
bot_commands = [omikuji_command, quiz_command]
for command in bot_commands:
client.add_command(command)
client.run(token)コマンド毎にファイルを分ける
次に、分離した各コマンド用の関数を、管理を容易にするため別ファイルに分けましょう。
commandsディレクトリを作成し、そこにコマンド別のファイルを作成して分離します。
-
commands/omikuji.pyimport random from discord.ext import commands @commands.command(name='omikuji') async def omikuji_command(ctx: commands.Context): """おみくじを引く.""" choice = random.choice(['大吉', '吉', '小吉', '凶', '大凶']) await ctx.channel.send(f"あなたの今日の運勢は **{choice}** です!") -
commands/quiz.pyimport asyncio import random import discord from discord.ext import commands @commands.command(name='quiz') async def quiz_command(ctx: commands.Context): """3択クイズを出題する.""" def quiz_reply_check(m: discord.Message): """クイズの返答を判定する関数.""" return m.author == ctx.author and m.content.isdigit() quiz = random.choice(ctx.bot.quiz_list) quiz_str = ( f'Q: {quiz["question"]}\n\n' + '\n'.join( [f'[{i+1}] {c}' for (i, c) in enumerate(quiz["choices"])] ) ) await ctx.channel.send(quiz_str) try: reply_message = await ctx.bot.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 ctx.channel.send(result) -
chatbot.py(カスタマイズしやすいようにbot_commandsを上に持ってきました。)import json import discord from discord.ext import commands from commands.omikuji import omikuji_command from commands.quiz import quiz_command BOT_COMMANDS = [omikuji_command, quiz_command] class MyClient(commands.Bot): def __init__(self, command_prefix='!', *args, **kwargs): super().__init__(command_prefix, *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}') await super().on_message(message) with open('.discord_token') as f: token = f.read().strip() intents = discord.Intents.default() intents.message_content = True client = MyClient(intents=intents) for command in BOT_COMMANDS: client.add_command(command) client.run(token)
クイズデータの読み込みを __init__ から取り除きたい
ここまで、機能別にファイルを分離してリファクタリングを行ってきました。
しかし現状のファイルを見たところ、MyClientの __init__ でクイズデータを読み込んでおり、
この部分だけ機能別に分離できていないのが気になります。
いくつかの方法が考えられますが、今回は__init__のクイズファイル読み込み部分を取り除き、
quiz.py をロードした際に読み込むように、以下のように変更しましょう。
import asyncio
import json
import random
import discord
from discord.ext import commands
quiz_file = 'python_problems.json'
with open(quiz_file) as f:
quiz_list = json.load(f)
@commands.command(name='quiz')
async def quiz_command(ctx: commands.Context):
"""3択クイズを出題する."""
def quiz_reply_check(m: discord.Message):
"""クイズの返答を判定する関数."""
return m.author == ctx.author and m.content.isdigit()
quiz = random.choice(quiz_list)
quiz_str = (
f'Q: {quiz["question"]}\n\n' +
'\n'.join(
[f'[{i+1}] {c}' for (i, c) in enumerate(quiz["choices"])]
)
)
await ctx.channel.send(quiz_str)
try:
reply_message = await ctx.bot.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 ctx.channel.send(result)実行してみる
最後に、ちゃんとコマンドが機能するか確認しましょう。
また、!helpコマンドでコマンド一覧が表示されます。
まとめ
今回は、discord.pyの拡張機能であるdiscord.ext.commandsを使用して、
コードのリファクタリングを行いました。
機能別にファイルを分けることで、コードが少しスッキリして管理もしやすくなりました。
次回は、discord.ext.commandsの機能についてより掘り下げを行い、クイズ機能の強化を行います。
今回のコード
-
chatbot.pyimport discord from discord.ext import commands from commands.omikuji import omikuji_command from commands.quiz import quiz_command BOT_COMMANDS = [omikuji_command, quiz_command] class MyClient(commands.Bot): def __init__(self, command_prefix='!', *args, **kwargs): super().__init__(command_prefix, *args, **kwargs) 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}') await super().on_message(message) with open('.discord_token') as f: token = f.read().strip() intents = discord.Intents.default() intents.message_content = True client = MyClient(intents=intents) for command in BOT_COMMANDS: client.add_command(command) client.run(token) -
commands/omikuji.pyimport random from discord.ext import commands @commands.command(name='omikuji') async def omikuji_command(ctx: commands.Context): """おみくじを引く.""" choice = random.choice(['大吉', '吉', '小吉', '凶', '大凶']) await ctx.channel.send(f"あなたの今日の運勢は **{choice}** です!") -
commands/quiz.pyimport asyncio import json import random import discord from discord.ext import commands quiz_file = 'python_problems.json' with open(quiz_file) as f: quiz_list = json.load(f) @commands.command(name='quiz') async def quiz_command(ctx: commands.Context): """3択クイズを出題する.""" def quiz_reply_check(m: discord.Message): """クイズの返答を判定する関数.""" return m.author == ctx.author and m.content.isdigit() quiz = random.choice(quiz_list) quiz_str = ( f'Q: {quiz["question"]}\n\n' + '\n'.join( [f'[{i+1}] {c}' for (i, c) in enumerate(quiz["choices"])] ) ) await ctx.channel.send(quiz_str) try: reply_message = await ctx.bot.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 ctx.channel.send(result)