YusarinBot/functions.py

from gzip import open as gzipopen
from datetime import datetime
from os import listdir, makedirs, remove, stat, path
from shutil import copyfileobj
from typing import Any, Literal, Union
from discord import utils, Embed, TextChannel, VoiceChannel, PermissionOverwrite, Guild, CategoryChannel, Member, Client, ApplicationContext, Message

try:
    from ujson import loads, dumps
except ImportError:
    from json import loads, dumps

# path = Path(__file__).resolve().parent

log_size = 512

# This is the default option for "debug" key in
# file config.json, so if cebug is not set in it
# bot will use this value instead.
debug = False

try:
    with open("config.json", 'r', encoding="utf-8") as json_file:
        output = loads(json_file.read())
        json_file.close()
    debug = output["debug"]
except:
    debug = debug

# Check latest log size
def checkSize():

    try:
        if not path.isdir("logs"):
            makedirs("logs", exist_ok=True)
        
        log = stat("logs/latest.log")
        global log_size

        if (log.st_size / 1024) > log_size:
            with open("logs/latest.log", "rb") as f_in:
                with gzipopen(f'logs/{datetime.now().strftime("%d.%m.%Y_%H:%M:%S")}.log.gz', 'wb') as f_out:
                    copyfileobj(f_in, f_out)
                    print(f'Copied logs/{datetime.now().strftime("%d.%m.%Y_%H:%M:%S")}.log.gz')
            open("logs/latest.log", "w").close()
    except FileNotFoundError:
        print('Not found')
        pass

# Append string to log
def appendLog(message: str, guild: Union[Guild, None] = None, announce=True) -> None:
    global debug
    
    
    if guild == None:
        message_formatted = f'[{datetime.now().strftime("%d.%m.%Y")}] [{datetime.now().strftime("%H:%M:%S")}] {message}'
    else:
        if debug:
            message_formatted = f'[{datetime.now().strftime("%d.%m.%Y")}] [{datetime.now().strftime("%H:%M:%S")}] [{guild} | {str(guild.id)}] {message}'
        else:
            message_formatted = f'[{datetime.now().strftime("%d.%m.%Y")}] [{datetime.now().strftime("%H:%M:%S")}] [{guild}] {message}'
    
    if announce:
        print(message_formatted, flush=True)

    checkSize()

    log = open('logs/latest.log', 'a')  # type: ignore
    log.write(f'{message_formatted}\n')
    log.close()

def saveJson(value: Any, filename: str) -> None:
    with open(filename, 'w', encoding="utf-8") as f:
        f.write(dumps(value, indent=4, ensure_ascii=False))
        f.close()

def loadJson(filename: str) -> Any:
    global debug
    try:
        with open(filename, 'r', encoding="utf-8") as json_file:
            output = loads(json_file.read())
            json_file.close()
    except Exception as exp:
        if debug:
            appendLog(f"Could not load json file {filename} due to exception {exp}")
        output = {}
    return output

def colorToStr():
    pass

def strToColor(string: str) -> int:
    return int(hex(int(string.replace("#", ""), 16)), 0)

def gotCommand(message: Message) -> None:
    global debug
    if debug:
        appendLog(f"Command '{message.content}' from {message.author} ({str(message.author.id)})", message.guild)
    else:
        appendLog(f"Command '{message.content}' from {message.author}", message.guild)

def guildConfGet(guild: Guild, variable: str) -> Any:
    global debug
    try:
        config = loadJson(f"guilds/{str(guild.id)}/config.json")
        return config[variable]
    except Exception as exp:
        if debug:
            appendLog(f"Could not get guild config key '{variable}' due to {exp}", guild)
        return None

