import asyncio import logging.handlers import os import sys import aiohttp import discord from discord.ext import commands from discord.ext.commands import CommandError, Context from robocop_ng.helpers.notifications import report_critical_error if len(sys.argv[1:]) != 1: sys.stderr.write("usage: ") sys.exit(1) state_dir = os.path.abspath(sys.argv[1]) sys.path.append(state_dir) import config script_name = os.path.basename(__file__).split(".")[0] log_file_name = f"{script_name}.log" # Limit of discord (non-nitro) is 8MB (not MiB) max_file_size = 1000 * 1000 * 8 backup_count = 3 file_handler = logging.handlers.RotatingFileHandler( filename=log_file_name, maxBytes=max_file_size, backupCount=backup_count ) stdout_handler = logging.StreamHandler(sys.stdout) log_format = logging.Formatter( "[%(asctime)s] {%(filename)s:%(lineno)d} %(levelname)s - %(message)s" ) file_handler.setFormatter(log_format) stdout_handler.setFormatter(log_format) log = logging.getLogger("discord") log.setLevel(logging.INFO) log.addHandler(file_handler) log.addHandler(stdout_handler) def get_prefix(bot, message): prefixes = config.prefixes return commands.when_mentioned_or(*prefixes)(bot, message) wanted_jsons = [ "data/restrictions.json", "data/robocronptab.json", "data/userlog.json", "data/invites.json", "data/macros.json", "data/persistent_roles.json", "data/disabled_ids.json", ] if not os.path.exists(os.path.join(state_dir, "data")): os.makedirs(os.path.join(state_dir, "data")) for wanted_json_idx in range(len(wanted_jsons)): wanted_jsons[wanted_json_idx] = os.path.join( state_dir, wanted_jsons[wanted_json_idx] ) if not os.path.isfile(wanted_jsons[wanted_json_idx]): with open(wanted_jsons[wanted_json_idx], "w") as file: file.write("{}") intents = discord.Intents.all() intents.typing = False bot = commands.Bot( command_prefix=get_prefix, description=config.bot_description, intents=intents ) bot.help_command = commands.DefaultHelpCommand(dm_help=True) bot.log = log bot.config = config bot.script_name = script_name bot.state_dir = state_dir bot.wanted_jsons = wanted_jsons async def get_channel_safe(self, channel_id: int): res = self.get_channel(channel_id) if res is None: res = await self.fetch_channel(channel_id) return res commands.Bot.get_channel_safe = get_channel_safe @bot.event async def on_ready(): aioh = {"User-Agent": f"{script_name}/1.0'"} bot.aiosession = aiohttp.ClientSession(headers=aioh) bot.app_info = await bot.application_info() bot.botlog_channel = await bot.get_channel_safe(config.botlog_channel) log.info( f"\nLogged in as: {bot.user.name} - " f"{bot.user.id}\ndpy version: {discord.__version__}\n" ) game_name = f"{config.prefixes[0]}help" # Send "Robocop has started! x has y members!" guild = bot.botlog_channel.guild msg = ( f"{bot.user.name} has started! " f"{guild.name} has {guild.member_count} members!" ) data_files = [discord.File(fpath) for fpath in wanted_jsons] await bot.botlog_channel.send(msg, files=data_files) activity = discord.Activity(name=game_name, type=discord.ActivityType.listening) await bot.change_presence(activity=activity) @bot.event async def on_command(ctx): log_text = ( f"{ctx.message.author} ({ctx.message.author.id}): " f'"{ctx.message.content}" ' ) if ctx.guild: # was too long for tertiary if log_text += ( f'on "{ctx.channel.name}" ({ctx.channel.id}) ' f'at "{ctx.guild.name}" ({ctx.guild.id})' ) else: log_text += f"on DMs ({ctx.channel.id})" log.info(log_text) @bot.event async def on_error(event: str, *args, **kwargs): log.exception(f"Error on {event}:") exception = sys.exc_info()[1] is_report_allowed = any( [ not isinstance(exception, x) for x in [ discord.RateLimited, discord.GatewayNotFound, discord.InteractionResponded, discord.LoginFailure, ] ] ) if exception is not None and is_report_allowed: await report_critical_error( bot, exception, additional_info={"Event": event, "args": args, "kwargs": kwargs}, ) @bot.event async def on_command_error(ctx: Context, error: CommandError): error_text = str(error) err_msg = ( f'Error with "{ctx.message.content}" from ' f'"{ctx.message.author} ({ctx.message.author.id}) ' f"of type {type(error)}: {error_text}" ) log.exception(err_msg) if not isinstance(error, commands.CommandNotFound): err_msg = bot.escape_message(err_msg) await bot.botlog_channel.send(err_msg) if isinstance(error, commands.NoPrivateMessage): return await ctx.send("This command doesn't work on DMs.") elif isinstance(error, commands.MissingPermissions): roles_needed = "\n- ".join(error.missing_perms) return await ctx.send( f"{ctx.author.mention}: You don't have the right" " permissions to run this command. You need: " f"```- {roles_needed}```" ) elif isinstance(error, commands.BotMissingPermissions): roles_needed = "\n-".join(error.missing_perms) return await ctx.send( f"{ctx.author.mention}: Bot doesn't have " "the right permissions to run this command. " "Please add the following roles: " f"```- {roles_needed}```" ) elif isinstance(error, commands.CommandOnCooldown): return await ctx.send( f"{ctx.author.mention}: You're being " "ratelimited. Try in " f"{error.retry_after:.1f} seconds." ) elif isinstance(error, commands.CheckFailure): return await ctx.send( f"{ctx.author.mention}: Check failed. " "You might not have the right permissions " "to run this command, or you may not be able " "to run this command in the current channel." ) elif isinstance(error, commands.CommandInvokeError) and ( "Cannot send messages to this user" in error_text ): return await ctx.send( f"{ctx.author.mention}: I can't DM you.\n" "You might have me blocked or have DMs " f"blocked globally or for {ctx.guild.name}.\n" "Please resolve that, then " "run the command again." ) elif isinstance(error, commands.CommandNotFound): # Nothing to do when command is not found. return help_text = ( f"Usage of this command is: ```{ctx.prefix}{ctx.command.name} " f"{ctx.command.signature}```\nPlease see `{ctx.prefix}help " f"{ctx.command.name}` for more info about this command." ) # Keep a list of commands that involve mentioning users # and can involve users leaving/getting banned # noinspection NonAsciiCharacters,PyPep8Naming ಠ_ಠ = ["warn", "kick", "ban"] if isinstance(error, commands.BadArgument): # and if said commands get used, add a specific notice. if ctx.command.name in ಠ_ಠ: help_text = ( "This probably means that user left (or already got kicked/banned).\n" + help_text ) return await ctx.send( f"{ctx.author.mention}: You gave incorrect arguments. {help_text}" ) elif isinstance(error, commands.MissingRequiredArgument): return await ctx.send( f"{ctx.author.mention}: You gave incomplete arguments. {help_text}" ) @bot.event async def on_message(message): if message.author.bot: return if (message.guild) and (message.guild.id not in config.guild_whitelist): return # Ignore messages in newcomers channel, unless it's potentially # an allowed command welcome_allowed = ["reset", "kick", "ban", "warn"] if message.channel.id == config.welcome_channel and not any( cmd in message.content for cmd in welcome_allowed ): return ctx = await bot.get_context(message) await bot.invoke(ctx) async def main(): async with bot: if len(config.guild_whitelist) == 1: invite_url = discord.utils.oauth_url( config.client_id, guild=discord.Object(config.guild_whitelist[0]), disable_guild_select=True, ) else: invite_url = discord.utils.oauth_url(config.client_id) log.info(f"\nInvite URL: {invite_url}\n") for cog in config.initial_cogs: try: await bot.load_extension(f"robocop_ng.{cog}") except Exception as e: log.exception(f"Failed to load cog {cog}:", e) await bot.start(config.token) if __name__ == "__main__": asyncio.run(main())