insert2-cogs/speechbubble.py

import nextcord
from nextcord.ext import commands
from nextcord import TextInputStyle
from random import randint
import aiosqlite as sqlite3
import aiohttp
import asyncio
from urllib.parse import urlparse
import os


class ApplicationView(nextcord.ui.View):
    def __init__(self,bot):
        super().__init__(timeout=None)
        self.bot = bot

    @nextcord.ui.button(
        label="Approve", style=nextcord.ButtonStyle.green, custom_id="speechbubble:approve"
    )
    async def approve(self, button: nextcord.ui.Button, interaction: nextcord.Interaction):
        ogmsg = interaction.message.embeds
        embed = ogmsg[0]
        if interaction.user.id != int(embed.fields[0].value):
            await interaction.response.send_message("you may not modify this speech bubble!", ephemeral=True)
            return
        embed.add_field(
            name="Status",
            value="Approved",
            inline=False,
        )
        await self.bot.cur.execute(f"""
        INSERT INTO userinfo VALUES
            ({interaction.guild_id}, {int(embed.fields[0].value)}, '{embed.fields[1].value}', {int(embed.fields[2].value)})
        """)
        await self.bot.db.commit()
        await interaction.response.edit_message(embed=embed, view=None)
        return

    @nextcord.ui.button(
        label="Deny", style=nextcord.ButtonStyle.red, custom_id="speechbubble:deny"
    )
    async def deny(self, button: nextcord.ui.Button, interaction: nextcord.Interaction):
        ogmsg = interaction.message.embeds
        embed = ogmsg[0]
        if interaction.user.id != int(embed.fields[0].value):
            await interaction.response.send_message("you may not modify this speech bubble!", ephemeral=True)
            return
        embed.add_field(
            name="Status",
            value="Denied",
            inline=False,
        )
        await interaction.response.edit_message(embed=embed, view=None)
        return

class ApplicationModal(nextcord.ui.Modal):
    def __init__(self,user,bot):
        self.victim = user
        self.bot = bot
        super().__init__(
            f"create a speechbubble for {user.name}"
        )

        self.image = nextcord.ui.TextInput(
            label="link to image",
            placeholder="https://media.discordapp.net/goofyimg.png",
            style=TextInputStyle.short,
            max_length=400,
        )
        self.add_item(self.image)

        self.chance = nextcord.ui.TextInput(
            label="chance: a 1 in x chance",
            placeholder="10",
            style=TextInputStyle.short,
            max_length=5,
        )
        self.add_item(self.chance)
    
    async def callback(self, interaction: nextcord.Interaction) -> None:
        embed = nextcord.Embed(title="New speech bubble!", color=nextcord.Colour.green())
        embed.add_field(
            name="User ID",
            value=self.victim.id,
            inline=False,
        )
        print(self.image.value)
        if urlparse(self.image.value).netloc != 'media.discordapp.net' and urlparse(self.image.value).netloc != 'cdn.discordapp.com':
            await interaction.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(self.image.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 interaction.send(f"you did not supply an image {response.status} {response.headers['content-type']}", ephemeral=True)
                    return
            if not self.chance.value.isdigit():
                    await interaction.send("you supplied words for a number", ephemeral=True)
                    return
        embed.add_field(
            name="Image",
            value=self.image.value,
            inline=False,
        )
        embed.add_field(
            name="Chance, a 1 in x chance",
            value=self.chance.value,
            inline=False,
        )
        await interaction.send(f"<@{self.victim.id}> {interaction.user.name} proposes the following speech bubble for you:", embed=embed, view=ApplicationView(self.bot))


async def dbcheck(db,cur):
    res = await bot.cur.execute("SELECT name FROM sqlite_master WHERE name='userinfo'")
    if await res.fetchone() is None:
        await cur.execute("CREATE TABLE userinfo(serverid INTEGER, userid INTEGER, imagelink TEXT, chance INTEGER)")
        await db.commit()

class speechbubble(commands.Cog):

    def __init__(self, bot: commands.Bot):
        asyncio.run(dbcheck(bot.db,bot.cur))
        self.bot = bot
        bot.add_view(ApplicationView(bot))

    @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)

    @nextcord.slash_command(
        name="create",
        description="create a speech bubble for a user",
    )
    async def create(self, interaction: nextcord.Interaction, victim: nextcord.Member = nextcord.SlashOption(name="victim")):
        if interaction.user.id == victim.id:
            await interaction.response.send_message("you may not assign yourself a speech bubble, try someone else!", ephemeral=True)
            return
        await interaction.response.send_modal(ApplicationModal(victim,self.bot))

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