初めに
discord botを作成する際に使うと思われる機能群を逆引きで調べられるようにほぼ備忘録的に書きます。
ついでにbotをcog化する方法も書きます。ついでにコード内で文章では触れてない、陥りやすい罠にも触れます。
実行環境
python 3.9
discord.py 1.6.0
requests 2.24.0
psycopg2 2.8.5
Heroku (環境変数などはHerokuに保存して、それをos.environで取得して運用します。)
cog化
botを作っていくと全体のコードが長くなっていきます。これを続けると非常に見づらく、例えばバグが発生したときに発生箇所を探すのが面倒臭くなります。そこで、機能単位でファイルを分けられるcog, Extensionの考え方にとりあえず慣れましょう。
詳しくはこの記事にまとまっています。非常にわかりやすいので一読することをお勧めします。
https://qiita.com/Takkun0530/items/45b4a6acd7c74e651ec2
Extensionの考え方は以下から(公式リファレンス)
https://discordpy.readthedocs.io/ja/latest/ext/commands/extensions.html
私は簡単に実装するための下地を載せることにします。
cog化するための下地
import os
import discord
from discord import Embed
from discord.ext import commands
class BOT_NAME(commands.Bot):
def __init__(self, prefix):
intents = discord.Intents.all() # discordにはIntentの考えが最近追加されました。とりあえず全部許可する方向で書きます。
super().__init__(command_prefix=prefix, help_command=None, intents=intents)
self.cur = cur
for cog in os.listdir(f"./cogs"): # cogの読み込み
if cog.endswith(".py"):
try:
self.load_extension(f"cogs.{cog[:-3]}")
except Exception:
traceback.print_exc()
async def on_command_error(self, ctx, error):
db.commit()
"""すべてのコマンドで発生したエラーを拾う"""
if isinstance(error, commands.CommandInvokeError): # コマンド実行時にエラーが発生したら
orig_error = getattr(error, "original", error)
error_msg = ''.join(traceback.TracebackException.from_exception(orig_error).format())
error_message = f'```{error_msg}```'
ch = ctx.guild.get_channel(CHANNEL_ID) # botで不具合が発生したときここで指定したチャンネルにエラー文を出力します。
d = datetime.now() # 現在時刻の取得
time = d.strftime("%Y/%m/%d %H:%M:%S")
embed = Embed(title='Error_log', description=error_message, color=0xf04747)
embed.set_footer(text=f'channel:{ctx.channel}\ntime:{time}\nuser:{ctx.author.display_name}')
await ch.send(embed=embed)
if __name__ == '__main__':
bot = BOT_NAME(prefix="!") # prefix(コマンドの接頭辞)を定める。
bot.run(os.environ['TOKEN'])
そして、同ディレクトリ内に新しくフォルダを1つ作り、名前をcogsとします。(フォルダ名の理由はos.listdir(f"./cogs")の部分で察してください。)
そしてそのフォルダ内に実装する機能群をファイルに分けながら記述していきます。その下地も載せておきます。
例としてMinecraftのゲームに登場する概念である、1LC, 2stなどの記号を整数値に書き換えるコマンド!csコマンドを書いています。
機能実装用の下地
import traceback
from discord.ext import commands
import discord
import os
from datetime import datetime
"""Stackを計算するコマンド"""
# 機能群でclassに纏める。この例だとコマンドは1つしかないが複数置くことも可能。
class StackCalc(commands.Cog):
# 初期設定。基本はこれで問題ないが、実装内容によってはここを変える必要がある。(初期で変数に何か値を入れたい場合はここで操作する)
def __init__(self, bot):
self.bot = bot
# コマンドを配置する際はデコレーターを入れる。
# @commands.command() (コマンドの場合、コマンド名を複数定めたい場合(!csでも!check_stackでもこのコマンドを動かしたい場合)、
# @commands.command(aliases=["check_stack"])のように定める。),
# @commands.Cog.listener() (on_messageなど、discord.pyで既にある関数の場合),
# @staticmethod (ファイル外で使用する場合)
# @tasks.loop() (繰り返し処理する場合, seconds=20, minutes=2などの定め方ができる。) などが自分ではよく使うデコレータ。
@commands.command()
async def cs(self, ctx, amount):
try:
try:
if int(amount):
if self.bot.stack_check_reverse(amount) == 0:
await ctx.channel.send(f"入力した値が0または不正な値です。")
return
else:
await ctx.channel.send(f"{amount}はスタック表記で{self.bot.stack_check_reverse(amount)}です。")
except ValueError:
if self.bot.stack_check(amount) == 0:
await ctx.channel.send(f"入力した値が0または不正な値です。")
return
else:
await ctx.channel.send(f"{amount}は整数値で{self.bot.stack_check(amount)}です。")
except Exception as e:
orig_error = getattr(e, "original", e)
error_msg = ''.join(traceback.TracebackException.from_exception(orig_error).format())
error_message = f'```{error_msg}```'
ch = self.bot.get_channel(628807266753183754)
d = datetime.now() # 現在時刻の取得
time = d.strftime("%Y/%m/%d %H:%M:%S")
embed = discord.Embed(title='Error_log', description=error_message, color=0xf04747)
embed.set_footer(text=f'stack_calc\ntime:{time}\nuser:None')
await ch.send(embed=embed)
def setup(bot):
bot.add_cog(StackCalc(bot)) #
例えば、上記下地で登場したcsコマンドにて、self.bot.stack_check(amount)という関数が存在しますが、これは先ほど挙げたcogの下地ファイルに以下のように追記します。ついでにファイルごとに権限設定ができるのでその例も載せます。
cog化するための下地(stack_checkを載せたバージョン)
import os
import discord
import random
from discord import Embed
from discord.ext import commands
class BOT_NAME(commands.Bot):
def __init__(self, prefix):
intents = discord.Intents.all() # discordにはIntentの考えが最近追加されました。とりあえず全部許可する方向で書きます。
super().__init__(command_prefix=prefix, help_command=None, intents=intents)
self.cur = cur
for cog in os.listdir(f"./cogs"): # cogの読み込み
if cog.endswith(".py"):
try:
self.load_extension(f"cogs.{cog[:-3]}")
except Exception:
traceback.print_exc()
async def cog_check(self, ctx): # cog内のコマンド全てに適用されるcheck。例えばadminロールがついてないと利用できないようにしたい場合
if discord.utils.get(ctx.author.roles, name="admin"):
return True
else:
await ctx.send('運営以外のコマンド使用は禁止です')
return False
async def on_ready(self):
color = [0x126132, 0x82fc74, 0xfea283, 0x009497, 0x08fad4, 0x6ed843, 0x8005c0]
await self.get_channel(CHANNEL_ID).send(
embed=discord.Embed(description="起動しました", color=random.choice(color))) # channel_idに起動を知らせるembedを出力。
@staticmethod # 今回の場合、他のファイルでこの関数を使用したい&どのファイルからアクセスしても動作に変わりがないのでstaticmethodとする。
def stack_check(value) -> int:
"""
[a lc + b st + c]がvalueで来ることを想定する(関数使用前に文の構造確認を取る)
少数出来た場合、少数で計算して最後にintぐるみをして値を返す
:param value: [a lc + b st + c]の形の価格
:return: 価格をn個にしたもの(少数は丸め込む)
"""
value = str(value).replace("椎名", "").lower()
stack_frag = False
lc_frag = False
calc_result = [0, 0, 0]
if "lc" in value:
lc_frag = True
if "st" in value:
stack_frag = True
try:
data = value.replace("lc", "").replace("st", "").replace("個", "").split("+")
if lc_frag:
calc_result[0] = data[0]
data.pop(0)
if stack_frag:
calc_result[1] = data[0]
data.pop(0)
try:
calc_result[2] = data[0]
except IndexError:
pass
a = float(calc_result[0])
b = float(calc_result[1])
c = float(calc_result[2])
d = int(float(a * 3456 + b * 64 + c))
if d <= 0:
return 0
else:
return d
except ValueError:
return 0
async def on_command_error(self, ctx, error):
db.commit()
"""すべてのコマンドで発生したエラーを拾う"""
if isinstance(error, commands.CommandInvokeError): # コマンド実行時にエラーが発生したら
orig_error = getattr(error, "original", error)
error_msg = ''.join(traceback.TracebackException.from_exception(orig_error).format())
error_message = f'```{error_msg}```'
ch = ctx.guild.get_channel(CHANNEL_ID) # botで不具合が発生したときここで指定したチャンネルにエラー文を出力します。
d = datetime.now() # 現在時刻の取得
time = d.strftime("%Y/%m/%d %H:%M:%S")
embed = Embed(title='Error_log', description=error_message, color=0xf04747)
embed.set_footer(text=f'channel:{ctx.channel}\ntime:{time}\nuser:{ctx.author.display_name}')
await ch.send(embed=embed)
if __name__ == '__main__':
bot = BOT_NAME(prefix="!") # prefix(コマンドの接頭辞)を定める。
bot.run(os.environ['TOKEN'])
では各機能、それに必要なコードを記していきます。cog化されている前提で話を進めていきます。
各種機能実装例
Ⅰロールを操作する
実装例: サーバーにユーザーが入った時にロールを付与する
roleを付与
@commands.Cog.listener()
async def on_member_join(self, member):
if member.bot:
role = discord.utils.get(member.guild.roles, name="bot") # ロール名からRoleオブジェクトを取得できる
await member.add_roles(role) # ここで付与
return
else:
role = discord.utils.get(member.guild.roles, name="ユーザー")
await member.add_roles(role)
roleをはく奪
@commands.command()
async def rr(self, ctx, check_role: discord.Role, member_id):
i = 0
if member_id == "a":
for mem in check_role.members:
i += 1
await mem.remove_roles(check_role)
embed = discord.Embed(
description=f"{ctx.author.display_name}により、\n"
f"{i}人の{check_role.name}役職\n"
f"をはく奪しました。",
color=0x006400)
await ctx.channel.send(embed=embed)
else:
try:
member = discord.utils.get(ctx.guild.members, id=int(member_id))
if discord.utils.get(member.roles, id=check_role.id):
await member.remove_roles(check_role)
embed = discord.Embed(
description=f"{ctx.author.display_name}により、\n"
f"{member.display_name}から\n"
f"{check_role.name}をはく奪しました。",
color=0x006400)
await ctx.channel.send(embed=embed)
else:
await ctx.channel.send(f"{member.display_name}は{check_role.name}を所持していません。")
except ValueError:
await ctx.channel.send("引数が不正です。")
except discord.errors.HTTPException:
await ctx.channel.send("空白は一つだけ有効です。(Roleをメンションで挿入した際に空白を更に一つ開けていませんか？)")
Ⅱ メッセージを送信する
実装例: コマンド実行時刻からn時間後の時間を10回分連続出力するコマンド
メッセージを送信
# time引数にはn時間後のnを表す文字列が来ることを想定
@commands.command(aliases=["add_times"])
async def at(self, ctx, hour):
now = datetime.now()
description = "計算結果\n-----------------------\n"
for i in range(10):
description += f"{i + 1}: {now.strftime('%m/%d %H:%M')}\n"
now += timedelta(hours=int(hour))
await self.bot.get_channel(id=740183927729160252).send(description)
Ⅲ メッセージを削除する
実装例 指定数分、コマンドを入力したチャンネルにおける最新のメッセージから削除するコマンド
メッセージを削除
@commands.command(name='del') # python自体にdel関数が用意されており、名前が一致してしまう際には_delのように書き、nameで指定する。
async def _del(self, ctx, n): # メッセージ削除用
p = re.compile(r'^[0-9]+$')
if p.fullmatch(n):
count = int(n)
await ctx.channel.purge(limit=count + 1)
Ⅳ SQLを利用する
HerokuにはHeroku Postgreがあり、この機能を組み込むことでPostgre SQLが利用できます。
実装していきましょう。自分のプロジェクトにHeroku PostgreのAdd-onを追加する方法はここでは省略します。
クレジットカード情報の登録が必要だったはずです。但し無料プランがあるのでそれを利用すれば請求は起こりません。
DBを利用する
import traceback
from discord.ext import commands
import discord
import psycopg2
import os
from datetime import datetime
SQLpath = os.environ["DATABASE_URL"]
db = psycopg2.connect(SQLpath) # sqlに接続
cur = db.cursor() # なんか操作する時に使うやつ
# ここまで出来たら、要所要所でデータを取得、登録したい、操作したい部分で以下のように入力する
# データを取得したい(SELECT文。test_tableというテーブルから全データを取得する場合)
# ex. test_tableをカラム長2, レコード長2の2*2のテーブルだったとする
cur.execute("SELECT * FROM test_table")
data = cur.fetchall()
await ctx.channel.send(data)
# 出力: [[データ1,データ2],[データ3,データ4]]
# 2元配列になる
# データを登録、編集、操作したい場合
# CREATE, UPDATE, INSERTなどは基本的に同様の操作で実装できる。f-stringを用いて入力したデータの入力もできる。
# f-stringは処理が遅いという話があるが、正直運用で困ることは大規模でない限りほぼ無い
# 入力されたデータをinput_dataとする。ここで、input_dataを[5,6]の1元配列とする。
cur.execute(f"INSERT INTO test_table values ('{input_data[0]}, '{input_data[1]}'')")
db.commit()
# 確認用。db.commit()を通して問題なかったら一応データの挿入には成功している。その前にデータのフォーマットの確認位はしたほうがいいだろう。
await ctx.channel.send("処理完了")
※重要 SQL文は一つ一つのオーダー(SELECTでデータを取って来い！！、INSERTでデータを保存しろ！！など)で、SQL文が違う、データ型が一致しないなどが起きると、オーダーのトランザクションが閉じるというエラー(cursor already closed)が出てSQLが使えなくなる。
これを回避するために、全てのエラーが出た時にとりあえずdb.commit()を入れておけば解決する。
なので、最初のcogの基盤の部分でon_command_error関数はエラーが出たら必ず処理が通るので最初にdb.commit()を通している。
↑これはpsycopg2ライブラリを使用しているからである。cursorを作成する必要がないasyncpgライブラリを使うと非同期的処理が可能である。
以下のサイトを参考してください。
https://qiita.com/hoto17296/items/fcefdb308d95c01606c7
Ⅴ APIを利用する
APIは機能実装の幅を大きく広げるので覚えておくといい。
APIは、json形式、xml形式(割と希少)などがある。とりあえずjson形式で紹介する。
apiのレスポンスを読み込む
import requests
url = "APIのURL"
response = requests.get(url)
jsonData = response.json()
ここでjsonDataは多重配列になっている。
テストjson
{
success:true
lastUpdated:1618056025550
products:{
product_1:{
product_id:11023
sell_summary:[
0:{
amount:200
pricePerUnit:2.4
orders:2
}
1:{
amount:240 // これを知りたい
pricePerUnit:2.7
orders:100
}
]
}
}
}
上のようなjsonを読み込んだときに、知りたいとコメントしてある部分のデータを取得するなら、
jsonData[products][product_1][sell_summary][1][amount]と入力すると取得できる。
100件~1000件程度のデータ量だとそこまで時間はかからないが1万を越してくると少し時間がかかるので実装を工夫するといい。
スレッドをいくつか立てて並列処理を組ませるなどの手が有効である。
終わりに
基本的には上記5つの操作が分かれば簡単なbotなら作れる。
分からないことがあればdiscord botを作るコミュニティがあるので質問しよう。