Switch from using raw levelling db modification to library-based modification

cogs/economy.py

@@ -8,10 +8,10 @@
 import utils.logger
 import asyncio
 import framework.isobot.currency
+from framework.isobot.db import levelling
 from random import randint
 from discord import option, ApplicationContext
 from discord.ext import commands
-from cogs.levelling import get_level
 
 # Classes
 class ShopData:
@@ -32,6 +32,7 @@ def get_item_names(self) -> list:
 # Variables
 color = discord.Color.random()
 currency = framework.isobot.currency.CurrencyAPI("database/currency.json", "logs/currency.log")
+levelling = levelling.Levelling()
 shop_data = ShopData("config/shop.json")
 all_item_ids = shop_data.get_item_ids()
 jobs = [
@@ -463,12 +464,12 @@ async def work_list(self, ctx: ApplicationContext):
     @commands.cooldown(1, 1800, commands.BucketType.user)
     async def work_select(self, ctx: ApplicationContext, job: str):
         if job not in jobs: return await ctx.respond(f"This job does not exist. What kind of a job is even {job}??", ephemeral=True)
-        if job == "YouTuber" and get_level(ctx.author.id) < 3: return await ctx.respond("You currently do not have the required level to perform this job!", ephemeral=True)
-        elif job == "Streamer" and get_level(ctx.author.id) < 5: return await ctx.respond("You currently do not have the required level to perform this job!", ephemeral=True)
-        elif job == "Developer" and get_level(ctx.author.id) < 10: return await ctx.respond("You currently do not have the required level to perform this job!", ephemeral=True)
-        elif job == "Scientist" and get_level(ctx.author.id) < 20: return await ctx.respond("You currently do not have the required level to perform this job!", ephemeral=True)
-        elif job == "Engineer" and get_level(ctx.author.id) < 25: return await ctx.respond("You currently do not have the required level to perform this job!", ephemeral=True)
-        elif job == "Doctor" and get_level(ctx.author.id) < 40: return await ctx.respond("You currently do not have the required level to perform this job!", ephemeral=True)
+        if job == "YouTuber" and levelling.get_level(ctx.author.id) < 3: return await ctx.respond("You currently do not have the required level to perform this job!", ephemeral=True)
+        elif job == "Streamer" and levelling.get_level(ctx.author.id) < 5: return await ctx.respond("You currently do not have the required level to perform this job!", ephemeral=True)
+        elif job == "Developer" and levelling.get_level(ctx.author.id) < 10: return await ctx.respond("You currently do not have the required level to perform this job!", ephemeral=True)
+        elif job == "Scientist" and levelling.get_level(ctx.author.id) < 20: return await ctx.respond("You currently do not have the required level to perform this job!", ephemeral=True)
+        elif job == "Engineer" and levelling.get_level(ctx.author.id) < 25: return await ctx.respond("You currently do not have the required level to perform this job!", ephemeral=True)
+        elif job == "Doctor" and levelling.get_level(ctx.author.id) < 40: return await ctx.respond("You currently do not have the required level to perform this job!", ephemeral=True)
         userdat[str(ctx.author.id)]["work_job"] = job
         save()
         localembed = discord.Embed(title="New job!", description=f"You are now working as a {job}!")

cogs/levelling.py

@@ -5,27 +5,17 @@
 import json
 from discord import option, ApplicationContext
 from discord.ext import commands
+from framework.isobot.db import levelling
 
 # Variables
 color = discord.Color.random()
-
-with open("database/levels.json", 'r', encoding="utf-8") as f: levels = json.load(f)
-
-def save():
-    with open("database/levels.json", 'w+', encoding="utf-8") as f: json.dump(levels, f, indent=4)
-
-# Functions
-def get_xp(id: int) -> int:
-    return levels[str(id)]["xp"]
-
-def get_level(id: int) -> int:
-    return levels[str(id)]["level"]
+levelling = levelling.Levelling()
 
 # Commands
 class Levelling(commands.Cog):
     def __init__(self, bot):
         self.bot = bot
-    
+
     @commands.slash_command(
         name="rank",
         description="Shows your rank or another user's rank"
@@ -35,8 +25,8 @@ async def rank(self, ctx: ApplicationContext, user:discord.User=None):
         if user is None: user = ctx.author
         try:
             localembed = discord.Embed(title=f"{user.display_name}'s rank", color=color)
-            localembed.add_field(name="Level", value=levels[str(user.id)]["level"])
-            localembed.add_field(name="XP", value=levels[str(user.id)]["xp"])
+            localembed.add_field(name="Level", value=levelling.get_level(user.id))
+            localembed.add_field(name="XP", value=levelling.get_xp(user.id))
             localembed.set_footer(text="Keep chatting to earn levels!")
             await ctx.respond(embed = localembed)
         except KeyError: return await ctx.respond("Looks like that user isn't indexed yet. Try again later.", ephemeral=True)
@@ -50,8 +40,8 @@ async def rank(self, ctx: ApplicationContext, user:discord.User=None):
     async def edit_rank(self, ctx: ApplicationContext, user:discord.User, new_rank:int):
         if ctx.author.id != 738290097170153472: return await ctx.respond("This command isn't for you.", ephemeral=True)
         try:
-            levels[str(user.id)]["level"] = new_rank
-            await ctx.respond(f"{user.display_name}\'s rank successfully edited. `New Rank: {levels[str(user.id)]['level']}`")
+            levelling.set_level(user.id, new_rank)
+            await ctx.respond(f"{user.display_name}\'s rank successfully edited. `New Rank: {levelling.get_level(user.id)}`")
         except KeyError: return await ctx.respond("That user isn't indexed yet.", ephemeral=True)
 
     @commands.slash_command(
@@ -63,15 +53,16 @@ async def edit_rank(self, ctx: ApplicationContext, user:discord.User, new_rank:i
     async def edit_xp(self, ctx: ApplicationContext, user:discord.User, new_xp:int):
         if ctx.author.id != 738290097170153472: return await ctx.respond("This command isn't for you.", ephemeral=True)
         try:
-            levels[str(user.id)]["xp"] = new_xp
-            await ctx.respond(f"{user.display_name}\'s XP count successfully edited. `New XP: {levels[str(user.id)]['xp']}`")
+            levelling.set_xp(user.id, new_xp)
+            await ctx.respond(f"{user.display_name}\'s XP count successfully edited. `New XP: {levelling.get_xp(user.id)}`")
         except KeyError: return await ctx.respond("That user isn't indexed yet.", ephemeral=True)
 
     @commands.slash_command(
-        name="leaderboard_levels", 
+        name="leaderboard_levels",
         description="View the global leaderboard for user levelling ranks."
     )
     async def leaderboard_levels(self, ctx: ApplicationContext):
+        levels = levelling.get_raw()
         levels_dict = dict()
         for person in levels:
             levels_dict[str(person)] = levels[str(person)]["level"]

cogs/utils.py

@@ -7,14 +7,15 @@
 import math
 import openai
 from framework.isobot import currency, embedengine
+from framework.isobot.db import levelling
 from discord import option, ApplicationContext
 from discord.ext import commands
-from cogs.levelling import get_level, get_xp
 from cogs.afk import get_presence
 
 # Variables
 color = discord.Color.random()
 currency = currency.CurrencyAPI("database/currency.json", "logs/currency.log")
+levelling = levelling.Levelling()
 openai.api_key = os.getenv("chatgpt_API_KEY")
 chatgpt_conversation = dict()
 
@@ -94,7 +95,7 @@ async def profile(self, ctx: ApplicationContext, user: discord.User = None):
         if user is None: user = ctx.author
         localembed = discord.Embed(title=f"{user.display_name}'s isobot stats", color=color)
         localembed.set_thumbnail(url=user.avatar)
-        localembed.add_field(name="Level", value=f"Level {get_level(user.id)} ({get_xp(user.id)} XP)", inline=False)
+        localembed.add_field(name="Level", value=f"Level {levelling.get_level(user.id)} ({levelling.get_xp(user.id)} XP)", inline=False)
         localembed.add_field(name="Balance in Wallet", value=f"{currency.get_wallet(user.id)} coins", inline=True)
         localembed.add_field(name="Balance in Bank Account", value=f"{currency.get_bank(user.id)} coins", inline=True)
         localembed.add_field(name="Net-Worth", value=f"{currency.get_user_networth(user.id)} coins", inline=True)

framework/isobot/db/levelling.py

@@ -0,0 +1,86 @@
+"""The framework module library used for managing isobot's levelling database."""
+# Imports
+import json
+
+# Functions
+class Levelling():
+    """Used to initialize the levelling database."""
+    def __init__(self):
+        print("[framework/db/Levelling] Levelling db library initialized.")
+
+    def load(self) -> dict:
+        """Fetches and returns the latest data from the levelling database."""  
+        with open("database/levels.json", 'r', encoding="utf8") as f: db = json.load(f)
+        return db
+
+    def save(self, data: dict) -> int:
+        """Dumps all cached data to your local machine."""
+        with open("database/levels.json", 'w+', encoding="utf8") as f: json.dump(data, f, indent=4)
+        return 0
+
+    def generate(self, user_id: int) -> int:
+        """Generates a new data key for the specified user.\n
+        Returns `0` if the request was successful, returns `1` if the data key already exists."""
+        levels = self.load()
+        if str(user_id) not in levels:
+            levels[str(user_id)] = {"xp": 0, "level": 0}
+            self.save(levels)
+            return 0
+        return 1
+
+    def set_level(self, user_id: int, count: int) -> int:
+        """Sets a new level for the specified user."""
+        levels = self.load()
+        levels[str(user_id)]["level"] = count
+        self.save(levels)
+        return 0
+
+    def set_xp(self, user_id: int, count: int) -> int:
+        """Sets a new xp count for the specified user."""
+        levels = self.load()
+        levels[str(user_id)]["xp"] = count
+        self.save(levels)
+        return 0
+
+    def add_levels(self, user_id: int, count: int) -> int:
+        """Adds a specified amount of levels to the specified user."""
+        levels = self.load()
+        levels[str(user_id)]["levels"] += count
+        self.save(levels)
+        return 0
+
+    def remove_levels(self, user_id: int, count: int) -> int:
+        """Removes a specified amount of levels from the specified user."""
+        levels = self.load()
+        levels[str(user_id)]["levels"] -= count
+        self.save(levels)
+        return 0
+
+    def add_xp(self, user_id: int, count: int) -> int:
+        """Adds a specified amount of xp to the specified user."""
+        levels = self.load()
+        levels[str(user_id)]["xp"] += count
+        self.save(levels)
+        return 0
+
+    def remove_xp(self, user_id: int, count: int) -> int:
+        """Removes a specified amount of xp from the specified user."""
+        levels = self.load()
+        levels[str(user_id)]["xp"] -= count
+        self.save(levels)
+        return 0
+
+    def get_level(self, user_id: int) -> int:
+        """Fetches a user's current level."""
+        levels = self.load()
+        return levels[str(user_id)]["level"]
+
+    def get_xp(self, user_id: int) -> int:
+        """Fetches a user's current xp."""
+        levels = self.load()
+        return levels[str(user_id)]["xp"]
+
+    def get_raw(self):
+        """Fetches and returns the raw json data in the levelling database."""
+        levels = self.load()
+        return levels

main.py

@@ -11,6 +11,7 @@
 from math import floor
 from random import randint
 from framework.isobot import currency, colors, settings
+from framework.isobot.db import levelling
 from discord import ApplicationContext, option
 from discord.ext import commands
 from cogs.economy import new_userdat
@@ -27,7 +28,6 @@
 with open('database/items.json', 'r', encoding="utf-8") as f: items = json.load(f)
 with open('config/shop.json', 'r', encoding="utf-8") as f: shopitem = json.load(f)
 with open('database/presence.json', 'r', encoding="utf-8") as f: presence = json.load(f)
-with open('database/levels.json', 'r', encoding="utf-8") as f: levels = json.load(f)
 with open('config/commands.json', 'r', encoding="utf-8") as f: commandsdb = json.load(f)
 with open('database/automod.json', 'r', encoding="utf-8") as f: automod_config = json.load(f)
 cmd_list = commandsdb.keys()
@@ -37,7 +37,6 @@ def save():
     """Dumps all cached data to the local databases."""
     with open('database/items.json', 'w+', encoding="utf-8") as e: json.dump(items, e, indent=4)
     with open('database/presence.json', 'w+', encoding="utf-8") as e: json.dump(presence, e, indent=4)
-    with open('database/levels.json', 'w+', encoding="utf-8") as e: json.dump(levels, e, indent=4)
     with open('database/automod.json', 'w+', encoding="utf-8") as e: json.dump(automod_config, e, indent=4)
 
 if not os.path.isdir("logs"):
@@ -58,6 +57,7 @@ def save():
 colors = colors.Colors()
 currency = currency.CurrencyAPI("database/currency.json", "logs/currency.log")
 settings = settings.Configurator()
+levelling = levelling.Levelling()
 
 # Theme Loader
 themes = False  # True: enables themes; False: disables themes;
@@ -101,8 +101,7 @@ async def on_message(ctx):
     settings.generate(ctx.author.id)
     if str(ctx.author.id) not in items:
         items[str(ctx.author.id)] = {}
-    if str(ctx.author.id) not in levels:
-        levels[str(ctx.author.id)] = {"xp": 0, "level": 0}
+    levelling.generate(ctx.author.id)
     if str(ctx.guild.id) not in automod_config:
         automod_config[str(ctx.guild.id)] = {
             "swear_filter": {
@@ -133,22 +132,22 @@ async def on_message(ctx):
         await asyncio.sleep(5)
         await m1.delete()
     if not ctx.author.bot:
-        levels[str(ctx.author.id)]["xp"] += randint(1, 5)
+        levelling.add_xp(ctx.author.id, randint(1, 5))
         xpreq = 0
-        for level in range(int(levels[str(ctx.author.id)]["level"])):
+        for level in range(levelling.get_level(ctx.author.id)):
             xpreq += 50
             if xpreq >= 5000: break
-        if levels[str(ctx.author.id)]["xp"] >= xpreq:
-            levels[str(ctx.author.id)]["xp"] = 0
-            levels[str(ctx.author.id)]["level"] += 1
+        if levelling.get_xp(ctx.author.id) >= xpreq:
+            levelling.set_xp(ctx.author.id, 0)
+            levelling.add_levels(ctx.author.id, 1)
             if settings.fetch_setting(ctx.author.id, "levelup_messages") is True:
-                await ctx.author.send(f"{ctx.author.mention}, you just ranked up to **level {levels[str(ctx.author.id)]['level']}**. Nice!")
+                await ctx.author.send(f"{ctx.author.mention}, you just ranked up to **level {levelling.get_level(ctx.author.id)}**. Nice!")
         save()
         if automod_config[str(ctx.guild.id)]["swear_filter"]["enabled"] is True:
             if automod_config[str(ctx.guild.id)]["swear_filter"]["keywords"]["use_default"] and any(x in ctx.content.lower() for x in automod_config[str(ctx.guild.id)]["swear_filter"]["keywords"]["default"]):
                 await ctx.delete()
                 await ctx.channel.send(f'{ctx.author.mention} watch your language.', delete_after=5)
-            elif automod_config[str(ctx.guild.id)]["swear_filter"]["keywords"]["custom"] is not [] and any(x in ctx.content.lower() for x in automod_config[str(ctx.guild.id)]["swear_filter"]["keywords"]["custom"]):
+            elif automod_config[str(ctx.guild.id)]["swear_filter"]["keywords"]["custom"] != [] and any(x in ctx.content.lower() for x in automod_config[str(ctx.guild.id)]["swear_filter"]["keywords"]["custom"]):
                 await ctx.delete()
                 await ctx.channel.send(f'{ctx.author.mention} watch your language.', delete_after=5)