Created
April 9, 2021 14:25
-
-
Save AXVin/0be0b8997e24fa191e23e1dce651f37d to your computer and use it in GitHub Desktop.
Cog that allows you to have slash commands in d.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from enum import Enum | |
import discord | |
from discord.ext import commands | |
BASE_URL = "https://discord.com/api/v8" | |
class InteractionResponseType(Enum): | |
pong = 1 | |
channel_message_with_source = 4 | |
deferred_channel_message_with_source = 5 | |
class ApplicationCommandOptionType(Enum): | |
sub_command = 1 | |
sub_command_group = 2 | |
string = 3 | |
integer = 4 | |
boolean = 5 | |
user = 6 | |
channel = 7 | |
role = 8 | |
class FollowupMessage: | |
def __init__(self, data, interaction): | |
self.data = data | |
self.interaction = interaction | |
self.session = interaction.session | |
self.id = int(data["id"]) | |
self.content = data.get("content") | |
self.embeds = [ | |
discord.Embed.from_dict(embed) | |
for embed in data.get("embeds", []) | |
] | |
def __repr__(self): | |
return f"<FollowupMessage id={self.id}>" | |
def _update(self, data): | |
self.content = data.get("content") | |
self.embeds = [ | |
discord.Embed.from_dict(embed) | |
for embed in data.get("embeds", []) | |
] | |
async def edit(self, *, content=None, embed=None, embeds=None): | |
''' | |
Edits the message | |
''' | |
if embeds is None and embed is None: | |
embeds = [] | |
elif embeds is None and embed: | |
embeds = [embed] | |
url = f"{BASE_URL}/webhooks/{self.interaction.application_id}/{self.interaction.token}/messages/{self.id}" | |
payload = { | |
"content": content, | |
"embeds": [embed.to_dict() for embed in embeds] | |
} | |
async with self.session.patch(url, json=payload) as resp: | |
resp.raise_for_status() | |
data = await resp.json() | |
self._update(data) | |
return self | |
async def delete(self): | |
''' | |
Deletes this message | |
''' | |
embeds = embeds or [embed] or [] | |
url = f"{BASE_URL}/webhooks/{self.interaction.application_id}/{self.interaction.token}/messages/{self.id}" | |
async with self.session.delete(url) as resp: | |
resp.raise_for_status() | |
class Interaction: | |
def __init__(self, data, bot): | |
self.data = data | |
self.bot = bot | |
self.session = bot.session | |
self.id = int(data.get("id")) | |
self.application_id = int(data.get("application_id")) | |
self.token = data["token"] | |
self.guild = bot.get_guild(int(data.get("guild_id"))) | |
if data.get("user"): | |
self.author = discord.User(state=bot._connection, data=data["user"]) | |
else: | |
self.author = discord.Member(state=self.guild._state, data=data["member"], guild=self.guild) | |
if self.guild: | |
self.channel = self.guild.get_channel(int(data.get("channel_id"))) | |
self.command_name = self.data["data"]["name"] | |
self.command_id = int(self.data["data"]["id"]) | |
self.options = self.build_options(data["data"].get("options", [])) | |
def __repr__(self): | |
return f"<Interaction name={self.command_name} options={self.options}>" | |
def build_options(self, options): | |
ret = {} | |
for option in options: | |
name = option["name"] | |
value = option.get("value", option.get("options")) | |
type_ = ApplicationCommandOptionType(option["type"]) | |
if type_ in (ApplicationCommandOptionType.sub_command_group, | |
ApplicationCommandOptionType.sub_command): | |
value = self.build_options(value) | |
elif type_ is ApplicationCommandOptionType.integer: | |
value = int(value) | |
elif type_ is ApplicationCommandOptionType.channel: | |
value = self.guild.get_channel(int(value)) | |
elif type_ is ApplicationCommandOptionType.role: | |
value = self.guild.get_role(int(value)) | |
elif type_ is ApplicationCommandOptionType.user: | |
resolved = self.data["data"]["resolved"] | |
if value in resolved.get("members", {}): | |
member = resolved["members"][value] | |
member["user"] = resolved["users"][value] | |
value = discord.Member(state=self.guild._state, data=member, guild=self.guild) | |
else: | |
value = discord.User(state=self.bot._connection, data=resolved["users"][value]) | |
ret[name] = value | |
return ret | |
async def ack(self): | |
''' | |
Acknowledge the slash command to get more time | |
to issue followups | |
Note: You cannot use Message.respond after this | |
You must use Message.edit_original or Message.followup | |
after acknowledging | |
''' | |
url = f"{BASE_URL}/interactions/{self.id}/{self.token}/callback" | |
payload = { | |
"type": InteractionResponseType.deferred_channel_message_with_source.value | |
} | |
async with self.session.post(url, json=payload) as resp: | |
return | |
async def respond(self, content=None, *, tts=False, embed=None, embeds=None, hidden=False): | |
''' | |
Send the initial response message | |
''' | |
if embeds is None and embed is None: | |
embeds = [] | |
elif embeds is None and embed: | |
embeds = [embed] | |
url = f"{BASE_URL}/interactions/{self.id}/{self.token}/callback" | |
payload = { | |
"type": InteractionResponseType.channel_message_with_source.value, | |
"data": { | |
"content": content, | |
"tts": tts, | |
"embeds": [embed.to_dict() for embed in embeds] | |
} | |
} | |
if hidden: | |
payload["data"]["flags"] = 64 | |
async with self.session.post(url, json=payload) as resp: | |
resp.raise_for_status() | |
async def edit_original(self, *, content=None, embed=None, embeds=None): | |
''' | |
Edit the original response message | |
Can only be used after Message.ack or Message.respond | |
''' | |
if embeds is None and embed is None: | |
embeds = [] | |
elif embeds is None and embed: | |
embeds = [embed] | |
url = f"{BASE_URL}/webhooks/{self.application_id}/{self.token}/messages/@original" | |
payload = { | |
"content": content, | |
"embeds": [embed.to_dict() for embed in embeds] | |
} | |
async with self.session.patch(url, json=payload) as resp: | |
resp.raise_for_status() | |
async def delete_original(self): | |
''' | |
Deletes the original response message | |
''' | |
url = f"{BASE_URL}/webhooks/{self.application_id}/{self.token}/messages/@original" | |
async with self.session.delete(url) as resp: | |
resp.raise_for_status() | |
async def followup(self, content=None, *, tts=False, embed=None, embeds=None): | |
''' | |
Sends a follow up message | |
''' | |
if embeds is None and embed is None: | |
embeds = [] | |
elif embeds is None and embed: | |
embeds = [embed] | |
url = f"{BASE_URL}/webhooks/{self.application_id}/{self.token}" | |
payload = { | |
"tts": tts, | |
"content": content, | |
"embeds": [embed.to_dict() for embed in embeds] | |
} | |
async with self.session.post(url, json=payload) as resp: | |
resp.raise_for_status() | |
data = await resp.json() | |
return FollowupMessage(data, self) | |
class SlashCommands(commands.Cog): | |
def __init__(self, bot): | |
self.bot = bot | |
@commands.Cog.listener() | |
async def on_socket_response(self, payload): | |
if payload['t'] != 'INTERACTION_CREATE': | |
return | |
data = payload['d'] | |
self.bot.dispatch('slash_command', Interaction(data, self.bot)) | |
def setup(bot): | |
cog = SlashCommands(bot) | |
bot.add_cog(cog) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment