aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--bot/cogs/antispam.py21
-rw-r--r--bot/cogs/reddit.py75
-rw-r--r--bot/cogs/superstarify/__init__.py48
-rw-r--r--bot/cogs/superstarify/stars.py3
-rw-r--r--bot/cogs/sync/__init__.py5
-rw-r--r--bot/cogs/sync/cog.py2
-rw-r--r--bot/cogs/sync/syncers.py7
7 files changed, 68 insertions, 93 deletions
diff --git a/bot/cogs/antispam.py b/bot/cogs/antispam.py
index 0c6a02bf9..c7f6503a8 100644
--- a/bot/cogs/antispam.py
+++ b/bot/cogs/antispam.py
@@ -33,19 +33,24 @@ RULE_FUNCTION_MAPPING = {
class AntiSpam:
+ """Spam detection & mitigation measures."""
+
def __init__(self, bot: Bot):
self.bot = bot
self._muted_role = Object(Roles.muted)
@property
def mod_log(self) -> ModLog:
+ """Get currently loaded ModLog cog instance."""
return self.bot.get_cog("ModLog")
- async def on_ready(self):
+ async def on_ready(self) -> None:
+ """Instantiate punishment role."""
role_id = AntiSpamConfig.punishment['role_id']
self.muted_role = Object(role_id)
- async def on_message(self, message: Message):
+ async def on_message(self, message: Message) -> None:
+ """Monitor incoming messages & match against spam criteria for possible filtering."""
if (
not message.guild
or message.guild.id != GuildConfig.id
@@ -97,7 +102,8 @@ class AntiSpam:
await self.maybe_delete_messages(message.channel, relevant_messages)
break
- async def punish(self, msg: Message, member: Member, reason: str, messages: List[Message], rule_name: str):
+ async def punish(self, msg: Message, member: Member, reason: str, messages: List[Message], rule_name: str) -> None:
+ """Deal out punishment for author(s) of messages that meet our spam criteria."""
# Sanity check to ensure we're not lagging behind
if self.muted_role not in member.roles:
remove_role_after = AntiSpamConfig.punishment['remove_after']
@@ -136,7 +142,8 @@ class AntiSpam:
# Run a tempmute
await mod_log_ctx.invoke(Moderation.tempmute, member, f"{remove_role_after}S", reason=reason)
- async def maybe_delete_messages(self, channel: TextChannel, messages: List[Message]):
+ async def maybe_delete_messages(self, channel: TextChannel, messages: List[Message]) -> None:
+ """Determine whether flagged messages should be disabled & delete them if so."""
# Is deletion of offending messages actually enabled?
if AntiSpamConfig.clean_offending:
@@ -153,7 +160,8 @@ class AntiSpam:
await messages[0].delete()
-def validate_config():
+def validate_config() -> None:
+ """Validate loaded antispam filter configuration(s)."""
for name, config in AntiSpamConfig.rules.items():
if name not in RULE_FUNCTION_MAPPING:
raise ValueError(
@@ -169,7 +177,8 @@ def validate_config():
)
-def setup(bot: Bot):
+def setup(bot: Bot) -> None:
+ """Antispam cog load."""
validate_config()
bot.add_cog(AntiSpam(bot))
log.info("Cog loaded: AntiSpam")
diff --git a/bot/cogs/reddit.py b/bot/cogs/reddit.py
index b5bd26e3d..69d4adf76 100644
--- a/bot/cogs/reddit.py
+++ b/bot/cogs/reddit.py
@@ -16,9 +16,7 @@ log = logging.getLogger(__name__)
class Reddit:
- """
- Track subreddit posts and show detailed statistics about them.
- """
+ """Track subreddit posts and show detailed statistics about them."""
HEADERS = {"User-Agent": "Discord Bot: PythonDiscord (https://pythondiscord.com/)"}
URL = "https://www.reddit.com"
@@ -34,11 +32,8 @@ class Reddit:
self.new_posts_task = None
self.top_weekly_posts_task = None
- async def fetch_posts(self, route: str, *, amount: int = 25, params=None):
- """
- A helper method to fetch a certain amount of Reddit posts at a given route.
- """
-
+ async def fetch_posts(self, route: str, *, amount: int = 25, params: dict = None) -> dict:
+ """A helper method to fetch a certain amount of Reddit posts at a given route."""
# Reddit's JSON responses only provide 25 posts at most.
if not 25 >= amount > 0:
raise ValueError("Invalid amount of subreddit posts requested.")
@@ -57,11 +52,10 @@ class Reddit:
return posts[:amount]
- async def send_top_posts(self, channel: TextChannel, subreddit: Subreddit, content=None, time="all"):
- """
- Create an embed for the top posts, then send it in a given TextChannel.
- """
-
+ async def send_top_posts(
+ self, channel: TextChannel, subreddit: Subreddit, content: str = None, time: str = "all"
+ ) -> None:
+ """Create an embed for the top posts, then send it in a given TextChannel."""
# Create the new spicy embed.
embed = Embed()
embed.description = ""
@@ -115,11 +109,8 @@ class Reddit:
embed=embed
)
- async def poll_new_posts(self):
- """
- Periodically search for new subreddit posts.
- """
-
+ async def poll_new_posts(self) -> None:
+ """Periodically search for new subreddit posts."""
while True:
await asyncio.sleep(RedditConfig.request_delay)
@@ -179,11 +170,8 @@ class Reddit:
log.trace(f"Sent {len(new_posts)} new {subreddit} posts to channel {self.reddit_channel.id}.")
- async def poll_top_weekly_posts(self):
- """
- Post a summary of the top posts every week.
- """
-
+ async def poll_top_weekly_posts(self) -> None:
+ """Post a summary of the top posts every week."""
while True:
now = datetime.utcnow()
@@ -214,19 +202,13 @@ class Reddit:
await message.pin()
@group(name="reddit", invoke_without_command=True)
- async def reddit_group(self, ctx: Context):
- """
- View the top posts from various subreddits.
- """
-
+ async def reddit_group(self, ctx: Context) -> None:
+ """View the top posts from various subreddits."""
await ctx.invoke(self.bot.get_command("help"), "reddit")
@reddit_group.command(name="top")
- async def top_command(self, ctx: Context, subreddit: Subreddit = "r/Python"):
- """
- Send the top posts of all time from a given subreddit.
- """
-
+ async def top_command(self, ctx: Context, subreddit: Subreddit = "r/Python") -> None:
+ """Send the top posts of all time from a given subreddit."""
await self.send_top_posts(
channel=ctx.channel,
subreddit=subreddit,
@@ -235,11 +217,8 @@ class Reddit:
)
@reddit_group.command(name="daily")
- async def daily_command(self, ctx: Context, subreddit: Subreddit = "r/Python"):
- """
- Send the top posts of today from a given subreddit.
- """
-
+ async def daily_command(self, ctx: Context, subreddit: Subreddit = "r/Python") -> None:
+ """Send the top posts of today from a given subreddit."""
await self.send_top_posts(
channel=ctx.channel,
subreddit=subreddit,
@@ -248,11 +227,8 @@ class Reddit:
)
@reddit_group.command(name="weekly")
- async def weekly_command(self, ctx: Context, subreddit: Subreddit = "r/Python"):
- """
- Send the top posts of this week from a given subreddit.
- """
-
+ async def weekly_command(self, ctx: Context, subreddit: Subreddit = "r/Python") -> None:
+ """Send the top posts of this week from a given subreddit."""
await self.send_top_posts(
channel=ctx.channel,
subreddit=subreddit,
@@ -262,11 +238,8 @@ class Reddit:
@with_role(*STAFF_ROLES)
@reddit_group.command(name="subreddits", aliases=("subs",))
- async def subreddits_command(self, ctx: Context):
- """
- Send a paginated embed of all the subreddits we're relaying.
- """
-
+ async def subreddits_command(self, ctx: Context) -> None:
+ """Send a paginated embed of all the subreddits we're relaying."""
embed = Embed()
embed.title = "Relayed subreddits."
embed.colour = Colour.blurple()
@@ -279,7 +252,8 @@ class Reddit:
max_lines=15
)
- async def on_ready(self):
+ async def on_ready(self) -> None:
+ """Initiate reddit post event loop."""
self.reddit_channel = self.bot.get_channel(Channels.reddit)
if self.reddit_channel is not None:
@@ -291,6 +265,7 @@ class Reddit:
log.warning("Couldn't locate a channel for subreddit relaying.")
-def setup(bot):
+def setup(bot: Bot) -> None:
+ """Reddit cog load."""
bot.add_cog(Reddit(bot))
log.info("Cog loaded: Reddit")
diff --git a/bot/cogs/superstarify/__init__.py b/bot/cogs/superstarify/__init__.py
index 4b26f3f40..bd5211102 100644
--- a/bot/cogs/superstarify/__init__.py
+++ b/bot/cogs/superstarify/__init__.py
@@ -19,29 +19,29 @@ NICKNAME_POLICY_URL = "https://pythondiscord.com/about/rules#nickname-policy"
class Superstarify:
- """
- A set of commands to moderate terrible nicknames.
- """
+ """A set of commands to moderate terrible nicknames."""
def __init__(self, bot: Bot):
self.bot = bot
@property
def moderation(self) -> Moderation:
+ """Get currently loaded Moderation cog instance."""
return self.bot.get_cog("Moderation")
@property
def modlog(self) -> ModLog:
+ """Get currently loaded ModLog cog instance."""
return self.bot.get_cog("ModLog")
- async def on_member_update(self, before: Member, after: Member):
+ async def on_member_update(self, before: Member, after: Member) -> None:
"""
This event will trigger when someone changes their name.
- At this point we will look up the user in our database and check
- whether they are allowed to change their names, or if they are in
- superstar-prison. If they are not allowed, we will change it back.
- """
+ At this point we will look up the user in our database and check whether they are allowed to
+ change their names, or if they are in superstar-prison. If they are not allowed, we will
+ change it back.
+ """
if before.display_name == after.display_name:
return # User didn't change their nickname. Abort!
@@ -91,14 +91,13 @@ class Superstarify:
"to DM them, and a discord.errors.Forbidden error was incurred."
)
- async def on_member_join(self, member: Member):
+ async def on_member_join(self, member: Member) -> None:
"""
This event will trigger when someone (re)joins the server.
- At this point we will look up the user in our database and check
- whether they are in superstar-prison. If so, we will change their name
- back to the forced nickname.
- """
+ At this point we will look up the user in our database and check whether they are in
+ superstar-prison. If so, we will change their name back to the forced nickname.
+ """
active_superstarifies = await self.bot.api_client.get(
'bot/infractions',
params={
@@ -153,13 +152,14 @@ class Superstarify:
@with_role(*MODERATION_ROLES)
async def superstarify(
self, ctx: Context, member: Member, expiration: ExpirationDate, reason: str = None
- ):
+ ) -> None:
"""
- This command will force a random superstar name (like Taylor Swift) to be the user's
- nickname for a specified duration. An optional reason can be provided.
+ Force a random superstar name (like Taylor Swift) to be the user's nickname for a specified duration.
+
+ An optional reason can be provided.
+
If no reason is given, the original name will be shown in a generated reason.
"""
-
active_superstarifies = await self.bot.api_client.get(
'bot/infractions',
params={
@@ -222,15 +222,8 @@ class Superstarify:
@command(name='unsuperstarify', aliases=('release_nick', 'unstar'))
@with_role(*MODERATION_ROLES)
- async def unsuperstarify(self, ctx: Context, member: Member):
- """
- This command will remove the entry from our database, allowing the user
- to once again change their nickname.
-
- :param ctx: Discord message context
- :param member: The member to unsuperstarify
- """
-
+ async def unsuperstarify(self, ctx: Context, member: Member) -> None:
+ """This command will the superstarify entry from our database, allowing the user to change their nickname."""
log.debug(f"Attempting to unsuperstarify the following user: {member.display_name}")
embed = Embed()
@@ -268,6 +261,7 @@ class Superstarify:
await ctx.send(embed=embed)
-def setup(bot):
+def setup(bot: Bot) -> None:
+ """Superstarify cog load."""
bot.add_cog(Superstarify(bot))
log.info("Cog loaded: Superstarify")
diff --git a/bot/cogs/superstarify/stars.py b/bot/cogs/superstarify/stars.py
index 9b49d7175..dbac86770 100644
--- a/bot/cogs/superstarify/stars.py
+++ b/bot/cogs/superstarify/stars.py
@@ -81,6 +81,7 @@ STAR_NAMES = (
)
-def get_nick(infraction_id, member_id):
+def get_nick(infraction_id: int, member_id: int) -> str:
+ """Randomly select a nickname from the Superstarify nickname list."""
rng = random.Random(str(infraction_id) + str(member_id))
return rng.choice(STAR_NAMES)
diff --git a/bot/cogs/sync/__init__.py b/bot/cogs/sync/__init__.py
index e4f960620..d4565f848 100644
--- a/bot/cogs/sync/__init__.py
+++ b/bot/cogs/sync/__init__.py
@@ -1,10 +1,13 @@
import logging
+from discord.ext.commands import Bot
+
from .cog import Sync
log = logging.getLogger(__name__)
-def setup(bot):
+def setup(bot: Bot) -> None:
+ """Sync cog load."""
bot.add_cog(Sync(bot))
log.info("Cog loaded: Sync")
diff --git a/bot/cogs/sync/cog.py b/bot/cogs/sync/cog.py
index 79177b69e..928ffa418 100644
--- a/bot/cogs/sync/cog.py
+++ b/bot/cogs/sync/cog.py
@@ -170,7 +170,6 @@ class Sync:
@commands.has_permissions(administrator=True)
async def sync_roles_command(self, ctx: Context) -> None:
"""Manually synchronize the guild's roles with the roles on the site."""
-
initial_response = await ctx.send("📊 Synchronizing roles.")
total_created, total_updated, total_deleted = await syncers.sync_roles(self.bot, ctx.guild)
await initial_response.edit(
@@ -184,7 +183,6 @@ class Sync:
@commands.has_permissions(administrator=True)
async def sync_users_command(self, ctx: Context) -> None:
"""Manually synchronize the guild's users with the users on the site."""
-
initial_response = await ctx.send("📊 Synchronizing users.")
total_created, total_updated, total_deleted = await syncers.sync_users(self.bot, ctx.guild)
await initial_response.edit(
diff --git a/bot/cogs/sync/syncers.py b/bot/cogs/sync/syncers.py
index 414c24adb..689d3736e 100644
--- a/bot/cogs/sync/syncers.py
+++ b/bot/cogs/sync/syncers.py
@@ -34,7 +34,6 @@ def get_roles_for_sync(
to be deleted on the site, meaning the roles are present on
the API but not in the cached guild.
"""
-
guild_role_ids = {role.id for role in guild_roles}
api_role_ids = {role.id for role in api_roles}
new_role_ids = guild_role_ids - api_role_ids
@@ -66,7 +65,6 @@ async def sync_roles(bot: Bot, guild: Guild) -> Tuple[int, int, int]:
(element `0`) , how many roles were updated (element `1`), and how many
roles were deleted (element `2`) on the API.
"""
-
roles = await bot.api_client.get('bot/roles')
# Pack API roles and guild roles into one common format,
@@ -138,7 +136,6 @@ def get_users_for_sync(
guild, but where the attribute of a user on the API is not
equal to the attribute of the user on the guild.
"""
-
users_to_create = set()
users_to_update = set()
@@ -169,8 +166,7 @@ def get_users_for_sync(
async def sync_users(bot: Bot, guild: Guild) -> Tuple[int, int, None]:
"""
- Synchronize users found on the given
- `guild` with the ones on the API.
+ Synchronize users found on the given `guild` with the ones on the API.
Arguments:
bot (discord.ext.commands.Bot):
@@ -186,7 +182,6 @@ async def sync_users(bot: Bot, guild: Guild) -> Tuple[int, int, None]:
(element `0`) and how many users were updated (element `1`), and `None`
to indicate that a user sync never deletes entries from the API.
"""
-
current_users = await bot.api_client.get('bot/users')
# Pack API users and guild users into one common format,
generated by cgit v1.2.3 (git 2.39.1) at 2025-10-22 14:24:53 +0000