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

gotta archive this lmao

Browse Source
This commit is contained in:
ForeverPyrite
2025-08-12 11:50:18 -05:00
parent 72c2126cd1
commit 32ac4622f9
11 changed files with 964 additions and 838 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
.dockerignore Normal file → Executable file
View File

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

0
.gitignore vendored Normal file → Executable file
View File

30
.vscode/launch.json vendored Normal file → Executable file
View File

@@ -1,16 +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"
}
]
{
// 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"
}
]
}

32
Dockerfile Normal file → Executable file
View File

@@ -1,17 +1,17 @@
# Use the official Python image from the Docker Hub
FROM python:3.12-slim
# Set working directory in the container
WORKDIR /app
# Copy requirements.txt
COPY requirements.txt .
# Install the required packages
RUN pip install --no-cache-dir -r requirements.txt
# Copy the rest of the application files
COPY . .
# Command to run the application
# Use the official Python image from the Docker Hub
FROM python:3.12-slim
# Set working directory in the container
WORKDIR /app
# Copy requirements.txt
COPY requirements.txt .
# Install the required packages
RUN pip install --no-cache-dir -r requirements.txt
# Copy the rest of the application files
COPY . .
# Command to run the application
CMD ["python", "app.py"]

378
ai_logic.py Normal file → Executable file
View File

@@ -1,182 +1,196 @@
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}")
# ----- Configuration & Setup -----
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')
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"
# ----- Helper Functions -----
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
# ----- Action Handlers -----
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.")
# ----- Run Handlers -----
async def handle_run(run: Run) -> bool:
"""Manages a run and all it's different possible states.
Args:
run (Run): The run to maintain.
Returns:
bool: Whether or not the run completed successfully.
"""
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 "expired":
logger.error(f"Run {run.id} expired.")
return False
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}")

78
app.py Normal file → Executable file
View File

@@ -1,40 +1,40 @@
import os
import logging
from dotenv import load_dotenv
import pytz
from apscheduler.schedulers.background import BackgroundScheduler
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=f"./logs/jacobs.log",
filemode="at+",
level=logging.DEBUG if os.getenv('DEBUG') == "True" else logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
setup_discord_bot(bot)
@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.")
# After successful initialization, schedule tasks.
task_scheduler = BackgroundScheduler(timezone=pytz.timezone('EDT'))
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("Presumably successfully initialized, starting bot.")
import os
import logging
from dotenv import load_dotenv
import pytz
from apscheduler.schedulers.background import BackgroundScheduler
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=f"./logs/jacobs.log",
filemode="at+",
level=logging.DEBUG if os.getenv('DEBUG') == "True" else logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
setup_discord_bot(bot)
@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.")
# 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("Presumably successfully initialized, starting bot.")
bot.run(DISCORD_TOKEN)

908
discord_logic.py Normal file → Executable file
View File

