diff --git a/.gitignore b/.gitignore index 0211b65..81aca1b 100644 --- a/.gitignore +++ b/.gitignore @@ -1,6 +1,13 @@ -.vscode +# IDE configuration folder .idea +.vscode + +# Logs +logs/logbooks + +# Secret files +env -__pycache__/ -env/ -logs/logbooks/ +# Python environement +__pycache__ +venv diff --git a/.pylintrc b/.pylintrc index eb27d5e..525c7b9 100644 --- a/.pylintrc +++ b/.pylintrc @@ -74,85 +74,14 @@ confidence= # --enable=similarities". If you want to run only the classes checker, but have # no Warning level messages displayed, use "--disable=all --enable=classes # --disable=W". -disable=print-statement, - parameter-unpacking, - unpacking-in-except, - old-raise-syntax, - backtick, - long-suffix, - old-ne-operator, - old-octal-literal, - import-star-module-level, - non-ascii-bytes-literal, - raw-checker-failed, +disable=raw-checker-failed, bad-inline-option, locally-disabled, file-ignored, suppressed-message, useless-suppression, deprecated-pragma, - use-symbolic-message-instead, - apply-builtin, - basestring-builtin, - buffer-builtin, - cmp-builtin, - coerce-builtin, - execfile-builtin, - file-builtin, - long-builtin, - raw_input-builtin, - reduce-builtin, - standarderror-builtin, - unicode-builtin, - xrange-builtin, - coerce-method, - delslice-method, - getslice-method, - setslice-method, - no-absolute-import, - old-division, - dict-iter-method, - dict-view-method, - next-method-called, - metaclass-assignment, - indexing-exception, - raising-string, - reload-builtin, - oct-method, - hex-method, - nonzero-method, - cmp-method, - input-builtin, - round-builtin, - intern-builtin, - unichr-builtin, - map-builtin-not-iterating, - zip-builtin-not-iterating, - range-builtin-not-iterating, - filter-builtin-not-iterating, - using-cmp-argument, - eq-without-hash, - div-method, - idiv-method, - rdiv-method, - exception-message-attribute, - invalid-str-codec, - sys-max-int, - bad-python3-import, - deprecated-string-function, - deprecated-str-translate-call, - deprecated-itertools-function, - deprecated-types-field, - next-method-defined, - dict-items-not-iterating, - dict-keys-not-iterating, - dict-values-not-iterating, - deprecated-operator-function, - deprecated-urllib-function, - xreadlines-attribute, - deprecated-sys-function, - exception-escape, - comprehension-escape + use-symbolic-message-instead # Enable the message, report, category or checker with the given id(s). You can # either give multiple identifier separated by comma (,) or put this option diff --git a/cogs/fun.py b/cogs/fun.py index 3e0efe3..5d5ff49 100644 --- a/cogs/fun.py +++ b/cogs/fun.py @@ -3,15 +3,12 @@ from typing import Optional import aiohttp -from bs4 import BeautifulSoup import discord +from bs4 import BeautifulSoup from discord.ext import commands -from emoji import UNICODE_EMOJI - from src.utils import user_only - # Supported action commands with they template. # {author} -> command author's mention # {target} -> user pinged in the command @@ -22,7 +19,7 @@ "cuddle": "{author} cuddles {target} :heart:", "poke": "Hey {target}! {author} poked you", "baka": "{target} BAKA", - "slap": "{author} slapped {target}!" + "slap": "{author} slapped {target}!", } @@ -52,15 +49,18 @@ async def action(self, ctx, member: discord.User): # Make a request to the nekos.life API at the command's name endpoint. async with aiohttp.ClientSession() as session: - async with session.get(f"https://nekos.life/api/v2/img/{ctx.invoked_with}") as response: + async with session.get( + f"https://nekos.life/api/v2/img/{ctx.invoked_with}" + ) as response: if response.status == 200: data = await response.json() else: # Send an error embed. embed.title = "API error" - embed.description = ("Sorry, a problem has occurred when " - "trying to interact with the " - "nekos.life API") + embed.description = ( + "Sorry, a problem has occurred when trying to interact " + "with the nekos.life API" + ) return await ctx.send(embed=embed) # Place the nekos.life's gif in the embed. @@ -68,11 +68,13 @@ async def action(self, ctx, member: discord.User): # Make a custom message based on the command's template message. if ctx.author.id != member.id: - embed.description = actions[ctx.invoked_with].format(author=ctx.author.name, - target=member.name) + embed.description = actions[ctx.invoked_with].format( + author=ctx.author.name, target=member.name + ) else: - embed.description = ("Aww, I see you are lonely, I will " - f"{ctx.invoked_with} you") + embed.description = ( + f"Aww, I see you are lonely, I will {ctx.invoked_with} you" + ) await ctx.send(embed=embed) @@ -88,9 +90,11 @@ async def apod(self, ctx): apod = await response.text() else: # Send an error embed. - embed.description = ("Sorry, a problem has occurred when " - "trying to interact with the apod " - "website") + embed.description = ( + "Sorry, a problem has occurred when " + "trying to interact with the apod " + "website" + ) return await ctx.send(embed=embed) # Collect informations. @@ -121,10 +125,13 @@ def __init__(self, bot, config): self.confession_queue = {} + self.confession_is_confirm_e: Optional[bool] = None + self.confession_msg: Optional[str] = None + @commands.Cog.listener() @user_only() async def on_message(self, message): - """"Decide to store message and send a confirm message.""" + """ "Decide to store message and send a confirm message.""" self.confession_is_confirm_e = False # Check if Confession is enabled. @@ -136,24 +143,27 @@ async def on_message(self, message): if not message.guild: self.confession_is_confirm_e = True self.confession_msg = message - await message.channel.send("React to this message to send it to " - "the confession channel") + await message.channel.send( + "React to this message to send it to the confession channel" + ) @commands.Cog.listener() async def on_raw_reaction_add(self, reaction): """Decide to send stored confession message.""" reaction_user = await self.bot.fetch_user(reaction.user_id) reaction_channel = await self.bot.fetch_channel(reaction.channel_id) - confession_channel = await self.bot.fetch_channel(self.config["CONFESSION"]["CHANNEL"]) - user_in_chan_guild = await confession_channel.guild.fetch_member(reaction.user_id) - - if (reaction_user.bot - or not self.confession_is_confirm_e): + confession_channel = await self.bot.fetch_channel( + self.config["CONFESSION"]["CHANNEL"] + ) + user_in_chan_guild = await confession_channel.guild.fetch_member( + reaction.user_id + ) + + if reaction_user.bot or not self.confession_is_confirm_e: return # If reaction is in DM and user is present in confession channel, - # send the message. - if (reaction_channel.type.name == "private" - and user_in_chan_guild is not None): + # send the message. + if reaction_channel.type.name == "private" and user_in_chan_guild is not None: self.confession_is_confirm_e = False await confession_channel.send(self.confession_msg.content) diff --git a/cogs/moderation.py b/cogs/moderation.py index f963056..13e55ce 100644 --- a/cogs/moderation.py +++ b/cogs/moderation.py @@ -1,9 +1,7 @@ """Contains cog related to moderation.""" import re -import unicodedata -import discord from discord.ext import commands from src.utils import user_only @@ -16,14 +14,15 @@ def __init__(self, bot, config): self.bot = bot self.config = config - self.regex_patterns = {chan_id: re.compile(pattern) - for chan_id, pattern in config["REGEX_CHANNELS"].items()} + self.regex_patterns = { + chan_id: re.compile(pattern) + for chan_id, pattern in config["REGEX_CHANNELS"].items() + } @commands.Cog.listener() @user_only() async def on_message(self, message): """Delete message that does not match the channel regex.""" - pattern = self.regex_patterns.get(str(message.channel.id)) - if pattern: + if pattern := self.regex_patterns.get(str(message.channel.id)): if not pattern.fullmatch(message.content): await message.delete() diff --git a/cogs/omega.py b/cogs/omega.py index 1abb7b2..b31bae5 100644 --- a/cogs/omega.py +++ b/cogs/omega.py @@ -3,7 +3,7 @@ import asyncio from datetime import datetime import re -from typing import AsyncGenerator +from typing import AsyncGenerator, Optional import aiohttp import discord @@ -18,7 +18,7 @@ async def get_github_issues(message: discord.Message) -> AsyncGenerator[dict, No If a request error occurs, it sends a message and stops. """ - matches = re.findall("(?=((^| )#[0-9]+(e|u|l)?($| )))", message.content) + matches = re.findall("(?=((^| )#[0-9]+([eul])?($| )))", message.content) async with aiohttp.ClientSession() as session: for i in matches: @@ -35,36 +35,38 @@ async def get_github_issues(message: discord.Message) -> AsyncGenerator[dict, No else: repo = "omega-numworks/omega" - async with session.get(f"https://api.github.com/repos/{repo}/issues/{issue}") as response: + async with session.get( + f"https://api.github.com/repos/{repo}/issues/{issue}" + ) as response: if response.status != 200: - await message.channel.send("Erreur lors de la requête " - f"({response.status})") + await message.channel.send( + f"Erreur lors de la requête ({response.status})" + ) return yield await response.json() async def make_embed(data: dict) -> discord.Embed: """Return a formatted ``discord.Embed`` from given data.""" - embed = discord.Embed(title=data["title"], - url=data["html_url"], - description=data["body"]) + embed = discord.Embed( + title=data["title"], url=data["html_url"], description=data["body"] + ) # Truncate the description if it's above the maximum size. if len(embed.description) > 2048: - embed.description = embed.description[:2043] + "[...]" + embed.description = f"{embed.description[:2043]}[...]" author = data["user"] - embed.set_author(name=author["login"], - url=author["html_url"], - icon_url=author["avatar_url"]) + embed.set_author( + name=author["login"], url=author["html_url"], icon_url=author["avatar_url"] + ) additional_infos = [] if data.get("locked"): additional_infos.append(":lock: locked") - pull_request = data.get("pull_request") - if pull_request: + if pull_request := data.get("pull_request"): additional_infos.append(":arrows_clockwise: Pull request") async with aiohttp.ClientSession() as session: @@ -72,11 +74,12 @@ async def make_embed(data: dict) -> discord.Embed: commits_data = await response.json() # Format all commits data into strings. - formatted = ["[`{}`]({}) {} - {}".format(commit['sha'][:7], - commit['html_url'], - commit['commit']['message'], - commit['committer']['login']) - for commit in commits_data] + formatted = [ + ( + f"[`{commit['sha'][:7]}`]({commit['html_url']})" + f" {commit['commit']['message']} - {commit['committer']['login']}" + ) for commit in commits_data + ] result = "\n".join(formatted) @@ -95,51 +98,60 @@ async def make_embed(data: dict) -> discord.Embed: embed.add_field(name="Commits", value=result) if data["comments"]: - additional_infos.append(":speech_balloon: Comments : " - f"{data['comments']}") + additional_infos.append(f":speech_balloon: Comments : {data['comments']}") if data["state"] == "closed": - closed_at = datetime.strptime(data["closed_at"], - "%Y-%m-%dT%H:%M:%SZ").strftime("%b. %d %H:%M %Y") - additional_infos.append(":x: Closed by " - f"{data['closed_by']['login']} on " - f"{closed_at}") + closed_at = datetime.strptime(data["closed_at"], "%Y-%m-%dT%H:%M:%SZ").strftime( + "%b. %d %H:%M %Y" + ) + + additional_infos.append( + f":x: Closed by {data['closed_by']['login']} on {closed_at}" + ) + elif data["state"] == "open": additional_infos.append(":white_check_mark: Open") if data["labels"]: - labels = '` `'.join(i['name'] for i in data['labels']) + labels = "` `".join(i["name"] for i in data["labels"]) additional_infos.append(f":label: Labels: `{labels}`") - embed.add_field(name="Additional informations", - value="\n".join(additional_infos)) + embed.add_field(name="Additional informations", value="\n".join(additional_infos)) return embed -async def make_color_embed(hex_code: int) -> discord.Embed: +async def make_color_embed( + hex_code: int, message: discord.Message +) -> Optional[discord.Embed]: """Return a ``discord.Embed`` contains some informations about color of given ``hex_code``. """ async with aiohttp.ClientSession() as session: - async with session.get(f"https://www.thecolorapi.com/id?hex={hex_code}") as r: - if r.status == 200: - data = await r.json() + async with session.get( + f"https://www.thecolorapi.com/id?hex={hex_code}" + ) as response: + if response.status == 200: + data = await response.json() else: - return await message.channel.send("Erreur lors de la requête " - f"({r.status})") + await message.channel.send( + f"Erreur lors de la requête ({response.status})" + ) + return title = f"{data['name']['value']} color" description = f"**Hex:** #{hex_code}\n" - description += "\n".join("**{}:** {}, {}, {}".format( - color_format.capitalize(), - *[data[color_format][letter] - for letter in tuple(color_format)]) - for color_format in ("rgb", "hsl", "hsv")) + description += "\n".join( + ( + f"**{color_format.capitalize()}:**" + f"{', '.join(data[color_format][letter] for letter in tuple(color_format))}" + ) + for color_format in ("rgb", "hsl", "hsv") + ) - return discord.Embed(title=title, - description=description, - color=int(hex_code, base=16)) + return discord.Embed( + title=title, description=description, color=int(hex_code, base=16) + ) class Omega(commands.Cog): @@ -159,7 +171,7 @@ async def on_message(self, message): # Checks if the message is an hex code if re.match("^#([A-Fa-f0-9]{6})$", message.content): hex_code = message.content.lstrip("#") - color_embed = await make_color_embed(hex_code) + color_embed = await make_color_embed(hex_code, message) await message.channel.send(embed=color_embed) @@ -190,8 +202,9 @@ async def on_raw_reaction_add(self, reaction): # If the reaction is "🗑️" and on a message stored in issue_embeds, # it deletes it on discord and in the storage dictionary. - if (reaction.emoji.name == "🗑️" - and self.issue_embeds.pop(reaction.message_id, None)): + if reaction.emoji.name == "🗑️" and self.issue_embeds.pop( + reaction.message_id, None + ): channel = self.bot.get_channel(reaction.channel_id) message = await channel.fetch_message(reaction.message_id) await message.delete() diff --git a/logs/logger.py b/logs/logger.py index 1260f76..295e0e9 100644 --- a/logs/logger.py +++ b/logs/logger.py @@ -7,10 +7,10 @@ logger = logging.getLogger("Omega-robot logger") logger.setLevel(logging.INFO) -formatter = logging.Formatter('%(asctime)s :: %(levelname)s :: %(message)s') -file_handler = RotatingFileHandler("logs/logbooks/activity.log", - maxBytes=5000000, - backupCount=5) +formatter = logging.Formatter("%(asctime)s :: %(levelname)s :: %(message)s") +file_handler = RotatingFileHandler( + "logs/logbooks/activity.log", maxBytes=5000000, backupCount=5 +) file_handler.setLevel(logging.INFO) file_handler.setFormatter(formatter) diff --git a/main.py b/main.py index d858e6d..2d3dc02 100644 --- a/main.py +++ b/main.py @@ -5,6 +5,7 @@ import json from discord.ext import commands +from discord import Intents from cogs.omega import Omega from cogs.moderation import Moderation @@ -19,27 +20,21 @@ class Bot(commands.Bot): """Encapsulate a discord bot.""" - extensions = ( - Fun, - Moderation, - Omega - ) - optionals = { - Confession: config["CONFESSION"]["ENABLED"] - } + extensions = (Fun, Moderation, Omega) + + optionals = {Confession: config["CONFESSION"]["ENABLED"]} def __init__(self): - super().__init__(config["PREFIX"]) + super().__init__(config["PREFIX"], intents=Intents.all()) self.description = "A bot for two Omega Discord servers." self.token = config["TOKEN"] async def on_ready(self): """Log information about bot launching.""" - logger.info("Bot %s connected on %s servers", - self.user.name, - len(self.guilds)) + logger.info("Bot %s connected on %s servers", self.user.name, len(self.guilds)) + await self.load_extensions() async def on_command(self, msg): """Log each command submitted. Log message provides information @@ -47,22 +42,24 @@ async def on_command(self, msg): """ args = msg.args[2:] - args_info = ', '.join(repr(arg) for arg in args) if args else "" + args_info = ", ".join(repr(arg) for arg in args) if args else "" - log_msg = (f"{msg.command.name} called by {msg.author} with " - f"args {args_info}.") + log_msg = ( + f"{msg.command.name} called by {msg.author} with " f"args {args_info}." + ) logger.info(log_msg) - def run(self): + def run(self, **kwargs): """Start the bot and load one by one available cogs.""" + super().run(self.token) + + async def load_extensions(self): for cog in self.extensions: - self.add_cog(cog(self, config)) + await self.add_cog(cog(self, config)) for cog, requirement in self.optionals.items(): if requirement: - self.add_cog(cog(self, config)) - - super().run(self.token) + await self.add_cog(cog(self, config)) if __name__ == "__main__": diff --git a/src/utils.py b/src/utils.py index 9ac5da7..b247eee 100644 --- a/src/utils.py +++ b/src/utils.py @@ -3,8 +3,10 @@ from discord.ext import commands -def user_only() -> bool: +def user_only(): """A decorator to assume the author of message is not a bot.""" + async def predicate(ctx) -> bool: return not ctx.author.bot + return commands.check(predicate)