aboutsummaryrefslogtreecommitdiffstats
path: root/bot/seasons
diff options
context:
space:
mode:
Diffstat (limited to 'bot/seasons')
-rw-r--r--bot/seasons/evergreen/bookmark.py65
-rw-r--r--bot/seasons/evergreen/error_handler.py225
-rw-r--r--bot/seasons/evergreen/movie.py198
-rw-r--r--bot/seasons/evergreen/reddit.py130
-rw-r--r--bot/seasons/evergreen/trivia_quiz.py217
5 files changed, 642 insertions, 193 deletions
diff --git a/bot/seasons/evergreen/bookmark.py b/bot/seasons/evergreen/bookmark.py
new file mode 100644
index 00000000..7bdd362c
--- /dev/null
+++ b/bot/seasons/evergreen/bookmark.py
@@ -0,0 +1,65 @@
+import logging
+import random
+
+import discord
+from discord.ext import commands
+
+from bot.constants import Colours, ERROR_REPLIES, Emojis, bookmark_icon_url
+
+log = logging.getLogger(__name__)
+
+
+class Bookmark(commands.Cog):
+ """Creates personal bookmarks by relaying a message link to the user's DMs."""
+
+ def __init__(self, bot: commands.Bot):
+ self.bot = bot
+
+ @commands.command(name="bookmark", aliases=("bm", "pin"))
+ async def bookmark(
+ self,
+ ctx: commands.Context,
+ target_message: discord.Message,
+ *,
+ title: str = "Bookmark"
+ ) -> None:
+ """Send the author a link to `target_message` via DMs."""
+ # Prevent users from bookmarking a message in a channel they don't have access to
+ permissions = ctx.author.permissions_in(target_message.channel)
+ if not permissions.read_messages:
+ log.info(f"{ctx.author} tried to bookmark a message in #{target_message.channel} but has no permissions")
+ embed = discord.Embed(
+ title=random.choice(ERROR_REPLIES),
+ color=Colours.soft_red,
+ description="You don't have permission to view this channel."
+ )
+ await ctx.send(embed=embed)
+ return
+
+ embed = discord.Embed(
+ title=title,
+ colour=Colours.soft_green,
+ description=target_message.content
+ )
+ embed.add_field(name="Wanna give it a visit?", value=f"[Visit original message]({target_message.jump_url})")
+ embed.set_author(name=target_message.author, icon_url=target_message.author.avatar_url)
+ embed.set_thumbnail(url=bookmark_icon_url)
+
+ try:
+ await ctx.author.send(embed=embed)
+ except discord.Forbidden:
+ error_embed = discord.Embed(
+ title=random.choice(ERROR_REPLIES),
+ description=f"{ctx.author.mention}, please enable your DMs to receive the bookmark",
+ colour=Colours.soft_red
+ )
+ await ctx.send(embed=error_embed)
+ else:
+ log.info(f"{ctx.author} bookmarked {target_message.jump_url} with title '{title}'")
+ await ctx.message.add_reaction(Emojis.envelope)
+
+
+def setup(bot: commands.Bot) -> None:
+ """Load the Bookmark cog."""
+ bot.add_cog(Bookmark(bot))
+ log.info("Bookmark cog loaded")
diff --git a/bot/seasons/evergreen/error_handler.py b/bot/seasons/evergreen/error_handler.py
index 120462ee..0d8bb0bb 100644
--- a/bot/seasons/evergreen/error_handler.py
+++ b/bot/seasons/evergreen/error_handler.py
@@ -1,119 +1,106 @@
-import logging
-import math
-import random
-import sys
-import traceback
-
-from discord import Colour, Embed, Message
-from discord.ext import commands
-
-from bot.constants import NEGATIVE_REPLIES
-from bot.decorators import InChannelCheckFailure
-
-log = logging.getLogger(__name__)
-
-
-class CommandErrorHandler(commands.Cog):
- """A error handler for the PythonDiscord server."""
-
- def __init__(self, bot: commands.Bot):
- self.bot = bot
-
- @staticmethod
- def revert_cooldown_counter(command: commands.Command, message: Message) -> None:
- """Undoes the last cooldown counter for user-error cases."""
- if command._buckets.valid:
- bucket = command._buckets.get_bucket(message)
- bucket._tokens = min(bucket.rate, bucket._tokens + 1)
- logging.debug(
- "Cooldown counter reverted as the command was not used correctly."
- )
-
- @commands.Cog.listener()
- async def on_command_error(self, ctx: commands.Context, error: commands.CommandError) -> None:
- """Activates when a command opens an error."""
- if hasattr(ctx.command, 'on_error'):
- return logging.debug(
- "A command error occured but the command had it's own error handler."
- )
-
- error = getattr(error, 'original', error)
-
- if isinstance(error, InChannelCheckFailure):
- logging.debug(
- f"{ctx.author} the command '{ctx.command}', but they did not have "
- f"permissions to run commands in the channel {ctx.channel}!"
- )
- embed = Embed(colour=Colour.red())
- embed.title = random.choice(NEGATIVE_REPLIES)
- embed.description = str(error)
- return await ctx.send(embed=embed)
-
- if isinstance(error, commands.CommandNotFound):
- return logging.debug(
- f"{ctx.author} called '{ctx.message.content}' but no command was found."
- )
-
- if isinstance(error, commands.UserInputError):
- logging.debug(
- f"{ctx.author} called the command '{ctx.command}' but entered invalid input!"
- )
-
- self.revert_cooldown_counter(ctx.command, ctx.message)
-
- return await ctx.send(
- ":no_entry: The command you specified failed to run. "
- "This is because the arguments you provided were invalid."
- )
-
- if isinstance(error, commands.CommandOnCooldown):
- logging.debug(
- f"{ctx.author} called the command '{ctx.command}' but they were on cooldown!"
- )
- remaining_minutes, remaining_seconds = divmod(error.retry_after, 60)
-
- return await ctx.send(
- "This command is on cooldown, please retry in "
- f"{int(remaining_minutes)} minutes {math.ceil(remaining_seconds)} seconds."
- )
-
- if isinstance(error, commands.DisabledCommand):
- logging.debug(
- f"{ctx.author} called the command '{ctx.command}' but the command was disabled!"
- )
- return await ctx.send(":no_entry: This command has been disabled.")
-
- if isinstance(error, commands.NoPrivateMessage):
- logging.debug(
- f"{ctx.author} called the command '{ctx.command}' "
- "in a private message however the command was guild only!"
- )
- return await ctx.author.send(":no_entry: This command can only be used in the server.")
-
- if isinstance(error, commands.BadArgument):
- self.revert_cooldown_counter(ctx.command, ctx.message)
-
- logging.debug(
- f"{ctx.author} called the command '{ctx.command}' but entered a bad argument!"
- )
- return await ctx.send("The argument you provided was invalid.")
-
- if isinstance(error, commands.CheckFailure):
- logging.debug(f"{ctx.author} called the command '{ctx.command}' but the checks failed!")
- return await ctx.send(":no_entry: You are not authorized to use this command.")
-
- print(f"Ignoring exception in command {ctx.command}:", file=sys.stderr)
-
- logging.warning(
- f"{ctx.author} called the command '{ctx.command}' "
- "however the command failed to run with the error:"
- f"-------------\n{error}"
- )
-
- traceback.print_exception(type(error), error, error.__traceback__, file=sys.stderr)
-
-
-def setup(bot: commands.Bot) -> None:
- """Error handler Cog load."""
- bot.add_cog(CommandErrorHandler(bot))
- log.info("CommandErrorHandler cog loaded")
+import logging
+import math
+import random
+from typing import Iterable, Union
+
+from discord import Embed, Message
+from discord.ext import commands
+
+from bot.constants import Colours, ERROR_REPLIES, NEGATIVE_REPLIES
+from bot.decorators import InChannelCheckFailure
+
+log = logging.getLogger(__name__)
+
+
+class CommandErrorHandler(commands.Cog):
+ """A error handler for the PythonDiscord server."""
+
+ def __init__(self, bot: commands.Bot):
+ self.bot = bot
+
+ @staticmethod
+ def revert_cooldown_counter(command: commands.Command, message: Message) -> None:
+ """Undoes the last cooldown counter for user-error cases."""
+ if command._buckets.valid:
+ bucket = command._buckets.get_bucket(message)
+ bucket._tokens = min(bucket.rate, bucket._tokens + 1)
+ logging.debug("Cooldown counter reverted as the command was not used correctly.")
+
+ @staticmethod
+ def error_embed(message: str, title: Union[Iterable, str] = ERROR_REPLIES) -> Embed:
+ """Build a basic embed with red colour and either a random error title or a title provided."""
+ embed = Embed(colour=Colours.soft_red)
+ if isinstance(title, str):
+ embed.title = title
+ else:
+ embed.title = random.choice(title)
+ embed.description = message
+ return embed
+
+ @commands.Cog.listener()
+ async def on_command_error(self, ctx: commands.Context, error: commands.CommandError) -> None:
+ """Activates when a command opens an error."""
+ if hasattr(ctx.command, 'on_error'):
+ logging.debug("A command error occured but the command had it's own error handler.")
+ return
+
+ error = getattr(error, 'original', error)
+ logging.debug(
+ f"Error Encountered: {type(error).__name__} - {str(error)}, "
+ f"Command: {ctx.command}, "
+ f"Author: {ctx.author}, "
+ f"Channel: {ctx.channel}"
+ )
+
+ if isinstance(error, commands.CommandNotFound):
+ return
+
+ if isinstance(error, InChannelCheckFailure):
+ await ctx.send(embed=self.error_embed(str(error), NEGATIVE_REPLIES), delete_after=7.5)
+ return
+
+ if isinstance(error, commands.UserInputError):
+ self.revert_cooldown_counter(ctx.command, ctx.message)
+ embed = self.error_embed(
+ f"Your input was invalid: {error}\n\nUsage:\n```{ctx.prefix}{ctx.command} {ctx.command.signature}```"
+ )
+ await ctx.send(embed=embed)
+ return
+
+ if isinstance(error, commands.CommandOnCooldown):
+ mins, secs = divmod(math.ceil(error.retry_after), 60)
+ embed = self.error_embed(
+ f"This command is on cooldown:\nPlease retry in {mins} minutes {secs} seconds.",
+ NEGATIVE_REPLIES
+ )
+ await ctx.send(embed=embed, delete_after=7.5)
+ return
+
+ if isinstance(error, commands.DisabledCommand):
+ await ctx.send(embed=self.error_embed("This command has been disabled.", NEGATIVE_REPLIES))
+ return
+
+ if isinstance(error, commands.NoPrivateMessage):
+ await ctx.send(embed=self.error_embed("This command can only be used in the server.", NEGATIVE_REPLIES))
+ return
+
+ if isinstance(error, commands.BadArgument):
+ 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}```"
+ )
+ await ctx.send(embed=embed)
+ return
+
+ if isinstance(error, commands.CheckFailure):
+ await ctx.send(embed=self.error_embed("You are not authorized to use this command.", NEGATIVE_REPLIES))
+ return
+
+ log.exception(f"Unhandled command error: {str(error)}", exc_info=error)
+
+
+def setup(bot: commands.Bot) -> None:
+ """Error handler Cog load."""
+ bot.add_cog(CommandErrorHandler(bot))
+ log.info("CommandErrorHandler cog loaded")
diff --git a/bot/seasons/evergreen/movie.py b/bot/seasons/evergreen/movie.py
new file mode 100644
index 00000000..3c5a312d
--- /dev/null
+++ b/bot/seasons/evergreen/movie.py
@@ -0,0 +1,198 @@
+import logging
+import random
+from enum import Enum
+from typing import Any, Dict, List, Tuple
+from urllib.parse import urlencode
+
+from aiohttp import ClientSession
+from discord import Embed
+from discord.ext.commands import Bot, Cog, Context, group
+
+from bot.constants import Tokens
+from bot.pagination import ImagePaginator
+
+# Define base URL of TMDB
+BASE_URL = "https://api.themoviedb.org/3/"
+
+logger = logging.getLogger(__name__)
+
+# Define movie params, that will be used for every movie request
+MOVIE_PARAMS = {
+ "api_key": Tokens.tmdb,
+ "language": "en-US"
+}
+
+
+class MovieGenres(Enum):
+ """Movies Genre names and IDs."""
+
+ Action = "28"
+ Adventure = "12"
+ Animation = "16"
+ Comedy = "35"
+ Crime = "80"
+ Documentary = "99"
+ Drama = "18"
+ Family = "10751"
+ Fantasy = "14"
+ History = "36"
+ Horror = "27"
+ Music = "10402"
+ Mystery = "9648"
+ Romance = "10749"
+ Science = "878"
+ Thriller = "53"
+ Western = "37"
+
+
+class Movie(Cog):
+ """Movie Cog contains movies command that grab random movies from TMDB."""
+
+ def __init__(self, bot: Bot):
+ self.bot = bot
+ self.http_session: ClientSession = bot.http_session
+
+ @group(name='movies', aliases=['movie'], invoke_without_command=True)
+ async def movies(self, ctx: Context, genre: str = "", amount: int = 5) -> None:
+ """
+ Get random movies by specifying genre. Also support amount parameter, that define how much movies will be shown.
+
+ Default 5. Use .movies genres to get all available genres.
+ """
+ # Check is there more than 20 movies specified, due TMDB return 20 movies
+ # per page, so this is max. Also you can't get less movies than 1, just logic
+ if amount > 20:
+ await ctx.send("You can't get more than 20 movies at once. (TMDB limits)")
+ return
+ elif amount < 1:
+ await ctx.send("You can't get less than 1 movie.")
+ return
+
+ # Capitalize genre for getting data from Enum, get random page, send help when genre don't exist.
+ genre = genre.capitalize()
+ try:
+ result = await self.get_movies_list(self.http_session, MovieGenres[genre].value, 1)
+ except KeyError:
+ await ctx.send_help('movies')
+ return
+
+ # Check if "results" is in result. If not, throw error.
+ if "results" not in result.keys():
+ err_msg = f"There is problem while making TMDB API request. Response Code: {result['status_code']}, " \
+ f"{result['status_message']}."
+ await ctx.send(err_msg)
+ logger.warning(err_msg)
+
+ # Get random page. Max page is last page where is movies with this genre.
+ page = random.randint(1, result["total_pages"])
+
+ # Get movies list from TMDB, check if results key in result. When not, raise error.
+ movies = await self.get_movies_list(self.http_session, MovieGenres[genre].value, page)
+ if 'results' not in movies.keys():
+ err_msg = f"There is problem while making TMDB API request. Response Code: {result['status_code']}, " \
+ f"{result['status_message']}."
+ await ctx.send(err_msg)
+ logger.warning(err_msg)
+
+ # Get all pages and embed
+ pages = await self.get_pages(self.http_session, movies, amount)
+ embed = await self.get_embed(genre)
+
+ await ImagePaginator.paginate(pages, ctx, embed)
+
+ @movies.command(name='genres', aliases=['genre', 'g'])
+ async def genres(self, ctx: Context) -> None:
+ """Show all currently available genres for .movies command."""
+ await ctx.send(f"Current available genres: {', '.join('`' + genre.name + '`' for genre in MovieGenres)}")
+
+ async def get_movies_list(self, client: ClientSession, genre_id: str, page: int) -> Dict[str, Any]:
+ """Return JSON of TMDB discover request."""
+ # Define params of request
+ params = {
+ "api_key": Tokens.tmdb,
+ "language": "en-US",
+ "sort_by": "popularity.desc",
+ "include_adult": "false",
+ "include_video": "false",
+ "page": page,
+ "with_genres": genre_id
+ }
+
+ url = BASE_URL + "discover/movie?" + urlencode(params)
+
+ # Make discover request to TMDB, return result
+ async with client.get(url) as resp:
+ return await resp.json()
+
+ async def get_pages(self, client: ClientSession, movies: Dict[str, Any], amount: int) -> List[Tuple[str, str]]:
+ """Fetch all movie pages from movies dictionary. Return list of pages."""
+ pages = []
+
+ for i in range(amount):
+ movie_id = movies['results'][i]['id']
+ movie = await self.get_movie(client, movie_id)
+
+ page, img = await self.create_page(movie)
+ pages.append((page, img))
+
+ return pages
+
+ async def get_movie(self, client: ClientSession, movie: int) -> Dict:
+ """Get Movie by movie ID from TMDB. Return result dictionary."""
+ url = BASE_URL + f"movie/{movie}?" + urlencode(MOVIE_PARAMS)
+
+ async with client.get(url) as resp:
+ return await resp.json()
+
+ async def create_page(self, movie: Dict[str, Any]) -> Tuple[str, str]:
+ """Create page from TMDB movie request result. Return formatted page + image."""
+ text = ""
+
+ # Add title + tagline (if not empty)
+ text += f"**{movie['title']}**\n"
+ if movie['tagline']:
+ text += f"{movie['tagline']}\n\n"
+ else:
+ text += "\n"
+
+ # Add other information
+ text += f"**Rating:** {movie['vote_average']}/10 :star:\n"
+ text += f"**Release Date:** {movie['release_date']}\n\n"
+
+ text += "__**Production Information**__\n"
+
+ companies = movie['production_companies']
+ countries = movie['production_countries']
+
+ text += f"**Made by:** {', '.join(company['name'] for company in companies)}\n"
+ text += f"**Made in:** {', '.join(country['name'] for country in countries)}\n\n"
+
+ text += "__**Some Numbers**__\n"
+
+ budget = f"{movie['budget']:,d}" if movie['budget'] else "?"
+ revenue = f"{movie['revenue']:,d}" if movie['revenue'] else "?"
+
+ if movie['runtime'] is not None:
+ duration = divmod(movie['runtime'], 60)
+ else:
+ duration = ("?", "?")
+
+ text += f"**Budget:** ${budget}\n"
+ text += f"**Revenue:** ${revenue}\n"
+ text += f"**Duration:** {f'{duration[0]} hour(s) {duration[1]} minute(s)'}\n\n"
+
+ text += movie['overview']
+
+ img = f"http://image.tmdb.org/t/p/w200{movie['poster_path']}"
+
+ # Return page content and image
+ return text, img
+
+ async def get_embed(self, name: str) -> Embed:
+ """Return embed of random movies. Uses name in title."""
+ return Embed(title=f'Random {name} Movies').set_footer(text='Powered by TMDB (themoviedb.org)')
+
+
+def setup(bot: Bot) -> None:
+ """Load Movie Cog."""
+ bot.add_cog(Movie(bot))
diff --git a/bot/seasons/evergreen/reddit.py b/bot/seasons/evergreen/reddit.py
new file mode 100644
index 00000000..32ca419a
--- /dev/null
+++ b/bot/seasons/evergreen/reddit.py
@@ -0,0 +1,130 @@
+import logging
+import random
+
+import discord
+from discord.ext import commands
+from discord.ext.commands.cooldowns import BucketType
+
+from bot.pagination import ImagePaginator
+
+
+log = logging.getLogger(__name__)
+
+
+class Reddit(commands.Cog):
+ """Fetches reddit posts."""
+
+ def __init__(self, bot: commands.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.
+
+ Gets a post from r/python by default.
+ Usage:
+ --> .reddit [subreddit_name] [hot/top/new]
+ """
+ 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"
+
+ data = await self.fetch(f'https://www.reddit.com/r/{subreddit}/{sort}/.json')
+
+ try:
+ posts = data["data"]["children"]
+ except KeyError:
+ return await ctx.send('Subreddit not found!')
+ if not posts:
+ return await ctx.send('No posts available!')
+
+ 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."
+ )
+
+ embed_titles = ""
+
+ # Chooses k unique random elements from a population sequence or set.
+ random_posts = random.sample(posts, k=5)
+
+ # -----------------------------------------------------------
+ # This code below is bound of change when the emojis are added.
+
+ upvote_emoji = self.bot.get_emoji(638729835245731840)
+ comment_emoji = self.bot.get_emoji(638729835073765387)
+ user_emoji = self.bot.get_emoji(638729835442602003)
+ 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)
+
+ # ------------------------------------------------------------
+
+ 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."
+
+ # ------------------------------------------------------------------
+ # Embed building.
+
+ embed_titles += f"**{i}.[{post_title}]({post_url})**\n"
+ image_url = " "
+ post_stats = f"{text_emoji}" # Set default content type to text.
+
+ 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} "
+
+ 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'
+ )
+ 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)
+
+ # ------------------------------------------------------------------
+
+ 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)
+
+
+def setup(bot: commands.Bot) -> None:
+ """Load the Cog."""
+ bot.add_cog(Reddit(bot))
+ log.debug('Loaded')
diff --git a/bot/seasons/evergreen/trivia_quiz.py b/bot/seasons/evergreen/trivia_quiz.py
index 798523e6..99b64497 100644
--- a/bot/seasons/evergreen/trivia_quiz.py
+++ b/bot/seasons/evergreen/trivia_quiz.py
@@ -14,10 +14,8 @@ from bot.constants import Roles
logger = logging.getLogger(__name__)
-ANNOYED_EXPRESSIONS = ["-_-", "-.-"]
-
WRONG_ANS_RESPONSE = [
- "No one gave the correct answer",
+ "No one answered correctly!",
"Better luck next time"
]
@@ -28,10 +26,11 @@ class TriviaQuiz(commands.Cog):
def __init__(self, bot: commands.Bot) -> None:
self.bot = bot
self.questions = self.load_questions()
- self.game_status = {}
- self.game_owners = {}
+ self.game_status = {} # A variable to store the game status: either running or not running.
+ self.game_owners = {} # A variable to store the person's ID who started the quiz game in a channel.
self.question_limit = 4
- self.player_dict = {}
+ self.player_scores = {} # A variable to store all player's scores for a bot session.
+ self.game_player_scores = {} # A variable to store temporary game player's scores.
self.categories = {
"general": "Test your general knowledge"
# "retro": "Questions related to retro gaming."
@@ -39,138 +38,205 @@ class TriviaQuiz(commands.Cog):
@staticmethod
def load_questions() -> dict:
- """Load the questions from json file."""
+ """Load the questions from the JSON file."""
p = Path("bot", "resources", "evergreen", "trivia_quiz.json")
with p.open() as json_data:
questions = json.load(json_data)
return questions
- @commands.command(name="quiz", aliases=["trivia"])
- async def quiz_game(self, ctx: commands.Context, category: str = "general") -> None:
+ @commands.group(name="quiz", aliases=["trivia"], invoke_without_command=True)
+ async def quiz_game(self, ctx: commands.Context, category: str = None) -> None:
"""
- Start/Stop a quiz!
-
- arguments:
- option:
- - start : to start a quiz in a channel
- - stop : stop the quiz running in that channel.
+ Start a quiz!
Questions for the quiz can be selected from the following categories:
- general : Test your general knowledge. (default)
- (we wil be adding more later)
+ (More to come!)
"""
- category = category.lower()
-
if ctx.channel.id not in self.game_status:
self.game_status[ctx.channel.id] = False
- self.player_dict[ctx.channel.id] = {}
- if not self.game_status[ctx.channel.id]:
- self.game_owners[ctx.channel.id] = ctx.author
- self.game_status[ctx.channel.id] = True
- start_embed = discord.Embed(colour=discord.Colour.red())
- start_embed.title = "Quiz game Starting!!"
- start_embed.description = "Each game consists of 5 questions.\n"
- start_embed.description += "**Rules :**\nNo cheating and have fun!"
- start_embed.set_footer(
- text="Points for a question reduces by 25 after 10s or after a hint. Total time is 30s per question"
+ if ctx.channel.id not in self.game_player_scores:
+ self.game_player_scores[ctx.channel.id] = {}
+
+ # Stop game if running.
+ if self.game_status[ctx.channel.id] is True:
+ return await ctx.send(
+ f"Game is already running..."
+ f"do `{self.bot.command_prefix}quiz stop`"
)
- await ctx.send(embed=start_embed) # send an embed with the rules
- await asyncio.sleep(1)
- else:
- if (
- ctx.author == self.game_owners[ctx.channel.id]
- or Roles.moderator in [role.id for role in ctx.author.roles]
- ):
- await ctx.send("Quiz is no longer running.")
- await self.declare_winner(ctx.channel, self.player_dict[ctx.channel.id])
- self.game_status[ctx.channel.id] = False
- del self.game_owners[ctx.channel.id]
- else:
- await ctx.send(f"{ctx.author.mention}, you are not authorised to stop this game :ghost: !")
+ # Send embed showing available categories if inputted category is invalid.
+ if category is None:
+ category = random.choice(list(self.categories))
+ category = category.lower()
if category not in self.categories:
- embed = self.category_embed
+ embed = self.category_embed()
await ctx.send(embed=embed)
return
+
+ # Start game if not running.
+ if self.game_status[ctx.channel.id] is False:
+ self.game_owners[ctx.channel.id] = ctx.author
+ self.game_status[ctx.channel.id] = True
+ start_embed = self.make_start_embed(category)
+
+ await ctx.send(embed=start_embed) # send an embed with the rules
+ await asyncio.sleep(1)
+
topic = self.questions[category]
- unanswered = 0
done_question = []
hint_no = 0
answer = None
while self.game_status[ctx.channel.id]:
+ # Exit quiz if number of questions for a round are already sent.
if len(done_question) > self.question_limit and hint_no == 0:
- await ctx.send("The round ends here.")
- await self.declare_winner(ctx.channel, self.player_dict[ctx.channel.id])
- break
- if unanswered > 3:
- await ctx.send("Game stopped due to inactivity.")
- await self.declare_winner(ctx.channel, self.player_dict[ctx.channel.id])
+ await ctx.send("The round has ended.")
+ await self.declare_winner(ctx.channel, self.game_player_scores[ctx.channel.id])
+
+ self.game_status[ctx.channel.id] = False
+ del self.game_owners[ctx.channel.id]
+ self.game_player_scores[ctx.channel.id] = {}
+
break
+
+ # If no hint has been sent or any time alert. Basically if hint_no = 0 means it is a new question.
if hint_no == 0:
+ # Select a random question which has not been used yet.
while True:
question_dict = random.choice(topic)
if question_dict["id"] not in done_question:
done_question.append(question_dict["id"])
break
+
q = question_dict["question"]
answer = question_dict["answer"]
embed = discord.Embed(colour=discord.Colour.gold())
embed.title = f"Question #{len(done_question)}"
embed.description = q
- await ctx.send(embed=embed)
+ await ctx.send(embed=embed) # Send question embed.
+ # A function to check whether user input is the correct answer(close to the right answer)
def check(m: discord.Message) -> bool:
ratio = fuzz.ratio(answer.lower(), m.content.lower())
return ratio > 85 and m.channel == ctx.channel
+
try:
msg = await self.bot.wait_for('message', check=check, timeout=10)
except asyncio.TimeoutError:
+ # In case of TimeoutError and the game has been stopped, then do nothing.
if self.game_status[ctx.channel.id] is False:
break
+
+ # if number of hints sent or time alerts sent is less than 2, then send one.
if hint_no < 2:
hint_no += 1
if "hints" in question_dict:
hints = question_dict["hints"]
await ctx.send(f"**Hint #{hint_no+1}\n**{hints[hint_no]}")
else:
- await ctx.send(f"Cmon guys, {30-hint_no*10}s left!")
+ await ctx.send(f"{30 - hint_no * 10}s left!")
+ # Once hint or time alerts has been sent 2 times, the hint_no value will be 3
+ # If hint_no > 2, then it means that all hints/time alerts have been sent.
+ # Also means that the answer is not yet given and the bot sends the answer and the next question.
else:
+ if self.game_status[ctx.channel.id] is False:
+ break
+
response = random.choice(WRONG_ANS_RESPONSE)
- expression = random.choice(ANNOYED_EXPRESSIONS)
- await ctx.send(f"{response} {expression}")
+ await ctx.send(response)
await self.send_answer(ctx.channel, question_dict)
await asyncio.sleep(1)
- hint_no = 0
- unanswered += 1
- await self.send_score(ctx.channel, self.player_dict[ctx.channel.id])
- await asyncio.sleep(2)
+ hint_no = 0 # init hint_no = 0 so that 2 hints/time alerts can be sent for the new question.
+
+ await self.send_score(ctx.channel, self.game_player_scores[ctx.channel.id])
+ await asyncio.sleep(2)
else:
+ if self.game_status[ctx.channel.id] is False:
+ break
+
+ # Reduce points by 25 for every hint/time alert that has been sent.
points = 100 - 25*hint_no
- if msg.author in self.player_dict[ctx.channel.id]:
- self.player_dict[ctx.channel.id][msg.author] += points
+ if msg.author in self.game_player_scores[ctx.channel.id]:
+ self.game_player_scores[ctx.channel.id][msg.author] += points
+ else:
+ self.game_player_scores[ctx.channel.id][msg.author] = points
+
+ # Also updating the overall scoreboard.
+ if msg.author in self.player_scores:
+ self.player_scores[msg.author] += points
else:
- self.player_dict[ctx.channel.id][msg.author] = points
+ self.player_scores[msg.author] = points
+
hint_no = 0
- unanswered = 0
- await ctx.send(f"{msg.author.mention} got the correct answer :tada: {points} points for ya.")
+
+ await ctx.send(f"{msg.author.mention} got the correct answer :tada: {points} points!")
await self.send_answer(ctx.channel, question_dict)
- await self.send_score(ctx.channel, self.player_dict[ctx.channel.id])
+ await self.send_score(ctx.channel, self.game_player_scores[ctx.channel.id])
await asyncio.sleep(2)
@staticmethod
+ def make_start_embed(category: str) -> discord.Embed:
+ """Generate a starting/introduction embed for the quiz."""
+ start_embed = discord.Embed(colour=discord.Colour.red())
+ start_embed.title = "Quiz game Starting!!"
+ start_embed.description = "Each game consists of 5 questions.\n"
+ start_embed.description += "**Rules :**\nNo cheating and have fun!"
+ start_embed.description += f"\n **Category** : {category}"
+ start_embed.set_footer(
+ text="Points for each question reduces by 25 after 10s or after a hint. Total time is 30s per question"
+ )
+ return start_embed
+
+ @quiz_game.command(name="stop")
+ async def stop_quiz(self, ctx: commands.Context) -> None:
+ """
+ Stop a quiz game if its running in the channel.
+
+ Note: Only mods or the owner of the quiz can stop it.
+ """
+ if self.game_status[ctx.channel.id] is True:
+ # Check if the author is the game starter or a moderator.
+ if (
+ ctx.author == self.game_owners[ctx.channel.id]
+ or any(Roles.moderator == role.id for role in ctx.author.roles)
+ ):
+ await ctx.send("Quiz stopped.")
+ await self.declare_winner(ctx.channel, self.game_player_scores[ctx.channel.id])
+
+ self.game_status[ctx.channel.id] = False
+ del self.game_owners[ctx.channel.id]
+ self.game_player_scores[ctx.channel.id] = {}
+ else:
+ await ctx.send(f"{ctx.author.mention}, you are not authorised to stop this game :ghost:!")
+ else:
+ await ctx.send("No quiz running.")
+
+ @quiz_game.command(name="leaderboard")
+ async def leaderboard(self, ctx: commands.Context) -> None:
+ """View everyone's score for this bot session."""
+ await self.send_score(ctx.channel, self.player_scores)
+
+ @staticmethod
async def send_score(channel: discord.TextChannel, player_data: dict) -> None:
"""A function which sends the score."""
+ if len(player_data) == 0:
+ await channel.send("No one has made it onto the leaderboard yet.")
+ return
+
embed = discord.Embed(colour=discord.Colour.blue())
embed.title = "Score Board"
embed.description = ""
- for k, v in player_data.items():
- embed.description += f"{k} : {v}\n"
+
+ sorted_dict = sorted(player_data.items(), key=lambda a: a[1], reverse=True)
+ for item in sorted_dict:
+ embed.description += f"{item[0]} : {item[1]}\n"
+
await channel.send(embed=embed)
@staticmethod
@@ -185,33 +251,34 @@ class TriviaQuiz(commands.Cog):
word = "You guys"
winners = []
points_copy = list(player_data.values()).copy()
+
for _ in range(no_of_winners):
index = points_copy.index(highest_points)
winners.append(list(player_data.keys())[index])
points_copy[index] = 0
- winners_mention = None
- for winner in winners:
- winners_mention += f"{winner.mention} "
+ winners_mention = " ".join(winner.mention for winner in winners)
else:
word = "You"
author_index = list(player_data.values()).index(highest_points)
winner = list(player_data.keys())[author_index]
winners_mention = winner.mention
+
await channel.send(
- f"Congratz {winners_mention} :tada: "
- f"{word} have won this quiz game with a grand total of {highest_points} points!!"
+ f"Congratulations {winners_mention} :tada: "
+ f"{word} have won this quiz game with a grand total of {highest_points} points!"
)
- @property
def category_embed(self) -> discord.Embed:
"""Build an embed showing all available trivia categories."""
embed = discord.Embed(colour=discord.Colour.blue())
embed.title = "The available question categories are:"
+ embed.set_footer(text="If a category is not chosen, a random one will be selected.")
embed.description = ""
+
for cat, description in self.categories.items():
embed.description += f"**- {cat.capitalize()}**\n{description.capitalize()}\n"
- embed.set_footer(text="If not category is chosen, then a random one will be selected.")
+
return embed
@staticmethod
@@ -222,13 +289,15 @@ class TriviaQuiz(commands.Cog):
embed = discord.Embed(color=discord.Colour.red())
embed.title = f"The correct answer is **{answer}**\n"
embed.description = ""
+
if info != "":
embed.description += f"**Information**\n{info}\n\n"
- embed.description += "Lets move to the next question.\nRemaining questions: "
+
+ embed.description += "Let's move to the next question.\nRemaining questions: "
await channel.send(embed=embed)
def setup(bot: commands.Bot) -> None:
- """Loading the cog."""
+ """Load the cog."""
bot.add_cog(TriviaQuiz(bot))
logger.debug("TriviaQuiz cog loaded")
generated by cgit v1.2.3 (git 2.39.1) at 2025-10-29 23:57:03 +0000