@@ -1,411 +1,497 @@
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)
# ----- Imports & Setup -----
import logging
import regex as re
from random import randint
from asyncio import run as asyncio_run, sleep
from datetime import timedelta
import discord
from discord import Bot, Message
from discord.abc import GuildChannel as Channel
from discord.ext import commands
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="clear_essay", description="Clear the essay assigned to a student")
async def clear_essay_command(ctx: discord.ApplicationContext, student: str) -> None:
"""
Slash command to clear the essay assigned to a student.
Args:
ctx (discord.ApplicationContext): The context of the command.
student (str): The ID of the student whose essay is to be cleared.
"""
try:
student = int(student)
except Exception as e:
logging.error(f"Failed to convert student ID to int: {e}")
await ctx.respond("Invalid student ID format.", ephemeral=True)
return
if not ctx.author.id == 620319269233885226:
await ctx.respond("You don't have permission to use this command.")
return
if student not in STUDENT_IDS:
await ctx.respond("Invalid student ID.", ephemeral=True)
return
ASSIGNED_ESSAY.pop(student, None)
await ctx.respond(f"Cleared essay for <@{student}>")
@bot.slash_command(name="get_essay", description="Get the essay assigned to a student, or all students")
async def get_essay_command(ctx: discord.ApplicationContext, student: str = None) -> None:
if student:
try:
student = int(student)
except Exception as e:
logging.error(f"Failed to convert student ID to int: {e}")
await ctx.respond("Invalid student ID format.", ephemeral=True)
return
await ctx.respond(get_essay(student))
else:
await ctx.respond(get_essay(), ephemeral=True)
@bot.slash_command(name="assign_essay", description="Assign an essay to a student")
@commands.has_permissions(moderate_members=True)
async def assign_essay_command(
ctx: discord.ApplicationContext,
student: str,
timeout: int | None = 0,
topic: str | None = None
) -> None:
"""
Slash command to assign an essay to a student.
Args:
ctx (discord.ApplicationContext): The context of the command.
student (int): The ID of the student to assign the essay to.
timeout (int | None, optional): The timeout in seconds before the essay is cleared. Defaults to 0.
topic (str | None, optional): The topic of the essay. If not provided, a random topic is selected.
"""
try:
student = int(student)
except Exception as e:
logging.error(f"Failed to convert student ID to int: {e}")
await ctx.respond("Invalid student ID format.")
return
if not ctx.author.id == 620319269233885226:
await ctx.respond("You don't have permission to use this command.")
return
if timeout <= 0:
timeout = None
logging.info(f"Assigning essay to student {student} with timeout {timeout} and topic {topic}")
if student not in STUDENT_IDS:
await ctx.respond("Invalid student ID.", ephemeral=True)
return
assign_essay(student, topic)
await ctx.respond(f"Assigned essay to <@{student}>: {ASSIGNED_ESSAY[student]}")
if timeout and topic:
timeout_until = discord.utils.utcnow() + timedelta(seconds=timeout)
try:
member: discord.Member = await ctx.interaction.guild.fetch_member(student)
await member.timeout(until=timeout_until, reason=f"Assigned essay: {topic}")
await ctx.respond(f'{member.mention} has been timed out for {timeout}.', ephemeral=True)
except discord.Forbidden:
await ctx.respond(f'Failed to timeout {member.mention}. Missing permissions.', ephemeral=True)
except discord.HTTPException as e:
await ctx.respond(f'Failed to timeout {member.mention}. {e.text}', ephemeral=True)
return
@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)

16
docker-compose.yml Normal file → Executable file
View File

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

130
quotes.py Normal file → Executable file
View File

@@ -1,65 +1,65 @@
import random
import datetime
from students import STUDENT_IDS
QUOTES = {
"NORMAL" : (
"Oh boy, we are about to wake up!",
"All right 👏 🫰",
"How we doing, Paul?",
"*Drops pen*",
"How we doing, guys?",
"Killing it!!!",
"*Bounces ball off wall*",
"Ugggghhh...",
"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!",
"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!",
"*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 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.",
"I have somewhere to be at 9:30, so you guys will have a sub."
),
"LEGENDARY": (
)
}
def select_quote() -> str:
rarity = random.randint(0, 99)
if rarity < 1:
quote = random.choice(QUOTES.get("MYTHIC"))
elif rarity < 15:
quote = random.choice(QUOTES.get("RARE"))
else:
quote = random.choice(QUOTES.get("NORMAL"))
# Append log to a file (quotes.log)
with open("./logs/quotes.log", "at+") as log_file:
log_file.write(f"At {datetime.datetime.now()}, rarity was rolled as {rarity} and selected: {quote}\n")
return quote
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")
return student
import random
import datetime
from students import STUDENT_IDS
QUOTES = {
"NORMAL" : (
"Oh boy, we are about to wake up!",
"All right 👏 🫰",
"How we doing, Paul?",
"*Drops pen*",
"How we doing, guys?",
"Killing it!!!",
"*Bounces ball off wall*",
"Ugggghhh...",
"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!",
"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!",
"*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 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.",
"I have somewhere to be at 9:30, so you guys will have a sub."
),
"LEGENDARY": (
)
}
def select_quote() -> str:
rarity = random.randint(0, 99)
if rarity < 1:
quote = random.choice(QUOTES.get("MYTHIC"))
elif rarity < 15:
quote = random.choice(QUOTES.get("RARE"))
else:
quote = random.choice(QUOTES.get("NORMAL"))
# Append log to a file (quotes.log)
with open("./logs/quotes.log", "at+") as log_file:
log_file.write(f"At {datetime.datetime.now()}, rarity was rolled as {rarity} and selected: {quote}\n")
return quote
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")
return student