def guildConfSet(guild: Guild, variable: str, value: Any) -> None:
    config = loadJson(f"guilds/{str(guild.id)}/config.json")
    config[variable] = value
    try:
        saveJson(config, f"guilds/{str(guild.id)}/config.json")
    except:
        makedirs(f"guilds/{str(guild.id)}", exist_ok=True)
        makedirs(f"guilds/{str(guild.id)}/channels", exist_ok=True)
        saveJson(config, f"guilds/{str(guild.id)}/config.json")
    appendLog(f"Guild config key '{variable}' is now set to '{value}'", guild)

def guildConfReset(guild: Guild, variable: str) -> None:
    try:
        config = loadJson(f"guilds/{str(guild.id)}/config.json")
        del config[variable]
        try:
            saveJson(config, f"guilds/{str(guild.id)}/config.json")
        except:
            makedirs(f"guilds/{str(guild.id)}", exist_ok=True)
            makedirs(f"guilds/{str(guild.id)}/channels", exist_ok=True)
            saveJson(config, f"guilds/{str(guild.id)}/config.json")
        appendLog(f"Guild config key '{variable}' has been reset", guild)
    except Exception as exp:
        appendLog(f"Could not reset guild config key '{variable}' due to {exp}", guild)

def guildLocaleGet(guild: Guild) -> str:
    config = loadJson(f"config.json")
    try:
        locale = guildConfGet(guild, "locale")
    except:
        return config["bot_locale"]
    if locale is None:
        return config["bot_locale"]
    else:
        return locale

def getMsg(string: str, guild: Union[Guild, None] = None) -> str:
    try:
        locale = loadJson(f'locale/{guildLocaleGet(guild)}.json')
        return locale["messages"][string]
    except Exception as exp:
        appendLog(f"Could not get locale string named {string} due to exception {exp}", guild)
        return string

def makeEmbed(title="", description="", footer="", color=0xffffff) -> Embed:
    embed=Embed(title=title, description=description, color=color)
    if footer is not None:
        embed.set_footer(text=footer)
    return embed

def channelExists(number: int, guild: Guild, type: Literal["Any", "Text", "Voice"] = "Any") -> bool:
    global debug
    if number == None:
        return False
    try:
        if type == "Voice":
            selected_channel = utils.get(guild.channels, id=number)
            if isinstance(selected_channel, VoiceChannel):
                return True
        elif type == "Text":
            selected_channel = utils.get(guild.channels, id=number)
            if isinstance(selected_channel, TextChannel):
                return True
        elif type == "Any":
            selected_channel = utils.get(guild.channels, id=number)
            return True
    except Exception as exp:
        if debug:
            appendLog(f"Channel ID {str(number)} is not a channel due to {exp}")
    return False

def channelGetName(number: int, guild: Guild):
    global debug
    try:
        selected_channel = utils.get(guild.channels, id=number)
        return selected_channel.name
    except Exception as exp:
        if debug:
            appendLog(f"Channel ID {str(number)} is not a channel due to {exp}")
    return "Channel doesn't exist"

def isUserVoice(vc: VoiceChannel) -> bool:
    try:
        channels_list = listdir(f"guilds/{str(vc.guild.id)}/channels/")
        if f"{str(vc.id)}.json" in channels_list:
            return True
        else:
            return False
    except:
        return False
    
async def removeUserVoice(vc: VoiceChannel) -> None:
    global debug
    channels_list = listdir(f"guilds/{str(vc.guild.id)}/channels/")
    if f"{vc.id}.json" in channels_list:
        vc_file = f"guilds/{str(vc.guild.id)}/channels/{str(vc.id)}.json"
        vc_conf = loadJson(vc_file)
        
        needed_channel = utils.get(vc.guild.channels, id=vc.id)
        if loadJson("config.json")["enable_nomic"]:
            nomic_channel = utils.get(vc.guild.channels, id=vc_conf["nomic"])
        
        remove(vc_file)
        
        await needed_channel.delete()
        if debug:
            appendLog(f"Removed voice channel '{needed_channel}' ({str(needed_channel.id)}) of user with id {str(vc_conf['ownerid'])}", guild=vc.guild)
        else:
            appendLog(f"Removed voice channel '{needed_channel}' of user with id {str(vc_conf['ownerid'])}", guild=vc.guild)
        
        if loadJson("config.json")["enable_nomic"]:
            await nomic_channel.delete()
            if debug:
                appendLog(f"Removed nomic channel {nomic_channel} ({str(nomic_channel.id)}) of channel with id {str(needed_channel.id)}", guild=vc.guild)
            else:
                appendLog(f"Removed nomic channel '{nomic_channel}' of channel with id {str(needed_channel.id)}", guild=vc.guild)
    else:
        return

