Crosspost cog looks at attachments as well

rocketbot/cogs/autokickcog.py

 
 	async def __kick_or_ban_if_needed(self, member: Member):
 		guild: Guild = member.guild
-		recent_kicks: AgeBoundDict = Storage.get_state_value(guild, AutoKickCog.STATE_KEY_RECENT_KICKS)
+		recent_kicks: AgeBoundDict[int, AutoKickContext, datetime, timedelta] = Storage.get_state_value(guild, AutoKickCog.STATE_KEY_RECENT_KICKS)
 		if recent_kicks is None:
 			recent_kicks = AgeBoundDict(timedelta(seconds=3600), lambda i, context : context.last_kick)
 			Storage.set_state_value(guild, self.STATE_KEY_RECENT_KICKS, recent_kicks)
 	@staticmethod
 	def ordinal(val: int):
 		"""Formats an integer with an ordinal suffix (English only)"""
-		if val % 100 < 10 or val % 100 > 20:
+		if (val // 10) % 10 != 1:
 			if val % 10 == 1:
 				return f'{val}st'
 			if val % 10 == 2:

rocketbot/cogs/basecog.py

 		issue warnings regardless. Call record_warning or record_warnings after
 		triggering a mod warning.
 		"""
-		recent_warns: AgeBoundDict = Storage.get_state_value(member.guild,
+		recent_warns: AgeBoundDict[int, WarningContext, datetime, timedelta] = Storage.get_state_value(member.guild,
 			BaseCog.STATE_KEY_RECENT_WARNINGS)
 		if recent_warns is None:
 			return False
 		Records that mods have been warned about a member and do not need to be
 		warned about them again for a short while.
 		"""
-		recent_warns: AgeBoundDict = Storage.get_state_value(member.guild,
+		recent_warns: AgeBoundDict[int, WarningContext, datetime, timedelta] = Storage.get_state_value(member.guild,
 			BaseCog.STATE_KEY_RECENT_WARNINGS)
 		if recent_warns is None:
 			recent_warns = AgeBoundDict(timedelta(seconds=CONFIG['squelch_warning_seconds']),
 	# Bot message handling
 
 	@classmethod
-	def __bot_messages(cls, guild: Guild) -> AgeBoundDict[int, BotMessage]:
-		bm = Storage.get_state_value(guild, 'bot_messages')
+	def __bot_messages(cls, guild: Guild) -> AgeBoundDict[int, BotMessage, datetime, timedelta]:
+		bm: AgeBoundDict[int, BotMessage, datetime, timedelta] = Storage.get_state_value(guild, 'bot_messages')
 		if bm is None:
 			far_future = datetime.now(timezone.utc) + timedelta(days=1000)
 			bm = AgeBoundDict(timedelta(seconds=600),

rocketbot/cogs/crosspostcog.py

 Cog for detecting spam messages posted in multiple channels.
 """
 from datetime import datetime, timedelta
-from discord import Member, Message, utils as discordutils
+from typing import Optional
+
+from discord import Member, Message, utils as discordutils, TextChannel
 from discord.ext import commands
 
 from config import CONFIG
 	"""
 	Data about a set of duplicate messages from a user.
 	"""
-	def __init__(self, member, message_hash):
-		self.member = member
-		self.message_hash = message_hash
-		self.age = datetime.now()
-		self.bot_message = None  # BotMessage
-		self.is_kicked = False
-		self.is_banned = False
-		self.is_autobanned = False
-		self.spam_messages = set()  # of Message
-		self.deleted_messages = set()  # of Message
-		self.unique_channels = set()  # of TextChannel
+	def __init__(self, member: Member, message_hash: int) -> None:
+		self.member: Member = member
+		self.message_hash: int = message_hash
+		self.age: datetime = datetime.now()
+		self.bot_message: Optional[BotMessage] = None
+		self.is_kicked: bool = False
+		self.is_banned: bool = False
+		self.is_autobanned: bool = False
+		self.spam_messages: set[Message] = set()
+		self.deleted_messages: set[Message] = set()
+		self.unique_channels: set[TextChannel] = set()
 
 class CrossPostCog(BaseCog, name='Crosspost Detection'):
 	"""
 		if message.channel.permissions_for(message.author).ban_members:
 			# User exempt from spam detection
 			return
-		if len(message.content) < self.get_guild_setting(message.guild, self.SETTING_MIN_LENGTH):
+		def compute_message_hash(m: Message) -> int:
+			to_hash = m.content
+			for attachment in m.attachments:
+				to_hash += f'\n[[ATT: ct={attachment.content_type} s={attachment.size} w={attachment.width} h={attachment.height}]]'
+			h = hash(to_hash)
+			return h
+		compute_message_hash(message)
+		if len(message.attachments) == 0 and 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.get_guild_setting(message.guild, self.SETTING_TIMESPAN))
-		warn_count = self.get_guild_setting(message.guild, self.SETTING_WARN_COUNT)
-		recent_messages = Storage.get_state_value(message.guild, self.STATE_KEY_RECENT_MESSAGES)
+		warn_count: int = self.get_guild_setting(message.guild, self.SETTING_WARN_COUNT)
+		recent_messages: AgeBoundList[Message, datetime, timedelta] = Storage.get_state_value(message.guild, self.STATE_KEY_RECENT_MESSAGES)
 		if recent_messages is None:
 			recent_messages = AgeBoundList(max_age, lambda index, message : message.created_at)
 			Storage.set_state_value(message.guild, self.STATE_KEY_RECENT_MESSAGES, recent_messages)
 			return
 
 		# Look for repeats
-		hash_to_channels = {}  # int --> set(TextChannel)
+		hash_to_channels: dict[int, set[TextChannel]] = {}
 		max_count = 0
 		for m in member_messages:
-			key = hash(m.content)
-			channels = hash_to_channels.get(key)
+			message_hash = compute_message_hash(m)
+			channels: set[TextChannel] = hash_to_channels.get(message_hash)
 			if channels is None:
 				channels = set()
-				hash_to_channels[key] = channels
+				hash_to_channels[message_hash] = channels
 			channels.add(m.channel)
 			max_count = max(max_count, len(channels))
 		if max_count < warn_count:
 			return
 
 		# Handle the spam
-		spam_lookup = Storage.get_state_value(message.guild, self.STATE_KEY_SPAM_CONTEXT)
+		spam_lookup: SizeBoundDict[str, SpamContext, datetime] = Storage.get_state_value(message.guild, self.STATE_KEY_SPAM_CONTEXT)
 		if spam_lookup is None:
 			spam_lookup = SizeBoundDict(
 				self.max_spam_contexts,
 					f'\u0007{message.author.name} ({message.author.id}) ' + \
 					f'posted the same message in {channel_count} or more channels.')
 			for m in member_messages:
-				if hash(m.content) == message_hash:
+				if compute_message_hash(m) == message_hash:
 					context.spam_messages.add(m)
 					context.unique_channels.add(m.channel)
 			await self.__update_from_context(context)
 			message_type: int = BotMessage.TYPE_INFO if self.was_warned_recently(context.member) \
 				else BotMessage.TYPE_MOD_WARNING
 			message = BotMessage(context.member.guild, '', message_type, context)
-			message.quote = discordutils.remove_markdown(first_spam_message.clean_content())
+			message.quote = discordutils.remove_markdown(first_spam_message.clean_content)
 			self.record_warning(context.member)
 		if context.is_autobanned:
 			text = f'User {context.member.mention} auto banned for ' + \
 				message.author.bot or \
 				message.channel is None or \
 				message.guild is None or \
-				message.content is None or \
-				message.content == '':
+				message.content is None:
 			return
 		if not self.get_guild_setting(message.guild, self.SETTING_ENABLED):
 			return

rocketbot/cogs/joinagecog.py

 	async def search(self, context: commands.Context, timespan: str):
 		"""Command handler"""
 		guild: Guild = context.guild
-		recent_joins: AgeBoundList = Storage.get_state_value(guild, self.STATE_KEY_RECENT_JOINS)
+		recent_joins: AgeBoundList[Member, datetime, timedelta] = Storage.get_state_value(guild, self.STATE_KEY_RECENT_JOINS)
 		if recent_joins is None:
 			max_age: timedelta = timedelta(seconds=self.get_guild_setting(guild, self.SETTING_JOIN_TIME))
 			recent_joins = AgeBoundList(max_age, lambda i, member0 : member0.joined_at)
 		guild: Guild = member.guild
 		if not self.get_guild_setting(guild, self.SETTING_ENABLED):
 			return
-		recent_joins: AgeBoundList = Storage.get_state_value(guild, self.STATE_KEY_RECENT_JOINS)
+		recent_joins: AgeBoundList[Member, datetime, timedelta] = Storage.get_state_value(guild, self.STATE_KEY_RECENT_JOINS)
 		if recent_joins is None:
 			max_age: timedelta = timedelta(seconds=self.get_guild_setting(guild, self.SETTING_JOIN_TIME))
 			recent_joins = AgeBoundList(max_age, lambda i, member : member.joined_at)

rocketbot/cogs/joinraidcog.py

 		timespan: timedelta = timedelta(seconds=seconds)
 
 		last_raid: JoinRaidContext = Storage.get_state_value(guild, self.STATE_KEY_LAST_RAID)
-		recent_joins: AgeBoundList = Storage.get_state_value(guild, self.STATE_KEY_RECENT_JOINS)
+		recent_joins: AgeBoundList[Member, datetime, timedelta] = Storage.get_state_value(guild, self.STATE_KEY_RECENT_JOINS)
 		if recent_joins is None:
 			recent_joins = AgeBoundList(timespan, lambda i, member : member.joined_at)
 			Storage.set_state_value(guild, self.STATE_KEY_RECENT_JOINS, recent_joins)
 		if setting is self.SETTING_JOIN_TIME:
 			seconds = self.get_guild_setting(guild, self.SETTING_JOIN_TIME)
 			timespan: timedelta = timedelta(seconds=seconds)
-			recent_joins: AgeBoundList = Storage.get_state_value(guild,
+			recent_joins: AgeBoundList[Member, datetime, timedelta] = Storage.get_state_value(guild,
 				self.STATE_KEY_RECENT_JOINS)
 			if recent_joins:
 				recent_joins.max_age = timespan

rocketbot/cogs/patterncog.py

 				type=message_type,
 				context=context)
 			self.record_warning(message.author)
-			bm.quote = discordutils.remove_markdown(message.clean_content())
+			bm.quote = discordutils.remove_markdown(message.clean_content)
 			await bm.set_reactions(BotMessageReaction.standard_set(
 				did_delete=context.is_deleted,
 				did_kick=context.is_kicked,

rocketbot/cogs/urlspamcog.py

 					f'{join_age_str} after joining.',
 					type = BotMessage.TYPE_MOD_WARNING if needs_attention else BotMessage.TYPE_INFO,
 					context = context)
-			bm.quote = discordutils.remove_markdown(message.clean_content())
+			bm.quote = discordutils.remove_markdown(message.clean_content)
 			await bm.set_reactions(BotMessageReaction.standard_set(
 				did_delete=context.is_deleted,
 				did_kick=context.is_kicked,

rocketbot/collections.py

 """
 
 from abc import ABCMeta, abstractmethod
-from typing import Generic, TypeVar
+from typing import Callable, Generic, TypeVar, Optional
 
 # Abstract collections
 
+# Dictionary key
 K = TypeVar('K')
+# Collection value
 V = TypeVar('V')
+# Element age
+A = TypeVar('A')
+# Age delta
+D = TypeVar('D')
 
 class AbstractMutableList(list[V], Generic[V], metaclass=ABCMeta):
 	"""
 
 # Collections with limited number of elements
 
-class SizeBoundList(AbstractMutableList[V], Generic[V]):
+class SizeBoundList(AbstractMutableList[V], Generic[V, A]):
 	"""
 	Subclass of `list` that enforces a maximum number of elements.
 
 	"""
 	def __init__(self,
 			max_element_count: int,
-			element_age,
+			element_age: Callable[[int, V], A],
 			*args, **kwargs):
 		super().__init__(*args, **kwargs)
 		self.element_age = element_age
 	def copy(self):
 		return SizeBoundList(self.max_element_count, self.element_age, super())
 
-class SizeBoundSet(AbstractMutableSet[V], Generic[V]):
+class SizeBoundSet(AbstractMutableSet[V], Generic[V, A]):
 	"""
 	Subclass of `set` that enforces a maximum number of elements.
 
 	"""
 	def __init__(self,
 			max_element_count: int,
-			element_age,
+			element_age: Callable[[int, V], A],
 			*args, **kwargs):
 		super().__init__(*args, **kwargs)
 		self.element_age = element_age
 	def copy(self):
 		return SizeBoundSet(self.max_element_count, self.element_age, super())
 
-class SizeBoundDict(AbstractMutableDict[K, V], Generic[K, V]):
+class SizeBoundDict(AbstractMutableDict[K, V], Generic[K, V, A]):
 	"""
 	Subclass of `dict` that enforces a maximum number of elements.
 
 	"""
 	def __init__(self,
 			max_element_count: int,
-			element_age,
+			element_age: Callable[[K, V], A],
 			*args, **kwargs):
 		super().__init__(*args, **kwargs)
 		self.element_age = element_age
 
 # Collections with limited age of elements
 
-class AgeBoundList(AbstractMutableList[V], Generic[V]):
+class AgeBoundList(AbstractMutableList[V], Generic[V, A, D]):
 	"""
 	Subclass of `list` that enforces a maximum "age" of elements.
 
 	however elements will only be discarded following the next mutating
 	operation. Call `self.purge_old_elements()` to force resizing.
 	"""
-	def __init__(self, max_age, element_age, *args, **kwargs):
+	def __init__(self, max_age: D, element_age: Callable[[int, V], A], *args, **kwargs):
 		super().__init__(*args, **kwargs)
 		self.max_age = max_age
 		self.element_age = element_age
 		if self.is_culling or len(self) <= 1:
 			return
 		self.is_culling = True
-		min_age = None
-		max_age = None
-		ages = {}
+		min_age: Optional[A] = None
+		max_age: Optional[A] = None
+		ages: dict[int, A] = {}
 		for i, elem in enumerate(self):
-			age = self.element_age(i, elem)
+			age: A = self.element_age(i, elem)
 			ages[i] = age
 			if min_age is None or age < min_age:
 				min_age = age
 			if max_age is None or age > max_age:
 				max_age = age
-		cutoff = max_age - self.max_age
+		cutoff: A = max_age - self.max_age
 		if min_age >= cutoff:
 			self.is_culling = False
 			return
 	def copy(self):
 		return AgeBoundList(self.max_age, self.element_age, super())
 
-class AgeBoundSet(AbstractMutableSet[V], Generic[V]):
+class AgeBoundSet(AbstractMutableSet[V], Generic[V, A, D]):
 	"""
 	Subclass of `set` that enforces a maximum "age" of elements.
 
 	however elements will only be discarded following the next mutating
 	operation. Call `self.purge_old_elements()` to force resizing.
 	"""
-	def __init__(self, max_age, element_age, *args, **kwargs):
+	def __init__(self, max_age: D, element_age: Callable[[int, V], A], *args, **kwargs):
 		super().__init__(*args, **kwargs)
 		self.max_age = max_age
 		self.element_age = element_age
 	def copy(self):
 		return AgeBoundSet(self.max_age, self.element_age, super())
 
-class AgeBoundDict(AbstractMutableDict[K, V], Generic[K, V]):
+class AgeBoundDict(AbstractMutableDict[K, V], Generic[K, V, A, D]):
 	"""
 	Subclass of `dict` that enforces a maximum "age" of elements.
 
 	however elements will only be discarded following the next mutating
 	operation. Call `self.purge_old_elements()` to force resizing.
 	"""
-	def __init__(self, max_age, element_age, *args, **kwargs):
+	def __init__(self, max_age: D, element_age: Callable[[int, V], A], *args, **kwargs):
 		super().__init__(*args, **kwargs)
 		self.max_age = max_age
 		self.element_age = element_age

rocketbot/pattern.py

 		if self.field in ('content.markdown', 'content'):
 			return message.content
 		if self.field == 'content.plain':
-			return discordutils.remove_markdown(message.clean_content())
+			return discordutils.remove_markdown(message.clean_content)
 		if self.field == 'author':
 			return str(message.author.id)
 		if self.field == 'author.id':

rocketbot/storage.py

 		cls.__write_guild_config(guild, config)
 
 	@classmethod
-	def get_bot_messages(cls, guild: Guild) -> AgeBoundDict:
+	def get_bot_messages(cls, guild: Guild) -> AgeBoundDict[int, Any, datetime, timedelta]:
 		"""Returns all the bot messages for a guild."""
 		bm = cls.get_state_value(guild, 'bot_messages')
 		if bm is None: