Migrating to new CogSettings. Cleaning up CogSetting code. Brainstorming pattern matching expression language.

cogs/basecog.py

 
 	@commands.Cog.listener()
 	async def on_ready(self):
+		self.__set_up_setting_commands()
+
+	def __set_up_setting_commands(self):
+		"""
+		Sets up getter and setter commands for all registered cog settings.
+		Only runs once.
+		"""
 		if self.are_settings_setup:
 			return
+		self.are_settings_setup = True
+
+		# See if the cog has a command group. Currently only supporting one max.
 		group: commands.core.Group = None
 		for member_name in dir(self):
 			member = getattr(self, member_name)
 			if isinstance(member, commands.core.Group):
 				group = member
 				break
-		lookup = {}
+
 		for setting in self.settings:
-			# Manually constructing equivalent of:
-			# 	@commands.command(
-			# 		brief='Posts a test warning in the configured warning channel.'
-			# 	)
-			# 	@commands.has_permissions(ban_members=True)
-			# 	@commands.guild_only()
-			# 	async def get/setvar(self, context, ...):
-			async def _set_setting(self, context, new_value):
-				s = lookup[context.command.name]
-				await self.__set_setting(context, new_value, s)
-			async def _get_setting(self, context):
-				s = lookup[context.command.name]
-				await self.__get_setting(context, s)
-
-			set_command = commands.Command(
-				_set_setting,
-				name=f'set{setting.name}',
-				brief=f'Sets {setting.brief}',
-				description=setting.description,
-				usage=setting.usage,
-				checks=[
-					commands.has_permissions(ban_members=True),
-					commands.guild_only(),
-				])
-			# XXX: Passing `cog` in init gets ignored and set to `None`.
-			set_command.cog = self
-			get_command = commands.Command(
-				_get_setting,
-				name=f'get{setting.name}',
-				brief=f'Shows {setting.brief}',
-				description=setting.description,
-				checks=[
-					commands.has_permissions(ban_members=True),
-					commands.guild_only(),
-				])
-			get_command.cog = self
-
-			if group:
-				group.add_command(get_command)
-				group.add_command(set_command)
-			else:
-				self.bot.add_command(get_command)
-				self.bot.add_command(set_command)
+			self.__make_getter_setter_commands(setting, group)
 
-			lookup[set_command.name] = setting
-			lookup[get_command.name] = setting
-		self.are_settings_setup = True
+	def __make_getter_setter_commands(self,
+			setting: CogSetting,
+			group: commands.core.Group) -> None:
+		"""
+		Creates a "get..." and "set..." command for the given setting and
+		either registers them as subcommands under the given command group or
+		under the bot if `None`.
+		"""
+		# Manually constructing equivalent of:
+		# 	@commands.command(
+		# 		brief='Posts a test warning in the configured warning channel.'
+		# 	)
+		# 	@commands.has_permissions(ban_members=True)
+		# 	@commands.guild_only()
+		# 	async def getvar(self, context):
+		async def getter(self, context):
+			await self.__get_setting_command(context, setting)
+		async def setter(self, context, new_value):
+			await self.__set_setting_command(context, new_value, setting)
+
+		get_command = commands.Command(
+			getter,
+			name=f'get{setting.name}',
+			brief=f'Shows {setting.brief}',
+			description=setting.description,
+			checks=[
+				commands.has_permissions(ban_members=True),
+				commands.guild_only(),
+			])
+		set_command = commands.Command(
+			setter,
+			name=f'set{setting.name}',
+			brief=f'Sets {setting.brief}',
+			description=setting.description,
+			usage=setting.usage,
+			checks=[
+				commands.has_permissions(ban_members=True),
+				commands.guild_only(),
+			])
+
+		# XXX: Passing `cog` in init gets ignored and set to `None` so set after.
+		# This ensures the callback is passed `self`.
+		get_command.cog = self
+		set_command.cog = self
+
+		if group:
+			group.add_command(get_command)
+			group.add_command(set_command)
+		else:
+			self.bot.add_command(get_command)
+			self.bot.add_command(set_command)
 
