aboutsummaryrefslogtreecommitdiffstats
path: root/bot/exts/evergreen
diff options
context:
space:
mode:
authorGravatar ChrisJL <[email protected]>2021-05-13 20:28:20 +0100
committerGravatar GitHub <[email protected]>2021-05-13 20:28:20 +0100
commit196c435c586b3bc2d30aa6a9865b8d4498d9c5b8 (patch)
treec9585d2a668d75bbbda0fab5c1b80962a9946bb7 /bot/exts/evergreen
parentRemove redundant comments in bookmark cog (diff)
parentMerge pull request #727 from python-discord/reddit-revoke (diff)
Merge branch 'main' into bookmark-react-for-copy
Diffstat (limited to 'bot/exts/evergreen')
-rw-r--r--bot/exts/evergreen/8bitify.py54
-rw-r--r--bot/exts/evergreen/avatar_modification/__init__.py0
-rw-r--r--bot/exts/evergreen/avatar_modification/_effects.py287
-rw-r--r--bot/exts/evergreen/avatar_modification/avatar_modify.py370
-rw-r--r--bot/exts/evergreen/battleship.py2
-rw-r--r--bot/exts/evergreen/catify.py88
-rw-r--r--bot/exts/evergreen/error_handler.py10
-rw-r--r--bot/exts/evergreen/fun.py3
-rw-r--r--bot/exts/evergreen/githubinfo.py143
-rw-r--r--bot/exts/evergreen/help.py4
-rw-r--r--bot/exts/evergreen/issues.py302
-rw-r--r--bot/exts/evergreen/latex.py94
-rw-r--r--bot/exts/evergreen/ping.py44
-rw-r--r--bot/exts/evergreen/reddit.py425
-rw-r--r--bot/exts/evergreen/timed.py46
-rw-r--r--bot/exts/evergreen/uptime.py33
-rw-r--r--bot/exts/evergreen/wolfram.py3
17 files changed, 1572 insertions, 336 deletions
diff --git a/bot/exts/evergreen/8bitify.py b/bot/exts/evergreen/8bitify.py
deleted file mode 100644
index 54e68f80..00000000
--- a/bot/exts/evergreen/8bitify.py
+++ /dev/null
@@ -1,54 +0,0 @@
-from io import BytesIO
-
-import discord
-from PIL import Image
-from discord.ext import commands
-
-
-class EightBitify(commands.Cog):
- """Make your avatar 8bit!"""
-
- def __init__(self, bot: commands.Bot) -> None:
- self.bot = bot
-
- @staticmethod
- def pixelate(image: Image) -> Image:
- """Takes an image and pixelates it."""
- return image.resize((32, 32), resample=Image.NEAREST).resize((1024, 1024), resample=Image.NEAREST)
-
- @staticmethod
- def quantize(image: Image) -> Image:
- """Reduces colour palette to 256 colours."""
- return image.quantize()
-
- @commands.command(name="8bitify")
- async def eightbit_command(self, ctx: commands.Context) -> None:
- """Pixelates your avatar and changes the palette to an 8bit one."""
- async with ctx.typing():
- image_bytes = await ctx.author.avatar_url.read()
- avatar = Image.open(BytesIO(image_bytes))
- avatar = avatar.convert("RGBA").resize((1024, 1024))
-
- eightbit = self.pixelate(avatar)
- eightbit = self.quantize(eightbit)
-
- bufferedio = BytesIO()
- eightbit.save(bufferedio, format="PNG")
- bufferedio.seek(0)
-
- file = discord.File(bufferedio, filename="8bitavatar.png")
-
- embed = discord.Embed(
- title="Your 8-bit avatar",
- description='Here is your avatar. I think it looks all cool and "retro"'
- )
-
- embed.set_image(url="attachment://8bitavatar.png")
- embed.set_footer(text=f"Made by {ctx.author.display_name}", icon_url=ctx.author.avatar_url)
-
- await ctx.send(file=file, embed=embed)
-
-
-def setup(bot: commands.Bot) -> None:
- """Cog load."""
- bot.add_cog(EightBitify(bot))
diff --git a/bot/exts/evergreen/avatar_modification/__init__.py b/bot/exts/evergreen/avatar_modification/__init__.py
new file mode 100644
index 00000000..e69de29b
--- /dev/null
+++ b/bot/exts/evergreen/avatar_modification/__init__.py
diff --git a/bot/exts/evergreen/avatar_modification/_effects.py b/bot/exts/evergreen/avatar_modification/_effects.py
new file mode 100644
index 00000000..d2370b4b
--- /dev/null
+++ b/bot/exts/evergreen/avatar_modification/_effects.py
@@ -0,0 +1,287 @@
+import math
+import random
+import typing as t
+from io import BytesIO
+from pathlib import Path
+
+import discord
+from PIL import Image, ImageDraw, ImageOps
+
+from bot.constants import Colours
+
+
+class PfpEffects:
+ """
+ Implements various image modifying effects, for the PfpModify cog.
+
+ All of these fuctions are slow, and blocking, so they should be ran in executors.
+ """
+
+ @staticmethod
+ def apply_effect(image_bytes: bytes, effect: t.Callable, filename: str, *args) -> discord.File:
+ """Applies the given effect to the image passed to it."""
+ im = Image.open(BytesIO(image_bytes))
+ im = im.convert("RGBA")
+ im = effect(im, *args)
+
+ bufferedio = BytesIO()
+ im.save(bufferedio, format="PNG")
+ bufferedio.seek(0)
+
+ return discord.File(bufferedio, filename=filename)
+
+ @staticmethod
+ def closest(x: t.Tuple[int, int, int]) -> t.Tuple[int, int, int]:
+ """
+ Finds the closest "easter" colour to a given pixel.
+
+ Returns a merge between the original colour and the closest colour.
+ """
+ r1, g1, b1 = x
+
+ def distance(point: t.Tuple[int, int, int]) -> t.Tuple[int, int, int]:
+ """Finds the difference between a pastel colour and the original pixel colour."""
+ r2, g2, b2 = point
+ return (r1 - r2) ** 2 + (g1 - g2) ** 2 + (b1 - b2) ** 2
+
+ closest_colours = sorted(Colours.easter_like_colours, key=distance)
+ r2, g2, b2 = closest_colours[0]
+ r = (r1 + r2) // 2
+ g = (g1 + g2) // 2
+ b = (b1 + b2) // 2
+
+ return r, g, b
+
+ @staticmethod
+ def crop_avatar_circle(avatar: Image.Image) -> Image.Image:
+ """This crops the avatar given into a circle."""
+ mask = Image.new("L", avatar.size, 0)
+ draw = ImageDraw.Draw(mask)
+ draw.ellipse((0, 0) + avatar.size, fill=255)
+ avatar.putalpha(mask)
+ return avatar
+
+ @staticmethod
+ def crop_ring(ring: Image.Image, px: int) -> Image.Image:
+ """This crops the given ring into a circle."""
+ mask = Image.new("L", ring.size, 0)
+ draw = ImageDraw.Draw(mask)
+ draw.ellipse((0, 0) + ring.size, fill=255)
+ draw.ellipse((px, px, 1024-px, 1024-px), fill=0)
+ ring.putalpha(mask)
+ return ring
+
+ @staticmethod
+ def pridify_effect(image: Image.Image, pixels: int, flag: str) -> Image.Image:
+ """Applies the given pride effect to the given image."""
+ image = image.resize((1024, 1024))
+ image = PfpEffects.crop_avatar_circle(image)
+
+ ring = Image.open(Path(f"bot/resources/pride/flags/{flag}.png")).resize((1024, 1024))
+ ring = ring.convert("RGBA")
+ ring = PfpEffects.crop_ring(ring, pixels)
+
+ image.alpha_composite(ring, (0, 0))
+ return image
+
+ @staticmethod
+ def eight_bitify_effect(image: Image.Image) -> Image.Image:
+ """
+ Applies the 8bit effect to the given image.
+
+ This is done by reducing the image to 32x32 and then back up to 1024x1024.
+ We then quantize the image before returning too.
+ """
+ image = image.resize((32, 32), resample=Image.NEAREST)
+ image = image.resize((1024, 1024), resample=Image.NEAREST)
+ return image.quantize()
+
+ @staticmethod
+ def easterify_effect(image: Image.Image, overlay_image: t.Optional[Image.Image] = None) -> Image.Image:
+ """
+ Applies the easter effect to the given image.
+
+ This is done by getting the closest "easter" colour to each pixel and changing the colour
+ to the half-way RGBvalue.
+
+ We also then add an overlay image on top in middle right, a chocolate bunny by default.
+ """
+ if overlay_image:
+ ratio = 64 / overlay_image.height
+ overlay_image = overlay_image.resize((
+ round(overlay_image.width * ratio),
+ round(overlay_image.height * ratio)
+ ))
+ overlay_image = overlay_image.convert("RGBA")
+ else:
+ overlay_image = Image.open(Path("bot/resources/easter/chocolate_bunny.png"))
+
+ alpha = image.getchannel("A").getdata()
+ image = image.convert("RGB")
+ image = ImageOps.posterize(image, 6)
+
+ data = image.getdata()
+ data_set = set(data)
+ easterified_data_set = {}
+
+ for x in data_set:
+ easterified_data_set[x] = PfpEffects.closest(x)
+ new_pixel_data = [
+ (*easterified_data_set[x], alpha[i])
+ if x in easterified_data_set else x
+ for i, x in enumerate(data)
+ ]
+
+ im = Image.new("RGBA", image.size)
+ im.putdata(new_pixel_data)
+ im.alpha_composite(
+ overlay_image,
+ (im.width - overlay_image.width, (im.height - overlay_image.height) // 2)
+ )
+ return im
+
+ @staticmethod
+ def split_image(img: Image.Image, squares: int) -> list:
+ """
+ Split an image into a selection of squares, specified by the squares argument.
+
+ Explanation:
+
+ 1. It gets the width and the height of the Image passed to the function.
+
+ 2. It gets the root of a number of squares (number of squares) passed, which is called xy. Reason: if let's say
+ 25 squares (number of squares) were passed, that is the total squares (split pieces) that the image is supposed
+ to be split into. As it is known, a 2D shape has a height and a width, and in this case the program thinks of it
+ as rows and columns. Rows multiplied by columns is equal to the passed squares (number of squares). To get rows
+ and columns, since in this case, each square (split piece) is identical, rows are equal to columns and the
+ program treats the image as a square-shaped, it gets the root out of the squares (number of squares) passed.
+
+ 3. Now width and height are both of the original Image, Discord PFP, so when it comes to forming the squares,
+ the program divides the original image's height and width by the xy. In a case of 25 squares (number of squares)
+ passed, xy would be 5, so if an image was 250x300, x_frac would be 50 and y_frac - 60. Note:
+ x_frac stands for a fracture of width. The reason it's called that is because it is shorter to use x for width
+ in mind and then it's just half of the word fracture, same applies to y_frac, just height instead of width.
+ x_frac and y_frac are width and height of a single square (split piece).
+
+ 4. With left, top, right, bottom, = 0, 0, x_frac, y_frac, the program sets these variables to create the initial
+ square (split piece). Explanation: all of these 4 variables start at the top left corner of the Image, by adding
+ value to right and bottom, it's creating the initial square (split piece).
+
+ 5. In the for loop, it keeps adding those squares (split pieces) in a row and once (index + 1) % xy == 0 is
+ True, it adds to top and bottom to lower them and reset right and left to recreate the initial space between
+ them, forming a square (split piece), it also adds the newly created square (split piece) into the new_imgs list
+ where it stores them. The program keeps repeating this process till all 25 squares get added to the list.
+
+ 6. It returns new_imgs, a list of squares (split pieces).
+ """
+ width, heigth = img.size
+
+ xy = math.sqrt(squares)
+
+ x_frac = width // xy
+ y_frac = heigth // xy
+
+ left, top, right, bottom, = 0, 0, x_frac, y_frac
+
+ new_imgs = []
+
+ for index in range(squares):
+ new_img = img.crop((left, top, right, bottom))
+ new_imgs.append(new_img)
+
+ if (index + 1) % xy == 0:
+ top += y_frac
+ bottom += y_frac
+ left = 0
+ right = x_frac
+ else:
+ left += x_frac
+ right += x_frac
+
+ return new_imgs
+
+ @staticmethod
+ def join_images(images: t.List[Image.Image]) -> Image.Image:
+ """
+ Stitches all the image squares into a new image.
+
+ Explanation:
+
+ 1. Shuffles the passed images to randomize the pieces.
+
+ 2. The program gets a single square (split piece) out of the list and defines single_width as the square's width
+ and single_height as the square's height.
+
+ 3. It gets the root of type integer of the number of images (split pieces) in the list and calls it multiplier.
+ Program then proceeds to calculate total height and width of the new image that it's creating using the same
+ multiplier.
+
+ 4. The program then defines new_image as the image that it's creating, using the previously obtained total_width
+ and total_height.
+
+ 5. Now it defines width_multiplier as well as height with values of 0. These will be used to correctly position
+ squares (split pieces) onto the new_image canvas.
+
+ 6. Similar to how in the split_image function, the program gets the root of number of images in the list.
+ In split_image function, it was the passed squares (number of squares) instead of a number of imgs in the
+ list that it got the square of here.
+
+ 7. In the for loop, as it iterates, the program multiplies single_width by width_multiplier to correctly
+ position a square (split piece) width wise. It then proceeds to paste the newly positioned square (split piece)
+ onto the new_image. The program increases the width_multiplier by 1 every iteration so the image wouldn't get
+ pasted in the same spot and the positioning would move accordingly. It makes sure to increase the
+ width_multiplier before the check, which checks if the end of a row has been reached, -
+ (index + 1) % pieces == 0, so after it, if it was True, width_multiplier would have been reset to 0 (start of
+ the row). If the check returns True, the height gets increased by a single square's (split piece) height to
+ lower the positioning height wise and, as mentioned, the width_multiplier gets reset to 0 and width will
+ then be calculated from the start of the new row. The for loop finishes once all the squares (split pieces) were
+ positioned accordingly.
+
+ 8. Finally, it returns the new_image, the randomized squares (split pieces) stitched back into the format of the
+ original image - user's PFP.
+ """
+ random.shuffle(images)
+ single_img = images[0]
+
+ single_wdith = single_img.size[0]
+ single_height = single_img.size[1]
+
+ multiplier = int(math.sqrt(len(images)))
+
+ total_width = multiplier * single_wdith
+ total_height = multiplier * single_height
+
+ new_image = Image.new('RGBA', (total_width, total_height), (250, 250, 250))
+
+ width_multiplier = 0
+ height = 0
+
+ squares = math.sqrt(len(images))
+
+ for index, image in enumerate(images):
+ width = single_wdith * width_multiplier
+
+ new_image.paste(image, (width, height))
+
+ width_multiplier += 1
+
+ if (index + 1) % squares == 0:
+ width_multiplier = 0
+ height += single_height
+
+ return new_image
+
+ @staticmethod
+ def mosaic_effect(img_bytes: bytes, squares: int, file_name: str) -> discord.File:
+ """Seperate function run from an executor which turns an image into a mosaic."""
+ avatar = Image.open(BytesIO(img_bytes))
+ avatar = avatar.convert('RGBA').resize((1024, 1024))
+
+ img_squares = PfpEffects.split_image(avatar, squares)
+ new_img = PfpEffects.join_images(img_squares)
+
+ bufferedio = BytesIO()
+ new_img.save(bufferedio, format='PNG')
+ bufferedio.seek(0)
+
+ return discord.File(bufferedio, filename=file_name)
diff --git a/bot/exts/evergreen/avatar_modification/avatar_modify.py b/bot/exts/evergreen/avatar_modification/avatar_modify.py
new file mode 100644
index 00000000..693d15c7
--- /dev/null
+++ b/bot/exts/evergreen/avatar_modification/avatar_modify.py
@@ -0,0 +1,370 @@
+import asyncio
+import json
+import logging
+import math
+import string
+import typing as t
+import unicodedata
+from concurrent.futures import ThreadPoolExecutor
+
+import discord
+from aiohttp import client_exceptions
+from discord.ext import commands
+from discord.ext.commands.errors import BadArgument
+
+from bot.constants import Colours, Emojis
+from bot.exts.evergreen.avatar_modification._effects import PfpEffects
+from bot.utils.extensions import invoke_help_command
+from bot.utils.halloween import spookifications
+
+log = logging.getLogger(__name__)
+
+_EXECUTOR = ThreadPoolExecutor(10)
+
+FILENAME_STRING = "{effect}_{author}.png"
+
+MAX_SQUARES = 10_000
+
+T = t.TypeVar("T")
+
+with open("bot/resources/pride/gender_options.json") as f:
+ GENDER_OPTIONS = json.load(f)
+
+
+async def in_executor(func: t.Callable[..., T], *args) -> T:
+ """
+ Runs the given synchronus function `func` in an executor.
+
+ This is useful for running slow, blocking code within async
+ functions, so that they don't block the bot.
+ """
+ log.trace(f"Running {func.__name__} in an executor.")
+ loop = asyncio.get_event_loop()
+ return await loop.run_in_executor(_EXECUTOR, func, *args)
+
+
+def file_safe_name(effect: str, display_name: str) -> str:
+ """Returns a file safe filename based on the given effect and display name."""
+ valid_filename_chars = f"-_. {string.ascii_letters}{string.digits}"
+
+ file_name = FILENAME_STRING.format(effect=effect, author=display_name)
+
+ # Replace spaces
+ file_name = file_name.replace(" ", "_")
+
+ # Normalize unicode characters
+ cleaned_filename = unicodedata.normalize("NFKD", file_name).encode("ASCII", "ignore").decode()
+
+ # Remove invalid filename characters
+ cleaned_filename = "".join(c for c in cleaned_filename if c in valid_filename_chars)
+ return cleaned_filename
+
+
+class AvatarModify(commands.Cog):
+ """Various commands for users to apply affects to their own avatars."""
+
+ def __init__(self, bot: commands.Bot) -> None:
+ self.bot = bot
+
+ async def _fetch_user(self, user_id: int) -> t.Optional[discord.User]:
+ """
+ Fetches a user and handles errors.
+
+ This helper function is required as the member cache doesn't always have the most up to date
+ profile picture. This can lead to errors if the image is delted from the Discord CDN.
+ fetch_member can't be used due to the avatar url being part of the user object, and
+ some weird caching that D.py does
+ """
+ try:
+ user = await self.bot.fetch_user(user_id)
+ except discord.errors.NotFound:
+ log.debug(f"User {user_id} could not be found.")
+ return None
+ except discord.HTTPException:
+ log.exception(f"Exception while trying to retrieve user {user_id} from Discord.")
+ return None
+
+ return user
+
+ @commands.group(aliases=("avatar_mod", "pfp_mod", "avatarmod", "pfpmod"))
+ async def avatar_modify(self, ctx: commands.Context) -> None:
+ """Groups all of the pfp modifying commands to allow a single concurrency limit."""
+ if not ctx.invoked_subcommand:
+ await invoke_help_command(ctx)
+
+ @avatar_modify.command(name="8bitify", root_aliases=("8bitify",))
+ async def eightbit_command(self, ctx: commands.Context) -> None:
+ """Pixelates your avatar and changes the palette to an 8bit one."""
+ async with ctx.typing():
+ user = await self._fetch_user(ctx.author.id)
+ if not user:
+ await ctx.send(f"{Emojis.cross_mark} Could not get user info.")
+ return
+
+ image_bytes = await user.avatar_url_as(size=1024).read()
+ file_name = file_safe_name("eightbit_avatar", ctx.author.display_name)
+
+ file = await in_executor(
+ PfpEffects.apply_effect,
+ image_bytes,
+ PfpEffects.eight_bitify_effect,
+ file_name
+ )
+
+ embed = discord.Embed(
+ title="Your 8-bit avatar",
+ description="Here is your avatar. I think it looks all cool and 'retro'."
+ )
+
+ embed.set_image(url=f"attachment://{file_name}")
+ embed.set_footer(text=f"Made by {ctx.author.display_name}.", icon_url=user.avatar_url)
+
+ await ctx.send(embed=embed, file=file)
+
+ @avatar_modify.command(aliases=("easterify",), root_aliases=("easterify", "avatareasterify"))
+ async def avatareasterify(self, ctx: commands.Context, *colours: t.Union[discord.Colour, str]) -> None:
+ """
+ This "Easterifies" the user's avatar.
+
+ Given colours will produce a personalised egg in the corner, similar to the egg_decorate command.
+ If colours are not given, a nice little chocolate bunny will sit in the corner.
+ Colours are split by spaces, unless you wrap the colour name in double quotes.
+ Discord colour names, HTML colour names, XKCD colour names and hex values are accepted.
+ """
+ async def send(*args, **kwargs) -> str:
+ """
+ This replaces the original ctx.send.
+
+ When invoking the egg decorating command, the egg itself doesn't print to to the channel.
+ Returns the message content so that if any errors occur, the error message can be output.
+ """
+ if args:
+ return args[0]
+
+ async with ctx.typing():
+ user = await self._fetch_user(ctx.author.id)
+ if not user:
+ await ctx.send(f"{Emojis.cross_mark} Could not get user info.")
+ return
+
+ egg = None
+ if colours:
+ send_message = ctx.send
+ ctx.send = send # Assigns ctx.send to a fake send
+ egg = await ctx.invoke(self.bot.get_command("eggdecorate"), *colours)
+ if isinstance(egg, str): # When an error message occurs in eggdecorate.
+ await send_message(egg)
+ return
+ ctx.send = send_message # Reassigns ctx.send
+
+ image_bytes = await user.avatar_url_as(size=256).read()
+ file_name = file_safe_name("easterified_avatar", ctx.author.display_name)
+
+ file = await in_executor(
+ PfpEffects.apply_effect,
+ image_bytes,
+ PfpEffects.easterify_effect,
+ file_name,
+ egg
+ )
+
+ embed = discord.Embed(
+ name="Your Lovely Easterified Avatar!",
+ description="Here is your lovely avatar, all bright and colourful\nwith Easter pastel colours. Enjoy :D"
+ )
+ embed.set_image(url=f"attachment://{file_name}")
+ embed.set_footer(text=f"Made by {ctx.author.display_name}.", icon_url=user.avatar_url)
+
+ await ctx.send(file=file, embed=embed)
+
+ @staticmethod
+ async def send_pride_image(
+ ctx: commands.Context,
+ image_bytes: bytes,
+ pixels: int,
+ flag: str,
+ option: str
+ ) -> None:
+ """Gets and sends the image in an embed. Used by the pride commands."""
+ async with ctx.typing():
+ file_name = file_safe_name("pride_avatar", ctx.author.display_name)
+
+ file = await in_executor(
+ PfpEffects.apply_effect,
+ image_bytes,
+ PfpEffects.pridify_effect,
+ file_name,
+ pixels,
+ flag
+ )
+
+ embed = discord.Embed(
+ name="Your Lovely Pride Avatar!",
+ description=f"Here is your lovely avatar, surrounded by\n a beautiful {option} flag. Enjoy :D"
+ )
+ embed.set_image(url=f"attachment://{file_name}")
+ embed.set_footer(text=f"Made by {ctx.author.display_name}.", icon_url=ctx.author.avatar_url)
+ await ctx.send(file=file, embed=embed)
+
+ @avatar_modify.group(
+ aliases=("avatarpride", "pridepfp", "prideprofile"),
+ root_aliases=("prideavatar", "avatarpride", "pridepfp", "prideprofile"),
+ invoke_without_command=True
+ )
+ async def prideavatar(self, ctx: commands.Context, option: str = "lgbt", pixels: int = 64) -> None:
+ """
+ This surrounds an avatar with a border of a specified LGBT flag.
+
+ This defaults to the LGBT rainbow flag if none is given.
+ The amount of pixels can be given which determines the thickness of the flag border.
+ This has a maximum of 512px and defaults to a 64px border.
+ The full image is 1024x1024.
+ """
+ option = option.lower()
+ pixels = max(0, min(512, pixels))
+ flag = GENDER_OPTIONS.get(option)
+ if flag is None:
+ await ctx.send("I don't have that flag!")
+ return
+
+ async with ctx.typing():
+ user = await self._fetch_user(ctx.author.id)
+ if not user:
+ await ctx.send(f"{Emojis.cross_mark} Could not get user info.")
+ return
+ image_bytes = await user.avatar_url_as(size=1024).read()
+ await self.send_pride_image(ctx, image_bytes, pixels, flag, option)
+
+ @prideavatar.command()
+ async def image(self, ctx: commands.Context, url: str, option: str = "lgbt", pixels: int = 64) -> None:
+ """
+ This surrounds the image specified by the URL with a border of a specified LGBT flag.
+
+ This defaults to the LGBT rainbow flag if none is given.
+ The amount of pixels can be given which determines the thickness of the flag border.
+ This has a maximum of 512px and defaults to a 64px border.
+ The full image is 1024x1024.
+ """
+ option = option.lower()
+ pixels = max(0, min(512, pixels))
+ flag = GENDER_OPTIONS.get(option)
+ if flag is None:
+ await ctx.send("I don't have that flag!")
+ return
+
+ async with ctx.typing():
+ try:
+ async with self.bot.http_session.get(url) as response:
+ if response.status != 200:
+ await ctx.send("Bad response from provided URL!")
+ return
+ image_bytes = await response.read()
+ except client_exceptions.ClientConnectorError:
+ raise BadArgument("Cannot connect to provided URL!")
+ except client_exceptions.InvalidURL:
+ raise BadArgument("Invalid URL!")
+
+ await self.send_pride_image(ctx, image_bytes, pixels, flag, option)
+
+ @prideavatar.command()
+ async def flags(self, ctx: commands.Context) -> None:
+ """This lists the flags that can be used with the prideavatar command."""
+ choices = sorted(set(GENDER_OPTIONS.values()))
+ options = "• " + "\n• ".join(choices)
+ embed = discord.Embed(
+ title="I have the following flags:",
+ description=options,
+ colour=Colours.soft_red
+ )
+ await ctx.send(embed=embed)
+
+ @avatar_modify.command(
+ aliases=("savatar", "spookify"),
+ root_aliases=("spookyavatar", "spookify", "savatar"),
+ brief="Spookify an user's avatar."
+ )
+ async def spookyavatar(self, ctx: commands.Context, member: discord.Member = None) -> None:
+ """This "spookifies" the given user's avatar, with a random *spooky* effect."""
+ if member is None:
+ member = ctx.author
+
+ user = await self._fetch_user(member.id)
+ if not user:
+ await ctx.send(f"{Emojis.cross_mark} Could not get user info.")
+ return
+
+ async with ctx.typing():
+ image_bytes = await user.avatar_url_as(size=1024).read()
+
+ file_name = file_safe_name("spooky_avatar", member.display_name)
+
+ file = await in_executor(
+ PfpEffects.apply_effect,
+ image_bytes,
+ spookifications.get_random_effect,
+ file_name
+ )
+
+ embed = discord.Embed(
+ title="Is this you or am I just really paranoid?",
+ colour=Colours.soft_red
+ )
+ embed.set_author(name=member.name, icon_url=member.avatar_url)
+ embed.set_image(url=f"attachment://{file_name}")
+ embed.set_footer(text=f"Made by {ctx.author.display_name}.", icon_url=ctx.author.avatar_url)
+
+ await ctx.send(file=file, embed=embed)
+
+ @avatar_modify.command(name="mosaic", root_aliases=("mosaic",))
+ async def mosaic_command(self, ctx: commands.Context, squares: int = 16) -> None:
+ """Splits your avatar into x squares, randomizes them and stitches them back into a new image!"""
+ async with ctx.typing():
+ user = await self._fetch_user(ctx.author.id)
+ if not user:
+ await ctx.send(f"{Emojis.cross_mark} Could not get user info.")
+ return
+
+ if not 1 <= squares <= MAX_SQUARES:
+ raise commands.BadArgument(f"Squares must be a positive number less than or equal to {MAX_SQUARES:,}.")
+
+ sqrt = math.sqrt(squares)
+
+ if not sqrt.is_integer():
+ squares = math.ceil(sqrt) ** 2 # Get the next perfect square
+
+ file_name = file_safe_name("mosaic_avatar", ctx.author.display_name)
+
+ img_bytes = await user.avatar_url_as(size=1024).read()
+
+ file = await in_executor(
+ PfpEffects.mosaic_effect,
+ img_bytes,
+ squares,
+ file_name
+ )
+
+ if squares == 1:
+ title = "Hooh... that was a lot of work"
+ description = "I present to you... Yourself!"
+ elif squares == MAX_SQUARES:
+ title = "Testing the limits I see..."
+ description = "What a masterpiece. :star:"
+ else:
+ title = "Your mosaic avatar"
+ description = f"Here is your avatar. I think it looks a bit *puzzling*\nMade with {squares} squares."
+
+ embed = discord.Embed(
+ title=title,
+ description=description,
+ colour=Colours.blue
+ )
+
+ embed.set_image(url=f"attachment://{file_name}")
+ embed.set_footer(text=f"Made by {ctx.author.display_name}", icon_url=user.avatar_url)
+
+ await ctx.send(file=file, embed=embed)
+
+
+def setup(bot: commands.Bot) -> None:
+ """Load the AvatarModify cog."""
+ bot.add_cog(AvatarModify(bot))
diff --git a/bot/exts/evergreen/battleship.py b/bot/exts/evergreen/battleship.py
index fa3fb35c..1681434f 100644
--- a/bot/exts/evergreen/battleship.py
+++ b/bot/exts/evergreen/battleship.py
@@ -227,7 +227,7 @@ class Game:
if message.content.lower() == "surrender":
self.surrender = True
return True
- self.match = re.match("([A-J]|[a-j]) ?((10)|[1-9])", message.content.strip())
+ self.match = re.fullmatch("([A-J]|[a-j]) ?((10)|[1-9])", message.content.strip())
if not self.match:
self.bot.loop.create_task(message.add_reaction(CROSS_EMOJI))
return bool(self.match)
diff --git a/bot/exts/evergreen/catify.py b/bot/exts/evergreen/catify.py
new file mode 100644
index 00000000..a175602f
--- /dev/null
+++ b/bot/exts/evergreen/catify.py
@@ -0,0 +1,88 @@
+import random
+from contextlib import suppress
+from typing import Optional
+
+from discord import AllowedMentions, Embed, Forbidden
+from discord.ext import commands
+
+from bot.constants import Cats, Colours, NEGATIVE_REPLIES
+from bot.utils import helpers
+
+
+class Catify(commands.Cog):
+ """Cog for the catify command."""
+
+ def __init__(self, bot: commands.Bot):
+ self.bot = bot
+
+ @commands.command(aliases=["ᓚᘏᗢify", "ᓚᘏᗢ"])
+ @commands.cooldown(1, 5, commands.BucketType.user)
+ async def catify(self, ctx: commands.Context, *, text: Optional[str]) -> None:
+ """
+ Convert the provided text into a cat themed sentence by interspercing cats throughout text.
+
+ If no text is given then the users nickname is edited.
+ """
+ if not text:
+ display_name = ctx.author.display_name
+
+ if len(display_name) > 26:
+ embed = Embed(
+ title=random.choice(NEGATIVE_REPLIES),
+ description=(
+ "Your display name is too long to be catified! "
+ "Please change it to be under 26 characters."
+ ),
+ color=Colours.soft_red
+ )
+ await ctx.send(embed=embed)
+ return
+
+ else:
+ display_name += f" | {random.choice(Cats.cats)}"
+
+ await ctx.send(f"Your catified nickname is: `{display_name}`", allowed_mentions=AllowedMentions.none())
+
+ with suppress(Forbidden):
+ await ctx.author.edit(nick=display_name)
+ else:
+ if len(text) >= 1500:
+ embed = Embed(
+ title=random.choice(NEGATIVE_REPLIES),
+ description="Submitted text was too large! Please submit something under 1500 characters.",
+ color=Colours.soft_red
+ )
+ await ctx.send(embed=embed)
+ return
+
+ string_list = text.split()
+ for index, name in enumerate(string_list):
+ name = name.lower()
+ if "cat" in name:
+ if random.randint(0, 5) == 5:
+ string_list[index] = name.replace("cat", f"**{random.choice(Cats.cats)}**")
+ else:
+ string_list[index] = name.replace("cat", random.choice(Cats.cats))
+ for element in Cats.cats:
+ if element in name:
+ string_list[index] = name.replace(element, "cat")
+
+ string_len = len(string_list) // 3 or len(string_list)
+
+ for _ in range(random.randint(1, string_len)):
+ # insert cat at random index
+ if random.randint(0, 5) == 5:
+ string_list.insert(random.randint(0, len(string_list)), f"**{random.choice(Cats.cats)}**")
+ else:
+ string_list.insert(random.randint(0, len(string_list)), random.choice(Cats.cats))
+
+ text = helpers.suppress_links(" ".join(string_list))
+ await ctx.send(
+ f">>> {text}",
+ allowed_mentions=AllowedMentions.none()
+ )
+
+
+def setup(bot: commands.Bot) -> None:
+ """Loads the catify cog."""
+ bot.add_cog(Catify(bot))
diff --git a/bot/exts/evergreen/error_handler.py b/bot/exts/evergreen/error_handler.py
index 28902503..8db49748 100644
--- a/bot/exts/evergreen/error_handler.py
+++ b/bot/exts/evergreen/error_handler.py
@@ -46,6 +46,11 @@ class CommandErrorHandler(commands.Cog):
logging.debug(f"Command {ctx.command} had its error already handled locally; ignoring.")
return
+ parent_command = ""
+ if subctx := getattr(ctx, "subcontext", None):
+ parent_command = f"{ctx.command} "
+ ctx = subctx
+
error = getattr(error, 'original', error)
logging.debug(
f"Error Encountered: {type(error).__name__} - {str(error)}, "
@@ -63,8 +68,9 @@ class CommandErrorHandler(commands.Cog):
if isinstance(error, commands.UserInputError):
self.revert_cooldown_counter(ctx.command, ctx.message)
+ usage = f"```{ctx.prefix}{parent_command}{ctx.command} {ctx.command.signature}```"
embed = self.error_embed(
- f"Your input was invalid: {error}\n\nUsage:\n```{ctx.prefix}{ctx.command} {ctx.command.signature}```"
+ f"Your input was invalid: {error}\n\nUsage:{usage}"
)
await ctx.send(embed=embed)
return
@@ -95,7 +101,7 @@ class CommandErrorHandler(commands.Cog):
self.revert_cooldown_counter(ctx.command, ctx.message)
embed = self.error_embed(
"The argument you provided was invalid: "
- f"{error}\n\nUsage:\n```{ctx.prefix}{ctx.command} {ctx.command.signature}```"
+ f"{error}\n\nUsage:\n```{ctx.prefix}{parent_command}{ctx.command} {ctx.command.signature}```"
)
await ctx.send(embed=embed)
return
diff --git a/bot/exts/evergreen/fun.py b/bot/exts/evergreen/fun.py
index 101725da..7152d0cb 100644
--- a/bot/exts/evergreen/fun.py
+++ b/bot/exts/evergreen/fun.py
@@ -11,6 +11,7 @@ from discord.ext.commands import BadArgument, Bot, Cog, Context, MessageConverte
from bot import utils
from bot.constants import Client, Colours, Emojis
+from bot.utils import helpers
log = logging.getLogger(__name__)
@@ -83,6 +84,7 @@ class Fun(Cog):
if embed is not None:
embed = Fun._convert_embed(conversion_func, embed)
converted_text = conversion_func(text)
+ converted_text = helpers.suppress_links(converted_text)
# Don't put >>> if only embed present
if converted_text:
converted_text = f">>> {converted_text.lstrip('> ')}"
@@ -101,6 +103,7 @@ class Fun(Cog):
if embed is not None:
embed = Fun._convert_embed(conversion_func, embed)
converted_text = conversion_func(text)
+ converted_text = helpers.suppress_links(converted_text)
# Don't put >>> if only embed present
if converted_text:
converted_text = f">>> {converted_text.lstrip('> ')}"
diff --git a/bot/exts/evergreen/githubinfo.py b/bot/exts/evergreen/githubinfo.py
index 2e38e3ab..c8a6b3f7 100644
--- a/bot/exts/evergreen/githubinfo.py
+++ b/bot/exts/evergreen/githubinfo.py
@@ -1,16 +1,19 @@
import logging
import random
from datetime import datetime
-from typing import Optional
+from urllib.parse import quote
import discord
from discord.ext import commands
from discord.ext.commands.cooldowns import BucketType
-from bot.constants import NEGATIVE_REPLIES
+from bot.constants import Colours, NEGATIVE_REPLIES
+from bot.exts.utils.extensions import invoke_help_command
log = logging.getLogger(__name__)
+GITHUB_API_URL = "https://api.github.com"
+
class GithubInfo(commands.Cog):
"""Fetches info from GitHub."""
@@ -23,27 +26,28 @@ class GithubInfo(commands.Cog):
async with self.bot.http_session.get(url) as r:
return await r.json()
- @commands.command(name='github', aliases=['gh'])
- @commands.cooldown(1, 60, BucketType.user)
- async def get_github_info(self, ctx: commands.Context, username: Optional[str]) -> None:
- """
- Fetches a user's GitHub information.
-
- Username is optional and sends the help command if not specified.
- """
- if username is None:
- await ctx.invoke(self.bot.get_command('help'), 'github')
- ctx.command.reset_cooldown(ctx)
- return
+ @commands.group(name='github', aliases=('gh', 'git'))
+ @commands.cooldown(1, 10, BucketType.user)
+ async def github_group(self, ctx: commands.Context) -> None:
+ """Commands for finding information related to GitHub."""
+ if ctx.invoked_subcommand is None:
+ await invoke_help_command(ctx)
+ @github_group.command(name='user', aliases=('userinfo',))
+ async def github_user_info(self, ctx: commands.Context, username: str) -> None:
+ """Fetches a user's GitHub information."""
async with ctx.typing():
- user_data = await self.fetch_data(f"https://api.github.com/users/{username}")
+ user_data = await self.fetch_data(f"{GITHUB_API_URL}/users/{username}")
# User_data will not have a message key if the user exists
- if user_data.get('message') is not None:
- await ctx.send(embed=discord.Embed(title=random.choice(NEGATIVE_REPLIES),
- description=f"The profile for `{username}` was not found.",
- colour=discord.Colour.red()))
+ if "message" in user_data:
+ embed = discord.Embed(
+ title=random.choice(NEGATIVE_REPLIES),
+ description=f"The profile for `{username}` was not found.",
+ colour=Colours.soft_red
+ )
+
+ await ctx.send(embed=embed)
return
org_data = await self.fetch_data(user_data['organizations_url'])
@@ -63,7 +67,7 @@ class GithubInfo(commands.Cog):
embed = discord.Embed(
title=f"`{user_data['login']}`'s GitHub profile info",
description=f"```{user_data['bio']}```\n" if user_data['bio'] is not None else "",
- colour=0x7289da,
+ colour=discord.Colour.blurple(),
url=user_data['html_url'],
timestamp=datetime.strptime(user_data['created_at'], "%Y-%m-%dT%H:%M:%SZ")
)
@@ -72,26 +76,99 @@ class GithubInfo(commands.Cog):
if user_data['type'] == "User":
- embed.add_field(name="Followers",
- value=f"[{user_data['followers']}]({user_data['html_url']}?tab=followers)")
- embed.add_field(name="\u200b", value="\u200b")
- embed.add_field(name="Following",
- value=f"[{user_data['following']}]({user_data['html_url']}?tab=following)")
-
- embed.add_field(name="Public repos",
- value=f"[{user_data['public_repos']}]({user_data['html_url']}?tab=repositories)")
- embed.add_field(name="\u200b", value="\u200b")
+ embed.add_field(
+ name="Followers",
+ value=f"[{user_data['followers']}]({user_data['html_url']}?tab=followers)"
+ )
+ embed.add_field(
+ name="Following",
+ value=f"[{user_data['following']}]({user_data['html_url']}?tab=following)"
+ )
+
+ embed.add_field(
+ name="Public repos",
+ value=f"[{user_data['public_repos']}]({user_data['html_url']}?tab=repositories)"
+ )
if user_data['type'] == "User":
- embed.add_field(name="Gists", value=f"[{gists}](https://gist.github.com/{username})")
+ embed.add_field(name="Gists", value=f"[{gists}](https://gist.github.com/{quote(username, safe='')})")
- embed.add_field(name=f"Organization{'s' if len(orgs)!=1 else ''}",
- value=orgs_to_add if orgs else "No organizations")
- embed.add_field(name="\u200b", value="\u200b")
+ embed.add_field(
+ name=f"Organization{'s' if len(orgs)!=1 else ''}",
+ value=orgs_to_add if orgs else "No organizations"
+ )
embed.add_field(name="Website", value=blog)
await ctx.send(embed=embed)
+ @github_group.command(name='repository', aliases=('repo',))
+ async def github_repo_info(self, ctx: commands.Context, *repo: str) -> None:
+ """
+ Fetches a repositories' GitHub information.
+
+ The repository should look like `user/reponame` or `user reponame`.
+ """
+ repo = '/'.join(repo)
+ if repo.count('/') != 1:
+ embed = discord.Embed(
+ title=random.choice(NEGATIVE_REPLIES),
+ description="The repository should look like `user/reponame` or `user reponame`.",
+ colour=Colours.soft_red
+ )
+
+ await ctx.send(embed=embed)
+ return
+
+ async with ctx.typing():
+ repo_data = await self.fetch_data(f"{GITHUB_API_URL}/repos/{quote(repo)}")
+
+ # There won't be a message key if this repo exists
+ if "message" in repo_data:
+ embed = discord.Embed(
+ title=random.choice(NEGATIVE_REPLIES),
+ description="The requested repository was not found.",
+ colour=Colours.soft_red
+ )
+
+ await ctx.send(embed=embed)
+ return
+
+ embed = discord.Embed(
+ title=repo_data['name'],
+ description=repo_data["description"],
+ colour=discord.Colour.blurple(),
+ url=repo_data['html_url']
+ )
+
+ # If it's a fork, then it will have a parent key
+ try:
+ parent = repo_data["parent"]
+ embed.description += f"\n\nForked from [{parent['full_name']}]({parent['html_url']})"
+ except KeyError:
+ log.debug("Repository is not a fork.")
+
+ repo_owner = repo_data['owner']
+
+ embed.set_author(
+ name=repo_owner["login"],
+ url=repo_owner["html_url"],
+ icon_url=repo_owner["avatar_url"]
+ )
+
+ repo_created_at = datetime.strptime(repo_data['created_at'], "%Y-%m-%dT%H:%M:%SZ").strftime("%d/%m/%Y")
+ last_pushed = datetime.strptime(repo_data['pushed_at'], "%Y-%m-%dT%H:%M:%SZ").strftime("%d/%m/%Y at %H:%M")
+
+ embed.set_footer(
+ text=(
+ f"{repo_data['forks_count']} ⑂ "
+ f"• {repo_data['stargazers_count']} ⭐ "
+ f"• Created At {repo_created_at} "
+ f"• Last Commit {last_pushed}"
+ )
+ )
+
+ await ctx.send(embed=embed)
+
def setup(bot: commands.Bot) -> None:
"""Adding the cog to the bot."""
diff --git a/bot/exts/evergreen/help.py b/bot/exts/evergreen/help.py
index 91147243..f557e42e 100644
--- a/bot/exts/evergreen/help.py
+++ b/bot/exts/evergreen/help.py
@@ -289,7 +289,9 @@ class HelpSession:
parent = self.query.full_parent_name + ' ' if self.query.parent else ''
paginator.add_line(f'**```{prefix}{parent}{signature}```**')
- aliases = ', '.join(f'`{a}`' for a in self.query.aliases)
+ aliases = [f"`{alias}`" if not parent else f"`{parent} {alias}`" for alias in self.query.aliases]
+ aliases += [f"`{alias}`" for alias in getattr(self.query, "root_aliases", ())]
+ aliases = ", ".join(sorted(aliases))
if aliases:
paginator.add_line(f'**Can also use:** {aliases}\n')
diff --git a/bot/exts/evergreen/issues.py b/bot/exts/evergreen/issues.py
index bbcbf611..a0316080 100644
--- a/bot/exts/evergreen/issues.py
+++ b/bot/exts/evergreen/issues.py
@@ -2,12 +2,23 @@ import logging
import random
import re
import typing as t
-from enum import Enum
+from dataclasses import dataclass
import discord
-from discord.ext import commands, tasks
-
-from bot.constants import Categories, Channels, Colours, ERROR_REPLIES, Emojis, Tokens, WHITELISTED_CHANNELS
+from discord.ext import commands
+
+from bot.constants import (
+ Categories,
+ Channels,
+ Colours,
+ ERROR_REPLIES,
+ Emojis,
+ NEGATIVE_REPLIES,
+ Tokens,
+ WHITELISTED_CHANNELS
+)
+from bot.utils.decorators import whitelist_override
+from bot.utils.extensions import invoke_help_command
log = logging.getLogger(__name__)
@@ -15,20 +26,20 @@ BAD_RESPONSE = {
404: "Issue/pull request not located! Please enter a valid number!",
403: "Rate limit has been hit! Please try again later!"
}
+REQUEST_HEADERS = {
+ "Accept": "application/vnd.github.v3+json"
+}
-MAX_REQUESTS = 10
-REQUEST_HEADERS = dict()
+REPOSITORY_ENDPOINT = "https://api.github.com/orgs/{org}/repos?per_page=100&type=public"
+ISSUE_ENDPOINT = "https://api.github.com/repos/{user}/{repository}/issues/{number}"
+PR_ENDPOINT = "https://api.github.com/repos/{user}/{repository}/pulls/{number}"
-REPOS_API = "https://api.github.com/orgs/{org}/repos"
if GITHUB_TOKEN := Tokens.github:
REQUEST_HEADERS["Authorization"] = f"token {GITHUB_TOKEN}"
WHITELISTED_CATEGORIES = (
Categories.development, Categories.devprojects, Categories.media, Categories.staff
)
-WHITELISTED_CHANNELS_ON_MESSAGE = (
- Channels.organisation, Channels.mod_meta, Channels.mod_tools, Channels.staff_voice
-)
CODE_BLOCK_RE = re.compile(
r"^`([^`\n]+)`" # Inline codeblock
@@ -36,12 +47,45 @@ CODE_BLOCK_RE = re.compile(
re.DOTALL | re.MULTILINE
)
+# Maximum number of issues in one message
+MAXIMUM_ISSUES = 5
+
+# Regex used when looking for automatic linking in messages
+# regex101 of current regex https://regex101.com/r/V2ji8M/6
+AUTOMATIC_REGEX = re.compile(
+ r"((?P<org>[a-zA-Z0-9][a-zA-Z0-9\-]{1,39})\/)?(?P<repo>[\w\-\.]{1,100})#(?P<number>[0-9]+)"
+)
+
+
+@dataclass
+class FoundIssue:
+ """Dataclass representing an issue found by the regex."""
+
+ organisation: t.Optional[str]
+ repository: str
+ number: str
+
+ def __hash__(self) -> int:
+ return hash((self.organisation, self.repository, self.number))
+
+
+@dataclass
+class FetchError:
+ """Dataclass representing an error while fetching an issue."""
+
+ return_code: int
+ message: str
-class FetchIssueErrors(Enum):
- """Errors returned in fetch issues."""
- value_error = "Numbers not found."
- max_requests = "Max requests hit."
+@dataclass
+class IssueState:
+ """Dataclass representing the state of an issue."""
+
+ repository: str
+ number: int
+ url: str
+ title: str
+ emoji: str
class Issues(commands.Cog):
@@ -50,97 +94,96 @@ class Issues(commands.Cog):
def __init__(self, bot: commands.Bot):
self.bot = bot
self.repos = []
- self.get_pydis_repos.start()
-
- @tasks.loop(minutes=30)
- async def get_pydis_repos(self) -> None:
- """Get all python-discord repositories on github."""
- async with self.bot.http_session.get(REPOS_API.format(org="python-discord")) as resp:
- if resp.status == 200:
- data = await resp.json()
- for repo in data:
- self.repos.append(repo["full_name"].split("/")[1])
- self.repo_regex = "|".join(self.repos)
- else:
- log.debug(f"Failed to get latest Pydis repositories. Status code {resp.status}")
@staticmethod
- def check_in_block(message: discord.Message, repo_issue: str) -> bool:
- """Check whether the <repo>#<issue> is in codeblocks."""
- block = re.findall(CODE_BLOCK_RE, message.content)
-
- if not block:
- return False
- elif "#".join(repo_issue.split(" ")) in "".join([*block[0]]):
- return True
- return False
+ def remove_codeblocks(message: str) -> str:
+ """Remove any codeblock in a message."""
+ return re.sub(CODE_BLOCK_RE, "", message)
async def fetch_issues(
self,
- numbers: set,
+ number: int,
repository: str,
user: str
- ) -> t.Union[FetchIssueErrors, str, list]:
- """Retrieve issue(s) from a GitHub repository."""
- links = []
- if not numbers:
- return FetchIssueErrors.value_error
-
- if len(numbers) > MAX_REQUESTS:
- return FetchIssueErrors.max_requests
-
- for number in numbers:
- url = f"https://api.github.com/repos/{user}/{repository}/issues/{number}"
- merge_url = f"https://api.github.com/repos/{user}/{repository}/pulls/{number}/merge"
- log.trace(f"Querying GH issues API: {url}")
- async with self.bot.http_session.get(url, headers=REQUEST_HEADERS) as r:
- json_data = await r.json()
-
- if r.status in BAD_RESPONSE:
- log.warning(f"Received response {r.status} from: {url}")
- return f"[{str(r.status)}] #{number} {BAD_RESPONSE.get(r.status)}"
-
- # The initial API request is made to the issues API endpoint, which will return information
- # if the issue or PR is present. However, the scope of information returned for PRs differs
- # from issues: if the 'issues' key is present in the response then we can pull the data we
- # need from the initial API call.
- if "issues" in json_data.get("html_url"):
- if json_data.get("state") == "open":
- icon_url = Emojis.issue
- else:
- icon_url = Emojis.issue_closed
-
- # If the 'issues' key is not contained in the API response and there is no error code, then
- # we know that a PR has been requested and a call to the pulls API endpoint is necessary
- # to get the desired information for the PR.
+ ) -> t.Union[IssueState, FetchError]:
+ """
+ Retrieve an issue from a GitHub repository.
+
+ Returns IssueState on success, FetchError on failure.
+ """
+ url = ISSUE_ENDPOINT.format(user=user, repository=repository, number=number)
+ pulls_url = PR_ENDPOINT.format(user=user, repository=repository, number=number)
+ log.trace(f"Querying GH issues API: {url}")
+
+ async with self.bot.http_session.get(url, headers=REQUEST_HEADERS) as r:
+ json_data = await r.json()
+
+ if r.status == 403:
+ if r.headers.get("X-RateLimit-Remaining") == "0":
+ log.info(f"Ratelimit reached while fetching {url}")
+ return FetchError(403, "Ratelimit reached, please retry in a few minutes.")
+ return FetchError(403, "Cannot access issue.")
+ elif r.status in (404, 410):
+ return FetchError(r.status, "Issue not found.")
+ elif r.status != 200:
+ return FetchError(r.status, "Error while fetching issue.")
+
+ # The initial API request is made to the issues API endpoint, which will return information
+ # if the issue or PR is present. However, the scope of information returned for PRs differs
+ # from issues: if the 'issues' key is present in the response then we can pull the data we
+ # need from the initial API call.
+ if "issues" in json_data["html_url"]:
+ if json_data.get("state") == "open":
+ emoji = Emojis.issue
else:
- log.trace(f"PR provided, querying GH pulls API for additional information: {merge_url}")
- async with self.bot.http_session.get(merge_url) as m:
- if json_data.get("state") == "open":
- icon_url = Emojis.pull_request
- # When the status is 204 this means that the state of the PR is merged
- elif m.status == 204:
- icon_url = Emojis.merge
- else:
- icon_url = Emojis.pull_request_closed
+ emoji = Emojis.issue_closed
+
+ # If the 'issues' key is not contained in the API response and there is no error code, then
+ # we know that a PR has been requested and a call to the pulls API endpoint is necessary
+ # to get the desired information for the PR.
+ else:
+ log.trace(f"PR provided, querying GH pulls API for additional information: {pulls_url}")
+ async with self.bot.http_session.get(pulls_url) as p:
+ pull_data = await p.json()
+ if pull_data["draft"]:
+ emoji = Emojis.pull_request_draft
+ elif pull_data["state"] == "open":
+ emoji = Emojis.pull_request
+ # When 'merged_at' is not None, this means that the state of the PR is merged
+ elif pull_data["merged_at"] is not None:
+ emoji = Emojis.merge
+ else:
+ emoji = Emojis.pull_request_closed
- issue_url = json_data.get("html_url")
- links.append([icon_url, f"[{repository}] #{number} {json_data.get('title')}", issue_url])
+ issue_url = json_data.get("html_url")
- return links
+ return IssueState(repository, number, issue_url, json_data.get('title', ''), emoji)
@staticmethod
- def get_embed(result: list, user: str = "python-discord", repository: str = "") -> discord.Embed:
- """Get Response Embed."""
- description_list = ["{0} [{1}]({2})".format(*link) for link in result]
+ def format_embed(
+ results: t.List[t.Union[IssueState, FetchError]],
+ user: str,
+ repository: t.Optional[str] = None
+ ) -> discord.Embed:
+ """Take a list of IssueState or FetchError and format a Discord embed for them."""
+ description_list = []
+
+ for result in results:
+ if isinstance(result, IssueState):
+ description_list.append(f"{result.emoji} [{result.title}]({result.url})")
+ elif isinstance(result, FetchError):
+ description_list.append(f":x: [{result.return_code}] {result.message}")
+
resp = discord.Embed(
colour=Colours.bright_green,
description='\n'.join(description_list)
)
- resp.set_author(name="GitHub", url=f"https://github.com/{user}/{repository}")
+ embed_url = f"https://github.com/{user}/{repository}" if repository else f"https://github.com/{user}"
+ resp.set_author(name="GitHub", url=embed_url)
return resp
+ @whitelist_override(channels=WHITELISTED_CHANNELS, categories=WHITELISTED_CATEGORIES)
@commands.command(aliases=("pr",))
async def issue(
self,
@@ -150,56 +193,79 @@ class Issues(commands.Cog):
user: str = "python-discord"
) -> None:
"""Command to retrieve issue(s) from a GitHub repository."""
- if not(
- ctx.channel.category.id in WHITELISTED_CATEGORIES
- or ctx.channel.id in WHITELISTED_CHANNELS
- ):
- return
-
- result = await self.fetch_issues(set(numbers), repository, user)
+ # Remove duplicates
+ numbers = set(numbers)
- if result == FetchIssueErrors.value_error:
- await ctx.invoke(self.bot.get_command('help'), 'issue')
-
- elif result == FetchIssueErrors.max_requests:
+ if len(numbers) > MAXIMUM_ISSUES:
embed = discord.Embed(
title=random.choice(ERROR_REPLIES),
color=Colours.soft_red,
- description=f"Too many issues/PRs! (maximum of {MAX_REQUESTS})"
+ description=f"Too many issues/PRs! (maximum of {MAXIMUM_ISSUES})"
)
await ctx.send(embed=embed)
+ await invoke_help_command(ctx)
- elif isinstance(result, list):
- # Issue/PR format: emoji to show if open/closed/merged, number and the title as a singular link.
- resp = self.get_embed(result, user, repository)
- await ctx.send(embed=resp)
-
- elif isinstance(result, str):
- await ctx.send(result)
+ results = [await self.fetch_issues(number, repository, user) for number in numbers]
+ await ctx.send(embed=self.format_embed(results, user, repository))
@commands.Cog.listener()
async def on_message(self, message: discord.Message) -> None:
- """Command to retrieve issue(s) from a GitHub repository using automatic linking if matching <repo>#<issue>."""
- if not(
- message.channel.category.id in WHITELISTED_CATEGORIES
- or message.channel.id in WHITELISTED_CHANNELS_ON_MESSAGE
- ):
+ """
+ Automatic issue linking.
+
+ Listener to retrieve issue(s) from a GitHub repository using automatic linking if matching <org>/<repo>#<issue>.
+ """
+ # Ignore bots
+ if message.author.bot:
return
- message_repo_issue_map = re.findall(fr"({self.repo_regex})#(\d+)", message.content)
+ issues = [
+ FoundIssue(*match.group("org", "repo", "number"))
+ for match in AUTOMATIC_REGEX.finditer(self.remove_codeblocks(message.content))
+ ]
links = []
- if message_repo_issue_map:
- for repo_issue in message_repo_issue_map:
- if not self.check_in_block(message, " ".join(repo_issue)):
- result = await self.fetch_issues({repo_issue[1]}, repo_issue[0], "python-discord")
- if isinstance(result, list):
- links.extend(result)
+ if issues:
+ # Block this from working in DMs
+ if not message.guild:
+ await message.channel.send(
+ embed=discord.Embed(
+ title=random.choice(NEGATIVE_REPLIES),
+ description=(
+ "You can't retrieve issues from DMs. "
+ f"Try again in <#{Channels.community_bot_commands}>"
+ ),
+ colour=Colours.soft_red
+ )
+ )
+ return
+
+ log.trace(f"Found {issues = }")
+ # Remove duplicates
+ issues = set(issues)
+
+ if len(issues) > MAXIMUM_ISSUES:
+ embed = discord.Embed(
+ title=random.choice(ERROR_REPLIES),
+ color=Colours.soft_red,
+ description=f"Too many issues/PRs! (maximum of {MAXIMUM_ISSUES})"
+ )
+ await message.channel.send(embed=embed, delete_after=5)
+ return
+
+ for repo_issue in issues:
+ result = await self.fetch_issues(
+ int(repo_issue.number),
+ repo_issue.repository,
+ repo_issue.organisation or "python-discord"
+ )
+ if isinstance(result, IssueState):
+ links.append(result)
if not links:
return
- resp = self.get_embed(links, "python-discord")
+ resp = self.format_embed(links, "python-discord")
await message.channel.send(embed=resp)
diff --git a/bot/exts/evergreen/latex.py b/bot/exts/evergreen/latex.py
new file mode 100644
index 00000000..c4a8597c
--- /dev/null
+++ b/bot/exts/evergreen/latex.py
@@ -0,0 +1,94 @@
+import asyncio
+import hashlib
+import pathlib
+import re
+from concurrent.futures import ThreadPoolExecutor
+from io import BytesIO
+
+import discord
+import matplotlib.pyplot as plt
+from discord.ext import commands
+
+# configure fonts and colors for matplotlib
+plt.rcParams.update(
+ {
+ "font.size": 16,
+ "mathtext.fontset": "cm", # Computer Modern font set
+ "mathtext.rm": "serif",
+ "figure.facecolor": "36393F", # matches Discord's dark mode background color
+ "text.color": "white",
+ }
+)
+
+FORMATTED_CODE_REGEX = re.compile(
+ r"(?P<delim>(?P<block>```)|``?)" # code delimiter: 1-3 backticks; (?P=block) only matches if it's a block
+ r"(?(block)(?:(?P<lang>[a-z]+)\n)?)" # if we're in a block, match optional language (only letters plus newline)
+ r"(?:[ \t]*\n)*" # any blank (empty or tabs/spaces only) lines before the code
+ r"(?P<code>.*?)" # extract all code inside the markup
+ r"\s*" # any more whitespace before the end of the code markup
+ r"(?P=delim)", # match the exact same delimiter from the start again
+ re.DOTALL | re.IGNORECASE, # "." also matches newlines, case insensitive
+)
+
+CACHE_DIRECTORY = pathlib.Path("_latex_cache")
+CACHE_DIRECTORY.mkdir(exist_ok=True)
+
+
+class Latex(commands.Cog):
+ """Renders latex."""
+
+ @staticmethod
+ def _render(text: str, filepath: pathlib.Path) -> BytesIO:
+ """
+ Return the rendered image if latex compiles without errors, otherwise raise a BadArgument Exception.
+
+ Saves rendered image to cache.
+ """
+ fig = plt.figure()
+ rendered_image = BytesIO()
+ fig.text(0, 1, text, horizontalalignment="left", verticalalignment="top")
+
+ try:
+ plt.savefig(rendered_image, bbox_inches="tight", dpi=600)
+ except ValueError as e:
+ raise commands.BadArgument(str(e))
+
+ rendered_image.seek(0)
+
+ with open(filepath, "wb") as f:
+ f.write(rendered_image.getbuffer())
+
+ return rendered_image
+
+ @staticmethod
+ def _prepare_input(text: str) -> str:
+ text = text.replace(r"\\", "$\n$") # matplotlib uses \n for newlines, not \\
+
+ if match := FORMATTED_CODE_REGEX.match(text):
+ return match.group("code")
+ else:
+ return text
+
+ @commands.command()
+ @commands.max_concurrency(1, commands.BucketType.guild, wait=True)
+ async def latex(self, ctx: commands.Context, *, text: str) -> None:
+ """Renders the text in latex and sends the image."""
+ text = self._prepare_input(text)
+ query_hash = hashlib.md5(text.encode()).hexdigest()
+ image_path = CACHE_DIRECTORY.joinpath(f"{query_hash}.png")
+ async with ctx.typing():
+ if image_path.exists():
+ await ctx.send(file=discord.File(image_path))
+ return
+
+ with ThreadPoolExecutor() as pool:
+ image = await asyncio.get_running_loop().run_in_executor(
+ pool, self._render, text, image_path
+ )
+
+ await ctx.send(file=discord.File(image, "latex.png"))
+
+
+def setup(bot: commands.Bot) -> None:
+ """Load the Latex Cog."""
+ bot.add_cog(Latex(bot))
diff --git a/bot/exts/evergreen/ping.py b/bot/exts/evergreen/ping.py
new file mode 100644
index 00000000..07c13524
--- /dev/null
+++ b/bot/exts/evergreen/ping.py
@@ -0,0 +1,44 @@
+import arrow
+from dateutil.relativedelta import relativedelta
+from discord import Embed
+from discord.ext import commands
+
+from bot import start_time
+from bot.constants import Colours
+
+
+class Ping(commands.Cog):
+ """Get info about the bot's ping and uptime."""
+
+ def __init__(self, bot: commands.Bot):
+ self.bot = bot
+
+ @commands.command(name="ping")
+ async def ping(self, ctx: commands.Context) -> None:
+ """Ping the bot to see its latency and state."""
+ embed = Embed(
+ title=":ping_pong: Pong!",
+ colour=Colours.bright_green,
+ description=f"Gateway Latency: {round(self.bot.latency * 1000)}ms",
+ )
+
+ await ctx.send(embed=embed)
+
+ # Originally made in 70d2170a0a6594561d59c7d080c4280f1ebcd70b by lemon & gdude2002
+ @commands.command(name="uptime")
+ async def uptime(self, ctx: commands.Context) -> None:
+ """Get the current uptime of the bot."""
+ difference = relativedelta(start_time - arrow.utcnow())
+ uptime_string = start_time.shift(
+ seconds=-difference.seconds,
+ minutes=-difference.minutes,
+ hours=-difference.hours,
+ days=-difference.days
+ ).humanize()
+
+ await ctx.send(f"I started up {uptime_string}.")
+
+
+def setup(bot: commands.Bot) -> None:
+ """Load the Ping cog."""
+ bot.add_cog(Ping(bot))
diff --git a/bot/exts/evergreen/reddit.py b/bot/exts/evergreen/reddit.py
index 49127bea..e57fa2c0 100644
--- a/bot/exts/evergreen/reddit.py
+++ b/bot/exts/evergreen/reddit.py
@@ -1,128 +1,367 @@
+import asyncio
import logging
import random
+import textwrap
+from collections import namedtuple
+from datetime import datetime, timedelta
+from typing import List, Union
-import discord
-from discord.ext import commands
-from discord.ext.commands.cooldowns import BucketType
+from aiohttp import BasicAuth, ClientError
+from discord import Colour, Embed, TextChannel
+from discord.ext.commands import Cog, Context, group, has_any_role
+from discord.ext.tasks import loop
+from discord.utils import escape_markdown, sleep_until
-from bot.utils.pagination import ImagePaginator
+from bot.bot import Bot
+from bot.constants import Channels, ERROR_REPLIES, Emojis, Reddit as RedditConfig, STAFF_ROLES
+from bot.utils.converters import Subreddit
+from bot.utils.extensions import invoke_help_command
+from bot.utils.messages import sub_clyde
+from bot.utils.pagination import ImagePaginator, LinePaginator
log = logging.getLogger(__name__)
+AccessToken = namedtuple("AccessToken", ["token", "expires_at"])
-class Reddit(commands.Cog):
- """Fetches reddit posts."""
- def __init__(self, bot: commands.Bot):
+class Reddit(Cog):
+ """Track subreddit posts and show detailed statistics about them."""
+
+ HEADERS = {"User-Agent": "python3:python-discord/bot:1.0.0 (by /u/PythonDiscord)"}
+ URL = "https://www.reddit.com"
+ OAUTH_URL = "https://oauth.reddit.com"
+ MAX_RETRIES = 3
+
+ def __init__(self, bot: Bot):
self.bot = bot
- async def fetch(self, url: str) -> dict:
- """Send a get request to the reddit API and get json response."""
- session = self.bot.http_session
- params = {
- 'limit': 50
- }
- headers = {
- 'User-Agent': 'Iceman'
- }
-
- async with session.get(url=url, params=params, headers=headers) as response:
- return await response.json()
-
- @commands.command(name='reddit')
- @commands.cooldown(1, 10, BucketType.user)
- async def get_reddit(self, ctx: commands.Context, subreddit: str = 'python', sort: str = "hot") -> None:
- """
- Fetch reddit posts by using this command.
+ self.webhook = None
+ self.access_token = None
+ self.client_auth = BasicAuth(RedditConfig.client_id, RedditConfig.secret)
- Gets a post from r/python by default.
- Usage:
- --> .reddit [subreddit_name] [hot/top/new]
- """
+ bot.loop.create_task(self.init_reddit_ready())
+ self.auto_poster_loop.start()
+
+ def cog_unload(self) -> None:
+ """Stop the loop task and revoke the access token when the cog is unloaded."""
+ self.auto_poster_loop.cancel()
+ if self.access_token and self.access_token.expires_at > datetime.utcnow():
+ asyncio.create_task(self.revoke_access_token())
+
+ async def init_reddit_ready(self) -> None:
+ """Sets the reddit webhook when the cog is loaded."""
+ await self.bot.wait_until_guild_available()
+ if not self.webhook:
+ self.webhook = await self.bot.fetch_webhook(RedditConfig.webhook)
+
+ @property
+ def channel(self) -> TextChannel:
+ """Get the #reddit channel object from the bot's cache."""
+ return self.bot.get_channel(Channels.reddit)
+
+ def build_pagination_pages(self, posts: List[dict], paginate: bool) -> Union[List[tuple], str]:
+ """Build embed pages required for Paginator."""
pages = []
- sort_list = ["hot", "new", "top", "rising"]
- if sort.lower() not in sort_list:
- await ctx.send(f"Invalid sorting: {sort}\nUsing default sorting: `Hot`")
- sort = "hot"
+ first_page = ""
+ for post in posts:
+ post_page = ""
+ image_url = ""
- data = await self.fetch(f'https://www.reddit.com/r/{subreddit}/{sort}/.json')
+ data = post["data"]
- try:
- posts = data["data"]["children"]
- except KeyError:
- return await ctx.send('Subreddit not found!')
- if not posts:
- return await ctx.send('No posts available!')
+ title = textwrap.shorten(data["title"], width=50, placeholder="...")
+
+ # Normal brackets interfere with Markdown.
+ title = escape_markdown(title).replace("[", "⦋").replace("]", "⦌")
+ link = self.URL + data["permalink"]
+
+ first_page += f"**[{title.replace('*', '')}]({link})**\n"
+
+ text = data["selftext"]
+ if text:
+ first_page += textwrap.shorten(text, width=100, placeholder="...").replace("*", "") + "\n"
+
+ ups = data["ups"]
+ comments = data["num_comments"]
+ author = data["author"]
+
+ content_type = Emojis.reddit_post_text
+ if data["is_video"] or {"youtube", "youtu.be"}.issubset(set(data["url"].split("."))):
+ # This means the content type in the post is a video.
+ content_type = f"{Emojis.reddit_post_video}"
+
+ elif data["url"].endswith(("jpg", "png", "gif")):
+ # This means the content type in the post is an image.
+ content_type = f"{Emojis.reddit_post_photo}"
+ image_url = data["url"]
- if posts[1]["data"]["over_18"] is True:
- return await ctx.send(
- "You cannot access this Subreddit as it is ment for those who "
- "are 18 years or older."
+ first_page += (
+ f"{content_type}\u2003{Emojis.reddit_upvote}{ups}\u2003{Emojis.reddit_comments}"
+ f"\u2002{comments}\u2003{Emojis.reddit_users}{author}\n\n"
)
- embed_titles = ""
+ if paginate:
+ post_page += f"**[{title}]({link})**\n\n"
+ if text:
+ post_page += textwrap.shorten(text, width=252, placeholder="...") + "\n\n"
+ post_page += (
+ f"{content_type}\u2003{Emojis.reddit_upvote}{ups}\u2003{Emojis.reddit_comments}\u2002"
+ f"{comments}\u2003{Emojis.reddit_users}{author}"
+ )
- # Chooses k unique random elements from a population sequence or set.
- random_posts = random.sample(posts, k=5)
+ pages.append((post_page, image_url))
- # -----------------------------------------------------------
- # This code below is bound of change when the emojis are added.
+ if not paginate:
+ # Return the first summery page if pagination is not required
+ return first_page
- upvote_emoji = self.bot.get_emoji(755845219890757644)
- comment_emoji = self.bot.get_emoji(755845255001014384)
- user_emoji = self.bot.get_emoji(755845303822974997)
- text_emoji = self.bot.get_emoji(676030265910493204)
- video_emoji = self.bot.get_emoji(676030265839190047)
- image_emoji = self.bot.get_emoji(676030265734201344)
- reddit_emoji = self.bot.get_emoji(676030265734332427)
+ pages.insert(0, (first_page, "")) # Using image paginator, hence settings image url to empty string
+ return pages
- # ------------------------------------------------------------
+ async def get_access_token(self) -> None:
+ """
+ Get a Reddit API OAuth2 access token and assign it to self.access_token.
- for i, post in enumerate(random_posts, start=1):
- post_title = post["data"]["title"][0:50]
- post_url = post['data']['url']
- if post_title == "":
- post_title = "No Title."
- elif post_title == post_url:
- post_title = "Title is itself a link."
+ A token is valid for 1 hour. There will be MAX_RETRIES to get a token, after which the cog
+ will be unloaded and a ClientError raised if retrieval was still unsuccessful.
+ """
+ for i in range(1, self.MAX_RETRIES + 1):
+ response = await self.bot.http_session.post(
+ url=f"{self.URL}/api/v1/access_token",
+ headers=self.HEADERS,
+ auth=self.client_auth,
+ data={
+ "grant_type": "client_credentials",
+ "duration": "temporary"
+ }
+ )
- # ------------------------------------------------------------------
- # Embed building.
+ if response.status == 200 and response.content_type == "application/json":
+ content = await response.json()
+ expiration = int(content["expires_in"]) - 60 # Subtract 1 minute for leeway.
+ self.access_token = AccessToken(
+ token=content["access_token"],
+ expires_at=datetime.utcnow() + timedelta(seconds=expiration)
+ )
- embed_titles += f"**{i}.[{post_title}]({post_url})**\n"
- image_url = " "
- post_stats = f"{text_emoji}" # Set default content type to text.
+ log.debug(f"New token acquired; expires on UTC {self.access_token.expires_at}")
+ return
+ else:
+ log.debug(
+ f"Failed to get an access token: "
+ f"status {response.status} & content type {response.content_type}; "
+ f"retrying ({i}/{self.MAX_RETRIES})"
+ )
- if post["data"]["is_video"] is True or "youtube" in post_url.split("."):
- # This means the content type in the post is a video.
- post_stats = f"{video_emoji} "
+ await asyncio.sleep(3)
- elif post_url.endswith("jpg") or post_url.endswith("png") or post_url.endswith("gif"):
- # This means the content type in the post is an image.
- post_stats = f"{image_emoji} "
- image_url = post_url
-
- votes = f'{upvote_emoji}{post["data"]["ups"]}'
- comments = f'{comment_emoji}\u2002{ post["data"]["num_comments"]}'
- post_stats += (
- f"\u2002{votes}\u2003"
- f"{comments}"
- f'\u2003{user_emoji}\u2002{post["data"]["author"]}\n'
+ self.bot.remove_cog(self.qualified_name)
+ raise ClientError("Authentication with the Reddit API failed. Unloading the cog.")
+
+ async def revoke_access_token(self) -> None:
+ """
+ Revoke the OAuth2 access token for the Reddit API.
+
+ For security reasons, it's good practice to revoke the token when it's no longer being used.
+ """
+ response = await self.bot.http_session.post(
+ url=f"{self.URL}/api/v1/revoke_token",
+ headers=self.HEADERS,
+ auth=self.client_auth,
+ data={
+ "token": self.access_token.token,
+ "token_type_hint": "access_token"
+ }
+ )
+
+ if response.status in [200, 204] and response.content_type == "application/json":
+ self.access_token = None
+ else:
+ log.warning(f"Unable to revoke access token: status {response.status}.")
+
+ async def fetch_posts(self, route: str, *, amount: int = 25, params: dict = None) -> List[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.")
+
+ # Renew the token if necessary.
+ if not self.access_token or self.access_token.expires_at < datetime.utcnow():
+ await self.get_access_token()
+
+ url = f"{self.OAUTH_URL}/{route}"
+ for _ in range(self.MAX_RETRIES):
+ response = await self.bot.http_session.get(
+ url=url,
+ headers={**self.HEADERS, "Authorization": f"bearer {self.access_token.token}"},
+ params=params
+ )
+ if response.status == 200 and response.content_type == 'application/json':
+ # Got appropriate response - process and return.
+ content = await response.json()
+ posts = content["data"]["children"]
+
+ filtered_posts = [post for post in posts if not post["data"]["over_18"]]
+
+ return filtered_posts[:amount]
+
+ await asyncio.sleep(3)
+
+ log.debug(f"Invalid response from: {url} - status code {response.status}, mimetype {response.content_type}")
+ return list() # Failed to get appropriate response within allowed number of retries.
+
+ async def get_top_posts(
+ self, subreddit: Subreddit, time: str = "all", amount: int = 5, paginate: bool = False
+ ) -> Union[Embed, List[tuple]]:
+ """
+ Get the top amount of posts for a given subreddit within a specified timeframe.
+
+ A time of "all" will get posts from all time, "day" will get top daily posts and "week" will get the top
+ weekly posts.
+
+ The amount should be between 0 and 25 as Reddit's JSON requests only provide 25 posts at most.
+ """
+ embed = Embed()
+
+ posts = await self.fetch_posts(
+ route=f"{subreddit}/top",
+ amount=amount,
+ params={"t": time}
+ )
+ if not posts:
+ embed.title = random.choice(ERROR_REPLIES)
+ embed.colour = Colour.red()
+ embed.description = (
+ "Sorry! We couldn't find any SFW posts from that subreddit. "
+ "If this problem persists, please let us know."
)
- embed_titles += f"{post_stats}\n"
- page_text = f"**[{post_title}]({post_url})**\n{post_stats}\n{post['data']['selftext'][0:200]}"
- embed = discord.Embed()
- page_tuple = (page_text, image_url)
- pages.append(page_tuple)
+ return embed
+
+ if paginate:
+ return self.build_pagination_pages(posts, paginate=True)
+
+ # Use only starting summary page for #reddit channel posts.
+ embed.description = self.build_pagination_pages(posts, paginate=False)
+ embed.colour = Colour.blurple()
+ return embed
+
+ @loop()
+ async def auto_poster_loop(self) -> None:
+ """Post the top 5 posts daily, and the top 5 posts weekly."""
+ # once d.py get support for `time` parameter in loop decorator,
+ # this can be removed and the loop can use the `time=datetime.time.min` parameter
+ now = datetime.utcnow()
+ tomorrow = now + timedelta(days=1)
+ midnight_tomorrow = tomorrow.replace(hour=0, minute=0, second=0)
+
+ await sleep_until(midnight_tomorrow)
+
+ await self.bot.wait_until_guild_available()
+ if not self.webhook:
+ await self.bot.fetch_webhook(RedditConfig.webhook)
+
+ if datetime.utcnow().weekday() == 0:
+ await self.top_weekly_posts()
+ # if it's a monday send the top weekly posts
+
+ for subreddit in RedditConfig.subreddits:
+ top_posts = await self.get_top_posts(subreddit=subreddit, time="day")
+ username = sub_clyde(f"{subreddit} Top Daily Posts")
+ message = await self.webhook.send(username=username, embed=top_posts, wait=True)
+
+ if message.channel.is_news():
+ await message.publish()
+
+ async def top_weekly_posts(self) -> None:
+ """Post a summary of the top posts."""
+ for subreddit in RedditConfig.subreddits:
+ # Send and pin the new weekly posts.
+ top_posts = await self.get_top_posts(subreddit=subreddit, time="week")
+ username = sub_clyde(f"{subreddit} Top Weekly Posts")
+ message = await self.webhook.send(wait=True, username=username, embed=top_posts)
- # ------------------------------------------------------------------
+ if subreddit.lower() == "r/python":
+ if not self.channel:
+ log.warning("Failed to get #reddit channel to remove pins in the weekly loop.")
+ return
+
+ # Remove the oldest pins so that only 12 remain at most.
+ pins = await self.channel.pins()
+
+ while len(pins) >= 12:
+ await pins[-1].unpin()
+ del pins[-1]
+
+ await message.pin()
+
+ if message.channel.is_news():
+ await message.publish()
+
+ @group(name="reddit", invoke_without_command=True)
+ async def reddit_group(self, ctx: Context) -> None:
+ """View the top posts from various subreddits."""
+ await invoke_help_command(ctx)
+
+ @reddit_group.command(name="top")
+ async def top_command(self, ctx: Context, subreddit: Subreddit = "r/Python") -> None:
+ """Send the top posts of all time from a given subreddit."""
+ async with ctx.typing():
+ pages = await self.get_top_posts(subreddit=subreddit, time="all", paginate=True)
+
+ await ctx.send(f"Here are the top {subreddit} posts of all time!")
+ embed = Embed(
+ color=Colour.blurple()
+ )
- pages.insert(0, (embed_titles, " "))
- embed.set_author(name=f"r/{posts[0]['data']['subreddit']} - {sort}", icon_url=reddit_emoji.url)
await ImagePaginator.paginate(pages, ctx, embed)
+ @reddit_group.command(name="daily")
+ async def daily_command(self, ctx: Context, subreddit: Subreddit = "r/Python") -> None:
+ """Send the top posts of today from a given subreddit."""
+ async with ctx.typing():
+ pages = await self.get_top_posts(subreddit=subreddit, time="day", paginate=True)
+
+ await ctx.send(f"Here are today's top {subreddit} posts!")
+ embed = Embed(
+ color=Colour.blurple()
+ )
+
+ await ImagePaginator.paginate(pages, ctx, embed)
+
+ @reddit_group.command(name="weekly")
+ async def weekly_command(self, ctx: Context, subreddit: Subreddit = "r/Python") -> None:
+ """Send the top posts of this week from a given subreddit."""
+ async with ctx.typing():
+ pages = await self.get_top_posts(subreddit=subreddit, time="week", paginate=True)
+
+ await ctx.send(f"Here are this week's top {subreddit} posts!")
+ embed = Embed(
+ color=Colour.blurple()
+ )
+
+ await ImagePaginator.paginate(pages, ctx, embed)
+
+ @has_any_role(*STAFF_ROLES)
+ @reddit_group.command(name="subreddits", aliases=("subs",))
+ 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()
+
+ await LinePaginator.paginate(
+ RedditConfig.subreddits,
+ ctx, embed,
+ footer_text="Use the reddit commands along with these to view their posts.",
+ empty=False,
+ max_lines=15
+ )
+
-def setup(bot: commands.Bot) -> None:
- """Load the Cog."""
+def setup(bot: Bot) -> None:
+ """Load the Reddit cog."""
+ if not RedditConfig.secret or not RedditConfig.client_id:
+ log.error("Credentials not provided, cog not loaded.")
+ return
bot.add_cog(Reddit(bot))
diff --git a/bot/exts/evergreen/timed.py b/bot/exts/evergreen/timed.py
new file mode 100644
index 00000000..5f177fd6
--- /dev/null
+++ b/bot/exts/evergreen/timed.py
@@ -0,0 +1,46 @@
+from copy import copy
+from time import perf_counter
+
+from discord import Message
+from discord.ext import commands
+
+
+class TimedCommands(commands.Cog):
+ """Time the command execution of a command."""
+
+ @staticmethod
+ async def create_execution_context(ctx: commands.Context, command: str) -> commands.Context:
+ """Get a new execution context for a command."""
+ msg: Message = copy(ctx.message)
+ msg.content = f"{ctx.prefix}{command}"
+
+ return await ctx.bot.get_context(msg)
+
+ @commands.command(name="timed", aliases=["time", "t"])
+ async def timed(self, ctx: commands.Context, *, command: str) -> None:
+ """Time the command execution of a command."""
+ new_ctx = await self.create_execution_context(ctx, command)
+
+ ctx.subcontext = new_ctx
+
+ if not ctx.subcontext.command:
+ help_command = f"{ctx.prefix}help"
+ error = f"The command you are trying to time doesn't exist. Use `{help_command}` for a list of commands."
+
+ await ctx.send(error)
+ return
+
+ if new_ctx.command.qualified_name == "timed":
+ await ctx.send("You are not allowed to time the execution of the `timed` command.")
+ return
+
+ t_start = perf_counter()
+ await new_ctx.command.invoke(new_ctx)
+ t_end = perf_counter()
+
+ await ctx.send(f"Command execution for `{new_ctx.command}` finished in {(t_end - t_start):.4f} seconds.")
+
+
+def setup(bot: commands.Bot) -> None:
+ """Cog load."""
+ bot.add_cog(TimedCommands(bot))
diff --git a/bot/exts/evergreen/uptime.py b/bot/exts/evergreen/uptime.py
deleted file mode 100644
index a9ad9dfb..00000000
--- a/bot/exts/evergreen/uptime.py
+++ /dev/null
@@ -1,33 +0,0 @@
-import logging
-
-import arrow
-from dateutil.relativedelta import relativedelta
-from discord.ext import commands
-
-from bot import start_time
-
-log = logging.getLogger(__name__)
-
-
-class Uptime(commands.Cog):
- """A cog for posting the bot's uptime."""
-
- def __init__(self, bot: commands.Bot):
- self.bot = bot
-
- @commands.command(name="uptime")
- async def uptime(self, ctx: commands.Context) -> None:
- """Responds with the uptime of the bot."""
- difference = relativedelta(start_time - arrow.utcnow())
- uptime_string = start_time.shift(
- seconds=-difference.seconds,
- minutes=-difference.minutes,
- hours=-difference.hours,
- days=-difference.days
- ).humanize()
- await ctx.send(f"I started up {uptime_string}.")
-
-
-def setup(bot: commands.Bot) -> None:
- """Uptime Cog load."""
- bot.add_cog(Uptime(bot))
diff --git a/bot/exts/evergreen/wolfram.py b/bot/exts/evergreen/wolfram.py
index 437d9e1a..14ec1041 100644
--- a/bot/exts/evergreen/wolfram.py
+++ b/bot/exts/evergreen/wolfram.py
@@ -62,7 +62,8 @@ def custom_cooldown(*ignore: List[int]) -> Callable:
# if the invoked command is help we don't want to increase the ratelimits since it's not actually
# invoking the command/making a request, so instead just check if the user/guild are on cooldown.
guild_cooldown = not guildcd.get_bucket(ctx.message).get_tokens() == 0 # if guild is on cooldown
- if not any(r.id in ignore for r in ctx.author.roles): # check user bucket if user is not ignored
+ # check the message is in a guild, and check user bucket if user is not ignored
+ if ctx.guild and not any(r.id in ignore for r in ctx.author.roles):
return guild_cooldown and not usercd.get_bucket(ctx.message).get_tokens() == 0
return guild_cooldown