225 lines
9.7 KiB
Python
Executable File
225 lines
9.7 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
import argparse
|
|
import logging
|
|
import sys
|
|
import textwrap
|
|
from pathlib import Path
|
|
from typing import Any, List, Dict, Union
|
|
|
|
import discord
|
|
import discord.utils
|
|
|
|
from ChannelsConfigFile import ChannelsConfigFile
|
|
import utils
|
|
|
|
logger = logging.getLogger("VocalMaisBot")
|
|
|
|
|
|
class VocalMaisBot(discord.Client):
|
|
|
|
def __init__(self, channels_config: ChannelsConfigFile):
|
|
super().__init__(
|
|
intents = discord.Intents(voice_states = True, guild_messages = True, guilds = True)
|
|
)
|
|
self.channels_config = channels_config
|
|
self.owner = None # Fetched by on_ready
|
|
|
|
async def _check_administrator_for_command(self, message: discord.Message) -> bool:
|
|
if message.author.guild_permissions.administrator or message.author == self.owner:
|
|
return True
|
|
else:
|
|
await message.channel.send(f":dragon_face: Only an administrator can control me!")
|
|
return False
|
|
|
|
async def sorry_do_not_understand(self, message: discord.Message):
|
|
await message.channel.send(f"Sorry I did not understand this message. Try `@{self.user.display_name} help` for help")
|
|
|
|
async def on_ready(self):
|
|
logger.info("Connected and ready!")
|
|
logger.debug(f"Logged as {self.user}")
|
|
|
|
self.owner = (await self.application_info()).owner
|
|
|
|
oauth_url = discord.utils.oauth_url(
|
|
self.user.id,
|
|
discord.Permissions(manage_channels = True, read_messages = True, send_messages = True, move_members = True)
|
|
)
|
|
logger.info(f"You can invite the bot to your server using the following link: {oauth_url}")
|
|
|
|
async def on_message(self, message: discord.Message):
|
|
if self.user not in message.mentions: # We only ever react to messages that mention us
|
|
return
|
|
if message.author.bot or message.is_system(): # We ignore messages from automated sources
|
|
return
|
|
|
|
contents = message.content.split()
|
|
if utils.check_list_element_no_bounds(contents, 1, "help"):
|
|
return await self.print_help(message.channel)
|
|
elif utils.check_list_element_no_bounds(contents, 1, "ping"):
|
|
return await message.channel.send(":ping_pong:")
|
|
elif utils.check_list_element_no_bounds(contents, 1, "register"):
|
|
return await self.register_channel(message)
|
|
elif utils.check_list_element_no_bounds(contents, 1, "forget"):
|
|
return await self.forget_channel(message)
|
|
elif utils.check_list_element_no_bounds(contents, 1, "list"):
|
|
return await self.list_watched_channels(message)
|
|
elif utils.check_list_element_no_bounds(contents, 1, "clear"):
|
|
return await self.clear_watched_channels(message)
|
|
else:
|
|
return await self.sorry_do_not_understand(message)
|
|
|
|
async def print_help(self, channel: discord.TextChannel):
|
|
me = self.user.display_name
|
|
message = f"""
|
|
**·** `@{me} register channel_id` : make me watch the voice channel with id `channel_id`
|
|
**·** `@{me} forget channel_id` : make me stop watching the voice channel with id `channel_id`
|
|
**·** `@{me} list` : list watched channels
|
|
**·** `@{me} clear` : stop watching all channels
|
|
**·** `@{me} help` : print this help
|
|
**·** `@{me} ping` : pong
|
|
"""
|
|
await channel.send(embed = discord.Embed(title = "Available commands", description = textwrap.dedent(message)))
|
|
|
|
async def register_channel(self, message: discord.Message):
|
|
channel_id = await self._get_channel_id_for_command(message)
|
|
if channel_id is None:
|
|
return
|
|
if not await self._check_administrator_for_command(message):
|
|
return
|
|
|
|
# Retrieving the channel
|
|
channel = self.get_channel(channel_id)
|
|
if channel is None or channel.guild != message.guild:
|
|
return await message.channel.send(f"I could not find a channel with id {channel_id} :cry:")
|
|
|
|
# Adding the channel to the list of watched channels (if needed)
|
|
has_been_added = self.channels_config.add_channel_to_watched(message.guild, channel)
|
|
if has_been_added:
|
|
await message.channel.send(f":white_check_mark: I am now watching {channel.name}")
|
|
else:
|
|
await message.channel.send(":thumbsup: I was already watching this channel")
|
|
|
|
async def forget_channel(self, message: discord.Message):
|
|
channel_id = await self._get_channel_id_for_command(message)
|
|
if channel_id is None:
|
|
return
|
|
if not await self._check_administrator_for_command(message):
|
|
return
|
|
|
|
channel = self.get_channel(channel_id)
|
|
if channel is None: # The channel does not exist (maybe it has been deleted)
|
|
channel = discord.Object(channel_id)
|
|
channel_name = str(channel_id)
|
|
else:
|
|
channel_name = f"{channel.name} ({channel_id})"
|
|
|
|
has_been_removed = self.channels_config.remove_channel_from_watched(message.guild, channel)
|
|
if has_been_removed:
|
|
await message.channel.send(f":white_check_mark: I am no longer watching {channel_name}")
|
|
else:
|
|
await message.channel.send(f":thumbsup: I already was not watching {channel_name}")
|
|
|
|
async def _get_channel_id_for_command(self, message: discord.Message) -> Union[int, None]:
|
|
"""
|
|
Get the channel id from a command message in the form "@Bot command channel_id", or answer with an error message and return None if not possible.
|
|
"""
|
|
message_elements = message.content.split()
|
|
if len(message_elements) < 3:
|
|
await self.sorry_do_not_understand(message)
|
|
return None
|
|
channel_id = message_elements[2]
|
|
try:
|
|
return int(channel_id)
|
|
except ValueError:
|
|
await message.channel.send(f":x: {channel_id} can not be converted to number")
|
|
return None
|
|
|
|
async def list_watched_channels(self, message: discord.Message):
|
|
watched_channels_ids = self.channels_config.get_watched_channels_ids(message.guild)
|
|
if len(watched_channels_ids) > 0:
|
|
answer_lines = []
|
|
for channel_id in watched_channels_ids:
|
|
channel = self.get_channel(channel_id)
|
|
if channel is not None: # Does the channel still exist?
|
|
answer_lines.append(f":eyes: {channel.name} ({channel_id})")
|
|
else:
|
|
answer_lines.append(f":ninja: <deleted channel> ({channel_id})")
|
|
await message.channel.send(embed = discord.Embed(title = "I am watching the following channels", description = "\n".join(answer_lines)))
|
|
else:
|
|
await message.channel.send(f":see_no_evil: I am not watching any voice channel! Do not hesitate to add some by running `@{self.user.display_name} register channel_id`")
|
|
|
|
async def clear_watched_channels(self, message: discord.Message):
|
|
if not await self._check_administrator_for_command(message):
|
|
return
|
|
|
|
self.channels_config.clear_watched_channels(message.guild)
|
|
await message.channel.send(":white_check_mark: I am no longer watching any channel")
|
|
|
|
async def on_voice_state_update(self, member: discord.Member, before: discord.VoiceState, after: discord.VoiceState):
|
|
if before.channel is not None: # They left a channel
|
|
await self.handle_user_left_voice(before.channel)
|
|
if after.channel is not None: # They joined a channel. Note that there is an if and not an elif, because they could leave a channel to join another
|
|
await self.handle_user_connected_to_voice(member, after.channel)
|
|
|
|
async def handle_user_connected_to_voice(self, user: discord.Member, channel: discord.VoiceChannel):
|
|
if user.bot:
|
|
return # We ignore bots connecting to channels, we do not want them asking us to create voice channels
|
|
if not self.channels_config.is_watched(channel.guild, channel):
|
|
return # It's not one of our special watched join-to-create channels
|
|
|
|
# The user is not a bot and connected to one of our special watched channels
|
|
# We create a channel for them and move them into it
|
|
category = channel.category
|
|
user_name = user.display_name
|
|
user_channel = await category.create_voice_channel(f"{user_name}'s channel", overwrites = {user: discord.PermissionOverwrite(manage_channels = True)})
|
|
await user.move_to(user_channel)
|
|
logger.info(f"Created channel {user_channel.name}({user_channel.id}) for {user.display_name}({user.id})")
|
|
|
|
# Saving the channel in the configuration
|
|
self.channels_config.add_channel_to_created(user_channel.guild, user_channel)
|
|
|
|
async def handle_user_left_voice(self, channel: discord.VoiceChannel):
|
|
# We do not verify if the user is a bot, because a bot could be the last one to leave a channel (e.g. music bot)
|
|
|
|
if not self.channels_config.is_created(channel.guild, channel):
|
|
return # It is not a channel that we manage, nothing to do
|
|
|
|
logger.debug(f"User left temporary channel {channel.name} ({channel.id})")
|
|
# It is a channel that we manage
|
|
if len(channel.members) > 0:
|
|
return # There are still people inside it, nothing to do
|
|
try:
|
|
await channel.delete()
|
|
except discord.NotFound:
|
|
pass # The channel already does not exist, that's not a problem, that's what we want
|
|
logger.info(f"Deleted channel {channel.name}({channel.id}) because it was empty")
|
|
# updating the configuration
|
|
self.channels_config.remove_channel_from_created(channel.guild, channel)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
argparser = argparse.ArgumentParser(description = "Discord bot to automatically create temporary voice channels for users when they connect to a special channel", formatter_class = argparse.ArgumentDefaultsHelpFormatter)
|
|
argparser.add_argument("-t", "--token-file", default = ".token", help = "File where the discord bot token is stored")
|
|
argparser.add_argument("-w", "--watched-channels-file", default = "channels.json", help = "Used to store the list of channels that the bot watches")
|
|
argparser.add_argument("-v", "--verbose", action = "store_true", help = "Print debug messages")
|
|
|
|
ARGS = argparser.parse_args()
|
|
|
|
logging.basicConfig(
|
|
level = logging.DEBUG if ARGS.verbose else logging.INFO,
|
|
format = "%(asctime)s %(name)s [%(levelname)s] %(message)s"
|
|
)
|
|
|
|
logger.info(f"Using discord.py version {discord.__version__}")
|
|
|
|
token_file_path = Path(ARGS.token_file).absolute()
|
|
if not token_file_path.is_file():
|
|
logger.error(f"Token file {token_file_path} does not exist")
|
|
sys.exit(1)
|
|
with token_file_path.open("r") as token_file:
|
|
token = token_file.readline().strip()
|
|
|
|
channels_config_file = ChannelsConfigFile(ARGS.watched_channels_file)
|
|
bot = VocalMaisBot(channels_config_file)
|
|
bot.run(token)
|