python-app-rocketbot/cogs/generalcog.py

from datetime import datetime, timedelta
from discord import Member, Message
from discord.ext import commands
import re

from cogs.basecog import BaseCog, BotMessage
from config import CONFIG
from rbutils import parse_timedelta, describe_timedelta
from storage import ConfigKey, Storage

class GeneralCog(BaseCog):
	def __init__(self, bot: commands.Bot):
		super().__init__(bot)
		self.is_connected = False
		self.is_ready = False

	@commands.Cog.listener()
	async def on_connect(self):
		print('on_connect')
		self.is_connected = True

	@commands.Cog.listener()
	async def on_ready(self):
		print('on_ready')
		self.is_ready = True

	@commands.command(
		brief='Posts a test warning in the configured warning channel.'
	)
	@commands.has_permissions(ban_members=True)
	@commands.guild_only()
	async def testwarn(self, context):
		if Storage.get_config_value(context.guild, ConfigKey.WARNING_CHANNEL_ID) is None:
			await context.message.reply(
				f'{CONFIG["warning_emoji"]} No warning channel set!',
				mention_author=False)
		else:
			bm = BotMessage(
				context.guild,
				f'Test warning message (requested by {context.author.name})',
				type=BotMessage.TYPE_MOD_WARNING)
			await self.post_message(bm)

	@commands.command(
		brief='Simple test reply',
	)
	async def hello(self, context):
		await context.message.reply(
			f'Hey, {context.author.name}!',
		 	mention_author=False)

	@commands.command(
		brief='Shuts down the bot (admin only)',
	)
	@commands.has_permissions(administrator=True)
	@commands.guild_only()
	async def shutdown(self, context):
		await self.bot.close()

	@commands.command(
		brief='Mass deletes messages',
		description='Deletes recent messages by the given user. The age is ' +
			'a duration, such as "30s", "5m", "1h30m". Messages far back in ' +
			'the scrollback might not be deleted by this command.',
		usage='<user> <age>'
	)
	@commands.has_permissions(manage_messages=True)
	@commands.guild_only()
	async def deletemessages(self, context, user: str, age: str) -> None:
		member_id = self.__parse_member_id(user)
		if member_id is None:
			await context.message.reply(
				f'{CONFIG["failure_emoji"]} user must be a mention or numeric user id',
				mention_author=False)
			return
		try:
			age_delta: timedelta = parse_timedelta(age)
		except ValueError:
			await context.message.reply(
				f'{CONFIG["failure_emoji"]} age must be a timespan, like "30s", "10m", "1h30m"',
				mention_author=False)
			return
		cutoff: datetime = datetime.utcnow() - age_delta
		def predicate(message: Message) -> bool:
			return str(message.author.id) == member_id and message.created_at >= cutoff
		deleted_messages = []
		for channel in context.guild.text_channels:
			try:
				deleted_messages += await channel.purge(limit=100, check=predicate)
			except:
				# XXX: Sloppily glossing over access errors instead of checking access
				pass
		await context.message.reply(
			f'{CONFIG["success_emoji"]} Deleted {len(deleted_messages)} ' + \
			f'messages by <@!{member_id}> from the past {describe_timedelta(age_delta)}.',
			mention_author=False)

	def __parse_member_id(self, arg: str) -> str:
		p = re.compile('^<@!?([0-9]+)>$')
		m = p.match(arg)
		if m:
			return m.group(1)
		p = re.compile('^([0-9]+)$')
		m = p.match(arg)
		if m:
			return m.group(1)
		return None