insert2-cogs/speechbubble.py

import disnake
from disnake.ext import commands
from disnake import TextInputStyle
from dotenv import load_dotenv
from random import randint
import aiosqlite as sqlite3
import aiohttp
from urllib.parse import urlparse
import os

class ApplicationModal(disnake.ui.Modal):
    def __init__(self, user):
        self.victim = user
        components = [
            disnake.ui.TextInput(
                label="link to image",
                placeholder="https://media.discordapp.net/goofyimg.png",
                custom_id="image",
                style=TextInputStyle.short,
                max_length=400,
            ),
            disnake.ui.TextInput(
                label="chance: a 1 in x chance",
                placeholder="10",
                custom_id="chance: a 1 in x chance",
                style=TextInputStyle.short,
                max_length=5,
            ),
        ]
        super().__init__(title=f"create a speechbubble for {user.name}", components=components)

    async def callback(self, inter: disnake.ModalInteraction):
        speechbubble = inter.text_values.items()
        embed = disnake.Embed(title="New speech bubble!", color=disnake.Colour.green())
        embed.add_field(
            name="User ID",
            value=self.victim.id,
            inline=False,
        )
        for key, value in speechbubble:
            if key == "image":
                if urlparse(value).netloc != 'media.discordapp.net' and urlparse(value).netloc != 'cdn.discordapp.com':
                    await inter.response.send_message(f"for security reasons, the bot only accepts images from media.discordapp.net or cdn.discordapp.com", ephemeral=True)
                    return
                async with aiohttp.ClientSession() as session:
                    async with session.get(value) as response:
                        if (response.status == 200) and (response.headers['content-type'] == "image/png" or response.headers['content-type'] == "image/jpeg" or response.headers['content-type'] == "image/jpg"):
                            pass
                        else:
                            await inter.response.send_message(f"you did not supply an image {response.status} {response.headers['content-type']}", ephemeral=True)
                            return
            else:
                if not value.isdigit():
                    await inter.response.send_message("you supplied words for a number", ephemeral=True)
                    return
            embed.add_field(
                name=key,
                value=value,
                inline=False,
            )
        await inter.response.send_message(f"<@{self.victim.id}> {inter.user.name} proposes the following speech bubble for you:", embed=embed, components=[
            disnake.ui.Button(label="Approve", style=disnake.ButtonStyle.success, custom_id="Approve"),
            disnake.ui.Button(label="Deny", style=disnake.ButtonStyle.danger, custom_id="Deny"),],)


class speechbubble(commands.Cog):

    def __init__(self, bot: commands.Bot):
        self.bot = bot

    @commands.Cog.listener("on_button_click")
    async def button_listener(self, inter: disnake.MessageInteraction):
        ogmsg = inter.message.embeds
        embed = ogmsg[0]
        if inter.user.id != int(embed.fields[0].value):
            await inter.response.send_message("you may not modify this speech bubble!", ephemeral=True)
            return
        if inter.component.custom_id == "Approve":
            embed.add_field(
                name="Status",
                value="Approved",
                inline=False,
            )
            await inter.response.edit_message(embed=embed, components=[])
            await self.bot.cur.execute(f"""
            INSERT INTO userinfo VALUES
                ({inter.guild_id}, {int(embed.fields[0].value)}, '{embed.fields[1].value}', {int(embed.fields[2].value)})
            """)
            await self.bot.db.commit()
            return
        if inter.component.custom_id == "Deny":
            embed.add_field(
                name="Status",
                value="Denied",
                inline=False,
            )
            await inter.response.edit_message(embed=embed, components=[])
            return

    @commands.Cog.listener('on_message')
    async def on_message(self,message):
        res = await self.bot.cur.execute(f"SELECT imagelink, chance FROM userinfo WHERE serverid = {message.guild.id} AND userid = {message.author.id} ORDER BY RANDOM() LIMIT 1")
        try:
            imagelink, chance = await res.fetchone()
        except TypeError:
            return
        if randint(1, chance) == 1:
            await message.channel.send(imagelink)

    @commands.slash_command(
        name="create",
        description="create a speech bubble for a user",
    )
    async def create(self, inter: disnake.ApplicationCommandInteraction, victim: disnake.Member = commands.Param(name="victim")):
        if inter.user.id == victim.id:
            await inter.response.send_message("you may not assign yourself a speech bubble!", ephemeral=True)
            return
        await inter.response.send_modal(modal=ApplicationModal(victim))

    @commands.slash_command(
        name="removeme",
        description="Removes you from the database",
    )
    async def removeme(self, inter: disnake.ApplicationCommandInteraction, universal: str = commands.Param(choices=["yes", "no"])):
        if universal == "yes":
            await self.bot.cur.execute(f"DELETE FROM userinfo WHERE userid = {inter.user.id}")
        elif universal == "no":
            await self.bot.cur.execute(f"DELETE FROM userinfo WHERE userid = {inter.user.id} AND serverid = {inter.guild_id}")
        await self.bot.db.commit()
        await inter.response.send_message("Done!", ephemeral=True)
    
def setup(bot: commands.Bot):
    bot.add_cog(speechbubble(bot))