Add db management system libraries for rest of the databases

cogs/economy.py

@@ -9,6 +9,7 @@
 import utils.logger
 import asyncio
 import framework.isobot.currency
+import framework.isobot.db
 from random import randint
 from discord import option, ApplicationContext
 from discord.ext import commands
@@ -34,6 +35,7 @@ def get_item_names(self) -> list:
 wdir = os.getcwd()
 color = discord.Color.random()
 currency = framework.isobot.currency.CurrencyAPI("database/currency.json", "logs/currency.log")
+levels = framework.isobot.db.Levels("database/levels.json", None)
 shop_data = ShopData(f"{wdir}/config/shop.json")
 all_item_ids = shop_data.get_item_ids()
 jobs = [
@@ -467,12 +469,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 levels.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 levels.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 levels.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 levels.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 levels.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 levels.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

@@ -4,24 +4,26 @@
 import discord
 import json
 import os.path
+import framework.isobot.db
 from discord import option, ApplicationContext
 from discord.ext import commands
 
 # Variables
 wdir = os.getcwd()
 color = discord.Color.random()
+levels = framework.isobot.db.Levels(f"{wdir}/database/levels.json", None)
 
-with open(f"{wdir}/database/levels.json", 'r', encoding="utf-8") as f: levels = json.load(f)
+# with open(f"{wdir}/database/levels.json", 'r', encoding="utf-8") as f: levels = json.load(f)
 
-def save():
-    with open(f"{wdir}/database/levels.json", 'w+', encoding="utf-8") as f: json.dump(levels, f, indent=4)
+# def save():
+#     with open(f"{wdir}/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_xp(id: int) -> int:
+#     return levels[str(id)]["xp"]
 
-def get_level(id: int) -> int:
-    return levels[str(id)]["level"]
+# def get_level(id: int) -> int:
+#     return levels[str(id)]["level"]
 
 # Commands
 class Levelling(commands.Cog):
@@ -37,8 +39,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=levels.get_level(user.id))
+            localembed.add_field(name="XP", value=levels.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)
@@ -52,8 +54,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']}`")
+            levels.edit_level(user.id, new_rank)
+            await ctx.respond(f"{user.display_name}\'s rank successfully edited. `New Rank: {levels.get_level(user.id)}`")
         except KeyError: return await ctx.respond("That user isn't indexed yet.", ephemeral=True)
 
     @commands.slash_command(
@@ -65,8 +67,8 @@ 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']}`")
+            levels.edit_xp(user.id, new_xp)
+            await ctx.respond(f"{user.display_name}\'s XP count successfully edited. `New XP: {levels.get_xp(user.id)}`")
         except KeyError: return await ctx.respond("That user isn't indexed yet.", ephemeral=True)
 
     @commands.slash_command(
@@ -76,7 +78,7 @@ async def edit_xp(self, ctx: ApplicationContext, user:discord.User, new_xp:int):
     async def leaderboard_levels(self, ctx: ApplicationContext):
         levels_dict = dict()
         for person in levels:
-            levels_dict[str(person)] = levels[str(person)]["level"]
+            levels_dict[str(person)] = levels.get_level(person)
         undicted_leaderboard = sorted(levels_dict.items(), key=lambda x:x[1], reverse=True)
         dicted_leaderboard = dict(undicted_leaderboard)
         parsed_output = str()

cogs/utils.py

@@ -8,16 +8,17 @@
 import openai
 import framework.isobot.embedengine
 import framework.isobot.currency
+import framework.isobot.db
 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 = framework.isobot.currency.CurrencyAPI("database/currency.json", "logs/currency.log")
 openai.api_key = os.getenv("chatgpt_API_KEY")
 chatgpt_conversation = dict()
+levels = framework.isobot.db.Levels("database/levels.json", None)
 
 # Commands
 class Utils(commands.Cog):
@@ -95,7 +96,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 {levels.get_level(user.id)} ({levels.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)

cogs/weather.py

@@ -3,17 +3,13 @@
 import json
 import requests
 import os
+import framework.isobot.db
 from discord import ApplicationContext, option
 from discord.ext import commands
 
 # Variables
 api_key = os.environ['openweathermap_API_KEY']
-with open("database/weather.json", 'r', encoding="utf-8") as f: user_db = json.load(f)
-
-# Functions
-def save():
-    """Dumps all cached databases to storage."""
-    with open("database/weather.json", 'w+', encoding="utf-8") as f: json.dump(user_db, f, indent=4)
+weather_db = framework.isobot.db.Weather("database/weather.json", None)
 
 # Commands
 class Weather(commands.Cog):
@@ -26,13 +22,12 @@ def __init__(self, bot):
     )
     @option(name="location", description="What location do you want to set?", type=str)
     async def weather_set_location(self, ctx: ApplicationContext, location: str):
-        if str(ctx.author.id) not in user_db: user_db[str(ctx.author.id)] = {"location": None, "scale": "Celsius"}
+        weather_db.new(ctx.author.id)
         test_ping = requests.get(f"https://api.openweathermap.org/data/2.5/weather?q={location}&appid={api_key}").content
         test_ping_json = json.loads(test_ping)
         if test_ping_json["cod"] == '404': return await ctx.respond(":warning: This location does not exist.", ephemeral=True)
         else:
-            user_db[str(ctx.author.id)]["location"] = location.lower()
-            save()
+            weather_db.set_default_location(ctx.author.id, location.lower())
             localembed = discord.Embed(description="Your default location has been updated.", color=discord.Color.green())
             await ctx.respond(embed=localembed)
 
@@ -42,10 +37,9 @@ async def weather_set_location(self, ctx: ApplicationContext, location: str):
     )
     @option(name="scale", description="Which scale do you want to use?", type=str, choices=["Celsius", "Fahrenheit", "Kelvin"])
     async def weather_set_scale(self, ctx: ApplicationContext, scale: str):
-        if str(ctx.author.id) not in user_db: user_db[str(ctx.author.id)] = {"location": None, "scale": "Celsius"}
+        weather_db.new(ctx.author.id)
         if scale not in ["Celsius", "Fahrenheit", "Kelvin"]: return 1
-        user_db[str(ctx.author.id)]["scale"] = scale
-        save()
+        weather_db.set_scale(ctx.author.id, scale)
         localembed = discord.Embed(description="Your preferred unit scale has been updated.", color=discord.Color.green())
         await ctx.respond(embed=localembed)
 
@@ -55,10 +49,10 @@ async def weather_set_scale(self, ctx: ApplicationContext, scale: str):
     )
     @option(name="location", description="The location you want weather info about (leave empty for set location)", type=str, default=None)
     async def weather(self, ctx: ApplicationContext, location: str = None):
