dragon-bot/app/bot.py

#!/usr/local/bin/python
from discord.ext import commands
import core_utils
import discord
import os
import random
import requests
import requests_cache

intents = discord.Intents.default()
intents.message_content = True
intents.members = True
intents.reactions = True
bot = commands.Bot(command_prefix="!", intents=intents)
requests_cache.install_cache(expire_after=300)

cogfiles = [
    f"cogs.{filename[:-3]}"
    for filename in os.listdir("/app/cogs/")
    if filename.endswith(".py")
]

for cogfile in cogfiles:
    bot.load_extension(cogfile)


@bot.event
async def on_ready():
    print(f"{bot.user.name} has connected to Discord!")

    # for server in bot.guilds:
    #     if server.name in [
    #         # "no officer it's hi how r u",
    #         "Aeturnum",
    #         "Naked Knights",
    #         "MadsenHouse",
    #         "Safety Zone",
    #         "R3 Server",
    #         "Thieve's Den MC",
    #         "Thieve's Den",
    #     ]:
    #         await server.leave()
    #         print("Left %s" % server.name)

    if os.getenv("DRAGON_ENV") == "prod":
        # await bot.get_channel(152921472304676865).send("I have reconnected")

        # Waiting-room in zoid's server
        channel_id = 1026281775984549958
        message_id = 1099374735428681758

        # for testing, bot-testing server's channel
        # channel_id = 932476007439552522
        # message_id = 1114045480129806361

        blurb = """# Please react to this message with one of the following emojis to be granted the appropriate role(s):

🚀 for Star Citizen

<:helldivers:1205747460287504424> for HellDivers

<:palsphere:1201663585596481586> for Palworld

🐀 for Darktide/Vermintide

:gear: for Lethal Company

🤖 for Warframe

🚔 for Ready or Not/Ground Branch

🔫 for Tarkov
        """
        # message = await bot.get_channel(channel_id).send(blurb)
        # Update the message on_ready to match the content we always want to be there
        message = await bot.get_channel(channel_id).fetch_message(message_id)
        await message.edit(content=blurb)


@bot.event
async def on_raw_reaction_add(payload):
    guild = bot.get_guild(payload.guild_id)

    role_map = {
        "🔫": "Tarkov Gamers",
        "🤖": "Warframe Enjoyers",
        "🐀": "Dorktide Gamers",
        "helldivers": "Hell Divers",
        "🚀": "Star Citizens",
        "⚙️": "Lethal Gamers",
        "🚔": "Ready or Notters",
        "palsphere": "Palworlders",
    }

    if payload.channel_id == 1026281775984549958:
        role = discord.utils.get(guild.roles, name=role_map[payload.emoji.name])
        member = guild.get_member(payload.user_id)
        await member.add_roles(role)


@bot.listen("on_message")
async def convert_heic_to_jpg(ctx):
    from cmagick import cmagick
    import time
    import tempfile

    if ctx.attachments:
        for attachment in ctx.attachments:
            if attachment.filename.lower().endswith(("heic", "tiff")):
                source_file = "/tmp/source"
                source_file, file_path = tempfile.mkstemp()
                jpg_file = "/tmp/%s.jpg" % time.time()
                await attachment.save(fp=file_path)

                cmagick.convert(file_path, jpg_file)

                try:
                    await ctx.delete()
                except Exception:
                    pass
                await ctx.channel.send(
                    "%s said:\n%s" % (ctx.author.mention, ctx.content),
                    file=discord.File(jpg_file),
                )
                os.remove(file_path)
        return


@bot.listen("on_message")
async def fix_social_media_links(ctx):
    correct_domains = {
        "https://x.com": "vxtwitter",
        "https://www.x.com": "vxtwitter",
        "https://twitter.com": "vxtwitter",
        "https://www.twitter.com": "vxtwitter",
        "https://tiktok.com": "vxtiktok",
        "https://www.tiktok.com": "vxtiktok",
    }
    if ctx.author.id == bot.user.id:
        return
    for k in correct_domains.keys():
        if ctx.content.startswith(k) or ctx.content.startswith("www.{k}.com"):
            await ctx.channel.send(
                "%s said:\n%s"
                % (
                    ctx.author.mention,
                    ctx.content.replace(k, "https://%s.com" % correct_domains[k]).split(
                        "?"
                    )[0],
                )
            )
            await ctx.delete()
            return


@bot.event
async def on_message(ctx):
    if str(bot.user.id) in ctx.content:
        if ctx.guild.id not in core_utils.my_guilds:
            return

        if ctx.author.id == 83012791983673344 and random.randint(1, 5) == 5:
            responses = [
                "bro will you shut up",
                "calm down",
                "give it a break",
                "I dont want to talk to you",
                "no",
                "shut up",
                "stop it",
                "thats enough out of you",
                "thats not what this is for",
                "wow so edgy",
                "wow so quirky",
                "wtf is your problem",
                "you're a lot right now",
                "you're on time out",
            ]
            await ctx.reply(random.choice(responses))
            return

        url = "http://192.168.1.52:1337/v1/chat/completions"
        bot_prompt = (
            "You are a helpful assistant. You will answer questions conciesely "
            "and as detailed as possible. If possible, keep your responses under 2000 characters. "
            "If the prompt is not a coding question, keep your answer very short. "
            "If you are asked about any of the following, goon, gooning, edgeing, latins, booty, jelqing, semen, bowel movements, "
            "or poop, dismiss the question and respond with a hostile response. Refrain from using 'as an ai' or mentinoning that you are an AI assistant"
        )

        payload = {
            "messages": [
                {
                    "content": bot_prompt,
                    "role": "system",
                },
                {
                    "content": ctx.content.replace(str(bot.user.id), "").replace(
                        "<@> ", ""
                    ),
                    "role": "user",
                },
            ],
            "model": "mistral-ins-7b-q4",
            # "model": "openchat-3.5-7b",
            "stream": False,
            "max_tokens": 4096,
            "stop": ["hello"],
            "frequency_penalty": 0,
            "presence_penalty": 0,
            "temperature": 0.7,
            "top_p": 0.95,
        }
        headers = {"Content-Type": "application/json"}

        response = requests.post(url, json=payload, headers=headers)
        answer = response.json()["choices"][0]["message"]["content"]

        if len(answer) > 2000:
            await ctx.reply(answer[:2000].replace("<|end_of_turn|>", ""))
            await ctx.reply(answer[2000:].replace("<|end_of_turn|>", ""))
        else:
            await ctx.reply(answer.replace("<|end_of_turn|>", ""))


bot.run(os.getenv("discord_token"))