-	async def __set_setting(self, context, new_value, setting) -> None:
+	async def __set_setting_command(self, context, new_value, setting) -> None:
 		setting_name = setting.name
 		if context.command.parent:
 			setting_name = f'{context.command.parent.name}.{setting_name}'
 			f'{CONFIG["success_emoji"]} `{setting_name}` is now set to `{new_value}`',
 			mention_author=False)
 
-	async def __get_setting(self, context, setting) -> None:
+	async def __get_setting_command(self, context, setting) -> None:
 		setting_name = setting.name
 		if context.command.parent:
 			setting_name = f'{context.command.parent.name}.{setting_name}'

cogs/crosspostcog.py

 
 from config import CONFIG
 from rscollections import AgeBoundList, SizeBoundDict
-from cogs.basecog import BaseCog, BotMessage, BotMessageReaction
+from cogs.basecog import BaseCog, BotMessage, BotMessageReaction, CogSetting
 from storage import Storage
 
 class SpamContext:
 		self.deleted_messages = set()  # of Message
 
 class CrossPostCog(BaseCog):
+	SETTING_WARN_COUNT = CogSetting('warncount',
+		brief='number of messages to trigger a warning',
+		description='The number of identical messages to trigger a mod warning.',
+		usage='<count:int>',
+		min_value=2)
+	SETTING_BAN_COUNT = CogSetting('bancount',
+		brief='number of messages to trigger a ban',
+		description='The number of identical messages to trigger an ' + \
+			'automatic ban. Set to a large value to effectively disable, e.g. 9999.',
+		usage='<count:int>',
+		min_value=2)
+	SETTING_MIN_LENGTH = CogSetting('minlength',
+		brief='minimum message length',
+		description='The minimum number of characters in a message to be ' + \
+			'checked for duplicates. This can help ignore common short ' + \
+			'messages like "lol" or a single emoji. Set to 0 to count all ' + \
+			'message lengths.',
+		usage='<character_count:int>',
+		min_value=0)
+	SETTING_TIMESPAN = CogSetting('timespan',
+		brief='time window to look for dupe messages',
+		description='The number of seconds of message history to look at ' + \
+			'when looking for duplicates. Shorter values are preferred, ' + \
+			'both to detect bots and avoid excessive memory usage.',
+		usage='<seconds:int>',
+		min_value=1)
+
 	STATE_KEY_RECENT_MESSAGES = "crosspost_recent_messages"
 	STATE_KEY_SPAM_CONTEXT = "crosspost_spam_context"
 
 	CONFIG_KEY_MIN_MESSAGE_LENGTH = "crosspost_min_message_length"
 	CONFIG_KEY_MESSAGE_AGE = "crosspost_message_age"
 
-	MIN_WARN_COUNT = 2
-	MIN_BAN_COUNT = 2
-	MIN_MESSAGE_LENGTH = 0
-	MIN_TIME_SPAN = 1
-
 	def __init__(self, bot):
 		super().__init__(bot)
+		self.add_setting(CrossPostCog.SETTING_WARN_COUNT)
+		self.add_setting(CrossPostCog.SETTING_BAN_COUNT)
+		self.add_setting(CrossPostCog.SETTING_MIN_LENGTH)
+		self.add_setting(CrossPostCog.SETTING_TIMESPAN)
 		self.max_spam_contexts = 12
 
 	# Config
 
