240 lines
8.8 KiB
Python
240 lines
8.8 KiB
Python
import discord, json, os, copy
|
|
|
|
from discord.ext import commands
|
|
from datetime import datetime
|
|
from time import strptime
|
|
from io import BytesIO
|
|
from typing import Optional, Union, Dict, Any
|
|
from addons import Settings, TOKENS
|
|
|
|
from motor.motor_asyncio import (
|
|
AsyncIOMotorClient,
|
|
AsyncIOMotorCollection,
|
|
)
|
|
|
|
ROOT_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
|
|
if not os.path.exists(os.path.join(ROOT_DIR, "settings.json")):
|
|
raise Exception("Settings file not set!")
|
|
|
|
#--------------- Cache Var ---------------
|
|
tokens: TOKENS = TOKENS()
|
|
settings: Settings
|
|
|
|
MONGO_DB: AsyncIOMotorClient
|
|
SETTINGS_DB: AsyncIOMotorCollection
|
|
USERS_DB: AsyncIOMotorCollection
|
|
|
|
ERROR_LOGS: dict[int, dict[int, str]] = {} #Stores error that not a Voicelink Exception
|
|
LANGS: dict[str, dict[str, str]] = {} #Stores all the languages in ./langs
|
|
LOCAL_LANGS: dict[str, dict[str, str]] = {} #Stores all the localization languages in ./local_langs
|
|
SETTINGS_BUFFER: dict[int, dict[str, Any]] = {} #Cache guild language
|
|
USERS_BUFFER: dict[str, dict] = {}
|
|
|
|
USERS_BASE: dict[str, Any] = {
|
|
'playlist': {
|
|
'200': {
|
|
'tracks':[],
|
|
'perms': {'read': [], 'write':[], 'remove': []},
|
|
'name':'Favourite',
|
|
'type':'playlist'
|
|
}
|
|
},
|
|
'history': [],
|
|
'inbox':[]
|
|
}
|
|
|
|
ALLOWED_MENTIONS = discord.AllowedMentions().none()
|
|
|
|
#-------------- Vocard Functions --------------
|
|
def open_json(path: str) -> dict:
|
|
try:
|
|
with open(os.path.join(ROOT_DIR, path), encoding="utf8") as json_file:
|
|
return json.load(json_file)
|
|
except:
|
|
return {}
|
|
|
|
def update_json(path: str, new_data: dict) -> None:
|
|
data = open_json(path)
|
|
if not data:
|
|
return
|
|
|
|
data.update(new_data)
|
|
|
|
with open(os.path.join(ROOT_DIR, path), "w") as json_file:
|
|
json.dump(data, json_file, indent=4)
|
|
|
|
def langs_setup() -> None:
|
|
for language in os.listdir(os.path.join(ROOT_DIR, "langs")):
|
|
if language.endswith('.json'):
|
|
LANGS[language[:-5]] = {}
|
|
|
|
for language in os.listdir(os.path.join(ROOT_DIR, "local_langs")):
|
|
if language.endswith('.json'):
|
|
LOCAL_LANGS[language[:-5]] = open_json(os.path.join("local_langs", language))
|
|
|
|
return
|
|
|
|
def time(millis:int) -> str:
|
|
seconds=(millis/1000)%60
|
|
minutes=(millis/(1000*60))%60
|
|
hours=(millis/(1000*60*60))%24
|
|
if hours > 1:
|
|
return "%02d:%02d:%02d" % (hours, minutes, seconds)
|
|
else:
|
|
return "%02d:%02d" % (minutes, seconds)
|
|
|
|
def formatTime(number:str) -> Optional[int]:
|
|
try:
|
|
try:
|
|
num = strptime(number, '%M:%S')
|
|
except ValueError:
|
|
try:
|
|
num = strptime(number, '%S')
|
|
except ValueError:
|
|
num = strptime(number, '%H:%M:%S')
|
|
except:
|
|
return None
|
|
|
|
return (int(num.tm_hour) * 3600 + int(num.tm_min) * 60 + int(num.tm_sec)) * 1000
|
|
|
|
def get_source(source: str, type: str) -> str:
|
|
source_settings: dict = settings.sources_settings.get(source.lower(), {})
|
|
return source_settings.get(type, ("🔗" if type == "emoji" else settings.embed_color))
|
|
|
|
def gen_report() -> Optional[discord.File]:
|
|
if ERROR_LOGS:
|
|
errorText = ""
|
|
for guild_id, error in ERROR_LOGS.items():
|
|
errorText += f"Guild ID: {guild_id}\n" + "-" * 30 + "\n"
|
|
for index, (key, value) in enumerate(error.items() , start=1):
|
|
errorText += f"Error No: {index}, Time: {datetime.fromtimestamp(key)}\n" + value + "-" * 30 + "\n\n"
|
|
|
|
buffer = BytesIO(errorText.encode('utf-8'))
|
|
file = discord.File(buffer, filename='report.txt')
|
|
buffer.close()
|
|
|
|
return file
|
|
return None
|
|
|
|
def cooldown_check(ctx: commands.Context) -> Optional[commands.Cooldown]:
|
|
if ctx.author.id in settings.bot_access_user:
|
|
return None
|
|
cooldown = settings.cooldowns_settings.get(f"{ctx.command.parent.qualified_name} {ctx.command.name}" if ctx.command.parent else ctx.command.name)
|
|
if not cooldown:
|
|
return None
|
|
return commands.Cooldown(cooldown[0], cooldown[1])
|
|
|
|
def get_aliases(name: str) -> list:
|
|
return settings.aliases_settings.get(name, [])
|
|
|
|
def check_roles() -> tuple[str, int, int]:
|
|
return 'Normal', 5, 500
|
|
|
|
def truncate_string(text: str, length: int = 40) -> str:
|
|
return text[:length - 3] + "..." if len(text) > length else text
|
|
|
|
def get_lang_non_async(guild_id: int, *keys) -> Union[list[str], str]:
|
|
settings = SETTINGS_BUFFER.get(guild_id, {})
|
|
lang = settings.get("lang", "EN")
|
|
if lang in LANGS and not LANGS[lang]:
|
|
LANGS[lang] = open_json(os.path.join("langs", f"{lang}.json"))
|
|
|
|
if len(keys) == 1:
|
|
return LANGS.get(lang, {}).get(keys[0], "Language pack not found!")
|
|
return [LANGS.get(lang, {}).get(key, "Language pack not found!") for key in keys]
|
|
|
|
async def get_lang(guild_id:int, *keys) -> Union[list[str], str]:
|
|
settings = await get_settings(guild_id)
|
|
lang = settings.get("lang", "EN")
|
|
if lang in LANGS and not LANGS[lang]:
|
|
LANGS[lang] = open_json(os.path.join("langs", f"{lang}.json"))
|
|
|
|
if len(keys) == 1:
|
|
return LANGS.get(lang, {}).get(keys[0], "Language pack not found!")
|
|
return [LANGS.get(lang, {}).get(key, "Language pack not found!") for key in keys]
|
|
|
|
async def send(ctx: Union[commands.Context, discord.Interaction], key: str, *params, delete_after: float = None, ephemeral: bool = False) -> Optional[discord.Message]:
|
|
text = await get_lang(ctx.guild.id, key)
|
|
text = text.format(*params)
|
|
|
|
send_func = ctx.send if isinstance(ctx, commands.Context) else (ctx.followup.send if ctx.response.is_done() else ctx.response.send_message)
|
|
return await send_func(text, delete_after=delete_after, ephemeral=ephemeral, allowed_mentions=ALLOWED_MENTIONS)
|
|
|
|
async def update_db(db: AsyncIOMotorCollection, tempStore: dict, filter: dict, data: dict) -> bool:
|
|
for mode, action in data.items():
|
|
for key, value in action.items():
|
|
cursors = key.split(".")
|
|
|
|
nested_data = tempStore
|
|
for c in cursors[:-1]:
|
|
nested_data = nested_data.setdefault(c, {})
|
|
|
|
if mode == "$set":
|
|
try:
|
|
nested_data[cursors[-1]] = value
|
|
except TypeError:
|
|
nested_data[int(cursors[-1])] = value
|
|
|
|
elif mode == "$unset":
|
|
nested_data.pop(cursors[-1], None)
|
|
|
|
elif mode == "$inc":
|
|
nested_data[cursors[-1]] = nested_data.get(cursors[-1], 0) + value
|
|
|
|
elif mode == "$push":
|
|
if isinstance(value, dict) and "$each" in value:
|
|
nested_data.setdefault(cursors[-1], []).extend(value["$each"][value.get("$slice", len(value["$each"])):])
|
|
else:
|
|
nested_data.setdefault(cursors[-1], []).extend([value])
|
|
|
|
elif mode == "$push":
|
|
if isinstance(value, dict) and "$each" in value:
|
|
nested_data.setdefault(cursors[-1], []).extend(value["$each"])
|
|
nested_data[cursors[-1]] = nested_data[cursors[-1]][value.get("$slice", len(value["$each"])):]
|
|
else:
|
|
nested_data.setdefault(cursors[-1], []).extend([value])
|
|
|
|
elif mode == "$pull":
|
|
if cursors[-1] in nested_data:
|
|
value = value.get("$in", []) if isinstance(value, dict) else [value]
|
|
nested_data[cursors[-1]] = [item for item in nested_data[cursors[-1]] if item not in value]
|
|
|
|
else:
|
|
return False
|
|
|
|
result = await db.update_one(filter, data)
|
|
return result.modified_count > 0
|
|
|
|
async def get_settings(guild_id:int) -> dict[str, Any]:
|
|
settings = SETTINGS_BUFFER.get(guild_id, None)
|
|
if not settings:
|
|
settings = await SETTINGS_DB.find_one({"_id": guild_id})
|
|
if not settings:
|
|
await SETTINGS_DB.insert_one({"_id": guild_id})
|
|
|
|
settings = SETTINGS_BUFFER[guild_id] = settings or {}
|
|
return settings
|
|
|
|
async def update_settings(guild_id: int, data: dict[str, dict[str, Any]]) -> bool:
|
|
settings = await get_settings(guild_id)
|
|
return await update_db(SETTINGS_DB, settings, {"_id": guild_id}, data)
|
|
|
|
async def get_user(user_id: int, d_type: Optional[str] = None, need_copy: bool = True) -> Dict[str, Any]:
|
|
user = USERS_BUFFER.get(user_id)
|
|
if not user:
|
|
user = await USERS_DB.find_one({"_id": user_id})
|
|
if not user:
|
|
user = {"_id": user_id, **USERS_BASE}
|
|
await USERS_DB.insert_one(user)
|
|
|
|
USERS_BUFFER[user_id] = user
|
|
|
|
if d_type:
|
|
user = user.setdefault(d_type, copy.deepcopy(USERS_BASE.get(d_type)))
|
|
|
|
return copy.deepcopy(user) if need_copy else user
|
|
|
|
async def update_user(user_id:int, data:dict) -> bool:
|
|
playlist = await get_user(user_id, need_copy=False)
|
|
return await update_db(USERS_DB, playlist, {"_id": user_id}, data) |