349 lines
12 KiB
Python
349 lines
12 KiB
Python
import asyncio
|
|
import logging
|
|
from typing import Union
|
|
|
|
import discord
|
|
from redbot.core import Config, checks, commands
|
|
from redbot.core.bot import Red
|
|
from redbot.core.commands import Cog
|
|
from redbot.core.utils.chat_formatting import pagify
|
|
|
|
log = logging.getLogger("red.fox_v3.stealemoji")
|
|
# Replaced with discord.Asset.read()
|
|
# async def fetch_img(session: aiohttp.ClientSession, url: StrOrURL):
|
|
# async with session.get(url) as response:
|
|
# assert response.status == 200
|
|
# return await response.read()
|
|
|
|
|
|
async def check_guild(guild, emoji):
|
|
if len(guild.emojis) >= 2 * guild.emoji_limit:
|
|
return False
|
|
|
|
if len(guild.emojis) < guild.emoji_limit:
|
|
return True
|
|
|
|
if emoji.animated:
|
|
return sum(e.animated for e in guild.emojis) < guild.emoji_limit
|
|
else:
|
|
return sum(not e.animated for e in guild.emojis) < guild.emoji_limit
|
|
|
|
|
|
class StealEmoji(Cog):
|
|
"""
|
|
This cog steals emojis and creates servers for them
|
|
"""
|
|
|
|
default_stolemoji = {
|
|
"guildbank": None,
|
|
"name": None,
|
|
"require_colons": None,
|
|
"managed": None,
|
|
"guild_id": None,
|
|
"animated": None,
|
|
"saveid": None,
|
|
}
|
|
|
|
def __init__(self, red: Red):
|
|
super().__init__()
|
|
self.bot = red
|
|
self.config = Config.get_conf(self, identifier=11511610197108101109111106105)
|
|
default_global = {
|
|
"stolemoji": {},
|
|
"guildbanks": [],
|
|
"autobanked_guilds": [],
|
|
"on": False,
|
|
"notify": 0,
|
|
"autobank": False,
|
|
}
|
|
|
|
self.config.register_global(**default_global)
|
|
|
|
self.is_on = None
|
|
|
|
async def red_delete_data_for_user(self, **kwargs):
|
|
"""Nothing to delete"""
|
|
return
|
|
|
|
@commands.group()
|
|
async def stealemoji(self, ctx: commands.Context):
|
|
"""
|
|
Base command for this cog. Check help for the commands list.
|
|
"""
|
|
pass
|
|
|
|
@checks.is_owner()
|
|
@stealemoji.command(name="clearemojis")
|
|
async def se_clearemojis(self, ctx: commands.Context, confirm: bool = False):
|
|
"""Removes the history of all stolen emojis. Will not delete emojis from server banks"""
|
|
if not confirm:
|
|
await ctx.maybe_send_embed(
|
|
"This will reset all stolen emoji data.\n"
|
|
"If you want to continue, run this command again as:\n"
|
|
"`[p]stealemoji clearemojis True`"
|
|
)
|
|
return
|
|
|
|
await self.config.stolemoji.clear()
|
|
await ctx.tick()
|
|
|
|
@checks.is_owner()
|
|
@stealemoji.command(name="print")
|
|
async def se_print(self, ctx: commands.Context):
|
|
"""Prints all the emojis that have been stolen so far"""
|
|
stolen = await self.config.stolemoji()
|
|
id_list = [v.get("saveid") for k, v in stolen.items()]
|
|
|
|
emoj = " ".join(str(e) for e in self.bot.emojis if e.id in id_list)
|
|
|
|
if emoj == " ":
|
|
await ctx.maybe_send_embed("No stolen emojis yet")
|
|
return
|
|
|
|
for page in pagify(emoj, delims=[" "]):
|
|
await ctx.maybe_send_embed(page)
|
|
|
|
@checks.is_owner()
|
|
@stealemoji.command(name="notify")
|
|
async def se_notify(self, ctx: commands.Context):
|
|
"""Cycles between notification settings for when an emoji is stolen
|
|
|
|
None (Default)
|
|
DM Owner
|
|
Msg in server channel
|
|
"""
|
|
curr_setting = await self.config.notify()
|
|
|
|
if not curr_setting:
|
|
await self.config.notify.set(1)
|
|
await ctx.maybe_send_embed("Bot owner will now be notified when an emoji is stolen")
|
|
elif curr_setting == 1:
|
|
channel: discord.TextChannel = ctx.channel
|
|
await self.config.notify.set(channel.id)
|
|
await ctx.maybe_send_embed("This channel will now be notified when an emoji is stolen")
|
|
else:
|
|
await self.config.notify.set(0)
|
|
await ctx.maybe_send_embed("Notifications are now off")
|
|
|
|
@checks.is_owner()
|
|
@stealemoji.command(name="collect")
|
|
async def se_collect(self, ctx):
|
|
"""Toggles whether emoji's are collected or not"""
|
|
curr_setting = await self.config.on()
|
|
await self.config.on.set(not curr_setting)
|
|
|
|
self.is_on = await self.config.on()
|
|
|
|
await ctx.maybe_send_embed("Collection is now " + str(not curr_setting))
|
|
|
|
@checks.is_owner()
|
|
@stealemoji.command(name="autobank")
|
|
async def se_autobank(self, ctx):
|
|
"""Toggles automatically creating new guilds as emoji banks"""
|
|
curr_setting = await self.config.autobank()
|
|
await self.config.autobank.set(not curr_setting)
|
|
|
|
self.is_on = await self.config.autobank()
|
|
|
|
await ctx.maybe_send_embed("AutoBanking is now " + str(not curr_setting))
|
|
|
|
@checks.is_owner()
|
|
@commands.guild_only()
|
|
@stealemoji.command(name="deleteserver", aliases=["deleteguild"])
|
|
async def se_deleteserver(self, ctx: commands.Context, guild_id=None):
|
|
"""Delete servers the bot is the owner of.
|
|
|
|
Useful for auto-generated guildbanks."""
|
|
if guild_id is None:
|
|
guild = ctx.guild
|
|
else:
|
|
guild = await self.bot.get_guild(guild_id)
|
|
|
|
if guild is None:
|
|
await ctx.maybe_send_embed("Failed to get guild, cancelling")
|
|
return
|
|
guild: discord.Guild
|
|
await ctx.maybe_send_embed(
|
|
f"Will attempt to delete {guild.name} ({guild.id})\n" f"Okay to continue? (yes/no)"
|
|
)
|
|
|
|
def check(m):
|
|
return m.author == ctx.author and m.channel == ctx.channel
|
|
|
|
try:
|
|
answer = await self.bot.wait_for("message", timeout=120, check=check)
|
|
except asyncio.TimeoutError:
|
|
await ctx.send("Timed out, canceling")
|
|
return
|
|
|
|
if answer.content.upper() not in ["Y", "YES"]:
|
|
await ctx.maybe_send_embed("Cancelling")
|
|
return
|
|
try:
|
|
await guild.delete()
|
|
except discord.Forbidden:
|
|
log.exception("No permission to delete. I'm probably not the guild owner")
|
|
await ctx.maybe_send_embed("No permission to delete. I'm probably not the guild owner")
|
|
except discord.HTTPException:
|
|
log.exception("Unexpected error when deleting guild")
|
|
await ctx.maybe_send_embed("Unexpected error when deleting guild")
|
|
else:
|
|
await self.bot.send_to_owners(f"Guild {guild.name} deleted")
|
|
|
|
@checks.is_owner()
|
|
@commands.guild_only()
|
|
@stealemoji.command(name="bank")
|
|
async def se_bank(self, ctx):
|
|
"""Add or remove current server as emoji bank"""
|
|
|
|
def check(m):
|
|
return (
|
|
m.content.upper() in ["Y", "YES", "N", "NO"]
|
|
and m.channel == ctx.channel
|
|
and m.author == ctx.author
|
|
)
|
|
|
|
already_a_guildbank = ctx.guild.id in (await self.config.guildbanks())
|
|
|
|
if already_a_guildbank:
|
|
await ctx.maybe_send_embed(
|
|
"This is already an emoji bank\n"
|
|
"Are you sure you want to remove the current server from the emoji bank list? (y/n)"
|
|
)
|
|
else:
|
|
await ctx.maybe_send_embed(
|
|
"This will upload custom emojis to this server\n"
|
|
"Are you sure you want to make the current server an emoji bank? (y/n)"
|
|
)
|
|
|
|
msg = await self.bot.wait_for("message", check=check)
|
|
|
|
if msg.content.upper() in ["N", "NO"]:
|
|
await ctx.maybe_send_embed("Cancelled")
|
|
return
|
|
|
|
async with self.config.guildbanks() as guildbanks:
|
|
if already_a_guildbank:
|
|
guildbanks.remove(ctx.guild.id)
|
|
else:
|
|
guildbanks.append(ctx.guild.id)
|
|
|
|
if already_a_guildbank:
|
|
await ctx.maybe_send_embed("This server has been removed from being an emoji bank")
|
|
else:
|
|
await ctx.maybe_send_embed("This server has been added to be an emoji bank")
|
|
|
|
@commands.Cog.listener()
|
|
async def on_reaction_add(self, reaction: discord.Reaction, user: discord.User):
|
|
"""Event handler for reaction watching"""
|
|
if not reaction.custom_emoji:
|
|
# print("Not a custom emoji")
|
|
return
|
|
|
|
if self.is_on is None:
|
|
self.is_on = await self.config.on()
|
|
|
|
if not self.is_on:
|
|
# print("Collecting is off")
|
|
return
|
|
|
|
guild: discord.Guild = getattr(user, "guild", None)
|
|
if await self.bot.cog_disabled_in_guild(self, guild): # Handles None guild just fine
|
|
return
|
|
|
|
emoji: discord.Emoji = reaction.emoji
|
|
if emoji in self.bot.emojis:
|
|
# print("Emoji already in bot.emojis")
|
|
return
|
|
|
|
# This is now a custom emoji that the bot doesn't have access to, time to steal it
|
|
# First, do I have an available guildbank?
|
|
|
|
guildbank: Union[discord.Guild, None] = None
|
|
banklist = await self.config.guildbanks()
|
|
for guild_id in banklist:
|
|
guild: discord.Guild = self.bot.get_guild(guild_id)
|
|
# if len(guild.emojis) < 50:
|
|
if await check_guild(guild, emoji):
|
|
guildbank = guild
|
|
break
|
|
|
|
if guildbank is None:
|
|
if not await self.config.autobank():
|
|
return
|
|
|
|
try:
|
|
guildbank: discord.Guild = await self.bot.create_guild(
|
|
"StealEmoji Guildbank", code="S93bqTqKQ9rM"
|
|
)
|
|
except discord.HTTPException:
|
|
await self.config.autobank.set(False)
|
|
log.exception("Unable to create guilds, disabling autobank")
|
|
return
|
|
async with self.config.guildbanks() as guildbanks:
|
|
guildbanks.append(guildbank.id)
|
|
# Track generated guilds for easier deletion
|
|
async with self.config.autobanked_guilds() as autobanked_guilds:
|
|
autobanked_guilds.append(guildbank.id)
|
|
|
|
await asyncio.sleep(2)
|
|
|
|
if guildbank.text_channels:
|
|
channel = guildbank.text_channels[0]
|
|
else:
|
|
# Always hits the else.
|
|
# Maybe create_guild doesn't return guild object with
|
|
# the template channel?
|
|
channel = await guildbank.create_text_channel("invite-channel")
|
|
invite = await channel.create_invite()
|
|
|
|
await self.bot.send_to_owners(invite)
|
|
log.info(f"Guild created id {guildbank.id}. Invite: {invite}")
|
|
# Next, have I saved this emoji before (because uploaded emoji != orignal emoji)
|
|
|
|
if str(emoji.id) in await self.config.stolemoji():
|
|
# print("Emoji has already been stolen")
|
|
return
|
|
|
|
img = await emoji.url.read()
|
|
|
|
try:
|
|
uploaded_emoji = await guildbank.create_custom_emoji(
|
|
name=emoji.name, image=img, reason="Stole from " + str(user)
|
|
)
|
|
except discord.Forbidden as e:
|
|
# print("PermissionError - no permission to add emojis")
|
|
raise PermissionError("No permission to add emojis") from e
|
|
except discord.HTTPException as e:
|
|
# print("HTTPException exception")
|
|
raise e # Unhandled error
|
|
|
|
# If you get this far, YOU DID IT
|
|
|
|
save_dict = self.default_stolemoji.copy()
|
|
# e_attr_list = [a for a in dir(emoji) if not a.startswith("__")]
|
|
|
|
for k in save_dict.keys():
|
|
save_dict[k] = getattr(emoji, k, None)
|
|
|
|
# for k in e_attr_list:
|
|
# if k in save_dict:
|
|
# save_dict[k] = getattr(emoji, k, None)
|
|
|
|
save_dict["guildbank"] = guildbank.id
|
|
save_dict["saveid"] = uploaded_emoji.id
|
|
|
|
async with self.config.stolemoji() as stolemoji:
|
|
stolemoji[emoji.id] = save_dict
|
|
|
|
# Enable the below if you want to get notified when it works
|
|
notify_settings = await self.config.notify()
|
|
if notify_settings:
|
|
if notify_settings == 1:
|
|
owner = await self.bot.application_info()
|
|
target = owner.owner
|
|
else:
|
|
target = self.bot.get_channel(notify_settings)
|
|
|
|
await target.send(f"Just added emoji {uploaded_emoji} to server {guildbank}")
|