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

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

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

        llm_rules = requests.get(core_utils.json_endpoint + "rules.json").json()

        if ctx.author.id in llm_rules["disallowed_users"]:
            responses = [
                "You cant do that right now",
                "You cant use this feature right now",
                "You're on time out from this",
            ]
            await ctx.respond(random.choice(responses))
            return

        url = "http://192.168.1.137:1337/v1/chat/completions"
        instructions = llm_rules["prompt"]

        payload = {
            "messages": [
                {
                    "content": instructions,
                    "role": "system",
                },
                {
                    "content": ctx.content.replace(str(bot.user.id), "").replace(
                        "<@> ", ""
                    ),
                    "role": "user",
                },
            ],
            "model": llm_rules["model"],
            "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"))