df05a5b1bd
Ensures that get_prefix returns an empty string if the prefix is None, preventing potential issues with command parsing when no prefix is set in the settings.
243 lines
10 KiB
Python
243 lines
10 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 function as func
|
|
|
|
from discord.ext import commands
|
|
from logging.handlers import TimedRotatingFileHandler
|
|
from voicelink import Config, LangHandler, MongoDBHandler, IPCClient, VoicelinkException
|
|
from voicelink.utils import dispatch_message
|
|
|
|
class Translator(discord.app_commands.Translator):
|
|
MISSING_TRANSLATOR: dict[str, list[str]] = {}
|
|
|
|
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
|
|
) -> str | None:
|
|
locale_key = str(locale).upper()
|
|
local_translations = LangHandler._local_langs.get(locale_key)
|
|
if not local_translations:
|
|
return None
|
|
|
|
translated_text = local_translations.get(string.message)
|
|
if translated_text is not None:
|
|
return translated_text
|
|
|
|
missing = self.MISSING_TRANSLATOR.setdefault(locale_key, [])
|
|
if string.message not in missing:
|
|
missing.append(string.message)
|
|
|
|
return None
|
|
|
|
class Vocard(commands.Bot):
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
|
|
self.ipc_client: 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 MongoDBHandler.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:
|
|
if not ctx.prefix:
|
|
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 dispatch_message(ctx, str(e), ephemeral=True)
|
|
|
|
finally:
|
|
await message.delete()
|
|
|
|
await self.process_commands(message)
|
|
|
|
async def setup_hook(self) -> None:
|
|
# Connecting to MongoDB
|
|
await MongoDBHandler.init(bot_config.mongodb_url, bot_config.mongodb_name)
|
|
|
|
# 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_client: IPCClient = IPCClient(self, **bot_config.ipc_client)
|
|
if bot_config.ipc_client.get("enable", False):
|
|
try:
|
|
await self.ipc_client.connect()
|
|
except Exception as e:
|
|
func.logger.error(f"Cannot connected to dashboard! - Reason: {e}")
|
|
|
|
# Update version tracking
|
|
if not bot_config.version or bot_config.version != update.__version__:
|
|
await self.tree.sync()
|
|
func.update_json("settings.json", new_data={"version": update.__version__})
|
|
|
|
for locale_key, values in self.tree.translator.MISSING_TRANSLATOR.items():
|
|
func.logger.warning(f'Missing translation for "{", ".join(values)}" in "{locale_key}"')
|
|
self.tree.translator.MISSING_TRANSLATOR.clear()
|
|
|
|
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"Vocard Version: {update.__version__}")
|
|
func.logger.info(f"Discord Version: {discord.__version__}")
|
|
func.logger.info(f"Python Version: {sys.version}")
|
|
func.logger.info("------------------")
|
|
|
|
bot_config.client_id = self.user.id
|
|
LangHandler._local_langs.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=bot_config.embed_color)
|
|
embed.set_footer(icon_url=ctx.me.display_avatar.url, text=f"More Help: {bot_config.invite_link}")
|
|
return await ctx.reply(embed=embed)
|
|
|
|
elif not issubclass(error.__class__, VoicelinkException):
|
|
error = await Lang_handler.get_lang(ctx.guild.id, "common.errors.unknown") + bot_config.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 MongoDBHandler.get_settings(message.guild.id)
|
|
prefix = settings.get("prefix", bot_config.bot_prefix)
|
|
return prefix if prefix is not None else ""
|
|
|
|
# Loading settings and logger
|
|
bot_config = Config(func.open_json("settings.json"))
|
|
Lang_handler = LangHandler.init()
|
|
|
|
LOG_SETTINGS = bot_config.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 bot_config.bot_prefix is None else True
|
|
intents.members = bot_config.ipc_client.get("enable", False)
|
|
intents.voice_states = True
|
|
intents.presences = False
|
|
|
|
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(bot_config.token, root_logger=True)
|