foreverpyrite/jacobs-bot
1
0
Fork 0
Code Issues Pull Requests Activity

Everything seems to be stable, I think this is fair to be like 0.1.3 or something lmao. I don't know, really just toning his conversation is what'll make a v1.0.

Browse Source
This commit is contained in:
ForeverPyrite
2025-03-26 22:02:04 -04:00
parent f84605fb0d
commit 4178845057
10 changed files with 656 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
.dockerignore Normal file
View File

@@ -0,0 +1,6 @@
.gitignore
/logs
/.venv
/__pycache__
*.pyc
/data

1
.gitignore vendored
View File

@@ -2,3 +2,4 @@
/.venv
/logs
/__pycache__
/data

16
.vscode/launch.json vendored Normal file
View File

@@ -0,0 +1,16 @@
{
// Use IntelliSense to learn about possible attributes.
// Hover to view descriptions of existing attributes.
// For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
"version": "0.2.0",
"configurations": [
{
"name": "Python Debugger: Current File",
"type": "debugpy",
"request": "launch",
"program": "${file}",
"console": "integratedTerminal"
}
]
}

182
ai_logic.py Normal file
View File

@@ -0,0 +1,182 @@
import logging
import os
import asyncio
import discord # For type hintings
from json import loads as load_json
from openai import OpenAI
from openai.types.beta import Assistant, Thread
from openai.types.beta.threads import Run, Message, RequiredActionFunctionToolCall
from quotes import QUOTES
from students import *
logger = logging.getLogger(__name__)
OPENAI_KEY = os.getenv('OPENAI_KEY')
# OpenAI & Assistant configuration
client: OpenAI = OpenAI(
api_key=OPENAI_KEY,
project="proj_vUQ7duhKKc1jxIqllRhy6DVJ",
)
mr_jacobs: Assistant = client.beta.assistants.retrieve("asst_KdPdwqNAKijujfyCRrJCOgJN")
base_instructions: str = mr_jacobs.instructions
logger.info(f"Base instructions: {base_instructions}")
instructions: str = base_instructions + f"\n\nHere is a dictionary containing some of your most well known quotes, organized by their frequency. DO NOT USE THE QUOTES FREQUENTLY! JUST UTILIZE THEM TO KNOW THE TYPES OF THINGS YOU SAY! \n{QUOTES}\n\n"
def get_run_status(run: Run) -> str:
"""
Retrieves the status of a run.
Args:
run: The run object.
Returns:
str: The status of the run.
"""
status: str = client.beta.threads.runs.retrieve(
run.id,
thread_id=run.thread_id
).status
logger.info(f"Status of run {run.id} is {status}")
return status
def get_instructions(context: discord.abc.GuildChannel | discord.Message = None) -> str:
"""
Retrieves the instructions for the AI based on the context.
Args:
context (discord.abc.GuildChannel | discord.Message | None): The context from which to derive additional instructions.
If it's a GuildChannel, includes channel name and category.
If it's a Message, includes channel name and category from the message.
Returns:
str: The instructions for the AI.
"""
tuned_instructions: str = instructions
if context:
tuned_instructions += "\n\nNow I will give you the context of the channel that the message is sent in, keep it in mind. Please try to remain conversational depending on the topic of the channel, and don't be worshiping Cisco so religiously outside of on-topic. It's okay to joke about it, but you still wanna goof around with the students sometimes. However, if they have an essay, you should definately get on them about it outside of on-topic, so that you can still be helpful when they are talking about things on-topic, but can scold them to do their essay in any other channel. You should take the channel info fairly seriously, please. I really hate having to make such extreme channel descriptions to see a slight change in behavior from the Cisco worshipping. :\n"
if isinstance(context, discord.Message):
context = context.channel
if not isinstance(context, discord.abc.Messageable):
logger.critical(f"Literally how are we going to send a message in a channel that is not messageable? (bro how did we get here, wtf was passed as context? oh: {context}")
return tuned_instructions
if isinstance(context, discord.TextChannel):
tuned_instructions += f"\nChannel Category: {context.category.name}\nChannel Name:{context.name}\nChannel Description: {context.topic if isinstance(context, discord.TextChannel) else 'No Description'}"
elif isinstance(context, discord.DMChannel):
logger.warning(f"DM Channel detected, adding context to instructions.\nDM channel with {ID_TO_NAME.get(context.recipient.id, context.recipient.name)}")
tuned_instructions += f"\nThis is a direct message with {STUDENTS.get(context.recipient.id) if context.recipient.id in STUDENTS else context.recipient.name + 'No extra known info on user.'}"
return tuned_instructions
async def handle_action(run: Run) -> bool:
# Define the list to store tool outputs
tool_outputs = []
tool: RequiredActionFunctionToolCall
# Loop through each tool in the required action section
try:
logger.debug(f"Run.require_action is currently: {run.required_action} and tool calls are {run.required_action.submit_tool_outputs.tool_calls}")
for tool in run.required_action.submit_tool_outputs.tool_calls:
logger.info(f"Handling action for tool {tool.id}, function {tool.function.name}")
if tool.function.name == "assign_essay":
logger.debug(f"Handling action for assign_essay tool. Received arguments {tool.function.arguments}")
args = load_json(tool.function.arguments)
tool_outputs.append({
"tool_call_id": tool.id,
"output": assign_essay(int(run.metadata.get("user_id"))) if not tool.function.arguments else assign_essay(int(run.metadata.get("user_id")), args["topic"])
})
elif tool.function.name == "clear_essay":
logger.debug(f"Clearing the essay for student {run.metadata.get('user_id')}")
clear_essay(int(run.metadata.get("user_id")))
tool_outputs.append({
"tool_call_id": tool.id,
"output": "Essay cleared sucessfully, feel free to still give the student feedback on their essay because it couldn't have been perfect."
})
except AttributeError as e:
logger.error(f"Failed to handle action: {e}")
except Exception as e:
logger.critical(f"An unexpected error occurred while handling action: {e}")
finally:
logger.info(f"Tool outputs: {tool_outputs}")
if tool_outputs:
try:
run = client.beta.threads.runs.submit_tool_outputs_and_poll(
thread_id=run.thread_id,
run_id=run.id,
tool_outputs=tool_outputs
)
logger.info("Tool outputs submitted successfully.")
except Exception as e:
logger.error("Failed to submit tool outputs:", e)
else:
logger.warning("No tool outputs to submit.")
async def handle_run(run: Run) -> bool:
while True:
run = client.beta.threads.runs.retrieve(
run.id,
thread_id=run.thread_id
)
match run.status:
case "completed":
return True
case "failed":
logger.error(f"Run {run.id} failed.")
return False
case "requires_action":
logger.info(f"Run {run.id} requires action.")
await handle_action(run)
case _:
await asyncio.sleep(1)
await asyncio.sleep(1)
async def run(messages: list[dict], instructions: str = instructions, user_id : int = None) -> str:
"""
Runs the AI with the given messages and instructions.
Args:
messages (list[dict]): The list of messages.
instructions (str): The instructions for the AI.
user_id (int): The Discord ID of the user who initiated the run.
Returns:
str: The response from the AI.
"""
logger.debug(f"Running AI assistant with the following parameters:\nmessages={messages}\ninstructions={instructions}\nmetadata={str(user_id if user_id else -1)}")
run = client.beta.threads.create_and_run(
assistant_id=mr_jacobs.id,
instructions=instructions,
thread={
"messages": messages
},
metadata={
"user_id": str(user_id if user_id else -1)
}
)
response = await run_message(run)
return response
async def run_message(run) -> str:
"""
Retrieves the response message from a run.
Args:
run: The run object.
Returns:
str: The response message.
"""
logger.debug(f"Retrieving response message for run ID {run.id}")
# ew but Python doesn't have a do while and it's less ugly than the same statement twice.
await handle_run(run)
thread_messages = client.beta.threads.messages.list(run.thread_id)
for msg_ob in thread_messages.data:
if msg_ob.id == thread_messages.first_id:
response = msg_ob.content[0].text.value
return response
logger.critical(f"Couldn't find the msg that matched with the first message ID:\nThread Messages List:\n{thread_messages}")