-	def __warn_count(self, guild: Guild) -> int:
-		return Storage.get_config_value(guild, self.CONFIG_KEY_WARN_COUNT) or \
-			self.get_cog_default('warn_message_count')
-
-	def __ban_count(self, guild: Guild) -> int:
-		return Storage.get_config_value(guild, self.CONFIG_KEY_BAN_COUNT) or \
-			self.get_cog_default('ban_message_count')
-
-	def __min_message_length(self, guild: Guild) -> int:
-		return Storage.get_config_value(guild, self.CONFIG_KEY_MIN_MESSAGE_LENGTH) or \
-			self.get_cog_default('min_message_length')
-
-	def __message_age_seconds(self, guild: Guild) -> int:
-		return Storage.get_config_value(guild, self.CONFIG_KEY_MESSAGE_AGE) or \
-			self.get_cog_default('time_window_seconds')
-
 	async def __record_message(self, message: Message) -> None:
 		if message.author.permissions_in(message.channel).ban_members:
 			# User exempt from spam detection
 			return
-		if len(message.content) < self.__min_message_length(message.guild):
+		if len(message.content) < self.get_guild_setting(message.guild, self.SETTING_MIN_LENGTH):
 			# Message too short to count towards spam total
 			return
-		max_age = timedelta(seconds=self.__message_age_seconds(message.guild))
+		max_age = timedelta(seconds=self.get_guild_setting(message.guild, self.SETTING_TIMESPAN))
 		recent_messages = Storage.get_state_value(message.guild, self.STATE_KEY_RECENT_MESSAGES) \
 			or AgeBoundList(max_age, lambda index, message : message.created_at)
 		recent_messages.max_age = max_age
 
 		# Get all recent messages by user
 		member_messages = [m for m in recent_messages if m.author.id == message.author.id]
-		if len(member_messages) < self.__warn_count(message.guild):
+		if len(member_messages) < self.get_guild_setting(message.guild, self.SETTING_WARN_COUNT):
 			return
 
 		# Look for repeats
 			count = (hash_to_count.get(key) or 0) + 1
 			hash_to_count[key] = count
 			max_count = max(max_count, count)
-		if max_count < self.__warn_count(message.guild):
+		if max_count < self.get_guild_setting(message.guild, self.SETTING_WARN_COUNT):
 			return
 
 		# Handle the spam
 			or SizeBoundDict(self.max_spam_contexts, lambda key, context : context.age)
 		Storage.set_state_value(message.guild, self.STATE_KEY_SPAM_CONTEXT, spam_lookup)
 		for message_hash, count in hash_to_count.items():
-			if count < self.__warn_count(message.guild):
+			if count < self.get_guild_setting(message.guild, self.SETTING_WARN_COUNT):
 				continue
 			key = f'{message.author.id}|{message_hash}'
 			context = spam_lookup.get(key)
 			await self.__update_from_context(context)
 
 	async def __update_from_context(self, context: SpamContext):
-		if len(context.spam_messages) >= self.__ban_count(context.member.guild):
+		if len(context.spam_messages) >= self.get_guild_setting(context.member.guild, self.SETTING_BAN_COUNT):
 			if not context.is_banned:
 				count = len(context.spam_messages)
 				await context.member.ban(reason=f'Autobanned by Rocketbot for posting same message {count} times', delete_message_days=1)
 		'Command group'
 		if context.invoked_subcommand is None:
 			await context.send_help()
