Compare commits
17
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
8c2b3fa487 | ||
|
|
62e89960f7 | ||
|
|
3eed30567d | ||
|
|
565babd1d3 | ||
|
|
455b51a998 | ||
|
|
f136181216 | ||
|
|
463ef48052 | ||
|
|
64004329da | ||
|
|
f497928054 | ||
|
|
0911ebfa45 | ||
|
|
beedf946c9 | ||
|
|
5b31314f02 | ||
|
|
4b15fef30c | ||
|
|
e11a86af7b | ||
|
|
f4f8f426d0 | ||
|
|
b8eb84165e | ||
|
|
273ff757af |
@@ -6,8 +6,7 @@ from nextcord import Embed
|
||||
from nextcord.ext import commands
|
||||
|
||||
from highscores import highscores_data # pylint: disable=import-error
|
||||
from utilities.data_storage import save_highscores_data # pylint: disable=import-error
|
||||
from utilities.data_storage import create_highscore_skeleton # pylint: disable=import-error
|
||||
from utilities.highscores_data import save_highscores_data # pylint: disable=import-error
|
||||
|
||||
|
||||
class DeleteScores(commands.Cog, name="Delete Scores"):
|
||||
@@ -22,7 +21,7 @@ class DeleteScores(commands.Cog, name="Delete Scores"):
|
||||
@commands.command(name="delete-scores")
|
||||
@commands.has_permissions(administrator=True)
|
||||
async def delete_scores(self, ctx: commands.Context):
|
||||
"""A command which registers current channel as for various uses.
|
||||
"""A command which deletes all scores for this guild.
|
||||
Usage:
|
||||
```
|
||||
?delete_scores
|
||||
@@ -31,15 +30,7 @@ class DeleteScores(commands.Cog, name="Delete Scores"):
|
||||
# only allow AlwaysCompiling to use.
|
||||
# If you are a developer and want to use command, can add name to list
|
||||
if ctx.author.id == 206602713511886849:
|
||||
guild_data: dict = highscores_data[str(ctx.guild.id)]
|
||||
|
||||
for boss, boss_data in guild_data["tables"].items():
|
||||
for category, category_data in boss_data["categories"].items():
|
||||
category_data = []
|
||||
boss_data["categories"][category] = category_data
|
||||
guild_data["tables"][boss] = boss_data
|
||||
|
||||
save_highscores_data(highscores_data)
|
||||
highscores_data.reset_guild_data(ctx.guild.id)
|
||||
|
||||
|
||||
# This function will be called when this extension is loaded.
|
||||
|
||||
@@ -2,15 +2,10 @@
|
||||
Defines cog for guild listeners. Contains functions for various guild events.
|
||||
"""
|
||||
|
||||
import os
|
||||
import json
|
||||
|
||||
from nextcord.ext import commands
|
||||
import nextcord
|
||||
|
||||
from highscores import highscores_data # pylint: disable=import-error
|
||||
from utilities.data_storage import create_highscore_skeleton # pylint: disable=import-error
|
||||
from utilities.data_storage import save_highscores_data # pylint: disable=import-error
|
||||
|
||||
|
||||
class GuildListener(commands.Cog, name="Guild Listener"):
|
||||
@@ -22,9 +17,7 @@ class GuildListener(commands.Cog, name="Guild Listener"):
|
||||
@commands.Cog.listener()
|
||||
async def on_guild_join(self, guild: nextcord.Guild):
|
||||
"""Listener for Guild Join. Creates skeleton for new guild joined and saves it."""
|
||||
# turning guild id into string because json doesn't support ints as object keys.
|
||||
highscores_data[str(guild.id)] = create_highscore_skeleton()
|
||||
save_highscores_data(highscores_data)
|
||||
highscores_data.add_new_server(guild.id)
|
||||
|
||||
welcome_message: str = "Hello, I keep track of Highscores for you!"
|
||||
welcome_message += " Currently only supporting RuneScape boss highscores,"
|
||||
@@ -37,14 +30,7 @@ class GuildListener(commands.Cog, name="Guild Listener"):
|
||||
async def on_guild_remove(self, guild: nextcord.Guild):
|
||||
"""Listener for Guild Remove. Removes data from highscores_data."""
|
||||
print(f"Removing data for guild {guild.name} (id {guild.id}).")
|
||||
old_data = highscores_data.pop(str(guild.id))
|
||||
save_highscores_data(highscores_data)
|
||||
|
||||
if not os.path.exists("deleted_backups"):
|
||||
os.mkdir("deleted_backups")
|
||||
|
||||
with open(f"deleted_backups/{guild.name}_backup.json", encoding="utf-8", mode="w") as file:
|
||||
file.write(json.dumps(old_data, indent=4))
|
||||
highscores_data.remove_guild(guild.id, guild.name)
|
||||
|
||||
|
||||
# This function will be called when this extension is loaded.
|
||||
|
||||
@@ -9,7 +9,6 @@ import nextcord
|
||||
|
||||
from cogs.submission.submission_views import SubmissionState # pylint: disable=import-error
|
||||
from cogs.submission.submission_views import get_submission_embed # pylint: disable=import-error
|
||||
from highscores import highscores_config # pylint: disable=import-error
|
||||
from highscores import highscores_data # pylint: disable=import-error
|
||||
from highscores import submission_objects # pylint: disable=import-error
|
||||
|
||||
@@ -24,35 +23,34 @@ class SubmissionListener(commands.Cog, name="Message Listener"):
|
||||
async def on_message(self, message: nextcord.Message):
|
||||
"""Listener for message creation. Adds score and attachments to submission."""
|
||||
guild_id = message.guild.id
|
||||
guild_data = highscores_data[str(guild_id)]
|
||||
if message.author.id in submission_objects:
|
||||
message_id = submission_objects[message.author.id]["message_id"]
|
||||
submission_channel_id = highscores_data.get_guild_submission_channel(guild_id)
|
||||
submission_message: nextcord.Message = (
|
||||
await self.bot.get_guild(guild_id)
|
||||
.get_channel(guild_data["submission_channel_id"])
|
||||
.get_channel(submission_channel_id)
|
||||
.fetch_message(message_id)
|
||||
)
|
||||
|
||||
# need check if correct user and correct channel
|
||||
if message.channel.id == guild_data["submission_channel_id"]:
|
||||
if message.channel.id == submission_channel_id:
|
||||
state = submission_objects[message.author.id]["submission_state"]
|
||||
if state == SubmissionState.SCORE and message.content:
|
||||
|
||||
# check with regex if score matches correctly
|
||||
is_time_record = highscores_config["categories"][
|
||||
is_time_record = highscores_data.get_config_time_record(
|
||||
submission_objects[message.author.id]["category"]
|
||||
]["is_time_record"]
|
||||
)
|
||||
|
||||
time_regex = re.search(
|
||||
"^([0-9]{1,2}:)?[0-9]{2}(\.[0-9])?$", # pylint: disable=anomalous-backslash-in-string
|
||||
message.content,
|
||||
"^([0-9]{1,2}:[0-9]|[0-9])?[0-9](\\.[0-9])?$", message.content
|
||||
)
|
||||
int_regex = re.search("^[0-9]+$", message.content)
|
||||
if time_regex is None and is_time_record:
|
||||
embed = submission_message.embeds[0]
|
||||
embed.add_field(
|
||||
name="Error",
|
||||
value="Time record must follow format MM:SS.ms or SS.ms or SS.",
|
||||
value="Time record must follow format MM:SS.ms, SS.ms, or SS.",
|
||||
)
|
||||
elif int_regex is None and not is_time_record:
|
||||
embed = submission_message.embeds[0]
|
||||
@@ -61,11 +59,18 @@ class SubmissionListener(commands.Cog, name="Message Listener"):
|
||||
score = message.content
|
||||
if is_time_record:
|
||||
# standardize time score format.
|
||||
# strip leading 0s and colon
|
||||
# then add colon if its not there
|
||||
score = score.lstrip("0:")
|
||||
if score.find(":") == -1:
|
||||
if re.search("^[1-9]?[0-9]:[0-9]{2}(\\.[0-9])?$", score):
|
||||
# score is already fine.
|
||||
pass
|
||||
elif re.search("^[0-9]{2}(\\.[0-9])?$", score):
|
||||
# need to add leading colon and 0
|
||||
score = "0:" + score
|
||||
elif re.search("^[0-9](\\.[0-9])?$", score):
|
||||
# need to add 0 in seconds and leading colon and 0
|
||||
score = "0:0" + score
|
||||
elif re.search("^[0]?[0-9]:[0-9]{2}(\\.[0-9])?$", score):
|
||||
# if there is a minute 10ths place 0, remove
|
||||
score = score.lstrip("0")
|
||||
submission_objects[message.author.id]["score"] = score
|
||||
embed = get_submission_embed(message.author.id)
|
||||
|
||||
@@ -78,7 +83,7 @@ class SubmissionListener(commands.Cog, name="Message Listener"):
|
||||
)
|
||||
elif state == SubmissionState.NAME and message.content:
|
||||
submission_objects[message.author.id]["username"] = message.content[
|
||||
: guild_data["username_length"]
|
||||
: highscores_data.get_guild_username_length(guild_id)
|
||||
]
|
||||
embed = get_submission_embed(message.author.id)
|
||||
await submission_message.edit(
|
||||
|
||||
@@ -6,9 +6,7 @@ from nextcord.ext import commands
|
||||
import nextcord
|
||||
|
||||
from cogs.submission.submission_views import SubmissionCreateButton # pylint: disable=import-error
|
||||
from highscores import highscores_config # pylint: disable=import-error
|
||||
from highscores import highscores_data # pylint: disable=import-error
|
||||
from utilities.data_storage import save_highscores_data # pylint: disable=import-error
|
||||
from utilities.utils import send_highscore_message # pylint: disable=import-error
|
||||
|
||||
|
||||
@@ -29,24 +27,18 @@ class PostHighscores(commands.Cog, name="Post Highscores"):
|
||||
?post-scores
|
||||
```
|
||||
"""
|
||||
guild_data = highscores_data[str(ctx.guild.id)]
|
||||
highscore_channel_id = guild_data["highscore_channel_id"]
|
||||
highscore_channel_id = highscores_data.get_guild_highscores_channel(ctx.guild.id)
|
||||
print(f"Printing out highscores information in channel {highscore_channel_id}")
|
||||
channel = ctx.guild.get_channel(highscore_channel_id)
|
||||
|
||||
error_response = 'Registered Highscores channel does not exist or was never registered. \n\
|
||||
Register with "?register" command.'
|
||||
error_response = "Registered Highscores channel does not exist or was never registered. "
|
||||
error_response += 'Register with "?register" command.'
|
||||
assert channel is not None, await ctx.send(error_response)
|
||||
|
||||
for boss, categories in highscores_config[ # pylint: disable=unused-variable
|
||||
"tables"
|
||||
].items():
|
||||
message: nextcord.Message = await send_highscore_message(channel, boss, guild_data)
|
||||
for table in highscores_data.get_table_names(ctx.guild.id):
|
||||
message: nextcord.Message = await send_highscore_message(channel, table, ctx.guild.id)
|
||||
await message.edit(view=SubmissionCreateButton(self.bot))
|
||||
|
||||
# Saving in case new message was sent.
|
||||
save_highscores_data(highscores_data)
|
||||
|
||||
|
||||
# This function will be called when this extension is loaded.
|
||||
# It is necessary to add these functions to the bot.
|
||||
|
||||
@@ -0,0 +1,53 @@
|
||||
"""
|
||||
Functions for controlling the bot's presence.
|
||||
"""
|
||||
|
||||
import random
|
||||
|
||||
from nextcord.ext import commands
|
||||
from nextcord.ext import tasks
|
||||
import nextcord
|
||||
|
||||
|
||||
class BotPresence(commands.Cog, name="Bot Presence"):
|
||||
"""Class for bot presence functions."""
|
||||
|
||||
def __init__(self, bot: commands.Bot):
|
||||
self.bot = bot
|
||||
|
||||
@commands.Cog.listener()
|
||||
async def on_guild_join(self):
|
||||
"""Listener for Guild Join. Updated presence on guild join."""
|
||||
await self.change_presence()
|
||||
|
||||
@commands.Cog.listener()
|
||||
async def on_ready(self):
|
||||
"""Starts the loop when bot is ready."""
|
||||
await self.bot.wait_until_ready()
|
||||
# pylint doesn't recognize that periodic_change_presense is subclassed from tasks.loop.
|
||||
self.periodic_change_presence.start() # pylint: disable=no-member
|
||||
|
||||
@tasks.loop(minutes=2)
|
||||
async def periodic_change_presence(self):
|
||||
"""Loops every 2 minutes and changes presence."""
|
||||
await self.change_presence()
|
||||
|
||||
async def change_presence(self):
|
||||
"""Randomly chooses a presence and updates bot's presence."""
|
||||
statuses = [
|
||||
f"?help | {len(self.bot.guilds)} servers",
|
||||
f"Vuln Pls. | {len(self.bot.guilds)} servers",
|
||||
f"All your base are belong to us. | {len(self.bot.guilds)} servers",
|
||||
f"Double plus good. | {len(self.bot.guilds)} servers",
|
||||
]
|
||||
|
||||
status = random.choice(statuses)
|
||||
|
||||
await self.bot.change_presence(activity=nextcord.Game(status))
|
||||
|
||||
|
||||
# This function will be called when this extension is loaded.
|
||||
# It is necessary to add these functions to the bot.
|
||||
def setup(bot: commands.Bot):
|
||||
"""Adds function to bot"""
|
||||
bot.add_cog(BotPresence(bot))
|
||||
@@ -6,7 +6,6 @@ from nextcord import Embed
|
||||
from nextcord.ext import commands
|
||||
|
||||
from highscores import highscores_data # pylint: disable=import-error
|
||||
from utilities.data_storage import save_highscores_data # pylint: disable=import-error
|
||||
|
||||
|
||||
class RegisterVerificationChannel(commands.Cog, name="Register"):
|
||||
@@ -29,17 +28,16 @@ class RegisterVerificationChannel(commands.Cog, name="Register"):
|
||||
"""
|
||||
# respond to the message
|
||||
channel_id = ctx.channel.id
|
||||
guild_data = highscores_data[str(ctx.guild.id)]
|
||||
|
||||
if registration_type == "verify":
|
||||
await ctx.send(f"Registered <#{channel_id}> as channel to post verifications in.")
|
||||
guild_data["verification_channel_id"] = channel_id
|
||||
highscores_data.set_guild_verification_channel(ctx.guild.id, channel_id)
|
||||
elif registration_type == "highscore":
|
||||
await ctx.send(f"Registered <#{channel_id}> as channel to post highscores in.")
|
||||
guild_data["highscore_channel_id"] = channel_id
|
||||
highscores_data.set_guild_highscore_channel(ctx.guild.id, channel_id)
|
||||
elif registration_type == "submission":
|
||||
await ctx.send(f"Registered <#{channel_id}> as channel to post submissions in.")
|
||||
guild_data["submission_channel_id"] = channel_id
|
||||
highscores_data.set_guild_submission_channel(ctx.guild.id, channel_id)
|
||||
else:
|
||||
embed = Embed(
|
||||
title="Register command help", description="How to use the register command."
|
||||
@@ -58,8 +56,6 @@ class RegisterVerificationChannel(commands.Cog, name="Register"):
|
||||
embed.add_field(name="?register help", value="Use to show this message.")
|
||||
await ctx.reply(embed=embed)
|
||||
|
||||
save_highscores_data(highscores_data)
|
||||
|
||||
|
||||
# This function will be called when this extension is loaded.
|
||||
# It is necessary to add these functions to the bot.
|
||||
|
||||
@@ -7,7 +7,6 @@ import nextcord
|
||||
from nextcord import Embed
|
||||
from nextcord.ext import commands
|
||||
|
||||
from highscores import highscores_config # pylint: disable=import-error
|
||||
from highscores import highscores_data # pylint: disable=import-error
|
||||
from highscores import highscores_message_map # pylint: disable=import-error
|
||||
from highscores import submission_objects # pylint: disable=import-error
|
||||
@@ -53,10 +52,10 @@ def get_submission_embed(userid: int) -> nextcord.Embed:
|
||||
return embed
|
||||
|
||||
|
||||
def category_select_options(boss_name: str) -> "list[nextcord.SelectOption]":
|
||||
def category_select_options(guild_id: int, boss_name: str) -> "list[nextcord.SelectOption]":
|
||||
"""Returns list of categories for a given boss."""
|
||||
options: "list[nextcord.SelectOption]" = []
|
||||
category_list = highscores_config["tables"][boss_name]
|
||||
category_list = highscores_data.get_table_categories(guild_id, boss_name)
|
||||
for category in category_list:
|
||||
options.append(nextcord.SelectOption(label=category))
|
||||
|
||||
@@ -149,9 +148,9 @@ class SubmitButton(nextcord.ui.Button):
|
||||
self._bot = bot
|
||||
|
||||
async def callback(self, interaction: nextcord.Interaction):
|
||||
verification_channel_id = highscores_data[str(interaction.guild.id)][
|
||||
"verification_channel_id"
|
||||
]
|
||||
verification_channel_id = highscores_data.get_guild_verification_channel(
|
||||
interaction.guild_id
|
||||
)
|
||||
channel = interaction.guild.get_channel(verification_channel_id)
|
||||
|
||||
error_response = "Registered Verification channel does not exist or was never registered. \
|
||||
@@ -194,23 +193,23 @@ class SubmissionCreateButton(nextcord.ui.View):
|
||||
await interaction.send("Please finish in progress submission first.", ephemeral=True)
|
||||
return
|
||||
|
||||
guild_id_str = str(interaction.guild.id)
|
||||
guild_data = highscores_data[guild_id_str]
|
||||
|
||||
boss_name = highscores_message_map[guild_id_str][interaction.message.id]
|
||||
channel = interaction.guild.get_channel(guild_data["submission_channel_id"])
|
||||
boss_name = highscores_message_map[interaction.guild.id][interaction.message.id]
|
||||
channel = interaction.guild.get_channel(
|
||||
highscores_data.get_guild_submission_channel(interaction.guild_id)
|
||||
)
|
||||
|
||||
# a user should only have 1 submission active at once.
|
||||
# submitting with dummy message_id until message created
|
||||
# split username on pipe.
|
||||
submission_objects[interaction.user.id] = {
|
||||
"message_id": 0,
|
||||
"guild_id": interaction.guild_id,
|
||||
"submission_state": SubmissionState.CATEGORY,
|
||||
"boss_name": boss_name,
|
||||
"score": "",
|
||||
"category": "",
|
||||
"username": interaction.user.display_name.split("|")[0][
|
||||
: guild_data["username_length"]
|
||||
: highscores_data.get_guild_username_length(interaction.guild_id)
|
||||
],
|
||||
}
|
||||
|
||||
@@ -244,10 +243,13 @@ class SubmissionView(nextcord.ui.View):
|
||||
self._bot = bot
|
||||
self._state = submission_objects[self._submission_user]["submission_state"]
|
||||
self._boss_name = submission_objects[self._submission_user]["boss_name"]
|
||||
self._guild_id = submission_objects[self._submission_user]["guild_id"]
|
||||
|
||||
# add items depending on given state
|
||||
if self._state is SubmissionState.CATEGORY:
|
||||
self.add_item(SubmissionDropdown(bot, category_select_options(self._boss_name)))
|
||||
self.add_item(
|
||||
SubmissionDropdown(bot, category_select_options(self._guild_id, self._boss_name))
|
||||
)
|
||||
self.add_item(CancelButton())
|
||||
self.add_item(NextButton(self._bot))
|
||||
elif (
|
||||
|
||||
+9
-12
@@ -9,9 +9,7 @@ import nextcord
|
||||
from nextcord.ext import commands
|
||||
|
||||
import config
|
||||
from utilities.data_storage import open_highscores_config
|
||||
from utilities.data_storage import open_highscores_data
|
||||
from utilities.data_storage import create_message_map
|
||||
from utilities.highscores_data import HighscoresData
|
||||
|
||||
logger = logging.getLogger("nextcord")
|
||||
logger.setLevel(logging.DEBUG)
|
||||
@@ -20,9 +18,14 @@ handler.setFormatter(logging.Formatter("%(asctime)s:%(levelname)s:%(name)s: %(me
|
||||
logger.addHandler(handler)
|
||||
|
||||
|
||||
highscores_config = open_highscores_config()
|
||||
highscores_data = open_highscores_data()
|
||||
highscores_message_map = create_message_map(highscores_data)
|
||||
highscores_data: HighscoresData = HighscoresData()
|
||||
highscores_message_map = {}
|
||||
|
||||
for guild in highscores_data.get_guilds():
|
||||
highscores_message_map[guild] = {}
|
||||
for table in highscores_data.get_table_names(guild):
|
||||
message_id = highscores_data.get_table_message_id(guild, table)
|
||||
highscores_message_map[guild][message_id] = table
|
||||
|
||||
submission_objects: dict = {}
|
||||
|
||||
@@ -42,15 +45,9 @@ def main():
|
||||
# To enable the members priveleged intent:
|
||||
# intents.members = True
|
||||
|
||||
# Set custom status to "Listening to ?help"
|
||||
activity = nextcord.Activity(
|
||||
type=nextcord.ActivityType.listening, name=f"{config.BOT_PREFIX}help"
|
||||
)
|
||||
|
||||
bot = commands.Bot(
|
||||
commands.when_mentioned_or(config.BOT_PREFIX),
|
||||
intents=intents,
|
||||
activity=activity,
|
||||
)
|
||||
|
||||
# Get the modules of all cogs whose directory structure is ./cogs/<module_name>
|
||||
|
||||
@@ -1,79 +0,0 @@
|
||||
"""Functions for retreiving and storing highscores data and configurations."""
|
||||
|
||||
import os
|
||||
import json
|
||||
|
||||
HIGHSCORES_CONFIG_FILE = "highscores_config.json"
|
||||
HIGHSCORES_DATA_FILE = "highscores_data.json"
|
||||
|
||||
|
||||
def create_highscore_skeleton() -> dict:
|
||||
"""
|
||||
Creates a skeleton entry for new highscores data
|
||||
|
||||
@return: dict skeleton entry
|
||||
"""
|
||||
# create the dict
|
||||
skeleton = {"highscore_channel_id": -1, "verification_channel_id": -1, "username_length": 12}
|
||||
config: dict = open_highscores_config()
|
||||
skeleton["tables"] = {}
|
||||
for boss, categories in config["tables"].items():
|
||||
skeleton["tables"][boss] = {"message_id": 0}
|
||||
boss_categories = {}
|
||||
for category in categories:
|
||||
rank_list = []
|
||||
boss_categories[category] = rank_list
|
||||
skeleton["tables"][boss]["categories"] = boss_categories
|
||||
|
||||
return skeleton
|
||||
|
||||
|
||||
def create_message_map(data: dict):
|
||||
"""
|
||||
Creates a message map from input highscores data.
|
||||
|
||||
@param data: highscores_data dict to create map from.
|
||||
@return: dict message_map of message to boss.
|
||||
"""
|
||||
|
||||
message_map = {}
|
||||
|
||||
# guild id will be a string. Json object keys are strings.
|
||||
for guild, guild_scores in data.items():
|
||||
message_map[guild] = {}
|
||||
for boss, info in guild_scores["tables"].items():
|
||||
message_map[guild][info["message_id"]] = boss
|
||||
|
||||
return message_map
|
||||
|
||||
|
||||
def open_highscores_config():
|
||||
"""
|
||||
Opens the highscores config file and returns.
|
||||
@return: dict of highscores config.
|
||||
"""
|
||||
# Load bot config
|
||||
with open(HIGHSCORES_CONFIG_FILE, encoding="utf-8", mode="r") as file:
|
||||
return json.load(file)
|
||||
|
||||
|
||||
def open_highscores_data():
|
||||
"""
|
||||
Opens a highscores_data.json file if one exists, otherwise returns empty dict.
|
||||
@return: dict of highscores data.
|
||||
"""
|
||||
if os.path.exists(HIGHSCORES_DATA_FILE):
|
||||
with open(HIGHSCORES_DATA_FILE, encoding="utf-8", mode="r") as file:
|
||||
data = json.load(file)
|
||||
else:
|
||||
data = {}
|
||||
return data
|
||||
|
||||
|
||||
def save_highscores_data(data: dict):
|
||||
"""
|
||||
Saves the highscores data dict into a json file.
|
||||
@param data: dict of highscores data.
|
||||
"""
|
||||
with open(HIGHSCORES_DATA_FILE, encoding="utf-8", mode="w") as file:
|
||||
file.write(json.dumps(data, indent=4))
|
||||
@@ -0,0 +1,327 @@
|
||||
"""Class and functions for storing and retrieving highscores data."""
|
||||
|
||||
import os
|
||||
import json
|
||||
|
||||
HIGHSCORES_CONFIG_FILE = "highscores_config.json"
|
||||
HIGHSCORES_DATA_FILE = "highscores_data.json"
|
||||
|
||||
|
||||
class HighscoresData:
|
||||
"""Class for storing and retrieving highscores data."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._config = open_highscores_config()
|
||||
self._data = open_highscores_data()
|
||||
|
||||
def add_new_server(self, guild_id: int):
|
||||
"""
|
||||
Creates new skeleton data dict and adds to self for new guild.
|
||||
|
||||
@param guild_id: guild_id to add.
|
||||
"""
|
||||
# create the dict
|
||||
skeleton = {
|
||||
"highscore_channel_id": -1,
|
||||
"verification_channel_id": -1,
|
||||
"username_length": 12,
|
||||
}
|
||||
config: dict = open_highscores_config()
|
||||
skeleton["tables"] = {}
|
||||
for boss, categories in config["tables"].items():
|
||||
skeleton["tables"][boss] = {"message_id": 0}
|
||||
boss_categories = {}
|
||||
for category in categories:
|
||||
rank_list = []
|
||||
boss_categories[category] = rank_list
|
||||
skeleton["tables"][boss]["categories"] = boss_categories
|
||||
|
||||
self._data[str(guild_id)] = skeleton
|
||||
|
||||
def add_score(
|
||||
self, guild_id: int, boss_name: str, category: str, user: str, score: str
|
||||
) -> None:
|
||||
"""
|
||||
Adds a new score to guilds highscore data.
|
||||
Ensures that there is no user duplicate and then sorts the scores
|
||||
|
||||
@param guild_id: guild id to retrieve from.
|
||||
@param boss_name: boss to add score to.
|
||||
@param category: category to add score to.
|
||||
@param user: user that submitted score.
|
||||
@param score: score to add.
|
||||
"""
|
||||
scores: "list[tuple]" = self._data[str(guild_id)]["tables"][boss_name]["categories"][
|
||||
category
|
||||
]
|
||||
ascending: bool = self._config["categories"][category]["ascending"]
|
||||
time_record: bool = self._config["categories"][category]["is_time_record"]
|
||||
|
||||
# create score tuple for either time or int
|
||||
# define sort index
|
||||
if time_record:
|
||||
if score.find(":") == -1:
|
||||
score_seconds = float(score)
|
||||
else:
|
||||
minutes = score.split(":")[0]
|
||||
seconds = score.split(":")[1]
|
||||
score_seconds = int(minutes) * 60 + float(seconds)
|
||||
score_tuple = (user, score, score_seconds)
|
||||
sort_index = 2
|
||||
else:
|
||||
score_tuple = (user, int(score))
|
||||
sort_index = 1
|
||||
|
||||
# if user is in an existing tuple, delete it.
|
||||
users_scores: "list[tuple]" = [score_tuple]
|
||||
for tup in scores:
|
||||
if user == tup[0]:
|
||||
# Add tup to users_scores, remove from scores
|
||||
users_scores.append(tup)
|
||||
|
||||
# sort users_scores
|
||||
users_scores.sort(
|
||||
key=lambda x: x[sort_index],
|
||||
reverse=not ascending,
|
||||
)
|
||||
|
||||
for index, tup in enumerate(users_scores):
|
||||
if index == 0:
|
||||
# first score is the best score.
|
||||
if tup not in scores:
|
||||
scores.append(tup)
|
||||
else:
|
||||
# discard all other scores from scores list.
|
||||
if tup in scores:
|
||||
scores.remove(tup)
|
||||
|
||||
# sort scores.
|
||||
scores.sort(
|
||||
key=lambda x: x[sort_index],
|
||||
reverse=not ascending,
|
||||
)
|
||||
|
||||
self._data[str(guild_id)]["tables"][boss_name]["categories"][category] = scores
|
||||
save_highscores_data(self._data)
|
||||
|
||||
def get_categories_data(self, guild_id: int, table: str) -> "list[tuple]":
|
||||
"""
|
||||
Returns categories data in a list of tuples. Calls dict function items().
|
||||
|
||||
@param guild_id: guild's id to retrieve from.
|
||||
@param table: table to get category data from
|
||||
@returns: list of tuples containing all category and relative score data.
|
||||
"""
|
||||
return self._data[str(guild_id)]["tables"][table]["categories"].items()
|
||||
|
||||
def get_config_highscore_size(self) -> int:
|
||||
"""
|
||||
Gets the defined size of highscore for config.
|
||||
|
||||
@returns: int highscore size
|
||||
"""
|
||||
return self._config["highscore_size"]
|
||||
|
||||
def get_config_time_record(self, category: str) -> bool:
|
||||
"""
|
||||
Tells if the category is a time record
|
||||
|
||||
@param category: category name to retrieve from.
|
||||
@returns: bool indicating time record or not.
|
||||
"""
|
||||
return self._config["categories"][category]["is_time_record"]
|
||||
|
||||
def get_guilds(self) -> "list[int]":
|
||||
"""
|
||||
Gets a list of guilds tracked by data
|
||||
|
||||
@returns: list of ints representing guild ids
|
||||
"""
|
||||
guilds: "list[int]" = []
|
||||
|
||||
for guild, data in self._data.items(): # pylint: disable=unused-variable
|
||||
guilds.append(int(guild))
|
||||
|
||||
return guilds
|
||||
|
||||
def get_guild_highscores_channel(self, guild_id: int) -> int:
|
||||
"""
|
||||
Returns the channel id to post highscore messages to.
|
||||
|
||||
@param guild_id: guild id to retrieve from.
|
||||
@returns: channel id of highscores channel.
|
||||
"""
|
||||
return self._data[str(guild_id)]["highscore_channel_id"]
|
||||
|
||||
def get_guild_submission_channel(self, guild_id: int) -> int:
|
||||
"""
|
||||
Returns the channel id to post submission messages to.
|
||||
|
||||
@param guild_id: guild id to retrieve from.
|
||||
@returns: channel id of submission channel.
|
||||
"""
|
||||
return self._data[str(guild_id)]["submission_channel_id"]
|
||||
|
||||
def get_guild_username_length(self, guild_id: int) -> int:
|
||||
"""
|
||||
Returns username length as set by each guild.
|
||||
|
||||
@param guild_id: guild id to retrieve from.
|
||||
@returns: int of username length.
|
||||
"""
|
||||
return self._data[str(guild_id)]["username_length"]
|
||||
|
||||
def get_guild_verification_channel(self, guild_id: int) -> int:
|
||||
"""
|
||||
Returns the channel id to post verification messages to.
|
||||
|
||||
@param guild_id: guild id to retrieve from.
|
||||
@returns: channel id of verification channel.
|
||||
"""
|
||||
return self._data[str(guild_id)]["verification_channel_id"]
|
||||
|
||||
def get_table_categories(self, guild_id: int, table: str) -> "list[str]":
|
||||
"""
|
||||
Returns list of categories for a guild's table
|
||||
|
||||
@param guild_id: guild id to retrieve from.
|
||||
@param table: the table to retreive from.
|
||||
@returns: string list of categories in given table.
|
||||
"""
|
||||
categories: "list[str]" = []
|
||||
|
||||
for category, data in self._data[str(guild_id)][ # pylint: disable=unused-variable
|
||||
"tables"
|
||||
][table]["categories"].items():
|
||||
categories.append(category)
|
||||
|
||||
return categories
|
||||
|
||||
def get_table_message_id(self, guild_id: int, table: str) -> int:
|
||||
"""
|
||||
Returns the message id for given guild's table.
|
||||
|
||||
@param guild_id: guild id to retrieve from.
|
||||
@param table: the table to retrieve from.
|
||||
@returns: int id of message for table's highscores.
|
||||
"""
|
||||
return self._data[str(guild_id)]["tables"][table]["message_id"]
|
||||
|
||||
def get_table_names(self, guild_id) -> "list[str]":
|
||||
"""
|
||||
Returns a list of table names.
|
||||
|
||||
@param guild_id: guild id to retrieve from.
|
||||
@returns: list of table names
|
||||
"""
|
||||
tables: "list[str]" = []
|
||||
|
||||
for table, data in self._data[str(guild_id)][ # pylint: disable=unused-variable
|
||||
"tables"
|
||||
].items():
|
||||
tables.append(table)
|
||||
|
||||
return tables
|
||||
|
||||
def remove_guild(self, guild_id: int, guild_name: str) -> None:
|
||||
"""
|
||||
Removes a guild's data from object. Saves backup.
|
||||
|
||||
@param guild_id: guild id to remove.
|
||||
@param guild_name: name of guild for backup.
|
||||
"""
|
||||
old_data = self._data.pop(str(guild_id))
|
||||
save_highscores_data(self._data)
|
||||
|
||||
if not os.path.exists("deleted_backups"):
|
||||
os.mkdir("deleted_backups")
|
||||
|
||||
with open(f"deleted_backups/{guild_name}_backup.json", encoding="utf-8", mode="w") as file:
|
||||
file.write(json.dumps(old_data, indent=4))
|
||||
|
||||
def reset_guild_data(self, guild_id: int) -> None:
|
||||
"""
|
||||
Resets highscore data for a guild. Should only be used for development really...
|
||||
|
||||
@param guild_id: guild id to reset data for
|
||||
"""
|
||||
guild_data: dict = self._data[str(guild_id)]
|
||||
|
||||
for boss, boss_data in guild_data["tables"].items():
|
||||
for category, category_data in boss_data["categories"].items():
|
||||
category_data = []
|
||||
boss_data["categories"][category] = category_data
|
||||
guild_data["tables"][boss] = boss_data
|
||||
self._data[str(guild_id)] = guild_data
|
||||
save_highscores_data(self._data)
|
||||
|
||||
def set_guild_highscore_channel(self, guild_id: int, channel_id: int) -> None:
|
||||
"""
|
||||
Sets the guild's highscores channel.
|
||||
|
||||
@param guild_id: guild id to set to.
|
||||
@param channel_id: channel id to set.
|
||||
"""
|
||||
self._data[str(guild_id)]["highscore_channel_id"] = channel_id
|
||||
|
||||
def set_guild_submission_channel(self, guild_id: int, channel_id: int) -> None:
|
||||
"""
|
||||
Sets the guild's submission channel.
|
||||
|
||||
@param guild_id: guild id to set to.
|
||||
@param channel_id: channel id to set.
|
||||
"""
|
||||
self._data[str(guild_id)]["submission_channel_id"] = channel_id
|
||||
|
||||
def set_guild_verification_channel(self, guild_id: int, channel_id: int) -> None:
|
||||
"""
|
||||
Sets the guild's verification channel.
|
||||
|
||||
@param guild_id: guild id to set to.
|
||||
@param channel_id: channel id to set.
|
||||
"""
|
||||
self._data[str(guild_id)]["verification_channel_id"] = channel_id
|
||||
|
||||
def set_table_message_id(self, guild_id: int, table: str, message_id: int) -> None:
|
||||
"""
|
||||
Sets the message id for given guild's table.
|
||||
Saves highscores data.
|
||||
|
||||
@param guild_id: guild id to set to.
|
||||
@param table: the table to set to.
|
||||
@param message_id: message id to set.
|
||||
"""
|
||||
self._data[str(guild_id)]["tables"][table]["message_id"] = message_id
|
||||
save_highscores_data(self._data)
|
||||
|
||||
|
||||
def open_highscores_config():
|
||||
"""
|
||||
Opens the highscores config file and returns.
|
||||
@return: dict of highscores config.
|
||||
"""
|
||||
# Load bot config
|
||||
with open(HIGHSCORES_CONFIG_FILE, encoding="utf-8", mode="r") as file:
|
||||
return json.load(file)
|
||||
|
||||
|
||||
def open_highscores_data():
|
||||
"""
|
||||
Opens a highscores_data.json file if one exists, otherwise returns empty dict.
|
||||
@return: dict of highscores data.
|
||||
"""
|
||||
if os.path.exists(HIGHSCORES_DATA_FILE):
|
||||
with open(HIGHSCORES_DATA_FILE, encoding="utf-8", mode="r") as file:
|
||||
data = json.load(file)
|
||||
else:
|
||||
data = {}
|
||||
return data
|
||||
|
||||
|
||||
def save_highscores_data(data: dict):
|
||||
"""
|
||||
Saves the highscores data dict into a json file.
|
||||
@param data: dict of highscores data.
|
||||
"""
|
||||
with open(HIGHSCORES_DATA_FILE, encoding="utf-8", mode="w") as file:
|
||||
file.write(json.dumps(data, indent=4))
|
||||
+20
-73
@@ -3,42 +3,41 @@ Contains various utlity functions for highscores bot
|
||||
"""
|
||||
|
||||
import nextcord
|
||||
from highscores import highscores_config # pylint: disable=import-error
|
||||
from highscores import highscores_data # pylint: disable=import-error
|
||||
from highscores import highscores_message_map # pylint: disable=import-error
|
||||
from utilities.data_storage import save_highscores_data # pylint: disable=import-error
|
||||
|
||||
|
||||
def format_highscore_message(boss_name: str, guild_data: dict):
|
||||
def format_highscore_message(boss_name: str, guild_id: int):
|
||||
"""
|
||||
returns a formatted string for a boss's highscore
|
||||
@param boss: the dictionary object for the boss
|
||||
@param guild_data: the guild's highscore data
|
||||
@param guild_id: the guild id the message is for
|
||||
@return the formatted string
|
||||
"""
|
||||
ret_string = f"**{boss_name}**\n```"
|
||||
boss_data = guild_data["tables"][boss_name]
|
||||
|
||||
# Gather all categories and add to string
|
||||
for key, value in boss_data["categories"].items():
|
||||
for key, value in highscores_data.get_categories_data(guild_id, boss_name):
|
||||
ret_string += f"{key}".ljust(26, " ")
|
||||
ret_string += "\n"
|
||||
|
||||
highscore_size = highscores_config["highscore_size"]
|
||||
highscore_size = highscores_data.get_config_highscore_size()
|
||||
|
||||
# list rankings in a single line. Start with first place for each category and then continue
|
||||
for i in range(highscore_size):
|
||||
for key, value in boss_data["categories"].items():
|
||||
for key, value in highscores_data.get_categories_data(guild_id, boss_name):
|
||||
if len(value) > i:
|
||||
score_pair = value[i]
|
||||
|
||||
# pad username to max length.
|
||||
username: str = score_pair[0].ljust(guild_data["username_length"], " ")
|
||||
username: str = score_pair[0].ljust(
|
||||
highscores_data.get_guild_username_length(guild_id), " "
|
||||
)
|
||||
score: str = str(score_pair[1])
|
||||
|
||||
# pad score to correct length
|
||||
# MM:SS.ms is 7 characters long
|
||||
if highscores_config["categories"][key]["is_time_record"]:
|
||||
if highscores_data.get_config_time_record(key):
|
||||
# if there are single digit minutes, add a space for formatting.
|
||||
if score.find(":") and len(score.split(":", maxsplit=1)[0]) < 2:
|
||||
score = " " + score
|
||||
@@ -54,19 +53,19 @@ def format_highscore_message(boss_name: str, guild_data: dict):
|
||||
|
||||
|
||||
async def send_highscore_message(
|
||||
channel: nextcord.TextChannel, boss_name: str, guild_data: dict
|
||||
channel: nextcord.TextChannel, boss_name: str, guild_id: int
|
||||
) -> nextcord.Message:
|
||||
"""
|
||||
Function takes care of checking if a message exists for a boss, then
|
||||
creating and sending/editing the message
|
||||
@param channel: the channel to send the message in/search for message id
|
||||
@param boss: the boss to send the message about
|
||||
@param guild_data: the guild's highscore data
|
||||
@param guild_id: the guild id to send message to
|
||||
@return: The nextcord message
|
||||
"""
|
||||
highscore_string = format_highscore_message(boss_name, guild_data)
|
||||
highscore_string = format_highscore_message(boss_name, guild_id)
|
||||
|
||||
message_id = guild_data["tables"][boss_name]["message_id"]
|
||||
message_id = highscores_data.get_table_message_id(guild_id, boss_name)
|
||||
try:
|
||||
message: nextcord.Message = await channel.fetch_message(message_id)
|
||||
await message.edit(content=highscore_string)
|
||||
@@ -74,14 +73,14 @@ async def send_highscore_message(
|
||||
# message doesn't exist.
|
||||
message: nextcord.Message = await channel.send(highscore_string)
|
||||
message_id = message.id
|
||||
guild_data["tables"][boss_name]["message_id"] = message_id
|
||||
highscores_data.set_table_message_id(guild_id, boss_name, message_id)
|
||||
# save message -> boss_name in message_map
|
||||
# get guild id from channel
|
||||
# channel should come from the guild that is requesting this highscores message anyways.
|
||||
# check if guild is in message map already
|
||||
if not str(channel.guild.id) in highscores_message_map:
|
||||
highscores_message_map[str(channel.guild.id)] = {}
|
||||
highscores_message_map[str(channel.guild.id)][message_id] = boss_name
|
||||
if not channel.guild.id in highscores_message_map:
|
||||
highscores_message_map[channel.guild.id] = {}
|
||||
highscores_message_map[channel.guild.id][message_id] = boss_name
|
||||
|
||||
return message
|
||||
|
||||
@@ -97,58 +96,10 @@ async def submit_score(
|
||||
@param score: score to add
|
||||
"""
|
||||
|
||||
boss_category_config = highscores_config["categories"][category]
|
||||
guild_data = highscores_data[str(interaction.guild.id)]
|
||||
|
||||
# create score tuple for either time or int
|
||||
# define sort index
|
||||
if boss_category_config["is_time_record"]:
|
||||
if score.find(":") == -1:
|
||||
score_seconds = float(score)
|
||||
else:
|
||||
minutes = score.split(":")[0]
|
||||
seconds = score.split(":")[1]
|
||||
score_seconds = int(minutes) * 60 + float(seconds)
|
||||
score_tuple = (user, score, score_seconds)
|
||||
sort_index = 2
|
||||
|
||||
else:
|
||||
score_tuple = (user, int(score))
|
||||
sort_index = 1
|
||||
|
||||
scores: "list[tuple]" = guild_data["tables"][boss_name]["categories"][category]
|
||||
|
||||
# if user is in an existing tuple, delete it.
|
||||
users_scores: "list[tuple]" = [score_tuple]
|
||||
for tup in scores:
|
||||
if user == tup[0]:
|
||||
# Add tup to users_scores, remove from scores
|
||||
users_scores.append(tup)
|
||||
|
||||
# sort users_scores
|
||||
users_scores.sort(
|
||||
key=lambda x: x[sort_index],
|
||||
reverse=not boss_category_config["ascending"],
|
||||
)
|
||||
|
||||
for index, tup in enumerate(users_scores):
|
||||
if index == 0:
|
||||
# first score is the best score.
|
||||
if tup not in scores:
|
||||
scores.append(tup)
|
||||
else:
|
||||
# discard all other scores from scores list.
|
||||
if tup in scores:
|
||||
scores.remove(tup)
|
||||
|
||||
# sort scores.
|
||||
scores.sort(
|
||||
key=lambda x: x[sort_index],
|
||||
reverse=not boss_category_config["ascending"],
|
||||
)
|
||||
highscores_data.add_score(interaction.guild_id, boss_name, category, user, score)
|
||||
|
||||
# edit message
|
||||
highscore_channel_id = guild_data["highscore_channel_id"]
|
||||
highscore_channel_id = highscores_data.get_guild_highscores_channel(interaction.guild_id)
|
||||
print(f"Editting message {highscore_channel_id}")
|
||||
channel = interaction.guild.get_channel(highscore_channel_id)
|
||||
|
||||
@@ -156,8 +107,4 @@ async def submit_score(
|
||||
error_response += 'Register with "?register" command.'
|
||||
assert channel is not None, await interaction.send(error_response, ephemeral=True)
|
||||
|
||||
await send_highscore_message(channel, boss_name, guild_data)
|
||||
|
||||
# save data
|
||||
highscores_data[str(interaction.guild.id)] = guild_data
|
||||
save_highscores_data(highscores_data)
|
||||
await send_highscore_message(channel, boss_name, interaction.guild_id)
|
||||
|
||||
Reference in New Issue
Block a user