157 lines
5.4 KiB
Python
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)
|