async def createUserVoice(vc: VoiceChannel, category: CategoryChannel, member: Member) -> VoiceChannel:
    global debug
    chan = {}
    overwrites_channel = {
        vc.guild.default_role: PermissionOverwrite(view_channel=True),
        vc.guild.me: PermissionOverwrite(read_messages=True, view_channel=True, manage_channels=True),
        member: PermissionOverwrite(read_messages=True, view_channel=True, manage_channels=True)
    }
    overwrites_nomic = {
        vc.guild.default_role: PermissionOverwrite(view_channel=False, read_messages=False),
        vc.guild.me: PermissionOverwrite(read_messages=True, view_channel=True, manage_channels=True),
        member: PermissionOverwrite(read_messages=True, view_channel=True, manage_channels=True)
    }
    created_channel = await vc.guild.create_voice_channel(getMsg("name_voice", vc.guild).format(member.name), category=category, overwrites=overwrites_channel)
    if debug:
        appendLog(f"Created voice channel '{created_channel}' ({str(created_channel.id)}) for user {member} ({str(member.id)})", guild=vc.guild)
    else:
        appendLog(f"Created voice channel '{created_channel}' for user {member}", guild=vc.guild)
    if not path.isdir(f"guilds/{str(created_channel.guild.id)}/channels"):
        makedirs(f"guilds/{str(created_channel.guild.id)}/channels", exist_ok=True)
    vc_file = f"guilds/{str(created_channel.guild.id)}/channels/{str(created_channel.id)}.json"
    chan["ownerid"] = member.id
    saveJson(chan, vc_file)
    if loadJson("config.json")["enable_nomic"]:
        nomic_channel = await vc.guild.create_text_channel(getMsg("name_nomic", vc.guild).format(created_channel.id), category=category, overwrites=overwrites_nomic, topic=getMsg("description_nomic", vc.guild).format(str(created_channel.id)))
        if debug:
            appendLog(f"Created nomic channel '{nomic_channel}' ({str(nomic_channel.id)}) for channel '{created_channel}' ({str(created_channel.id)})", guild=vc.guild)
        else:
            appendLog(f"Created nomic channel '{nomic_channel}' for channel '{created_channel}'", guild=vc.guild)
        chan["nomic"] = nomic_channel.id
        saveJson(chan, vc_file)
    return created_channel

def isVoiceOfUser(vc: VoiceChannel, member: Member) -> bool:
    vc_file = f"guilds/{str(vc.guild.id)}/channels/{str(vc.id)}.json"
    vc_conf = loadJson(vc_file)
    if vc_conf["ownerid"] == member.id:
        return True
    else:
        return False

async def changeNomicPerms(mode: Literal["deny", "allow"], vc: VoiceChannel, member: Member) -> None:
    vc_file = f"guilds/{str(vc.guild.id)}/channels/{str(vc.id)}.json"
    vc_conf = loadJson(vc_file)
    if loadJson("config.json")["enable_nomic"]:
        nomic_channel = utils.get(vc.guild.channels, id=vc_conf["nomic"])
        if mode == "deny":
            await nomic_channel.set_permissions(member, view_channel=False)
        else:
            await nomic_channel.set_permissions(member, view_channel=True)

