5c973a76b9
Added a 'requires_fetch' parameter to the send function to ensure messages are fetched when needed. Enhanced command interaction checks to verify bot permissions for reading and sending messages in channels. Updated player controller logic to use the new fetch behavior for controller messages.
255 lines
11 KiB
Python
255 lines
11 KiB
Python
"""MIT License
|
|
|
|
Copyright (c) 2023 - present Vocard Development
|
|
|
|
Permission is hereby granted, free of charge, to any person obtaining a copy
|
|
of this software and associated documentation files (the "Software"), to deal
|
|
in the Software without restriction, including without limitation the rights
|
|
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
|
copies of the Software, and to permit persons to whom the Software is
|
|
furnished to do so, subject to the following conditions:
|
|
|
|
The above copyright notice and this permission notice shall be included in all
|
|
copies or substantial portions of the Software.
|
|
|
|
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
|
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
|
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
|
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
|
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
|
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
|
|
SOFTWARE.
|
|
"""
|
|
|
|
import discord
|
|
import sys
|
|
import os
|
|
import aiohttp
|
|
import update
|
|
import logging
|
|
import voicelink
|
|
import function as func
|
|
|
|
from discord.ext import commands
|
|
from ipc import IPCClient
|
|
from motor.motor_asyncio import AsyncIOMotorClient
|
|
from logging.handlers import TimedRotatingFileHandler
|
|
from addons import Settings
|
|
|
|
class Translator(discord.app_commands.Translator):
|
|
async def load(self):
|
|
func.logger.info("Loaded Translator")
|
|
|
|
async def unload(self):
|
|
func.logger.info("Unload Translator")
|
|
|
|
async def translate(self, string: discord.app_commands.locale_str, locale: discord.Locale, context: discord.app_commands.TranslationContext):
|
|
locale_key = str(locale)
|
|
|
|
if locale_key in func.LOCAL_LANGS:
|
|
translated_text = func.LOCAL_LANGS[locale_key].get(string.message)
|
|
|
|
if translated_text is None:
|
|
missing_translations = func.MISSING_TRANSLATOR.setdefault(locale_key, [])
|
|
if string.message not in missing_translations:
|
|
missing_translations.append(string.message)
|
|
|
|
return translated_text
|
|
|
|
return None
|
|
|
|
class Vocard(commands.Bot):
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
|
|
self.ipc: IPCClient
|
|
|
|
async def on_message(self, message: discord.Message, /) -> None:
|
|
# Ignore messages from bots or DMs
|
|
if message.author.bot or not message.guild:
|
|
return False
|
|
|
|
# Check if the bot is directly mentioned
|
|
if message.content.strip() == self.user.mention and not message.mention_everyone:
|
|
prefix = await self.command_prefix(self, message)
|
|
if not prefix:
|
|
return await message.channel.send("I don't have a bot prefix set.")
|
|
return await message.channel.send(f"My prefix is `{prefix}`")
|
|
|
|
# Fetch guild settings and check if the mesage is in the music request channel
|
|
settings = await func.get_settings(message.guild.id)
|
|
if settings and (request_channel := settings.get("music_request_channel")):
|
|
if message.channel.id == request_channel.get("text_channel_id"):
|
|
ctx = await self.get_context(message)
|
|
try:
|
|
cmd = self.get_command("play")
|
|
if message.content:
|
|
await cmd(ctx, query=message.content)
|
|
|
|
elif message.attachments:
|
|
for attachment in message.attachments:
|
|
await cmd(ctx, query=attachment.url)
|
|
|
|
except Exception as e:
|
|
await func.send(ctx, str(e), ephemeral=True)
|
|
|
|
finally:
|
|
return await message.delete()
|
|
|
|
await self.process_commands(message)
|
|
|
|
async def connect_db(self) -> None:
|
|
if not ((db_name := func.settings.mongodb_name) and (db_url := func.settings.mongodb_url)):
|
|
raise Exception("MONGODB_NAME and MONGODB_URL can't not be empty in settings.json")
|
|
|
|
try:
|
|
func.MONGO_DB = AsyncIOMotorClient(host=db_url)
|
|
await func.MONGO_DB.server_info()
|
|
func.logger.info(f"Successfully connected to [{db_name}] MongoDB!")
|
|
|
|
except Exception as e:
|
|
func.logger.error("Not able to connect MongoDB! Reason:", exc_info=e)
|
|
exit()
|
|
|
|
func.SETTINGS_DB = func.MONGO_DB[db_name]["Settings"]
|
|
func.USERS_DB = func.MONGO_DB[db_name]["Users"]
|
|
|
|
async def setup_hook(self) -> None:
|
|
func.langs_setup()
|
|
|
|
# Connecting to MongoDB
|
|
await self.connect_db()
|
|
|
|
# Set translator
|
|
await self.tree.set_translator(Translator())
|
|
|
|
# Loading all the module in `cogs` folder
|
|
for module in os.listdir(func.ROOT_DIR + '/cogs'):
|
|
if module.endswith('.py'):
|
|
try:
|
|
await self.load_extension(f"cogs.{module[:-3]}")
|
|
func.logger.info(f"Loaded {module[:-3]}")
|
|
except Exception as e:
|
|
func.logger.error(f"Something went wrong while loading {module[:-3]} cog.", exc_info=e)
|
|
|
|
self.ipc = IPCClient(self, **func.settings.ipc_client)
|
|
if func.settings.ipc_client.get("enable", False):
|
|
try:
|
|
await self.ipc.connect()
|
|
except Exception as e:
|
|
func.logger.error(f"Cannot connected to dashboard! - Reason: {e}")
|
|
|
|
# Update version tracking
|
|
if not func.settings.version or func.settings.version != update.__version__:
|
|
await self.tree.sync()
|
|
func.update_json("settings.json", new_data={"version": update.__version__})
|
|
for locale_key, values in func.MISSING_TRANSLATOR.items():
|
|
func.logger.warning(f'Missing translation for "{", ".join(values)}" in "{locale_key}"')
|
|
|
|
async def on_ready(self):
|
|
func.logger.info("------------------")
|
|
func.logger.info(f"Logging As {self.user}")
|
|
func.logger.info(f"Bot ID: {self.user.id}")
|
|
func.logger.info("------------------")
|
|
func.logger.info(f"Discord Version: {discord.__version__}")
|
|
func.logger.info(f"Python Version: {sys.version}")
|
|
func.logger.info("------------------")
|
|
|
|
func.settings.client_id = self.user.id
|
|
func.LOCAL_LANGS.clear()
|
|
func.MISSING_TRANSLATOR.clear()
|
|
|
|
async def on_command_error(self, ctx: commands.Context, exception, /) -> None:
|
|
error = getattr(exception, 'original', exception)
|
|
if ctx.interaction:
|
|
error = getattr(error, 'original', error)
|
|
|
|
if isinstance(error, (commands.CommandNotFound, aiohttp.client_exceptions.ClientOSError, discord.errors.NotFound)):
|
|
return
|
|
|
|
elif isinstance(error, (commands.CommandOnCooldown, commands.MissingPermissions, commands.RangeError, commands.BadArgument)):
|
|
pass
|
|
|
|
elif isinstance(error, (commands.MissingRequiredArgument, commands.MissingRequiredAttachment)):
|
|
command = f"{ctx.prefix}" + (f"{ctx.command.parent.qualified_name} " if ctx.command.parent else "") + f"{ctx.command.name} {ctx.command.signature}"
|
|
position = command.find(f"<{ctx.current_parameter.name}>") + 1
|
|
description = f"**Correct Usage:**\n```{command}\n" + " " * position + "^" * len(ctx.current_parameter.name) + "```\n"
|
|
if ctx.command.aliases:
|
|
description += f"**Aliases:**\n`{', '.join([f'{ctx.prefix}{alias}' for alias in ctx.command.aliases])}`\n\n"
|
|
description += f"**Description:**\n{ctx.command.help}\n\u200b"
|
|
|
|
embed = discord.Embed(description=description, color=func.settings.embed_color)
|
|
embed.set_footer(icon_url=ctx.me.display_avatar.url, text=f"More Help: {func.settings.invite_link}")
|
|
return await ctx.reply(embed=embed)
|
|
|
|
elif not issubclass(error.__class__, voicelink.VoicelinkException):
|
|
error = await func.get_lang(ctx.guild.id, "unknownException") + func.settings.invite_link
|
|
func.logger.error(f"An unexpected error occurred in the {ctx.command.name} command on the {ctx.guild.name}({ctx.guild.id}).", exc_info=exception)
|
|
|
|
try:
|
|
return await ctx.reply(error, ephemeral=True)
|
|
except:
|
|
pass
|
|
|
|
class CommandCheck(discord.app_commands.CommandTree):
|
|
async def interaction_check(self, interaction: discord.Interaction, /) -> bool:
|
|
if interaction.type == discord.InteractionType.application_command:
|
|
if not interaction.guild:
|
|
await interaction.response.send_message("This command can only be used in guilds!")
|
|
return False
|
|
|
|
channel_perm = interaction.channel.permissions_for(interaction.guild.me)
|
|
if not channel_perm.read_messages or not channel_perm.send_messages:
|
|
await interaction.response.send_message("I don't have permission to read or send messages in this channel.")
|
|
return False
|
|
|
|
return True
|
|
|
|
async def get_prefix(bot: commands.Bot, message: discord.Message) -> str:
|
|
settings = await func.get_settings(message.guild.id)
|
|
prefix = settings.get("prefix", func.settings.bot_prefix)
|
|
|
|
# Allow owner to use the bot without a prefix
|
|
if prefix and not message.content.startswith(prefix) and (await bot.is_owner(message.author) or message.author.id in func.settings.bot_access_user):
|
|
return ""
|
|
|
|
return prefix
|
|
|
|
# Loading settings and logger
|
|
func.settings = Settings(func.open_json("settings.json"))
|
|
|
|
LOG_SETTINGS = func.settings.logging
|
|
if (LOG_FILE := LOG_SETTINGS.get("file", {})).get("enable", True):
|
|
log_path = os.path.abspath(LOG_FILE.get("path", "./logs"))
|
|
if not os.path.exists(log_path):
|
|
os.makedirs(log_path)
|
|
|
|
file_handler = TimedRotatingFileHandler(filename=f'{log_path}/vocard.log', encoding="utf-8", backupCount=LOG_SETTINGS.get("max-history", 30), when="d")
|
|
file_handler.namer = lambda name: name.replace(".log", "") + ".log"
|
|
file_handler.setFormatter(logging.Formatter('{asctime} [{levelname:<8}] {name}: {message}', '%Y-%m-%d %H:%M:%S', style='{'))
|
|
logging.getLogger().addHandler(file_handler)
|
|
|
|
for log_name, log_level in LOG_SETTINGS.get("level", {}).items():
|
|
_logger = logging.getLogger(log_name)
|
|
_logger.setLevel(log_level)
|
|
|
|
# Setup the bot object
|
|
intents = discord.Intents.default()
|
|
intents.message_content = False if func.settings.bot_prefix is None else True
|
|
intents.members = func.settings.ipc_client.get("enable", False)
|
|
intents.voice_states = True
|
|
|
|
bot = Vocard(
|
|
command_prefix=get_prefix,
|
|
help_command=None,
|
|
tree_cls=CommandCheck,
|
|
chunk_guilds_at_startup=False,
|
|
activity=discord.Activity(type=discord.ActivityType.listening, name="Starting..."),
|
|
case_insensitive=True,
|
|
intents=intents
|
|
)
|
|
|
|
if __name__ == "__main__":
|
|
update.check_version(with_msg=True)
|
|
bot.run(func.settings.token, root_logger=True)
|