tuxbot-bot/tuxbot/core/utils/functions/extra.py

157 lines
5.4 KiB
Python

import asyncio
import random
from io import BytesIO
import aiohttp
import discord
from discord import Embed
from discord.ext import commands
TOKEN_REPLACEMENT = "" * random.randint(3, 15)
PASSWORD_REPLACEMENT = "" * random.randint(3, 15)
IP_REPLACEMENT = "" * random.randint(3, 15)
class ContextPlus(commands.Context):
# noinspection PyTypedDict
async def send(
self, content=None, *, embed=None, deletable=True, **kwargs
): # i know *args and **kwargs but, i prefer work with same values
from tuxbot.core.utils.functions.utils import (
replace_in_dict,
replace_in_list,
)
# todo: rewrite replacements
if content:
content = (
content.replace(self.bot.config.Core.token, TOKEN_REPLACEMENT)
.replace(
self.bot.config.Core.Database.password,
PASSWORD_REPLACEMENT,
)
.replace(self.bot.config.Core.ip, IP_REPLACEMENT)
.replace(self.bot.config.Core.ip6, IP_REPLACEMENT)
)
if len(content) > 1800:
content = "output too long..."
if embed:
e = embed.to_dict()
for key, value in e.items():
if isinstance(value, (str, bytes)):
# skipcq
e[key] = (
value.replace(
self.bot.config.Core.token, TOKEN_REPLACEMENT
)
.replace(
self.bot.config.Core.Database.password,
PASSWORD_REPLACEMENT,
)
.replace(self.bot.config.Core.ip, IP_REPLACEMENT)
.replace(self.bot.config.Core.ip6, IP_REPLACEMENT)
)
elif isinstance(value, list):
# skipcq
e[key] = replace_in_list(
value, self.bot.config.Core.token, TOKEN_REPLACEMENT
)
# skipcq
e[key] = replace_in_list(
e[key],
self.bot.config.Core.Database.password,
PASSWORD_REPLACEMENT,
)
# skipcq
e[key] = replace_in_list(
e[key], self.bot.config.Core.ip, IP_REPLACEMENT
)
# skipcq
e[key] = replace_in_list(
e[key], self.bot.config.Core.ip6, IP_REPLACEMENT
)
elif isinstance(value, dict):
# skipcq
e[key] = replace_in_dict(
value, self.bot.config.Core.token, TOKEN_REPLACEMENT
)
# skipcq
e[key] = replace_in_dict(
e[key],
self.bot.config.Core.Database.password,
PASSWORD_REPLACEMENT,
)
# skipcq
e[key] = replace_in_dict(
e[key], self.bot.config.Core.ip, IP_REPLACEMENT
)
# skipcq
e[key] = replace_in_dict(
e[key], self.bot.config.Core.ip6, IP_REPLACEMENT
)
embed = Embed.from_dict(e)
if (
hasattr(self.command, "deletable") and self.command.deletable
) and deletable:
message = await super().send(
content=content, embed=embed, **kwargs
)
await message.add_reaction("🗑")
def check(reaction: discord.Reaction, user: discord.User):
return (
user == self.author
and str(reaction.emoji) == "🗑"
and reaction.message.id == message.id
)
try:
await self.bot.wait_for(
"reaction_add", timeout=42.0, check=check
)
except asyncio.TimeoutError:
await message.remove_reaction("🗑", self.bot.user)
else:
await message.delete()
return message
return await super().send(content=content, embed=embed, **kwargs)
@property
def session(self) -> aiohttp.ClientSession:
return self.bot.session
def __repr__(self):
items = (
"message=%s" % self.message,
"channel=%s" % self.channel,
"guild=%s" % self.guild,
"author=%s" % self.author,
"prefix=%s" % self.prefix,
"args=%s" % self.args,
"kwargs=%s" % self.kwargs,
)
return "<%s %s>" % (self.__class__.__name__, ", ".join(items))
class CommandPLus(commands.Command):
def __init__(self, function, **kwargs):
super().__init__(function, **kwargs)
self.deletable = kwargs.pop("deletable", True)
def command_extra(*args, **kwargs):
return commands.command(*args, **kwargs, cls=CommandPLus)
class GroupPlus(commands.Group):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.deletable = kwargs.pop("deletable", True)
def group_extra(*args, **kwargs):
return commands.group(*args, **kwargs, cls=GroupPlus)