123
app.py
View File

@@ -1,123 +1,40 @@
import os
import logging
from dotenv import load_dotenv
import discord
from discord import Bot
from openai import OpenAI
from asyncio import run
import pytz
from apscheduler.schedulers.background import BackgroundScheduler
from students import STUDENTS
from quotes import QUOTES, select_quote, select_student
from discord_logic import bot, setup_discord_bot, send_quote, after_class
# Load environment variables
load_dotenv()
DISCORD_TOKEN = os.getenv('DISCORD_TOKEN')
# Logging.
logging.basicConfig(
filename="./logs/jacobs.log",
filename=f"./logs/jacobs.log",
filemode="at+",
level=logging.DEBUG
level=logging.DEBUG if os.getenv('DEBUG') == "True" else logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
DISCORD_TOKEN = os.getenv('DISCORD_TOKEN')
OPENAI_KEY = os.getenv('OPENAI_KEY')
setup_discord_bot(bot)
# Discord
bot = Bot(intents=discord.Intents.default())
SERVER_ID: int = 1339601046745645148
GENERAL_ID: int = 1339601047294840876
ch_general = None
@bot.event
async def on_ready() -> None:
"""Event handler for when the bot is initialized."""
logger.info(f"{bot.user.name} has connected to Discord and is ready.")
print(f"{bot.user.name} is ready.")
async def get_general():
global ch_general
global bot
if not ch_general:
ch_general = await bot.fetch_channel(GENERAL_ID)
if not ch_general:
raise(ValueError("General channel object should *not* be none!"))
return ch_general
async def send_quote(quote: str = None) -> None:
await get_general()
if not quote:
quote = select_quote()
logger.info(f"Sending quote {quote} in #general...")
await ch_general.send(quote)
async def after_class(student: int = None) -> None:
global ch_general
if not student:
student = select_quote()
logger.info(f"Sending mention to see {student} after class to #general...")
await ch_general.send(f"Come see me after class <@{student}>")
# OpenAI & Assistant configuration
client = OpenAI(
api_key=OPENAI_KEY,
project="proj_vUQ7duhKKc1jxIqllRhy6DVJ",
)
mr_jacobs = client.beta.assistants.retrieve("asst_KdPdwqNAKijujfyCRrJCOgJN")
base_instructions = mr_jacobs.instructions
instructions = base_instructions + f"\n\nHere is a dictionary containing some of your most well known quotes, organized by their frequency: \n{QUOTES}\n\nStudent: "
def get_run_status(run):
status = client.beta.threads.runs.retrieve(
run.id,
thread_id=run.thread_id
).status
logger.info(f"Status of run {run.id} is {status}")
return status
def get_instructions(student: int | None = None) -> str:
logging.info(f"Looking for {student} in students...")
if student in STUDENTS:
return instructions + STUDENTS.get(student)
else:
logging.warning(f"Couldn't find {student}")
return instructions + "Unknown"
async def run_message(run):
# ew but Python doesn't have a do while and it's less ugly than the same statement twice.
while True:
complete = get_run_status(run) == "completed"
if complete:
break
thread_messages = client.beta.threads.messages.list(run.thread_id)
for msg_ob in thread_messages.data:
if msg_ob.id == thread_messages.first_id:
response = msg_ob.content[0].text.value
return response
raise Exception(f"bro what did you do dumbass, figure it out.:\nThread Messages List:\n{thread_messages}")
@bot.slash_command(description="Talk to Mr. Jacobs!!!")
async def message(ctx: discord.ApplicationContext, text: str):
logging.info(f"User {ctx.author.global_name} sent message {text}")
instructions = get_instructions(ctx.author.id)
bot_reply = await ctx.respond("*hmmmm*...let me see here...")
run = client.beta.threads.create_and_run(
assistant_id=mr_jacobs.id,
instructions=instructions,
thread={
"messages": [
{
"role": "user",
"content": text
}
]
}
)
response = await run_message(run)
await bot_reply.edit_original_response(content=response)
# After successful initilization, schedule tasks.
task_scheduler = BackgroundScheduler()
# After successful initialization, schedule tasks.
task_scheduler = BackgroundScheduler(timezone=pytz.timezone('EST'))
task_scheduler.add_job(lambda: bot.loop.create_task(send_quote()), 'cron', day_of_week='mon-fri', hour=8, minute=50)
task_scheduler.add_job(lambda: bot.loop.create_task(send_quote()), 'cron', day_of_week='mon-fri', hour='8-14', minute="*/20", jitter=180)
task_scheduler.add_job(lambda: bot.loop.create_task(after_class()), 'cron', day_of_week='mon-fri', hour=10, minute=55)
task_scheduler.start()
logger.info(f"Presumably successfully initilized, starting bot.")
logger.info("Presumably successfully initialized, starting bot.")
bot.run(DISCORD_TOKEN)

411
discord_logic.py Normal file
View File

@@ -0,0 +1,411 @@
import logging
import regex as re
from random import randint
from asyncio import run as asyncio_run, sleep
import discord
from discord import Bot, Message
from discord.abc import GuildChannel as Channel
from ai_logic import run, get_instructions
from quotes import select_quote, select_student
from students import *
logger = logging.getLogger(__name__)
SERVER_ID: int = 1339601046745645148
GENERAL_ID: int = 1339601047294840876
bot = Bot(intents=discord.Intents.all())
# ---------- Discord Helper Functions ----------
async def get_channel(channel_id: int) -> Channel:
"""
Tries to get a cached channel object, if it fails it will send an API request to retrieve a new one.
Args:
channel_id (int): The ID of the channel to retrieve.
Raises:
ValueError: If the channel cannot be retrieved.
Returns:
discord.abc.GuildChannel: The retrieved discord channel object.
"""
logger.debug(f"Attempting to get channel with ID {channel_id}")
channel = bot.get_channel(channel_id)
if not channel:
logger.debug(f"Channel with ID {channel_id} not found in cache, fetching from API.")
channel = await bot.fetch_channel(channel_id)
if not channel:
logger.critical(f"Could not fetch channel with channel_id {channel_id}, fetch_channel returned None.")
return None
logger.info(f"Successfully retrieved {channel_id}, #{channel.name}")
return channel
async def get_message(message: Message | int, channel_id: int = None, attempts: int = 3) -> Message:
"""
Retrieves a message object either from cache or by fetching it from the channel.
Args:
message (Message | int): The message object or message ID to retrieve.
channel_id (int, optional): The ID of the channel to fetch the message from if not in cache.
attempts (int, optional): The number of attempts to retrieve the message. Defaults to 3.
Returns:
Message: The retrieved message object.
"""
logger.debug(f"Attempting to get message with ID {message if isinstance(message, int) else message.id}")
for attempt in range(attempts):
if isinstance(message, int):
got_message: Message = bot.get_message(message)
if not got_message and channel_id:
logger.debug(f"Message with ID {message} not found in cache, fetching from channel ID {channel_id}.")
channel: discord.abc.GuildChannel = await get_channel(channel_id)
got_message: Message = await channel.fetch_message(message)
elif not got_message:
logger.error(f"Message with ID {message} not found in cache and no channel ID provided.")
elif isinstance(message, Message):
got_message: Message = bot.get_message(message.id)
if not got_message:
logger.debug(f"Message with ID {message.id} not found in cache, fetching from channel.")
got_message: Message = await message.channel.fetch_message(message.id)
else:
logger.error(f"Couldn't retrieve message:\nMessage: {message}\nChannel ID: {channel_id}")
return None
if got_message:
return got_message
if attempt < attempts - 1:
logger.warning(f"Attempt {attempt + 1} failed. Retrying in 2 seconds...")
await sleep(2)
logger.error(f"Failed to retrieve message after {attempts} attempts.")
return None
async def send_message(message: str, channel: int | Channel = GENERAL_ID) -> Message:
"""
Has the bot send a message to a specified channel. If no channel is specified, the bot sends it to general.
Args:
message (str): The message to send.
channel (int | Message, optional): Channel to send message to, Defaults to GENERAL_ID.
Returns:
Message: The sent message object.
"""
logger.debug(f"Attempting to send message: {message}")
if isinstance(channel, int):
channel = await get_channel(channel)
if isinstance(channel, Channel):
sent_message: Message = await channel.send(message)
if isinstance(sent_message, Message):
logger.info(f"Message '{message}' successfully sent to {sent_message.channel}")
return sent_message
else:
logger.error(f"Message likely wasn't successfully sent, as channel.send did not return a Message.")
elif isinstance(channel, None):
logger.error(f"Message couldn't be sent as a channel wasn't found/messagable.")
async def edit_message(message: Message | int, channel: int = None):
"""
Edits a message in a channel.
Args:
message (Message | int): The message to edit.
channel (int, optional): The channel ID of the message. Defaults to None.
"""
logger.debug(f"Attempting to edit message {message if isinstance(message, int) else message.id}")
if isinstance(message, int):
message = await get_message(message, channel)
if isinstance(message, Message):
await message.edit(content=message.content)
logger.info(f"Message {message.id} successfully edited.")
else:
logger.error(f"Message couldn't be edited as it was not found.")
# ---------- Message Utility Functions ----------
async def msg_is_reply(message: Message) -> tuple[bool, Message]:
"""
Checks if a message is a reply to another message.
Args:
message (Message): The message to check.
Returns:
tuple[bool, Message]: A tuple containing a boolean indicating if the message is a reply and the replied message.
"""
logger.debug(f"Checking if message ID {message.id} is a reply.")
if message.reference is not None:
replied_msg = message.reference.cached_message
if replied_msg is None:
logger.debug(f"Replied message not found in cache, fetching from channel ID {message.reference.channel_id}.")
channel = await get_channel(message.reference.channel_id)
replied_msg = await channel.fetch_message(message.reference.message_id)
if isinstance(replied_msg, Message):
return True, replied_msg
return False, None
async def get_reply_chain(msg: Message) -> list[Message]:
"""
Retrieves the chain of replies for a given message.
Args:
msg (Message): The message to get the reply chain for.
Returns:
list[Message]: A list of messages in the reply chain.
"""
logger.debug(f"Getting reply chain for message ID {msg.id}.")
i = 0
reply_chain = [msg]
while i < 10:
is_reply, msg = await msg_is_reply(msg)
if not is_reply:
break
reply_chain.insert(0, msg)
i += 1
return reply_chain
async def gen_message_list(discord_messages: list[Message]) -> list[dict]:
"""
Generates a list of message dictionaries from a list of Discord messages.
Args:
discord_messages (list[Message]): The list of Discord messages.
Returns:
list[dict]: A list of message dictionaries.
"""
logger.debug("Generating message list from Discord messages.")
messages = []
if not isinstance(discord_messages, list):
discord_messages = [discord_messages]
for message in discord_messages:
students_mentioned = None
if isinstance(message, Message):
message_copy = {
"id": message.id,
"content": message.content,
"author": message.author,
"mentions": message.mentions,
"created_at": message.created_at,
"jump_url": message.jump_url
}
role = "assistant" if message_copy["author"].id == bot.user.id else "user"
students_mentioned = None
if message_copy["mentions"] and role == "user":
students_mentioned = [student.id for student in message_copy["mentions"] if student.id in STUDENT_IDS]
mentions = re.findall(r'<@\d{17,19}>', message_copy["content"])
for mention in mentions:
mention_id = mention.replace("<@", "").replace(">", "")
student_info = ID_TO_NAME.get(int(mention_id))
if student_info:
message_copy["content"] = message_copy["content"].replace(mention, student_info.split("\n")[0])
message_copy["content"] += '\n{'
if students_mentioned:
message_copy["content"] += f"\n\nInfo on students mentioned (DO NOT REPEAT!): "
for student in students_mentioned:
message_copy["content"] += f"\n{STUDENTS.get(student)}"
message_copy["content"] += f"\n\nSent by: {STUDENTS.get(message_copy['author'].id) if message_copy['author'].id in STUDENT_IDS else message_copy['author'].name + ' (Author\'s details not found)'}" + '}'
if not message_copy["content"]:
debug = True
logger.error(f"No message content for message {message_copy['id']}.")
fetched_message = await get_message(message_copy["id"])
message_copy["content"] = fetched_message.content
thread_message = {
"role": role,
"content": message_copy["content"]
}
debug = True # Delete after fix
if debug:
debug_dump = "Messages Dump:\n"
for msg_copy in discord_messages:
debug_dump += f"""
Message: {msg_copy}
Message Time: {msg_copy.created_at}
Message ID: {msg_copy.id}
Message Author: {msg_copy.author}
Message Content: {msg_copy.content}
Link: {msg_copy.jump_url}
"""
logger.debug(debug_dump)
messages.append(thread_message)
else:
logger.warning(f"Argument {message} is not of type Message and will be skipped.")
return messages
# ---------- AI Interaction Functions ----------
async def send_quote(quote: str = None) -> None:
"""
Sends a quote to the general channel.
Args:
quote (str, optional): The quote to send. If not provided, a random quote is selected.
"""
if not quote:
quote = select_quote()
logger.info(f"Sending quote '{quote}' in #general...")
await send_message(quote)
async def after_class(student: int = None) -> None:
"""
Sends a message to a student to see the bot after class.
Args:
student (int, optional): The ID of the student. If not provided, a random student is selected.
"""
if not student:
student = select_student()
logger.info(f"Sending mention to see {student} after class to #general...")
await send_message(f"Come see me after class <@{student}>")
async def has_essay(message: Message) -> bool:
if message.author.id in ASSIGNED_ESSAY:
async with message.channel.typing():
message = await get_message(message) if not message else message
message_list = await gen_message_list(message)
response = await run(
message_list,
f"{get_instructions(message)}\n\nTHIS USER HAS AN ESSAY!!! Yell at them and tell them to finish their essay on {ASSIGNED_ESSAY.get(message.author.id)}!!! Unless...they sent you the essay, in that case, you can decide whether or not it is satisfactory (it needs to be in MLA format and 300-2000 characters long, and to be relevant of course). If it is satisfactory, you can tell them they are free to go, after you run the function \"clear_essay\".",
message.author.id
)
await message.reply(response)
return True
else:
return False
# ---------- Discord Commands & Event Handlers ----------
def setup_discord_bot(bot: Bot) -> None:
"""
Sets up the Discord bot with commands and event listeners.
Args:
bot (Bot): The Discord bot instance.
"""
@bot.slash_command(description="Talk to Mr. Jacobs!!!")
async def message(ctx: discord.ApplicationContext, text: str) -> None:
"""
Slash command to talk to Mr. Jacobs. The bot will respond to the user's message using AI.
Args:
ctx (discord.ApplicationContext): The context of the command.
text (str): The message text.
"""
logging.info(f"User {ctx.author.global_name} sent message {text}")
await ctx.defer()
instructions = get_instructions(ctx.channel)
thread_messages = [{"role": "user", "content": text}]
response = await run(thread_messages, instructions, ctx.author.id)
await ctx.respond(content=response)
@bot.slash_command(
name="rps",
description="Play \"Rock, Paper, Scissors, Essay\" with Mr. Jacobs!!!",
)
async def rps_essay(
ctx: discord.ApplicationContext,
choice: str = discord.Option(
description="Your selection for the game",
choices=[
discord.OptionChoice("Rock"),
discord.OptionChoice("Paper"),
discord.OptionChoice("Scissors")
]
)
) -> None:
"""
Play Rock Paper Scissors with Mr. Jacobs.
Args:
ctx: Application command context
choice: Your selection for the game
"""
logging.info(f"{ctx.author} chose {choice}")
outcomes = {"Rock": "Paper", "Paper": "Scissors", "Scissors": "Rock"}
await ctx.respond(
f"I choose {outcomes[choice]}, you owe me a {randint(1,15)} page essay "
f"on {assign_essay(ctx.author.id)}. MLA format, double spaced, "
"12pt Times New Roman!"
)
@bot.event
async def on_message(message: Message) -> None:
"""
Event listener for new messages. The bot will respond to messages that mention it or reply to its messages.
Bot disregards its own messages.
Args:
message (Message): The new message.
"""
logger.debug(f"New message received: {message.content}")
if message.author.id == bot.user.id or message.interaction_metadata:
return
if await handle_reply(message):
return
if await handle_mentions(message):
return
async def handle_reply(message: Message) -> bool:
"""
Handles replies to the bot's messages.
Args:
message (Message): The new message.
Returns:
bool: True if the message was handled, False otherwise.
"""
is_reply, first_reply = await msg_is_reply(message)
if is_reply and first_reply.author.id == bot.user.id:
async with message.channel.typing():
if await has_essay(message):
return True
reply_chain = await get_reply_chain(message)
thread_messages = await gen_message_list(reply_chain)
await process_thread_messages(thread_messages, message)
return True
return False
async def handle_mentions(message: Message) -> bool:
"""
Handles messages that mention the bot.
Args:
message (Message): The new message.
Returns:
bool: True if the message was handled, False otherwise.
"""
if message.mentions:
for user in message.mentions:
if user.id == bot.user.id:
async with message.channel.typing():
if await has_essay(message):
return True
thread_messages = await gen_message_list(message)
await process_thread_messages(thread_messages, message)
return True
return False
async def process_thread_messages(thread_messages: list, message: Message) -> None:
"""
Processes a list of thread messages and sends a response.
Args:
thread_messages (list): The list of thread messages.
message (Message): The original message.
"""
instructions = get_instructions(message)
response = await run(thread_messages, instructions, message.author.id)
await message.reply(response)

4
docker-compose.yml
View File

@@ -1,7 +1,9 @@
services:
discord-bot:
build: .
container_name: jacobs-bot
image: docker.foreverpyrite.com/jacobs-bot:beta
volumes:
- ./logs/:/app/logs/
- ./data/:/app/data/
env_file: .env
restart: unless-stopped

22
quotes.py
View File

@@ -16,26 +16,36 @@ QUOTES = {
"Hmm... I see.",
"What are we doing over here?",
"Mmm... Okay! :clap:",
"*Loudly* We don't like stupid prizes, now do we? Then, don't play stupid games!",
"*LOUDLY* We don't like stupid prizes, now do we? Then, don't play stupid games!",
"You guys are killing it!",
"Do we need to go back over the module again?",
"Let's get it done!",
"Sorry, I can't hear over Devan and Connor talking.",
"That's what I like to hear!"
"That's what I like to hear!",
"*Paces*",
"*Sprints to the ringing phone*",
"*Spins lanyard*",
"Come on guys, you should know this already!",
"Let me just say this...",
"Not trying to be mean or anything...but..."
),
"RARE" : (
"Play stupid games, win big prizescat jacobs.sh! 🤑🤑",
"Play stupid games, win big prizes! 🤑🤑",
"Oooooo raahahah!",
"It's cherry-pickin' time, y'all!",
"What does the fox say?"
),
"MYTHIC" : (
"I'm proud of you.",
"You can take a 5-minute break."
"You can take a 5-minute break.",
"I have somewhere to be at 9:30, so you guys will have a sub."
),
"LEGENDARY": (
)
}
def select_quote():
def select_quote() -> str:
rarity = random.randint(0, 99)
if rarity < 1:
quote = random.choice(QUOTES.get("MYTHIC"))
@@ -48,7 +58,7 @@ def select_quote():
log_file.write(f"At {datetime.datetime.now()}, rarity was rolled as {rarity} and selected: {quote}\n")
return quote
def select_student():
def select_student() -> int:
student = random.choice(STUDENT_IDS)
with open("./logs/quotes.log", "at+") as log_file:
log_file.write(f"At {datetime.datetime.now()}, wanted to see {student} after class.\n")

BIN
requirements.txt
View File

Binary file not shown.

0
todo Normal file
View File