-
-	@crosspost.command(
-		name='setwarncount',
-		brief='Sets the number of duplicate messages to trigger a mod warning',
-		description='If the same user posts the exact same message ' +
-			'content this many times a warning will be posted to the mods,' +
-			'even if the messages are posted in different channels.',
-		usage='<warn_count:int>',
-	)
-	async def joinraid_setwarncount(self, context: commands.Context,
-			warn_count: int):
-		if not await self.validate_param(context, 'warn_count', warn_count,
-			allowed_types=(int, ), min_value=self.MIN_WARN_COUNT):
-			return
-		Storage.set_config_value(context.guild, self.CONFIG_KEY_WARN_COUNT, warn_count)
-		await context.message.reply(
-			CONFIG['success_emoji'] + ' ' +
-			'Mods will be warned if a user posts the exact same message ' +
-			f'{warn_count} or more times within ' +
-			f'{self.__message_age_seconds(context.guild)} seconds.',
-			mention_author=False)
-
-	@crosspost.command(
-		name='getwarncount',
-		brief='Returns the number of duplicate messages to trigger a mod warning',
-	)
-	async def joinraid_getwarncount(self, context: commands.Context):
-		await context.message.reply(f'ℹ️ Mods will be warned if a user posts ' +
-			f'the exact same message {self.__warn_count(context.guild)} or more ' +
-			f'times within {self.__message_age_seconds(context.guild)} seconds.',
-			mention_author=False)
-
-	@crosspost.command(
-		name='setbancount',
-		brief='Sets the number of duplicate messages to trigger an automatic ban',
-		description='If the same user posts the exact same message ' +
-			'content this many times they will be automatically banned and the ' +
-			'mods will be alerted.',
-		usage='<ban_count:int>',
-	)
-	async def joinraid_setbancount(self, context: commands.Context,
-			ban_count: int):
-		if not await self.validate_param(context, 'ban_count', ban_count,
-			allowed_types=(int, ), min_value=self.MIN_BAN_COUNT):
-			return
-		Storage.set_config_value(context.guild, self.CONFIG_KEY_BAN_COUNT, ban_count)
-		await context.message.reply(
-			CONFIG['success_emoji'] + ' ' +
-			'Users will be banned if they post the exact same message ' +
-			f'{ban_count} or more times within ' +
-			f'{self.__message_age_seconds(context.guild)} seconds.',
-			mention_author=False)
-
-	@crosspost.command(
-		name='getbancount',
-		brief='Returns the number of duplicate messages to trigger an automatic ban',
-	)
-	async def joinraid_getbancount(self, context: commands.Context):
-		await context.message.reply(
-			CONFIG['info_emoji'] + ' ' +
-			'Users will be banned if they post the exact same message ' +
-			f'{self.__ban_count(context.guild)} or more times within ' +
-			f'{self.__message_age_seconds(context.guild)} seconds.',
-			mention_author=False)
-
-	@crosspost.command(
-		name='setminlength',
-		brief='Sets the minimum number of characters for a message to count toward spamming',
-		description='Messages shorter than this number of characters will not ' +
-			'count toward spam counts. This helps prevent flagging common, ' +
-			'frequent, short responses like "lol". A value of 0 counts all messages.',
-		usage='<min_length:int>',
-	)
-	async def joinraid_setminlength(self, context: commands.Context,
-			min_length: int):
-		if not await self.validate_param(context, 'min_length', min_length,
-			allowed_types=(int, ), min_value=self.MIN_MESSAGE_LENGTH):
-			return
-		Storage.set_config_value(context.guild, self.CONFIG_KEY_MIN_MESSAGE_LENGTH, min_length)
-		if min_length == 0:
-			await context.message.reply(
-				CONFIG['success_emoji'] + ' ' +
-				f'All messages will count against spam counts, regardless ' +
-				'of length.', mention_author=False)
-		else:
-			await context.message.reply(
-				CONFIG['success_emoji'] + ' ' +
-				f'Only messages {min_length} characters or longer will ' +
-				'count against spam counts.',
-				mention_author=False)
-
-	@crosspost.command(
-		name='getminlength',
-		brief='Returns the number of duplicate messages to trigger an automatic ban',
-	)
-	async def joinraid_getminlength(self, context: commands.Context):
-		min_length = self.__min_message_length(context.guild)
-		if min_length == 0:
-			await context.message.reply(
-				CONFIG['info_emoji'] + ' ' +
-				f'All messages will count against spam counts, regardless ' +
-				'of length.', mention_author=False)
-		else:
-			await context.message.reply(
-				CONFIG['info_emoji'] + ' ' +
-				f'Only messages {min_length} characters or longer will ' +
-				'count against spam counts.',
-				mention_author=False)
-
-	@crosspost.command(
-		name='settimewindow',
-		brief='Sets the length of time recent messages are checked for duplicates',
-		description='Repeated messages are only checked against recent ' +
-			'messages. This sets the length of that window, in seconds. Lower ' +
-			'values save memory and prevent false positives.',
-		usage='<seconds:int>',
-	)
-	async def joinraid_settimewindow(self, context: commands.Context,
-			seconds: int):
-		if not await self.validate_param(context, 'seconds', seconds,
-			allowed_types=(int, ), min_value=self.MIN_TIME_SPAN):
-			return
-		Storage.set_config_value(context.guild, self.CONFIG_KEY_MESSAGE_AGE, seconds)
-		await context.message.reply(
-			CONFIG['success_emoji'] + ' ' +
-			f'Only messages in the past {seconds} seconds will be checked ' +
-			'for duplicates.',
-			mention_author=False)
-
-	@crosspost.command(
-		name='gettimewindow',
-		brief='Returns the length of time recent messages are checked for duplicates',
-	)
-	async def joinraid_gettimewindow(self, context: commands.Context):
-		seconds = self.__message_age_seconds(context.guild)
-		await context.message.reply(
-			CONFIG['info_emoji'] + ' ' +
-			f'Only messages in the past {seconds} seconds will be checked ' +
-			'for duplicates.',
-			mention_author=False)