async def clearTrash(client: Client):
    if not path.isdir(f"guilds/"):
        makedirs(f"guilds", exist_ok=True)
    guilds_list = listdir(f"guilds/")
    for guild in guilds_list:
        guild_object = client.get_guild(int(guild))
        if path.isdir(f"guilds/{guild}/channels"):
            channels_list = listdir(f"guilds/{guild}/channels/")
            for channel in channels_list:
                channel_id = channel[:-5]
                try:
                    selected_channel = utils.get(guild_object.voice_channels, id=int(channel_id))
                    channel_owner = loadJson(f"guilds/{guild}/channels/{channel}")["ownerid"]
                    remove_channel = True
                    for member in selected_channel.members:
                        if member.id == channel_owner:
                            remove_channel = False
                    if remove_channel:
                        await removeUserVoice(selected_channel)
                except:
                    remove(f"guilds/{guild}/channels/{channel_id}.json")

#async def autoClearTrash(client):
    # execute clearTrash every 120 seconds

def getHelpMessage(ctx: ApplicationContext, version: float) -> Embed:
    
    #channelExists(number, guild, type="Voice")
    
    config = loadJson("config.json")
    
    if ctx.guild is not None:
        if channelExists(guildConfGet(ctx.guild, 'channel'), ctx.guild, type="Voice"):
            desc_channel = getMsg("help_channel_set", guild=ctx.guild).format(channelGetName(guildConfGet(ctx.guild, 'channel'), ctx.guild))
        else:
            desc_channel = getMsg("help_channel_none", guild=ctx.guild)
        
        if channelExists(guildConfGet(ctx.guild, 'category'), ctx.guild, type="Any"):
            desc_category = getMsg("help_category_set", guild=ctx.guild).format(channelGetName(guildConfGet(ctx.guild, 'category'), ctx.guild))
        else:
            desc_category = getMsg("help_category_none", guild=ctx.guild)
        
        desc_locale = getMsg("help_locale", guild=ctx.guild).format(getMsg("locale_name", ctx.guild))

        description = "\n".join([desc_locale, desc_channel, desc_category])

        embed=Embed(title=getMsg("help_title", ctx.guild), description=description, color=strToColor(config["color_default"]))
    else:
        embed=Embed(title=getMsg("help_title_dm", ctx.guild), color=strToColor(config["color_default"]))
    
    embed.set_author(name=f'{config["bot_name"]} v{str(version)}', url=config["bot_site"], icon_url=config["bot_icon"])
    
    if ctx.author.id == config["owner"]:
        embed.add_field(name=f"/shutdown", value=getMsg("help_cmd_shutdown", ctx.guild), inline=False)
    
    embed.add_field(name=f"/channel set", value=getMsg("help_cmd_channel", ctx.guild), inline=False)
    embed.add_field(name=f"/category set", value=getMsg("help_cmd_category", ctx.guild), inline=False)
    embed.add_field(name=f"/locale set", value=getMsg("help_cmd_locale", ctx.guild), inline=False)
    
    if ctx.guild is None:
        embed.set_footer(text=getMsg("help_server", ctx.guild))
    else:
        embed.set_footer(text=getMsg("help_notice_id", ctx.guild))
    
    return embed

async def guildConfigured(guild: Guild) -> str:

    output = {}
    config = loadJson("config.json")

    for kind in ["channel", "category"]:
        if guildConfGet(guild, kind) is not None:
            try:
                guild_object = utils.get(guild.categories, id=guildConfGet(guild, kind))
                output[kind] = getMsg("configured_"+kind, guild).format(guild_object.name)
            except Exception as exp:
                output[kind] = getMsg("unconfigured_"+kind, guild)
        else:
            output[kind] = getMsg("unconfigured_"+kind, guild)

    return getMsg("server_config", guild).format(getMsg("info_locale", guild).format(getMsg("locale_name", guild)), output["channel"], output["category"])