-        if str(ctx.author.id) not in user_db: user_db[str(ctx.author.id)] = {"location": None, "scale": "Celsius"}
+        weather_db.new(ctx.author.id)
         if location == None:
-            if user_db[str(ctx.author.id)]["location"] == None: return await ctx.respond("You do not have a default location set yet.\nEnter a location name and try again.", ephemeral=True)
-            else: location = user_db[str(ctx.author.id)]["location"]
+            if weather_db.get_location(ctx.author.id) == None: return await ctx.respond("You do not have a default location set yet.\nEnter a location name and try again.", ephemeral=True)
+            else: location = weather_db.get_default_location(ctx.author.id)
         location = location.replace(" ", "%20")
         api_request = requests.get(f"https://api.openweathermap.org/data/2.5/weather?q={location}&appid={api_key}").content
         req: dict = json.loads(api_request)
@@ -71,11 +65,11 @@ async def weather(self, ctx: ApplicationContext, location: str = None):
 
         # Stripped API request data
         loc_name = req["name"]
-        if user_db[str(ctx.author.id)]["scale"] == "Celsius":
+        if weather_db.get_scale(ctx.author.id) == "Celsius":
             temp = round(req["main"]["temp"] - 273)
             temp_max = round(req["main"]["temp_max"] - 273)
             temp_min = round(req["main"]["temp_min"] - 273)
-        elif user_db[str(ctx.author.id)]["scale"] == "Fahrenheit":
+        elif weather_db.get_scale(ctx.author.id) == "Fahrenheit":
             temp = round(((req["main"]["temp"] - 273) * 9/5) + 32)
             temp_max = round(((req["main"]["temp_max"] - 273) * 9/5) + 32)
             temp_min = round(((req["main"]["temp_min"] - 273) * 9/5) + 32)
@@ -94,8 +88,8 @@ async def weather(self, ctx: ApplicationContext, location: str = None):
             description=f"**{forcast}**\n{forcast_description}",
             color=discord.Color.blue()
         )
-        if user_db[str(ctx.author.id)]["scale"] == "Celsius": localembed.add_field(name="Temperature", value=f"**{temp}C** (max: {temp_max}C,  min: {temp_min}C)")
-        elif user_db[str(ctx.author.id)]["scale"] == "Fahrenheit": localembed.add_field(name="Temperature", value=f"**{temp}F** (max: {temp_max}F,  min: {temp_min}F)")
+        if weather_db.get_scale(ctx.author.id) == "Celsius": localembed.add_field(name="Temperature", value=f"**{temp}C** (max: {temp_max}C,  min: {temp_min}C)")
+        elif weather_db.get_scale(ctx.author.id) == "Fahrenheit": localembed.add_field(name="Temperature", value=f"**{temp}F** (max: {temp_max}F,  min: {temp_min}F)")
         else: localembed.add_field(name="Temperature", value=f"**{temp}K** (max: {temp_max}K,  min: {temp_min}K)")
         localembed.add_field(name="Humidity", value=f"{humidity}%")
         localembed.add_field(name="Sunrise", value=f"<t:{sunrise}:f>", inline=False)