cogs/generalcog.py

 from discord.ext import commands
-from cogs.basecog import BaseCog
+from cogs.basecog import BaseCog, BotMessage
 from storage import ConfigKey, Storage
 
 class GeneralCog(BaseCog):
 	async def testwarn(self, context):
 		if Storage.get_config_value(context.guild, ConfigKey.WARNING_CHANNEL_ID) is None:
 			await context.message.reply(
-				'No warning channel set!',
+				f'{CONFIG["warning_emoji"]} No warning channel set!',
 				mention_author=False)
 		else:
-			await self.warn(context.guild,
-				f'Test warning message (requested by {context.author.name})')
+			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',

cogs/patterncog.py

 					m.quote = message.content
 					await self.post_message(m)
 				break
+
+	"""
+	Expression language samples:
+
+		content contains "poop"
+		content contains "poop" and content contains "tinkle"
+		joinage < 600s
+		(content contains "this" and content contains "that") or content contains "whatever"
+
+		<field> <op> <value>
+
+		Fields:	
+			content
+			author.id
+			author.name
+			author.joinage
+
+		Ops:
+			==
+			!=
+			<
+			>
+			<=
+			>=
+			contains, !contains  -- plain strings
+			matches, !matches  -- regexes
+
+		Value types:
+			timedelta  (600, 600s, 10m, 5m30s)
+			number
+			string
+			regex
+			mention
+	"""
+
+	@commands.group(
+		brief='Manages message pattern matching',
+	)
+	@commands.has_permissions(ban_members=True)
+	@commands.guild_only()
+	async def patterns(self, context: commands.Context):
+		'Message pattern matching'
+		if context.invoked_subcommand is None:
+			await context.send_help()
+
+	@patterns.command()
+	async def addpattern(self, context: commands.Context, name: str, expression: str, *args):
+		print(f'Pattern name: {name}')
+		tokens = []
+		tokens.append(expression)
+		tokens += args
+		print('Expression: ' + (' '.join(tokens)))

config.py.sample

 	'info_emoji': 'ℹ️',
 	'config_path': 'config/',
 	'cog_defaults': {
+		'CrossPostCog': {
+			'warncount': 3,
+			'bancount': 9999,
+			'minlength': 0,
+			'timespan': 60,
+		},
 		'JoinRaidCog': {
 			'enabled': False,
 			'warning_count': 5,
 			'warning_seconds': 5,
 		},
-		'CrossPostCog': {
-			'warn_message_count': 3,
-			'ban_message_count': 9999,
-			'time_window_seconds': 120,
-			'min_message_length': 0,
-		},
 		'URLSpamCog': {
 			'joinage': 900,  # Should be > 600 due to Discord-imposed waiting period
 			'action': 'nothing',  # "nothing" | "modwarn" | "delete" | "kick" | "ban"