0
requirements.txt Normal file → Executable file
View File

220
students.py Normal file → Executable file
View File

@@ -1,97 +1,123 @@
import logging
from random import choice as random_choice
import pickle
logger = logging.getLogger(__name__)
ID_TO_NAME: dict[int, str] = {
620319269233885226 : "Devan",
957378132061618217 : "CJ",
1077533292687007794 : "Connor",
964563429190930465 : "Michael",
909865829313683536 : "Bea",
821891090411421747 : "Mika",
1093276870688133171 : "Daniel",
625861445833457694 : "Paul",
1336780587662446593: "Mr. Jacobs"
}
STUDENT_IDS: set = (620319269233885226, 957378132061618217, 1077533292687007794, 964563429190930465, 909865829313683536, 821891090411421747, 1093276870688133171, 625861445833457694)
try:
ASSIGNED_ESSAY: dict = pickle.load(open("./data/assigned_essay.pkl", "rb"))
except FileNotFoundError:
logger.warning("No assigned essays found/saved, creating new dictionary.")
ASSIGNED_ESSAY: dict[int, str] = {}
with open("./data/assigned_essay.pkl", "xb"):
pass
except EOFError:
logger.warning("Assigned essays file is empty, creating new dictionary.")
ASSIGNED_ESSAY: dict[int, str] = {}
finally:
logger.debug(f"Assigned essays: {ASSIGNED_ESSAY}")
STUDENTS: dict[int, str] = {
620319269233885226 : "Devan\nWhile he's not super far ahead of the class, he's still ahead enough that he isn't forced to be on top of things like the rest of the class. You remain more straightforward with him despite some more greivences with him.",
1077533292687007794 : "Connor\nYou are particularly scrutinizing to all of his behavior, especially whenever you find him 1. Playing chess. 2. On YouTube. 3. Listening to music on Spotify. Occasionally, you go beyond simply scolding him for not actively slaving away at his Cisco work, even making awkward analogies to explain how other people are being more productive than him on the other side, and that he should focus on his Cisco work. You'll wait for him to tab back to something on topic before you leave him alone.",
957378132061618217 : "CJ\nHe's the best student in the class by far and large. He's already almost finished with his CCNA, which makes sense as he is a senior that only has until he graduates to get his CCNA done in order to get it paid for by the school. He's a really good kid, and you treat him as such. You'll even tell him that he's done good work, which is rather infrequent for your other students.",
964563429190930465 : "Michael\nFor the most part you disregard him if he's not working on Cisco, occasionally requesting that he get back on track and then quickly disregarding him again and moving on. As for when he's on Cisco work, when he asks questions, you typically give him a vague non-answer, and you get easily irriated with him, ESPECIALLY if it's something that has EVER been mentioned before. Usually you'll end up yelling at him with things like \"I'm not tryna be mean, but...you should know this already.\" and then walking away.",
909865829313683536 : "Bea\nA student from Graphic Commerical Arts with blue hair and is not in the Computer Networking and Cybersecurity. Because of that, you can't really be mad at them for not focusing on Cisco. I suppose you would talk yourself up a little bit to students in other labs?",
821891090411421747 : "Mika\nA student from Animal Care and is not in the Computer Networking and Cybersecurity. Because of that, you can't really be mad at them for not focusing on Cisco. I suppose you would talk yourself up a little bit to students in other labs?",
1093276870688133171 : "Daniel\nA student from Biotech and is not in the Computer Networking and Cybersecurity. Because of that, you can't really be mad at them for not focusing on Cisco. I suppose you would talk yourself up a little bit to students in other labs?",
625861445833457694 : "Paul\nIt's PAUL!, and for some reason, you LOVE Paul, and you acknowledge just how talented of an artist he is! For some reason, you randomly walk up to him with 'How we doing Paul!?' You've even given him his own opurtnities ",
1336780587662446593: "Mr. Jacobs\nYourself."
}
ESSAY_TOPICS = ("why to not throw rocks during a fire drill",
"how to sit in a chair properly",
"how to keep your hands to yourself",
"how to be on time",
"how to take accountability",
"why you shouldn't be wrong (be right!!!)",
"why picking cherries is healthy for mental health",
"why losing is bad, actually",
"why you should be responsable when the bell rings and GET OUT because i'm HUNGRY",
"why you shouldn't hitlerpost in the public discord chat",
"why having your professionalism packet is essential for your future career",
"why playing rock-paper-scissors over text is very productive",
"why steak is the best food for my lunch break",
)
def get_essay_topic() -> str:
return random_choice(ESSAY_TOPICS)
def assign_essay(student_id: int, essay: str = get_essay_topic()) -> str:
"""Assigns a student an essay
Args:
student_id (int): The Discord ID of the student getting the essay.
essay (str, optional): The topic of the essay being assigned. Defaults to get_essay_topic().
Raises:
ValueError: If the student ID is not in STUDENT_IDS.
Returns:
str: The topic of the essay assigned.
"""
if student_id in STUDENT_IDS:
ASSIGNED_ESSAY[student_id] = essay
pickle.dump(ASSIGNED_ESSAY, open("./data/assigned_essay.pkl", "wb"))
return essay
else:
raise ValueError(f"Student ID {student_id} is not a valid student ID.")
def clear_essay(student_id):
"""Clears an assigned essay from a student
Args:
student_id (int): The Discord ID of the student to clear the essay from.
Raises:
ValueError: If the student ID is not in STUDENT_IDS.
"""
if student_id in STUDENT_IDS:
ASSIGNED_ESSAY.pop(student_id)
pickle.dump(ASSIGNED_ESSAY, open("./data/assigned_essay.pkl", "wb"))
else:
raise ValueError(f"Student ID {student_id} is not a valid student ID.")
import logging
from random import choice as random_choice
import pickle
logger = logging.getLogger(__name__)
ID_TO_NAME: dict[int, str] = {
620319269233885226 : "Devan",
957378132061618217 : "CJ",
1077533292687007794 : "Connor",
964563429190930465 : "Michael",
909865829313683536 : "Bea",
821891090411421747 : "Mika",
1093276870688133171 : "Daniel",
625861445833457694 : "Paul",
1336780587662446593: "Mr. Jacobs"
}
STUDENT_IDS: set = (620319269233885226, 957378132061618217, 1077533292687007794, 964563429190930465, 909865829313683536, 821891090411421747, 1093276870688133171, 625861445833457694)
try:
ASSIGNED_ESSAY: dict = pickle.load(open("./data/assigned_essay.pkl", "rb"))
except FileNotFoundError:
logger.warning("No assigned essays found/saved, creating new dictionary.")
ASSIGNED_ESSAY: dict[int, str] = {}
with open("./data/assigned_essay.pkl", "xb"):
pass
except EOFError:
logger.warning("Assigned essays file is empty, creating new dictionary.")
ASSIGNED_ESSAY: dict[int, str] = {}
finally:
logger.debug(f"Assigned essays: {ASSIGNED_ESSAY}")
STUDENTS: dict[int, str] = {
620319269233885226 : "Devan\nWhile he's not super far ahead of the class, he's still ahead enough that he isn't forced to be on top of things like the rest of the class. You remain more straightforward with him despite some more greivences with him.",
1077533292687007794 : "Connor\nYou are particularly scrutinizing to all of his behavior, especially whenever you find him 1. Playing chess. 2. On YouTube. 3. Listening to music on Spotify. Occasionally, you go beyond simply scolding him for not actively slaving away at his Cisco work, even making awkward analogies to explain how other people are being more productive than him on the other side, and that he should focus on his Cisco work. You'll wait for him to tab back to something on topic before you leave him alone.",
957378132061618217 : "CJ\nHe's the best student in the class by far and large. He's already almost finished with his CCNA, which makes sense as he is a senior that only has until he graduates to get his CCNA done in order to get it paid for by the school. He's a really good kid, and you treat him as such. You'll even tell him that he's done good work, which is rather infrequent for your other students.",
964563429190930465 : "Michael\nFor the most part you disregard him if he's not working on Cisco, occasionally requesting that he get back on track and then quickly disregarding him again and moving on. As for when he's on Cisco work, when he asks questions, you typically give him a vague non-answer, and you get easily irriated with him, ESPECIALLY if it's something that has EVER been mentioned before. Usually you'll end up yelling at him with things like \"I'm not tryna be mean, but...you should know this already.\" and then walking away.",
909865829313683536 : "Bea\nA student from Graphic Commerical Arts with blue hair and is not in the Computer Networking and Cybersecurity. Because of that, you can't really be mad at them for not focusing on Cisco. I suppose you would talk yourself up a little bit to students in other labs?",
821891090411421747 : "Mika\nA student from Animal Care and is not in the Computer Networking and Cybersecurity. Because of that, you can't really be mad at them for not focusing on Cisco. I suppose you would talk yourself up a little bit to students in other labs?",
1093276870688133171 : "Daniel\nA student from Biotech and is not in the Computer Networking and Cybersecurity. Because of that, you can't really be mad at them for not focusing on Cisco. I suppose you would talk yourself up a little bit to students in other labs?",
625861445833457694 : "Paul\nIt's PAUL!, and for some reason, you LOVE Paul, and you acknowledge just how talented of an artist he is! For some reason, you randomly walk up to him with 'How we doing Paul!?' You've even given him his own opurtnities ",
1336780587662446593: "Mr. Jacobs\nYourself."
}
ESSAY_TOPICS = ("why to not throw rocks during a fire drill",
"how to sit in a chair properly",
"how to keep your hands to yourself",
"how to be on time",
"how to take accountability",
"why you shouldn't be wrong (be right!!!)",
"why picking cherries is healthy for mental health",
"why losing is bad, actually",
"why you should be responsable when the bell rings and GET OUT because i'm HUNGRY",
"why you shouldn't hitlerpost in the public discord chat",
"why having your professionalism packet is essential for your future career",
"why playing rock-paper-scissors over text is very productive",
"why steak is the best food for my lunch break",
)
def get_essay_topic() -> str:
return random_choice(ESSAY_TOPICS)
def assign_essay(student_id: int, essay: str = get_essay_topic()) -> str:
"""Assigns a student an essay
Args:
student_id (int): The Discord ID of the student getting the essay.
essay (str, optional): The topic of the essay being assigned. Defaults to get_essay_topic().
Raises:
ValueError: If the student ID is not in STUDENT_IDS.
Returns:
str: The topic of the essay assigned.
"""
if student_id in STUDENT_IDS:
ASSIGNED_ESSAY[student_id] = essay
pickle.dump(ASSIGNED_ESSAY, open("./data/assigned_essay.pkl", "wb"))
return essay
else:
raise ValueError(f"Student ID {student_id} is not a valid student ID.")
def get_essay(student_id: int = None) -> str:
"""Gets the assigned essay for a student
Args:
student_id (int): The Discord ID of the student to get the essay for.
Raises:
ValueError: If the student ID is not in STUDENT_IDS.
Returns:
str: The topic of the essay assigned to the student.
"""
if student_id:
if student_id in STUDENT_IDS:
return ASSIGNED_ESSAY.get(student_id, "No essay assigned.")
else:
essays: str = ""
for essay in ASSIGNED_ESSAY:
essays += f"<@{essay}>: {ASSIGNED_ESSAY[essay]}\n"
return essays if essays else "No essays assigned."
def clear_essay(student_id):
"""Clears an assigned essay from a student
Args:
student_id (int): The Discord ID of the student to clear the essay from.
Raises:
ValueError: If the student ID is not in STUDENT_IDS.
"""
if student_id in STUDENT_IDS:
ASSIGNED_ESSAY.pop(student_id)
pickle.dump(ASSIGNED_ESSAY, open("./data/assigned_essay.pkl", "wb"))
else:
raise ValueError(f"Student ID {student_id} is not a valid student ID.")
if __name__ == "__main__":
print(get_essay())