Compare commits

...

2 Commits

6 changed files with 272 additions and 144 deletions

View File

@@ -3,4 +3,5 @@
/.venv /.venv
/__pycache__ /__pycache__
*.pyc *.pyc
/data /data
students.pub.py

View File

@@ -13,7 +13,7 @@ from students import *
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
OPENAI_KEY = os.getenv('OPENAI_KEY') OPENAI_KEY = os.getenv("OPENAI_KEY")
client: OpenAI = OpenAI( client: OpenAI = OpenAI(
api_key=OPENAI_KEY, api_key=OPENAI_KEY,
@@ -23,7 +23,11 @@ client: OpenAI = OpenAI(
mr_jacobs: Assistant = client.beta.assistants.retrieve("asst_KdPdwqNAKijujfyCRrJCOgJN") mr_jacobs: Assistant = client.beta.assistants.retrieve("asst_KdPdwqNAKijujfyCRrJCOgJN")
base_instructions: str = mr_jacobs.instructions base_instructions: str = mr_jacobs.instructions
logger.info(f"Base instructions: {base_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" 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 ----- # ----- Helper Functions -----
def get_run_status(run: Run) -> str: def get_run_status(run: Run) -> str:
@@ -37,12 +41,12 @@ def get_run_status(run: Run) -> str:
str: The status of the run. str: The status of the run.
""" """
status: str = client.beta.threads.runs.retrieve( status: str = client.beta.threads.runs.retrieve(
run.id, run.id, thread_id=run.thread_id
thread_id=run.thread_id
).status ).status
logger.info(f"Status of run {run.id} is {status}") logger.info(f"Status of run {run.id} is {status}")
return status return status
def get_instructions(context: discord.abc.GuildChannel | discord.Message = None) -> str: def get_instructions(context: discord.abc.GuildChannel | discord.Message = None) -> str:
""" """
Retrieves the instructions for the AI based on the context. Retrieves the instructions for the AI based on the context.
@@ -56,22 +60,27 @@ def get_instructions(context: discord.abc.GuildChannel | discord.Message = None)
str: The instructions for the AI. str: The instructions for the AI.
""" """
tuned_instructions: str = instructions tuned_instructions: str = instructions
if context: 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" 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): if isinstance(context, discord.Message):
context = context.channel context = context.channel
if not isinstance(context, discord.abc.Messageable): 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}") 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 return tuned_instructions
if isinstance(context, discord.TextChannel): 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'}" 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): 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)}") 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.'}" 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 return tuned_instructions
# ----- Action Handlers ----- # ----- Action Handlers -----
async def handle_action(run: Run) -> bool: async def handle_action(run: Run) -> bool:
# Define the list to store tool outputs # Define the list to store tool outputs
@@ -79,23 +88,39 @@ async def handle_action(run: Run) -> bool:
tool: RequiredActionFunctionToolCall tool: RequiredActionFunctionToolCall
# Loop through each tool in the required action section # Loop through each tool in the required action section
try: try:
logger.debug(f"Run.require_action is currently: {run.required_action} and tool calls are {run.required_action.submit_tool_outputs.tool_calls}") 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: for tool in run.required_action.submit_tool_outputs.tool_calls:
logger.info(f"Handling action for tool {tool.id}, function {tool.function.name}") logger.info(
f"Handling action for tool {tool.id}, function {tool.function.name}"
)
if tool.function.name == "assign_essay": if tool.function.name == "assign_essay":
logger.debug(f"Handling action for assign_essay tool. Received arguments {tool.function.arguments}") logger.debug(
f"Handling action for assign_essay tool. Received arguments {tool.function.arguments}"
)
args = load_json(tool.function.arguments) args = load_json(tool.function.arguments)
tool_outputs.append({ 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"]) "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": elif tool.function.name == "clear_essay":
logger.debug(f"Clearing the essay for student {run.metadata.get('user_id')}") logger.debug(
f"Clearing the essay for student {run.metadata.get('user_id')}"
)
clear_essay(int(run.metadata.get("user_id"))) clear_essay(int(run.metadata.get("user_id")))
tool_outputs.append({ 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." "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: except AttributeError as e:
logger.error(f"Failed to handle action: {e}") logger.error(f"Failed to handle action: {e}")
except Exception as e: except Exception as e:
@@ -105,16 +130,14 @@ async def handle_action(run: Run) -> bool:
if tool_outputs: if tool_outputs:
try: try:
run = client.beta.threads.runs.submit_tool_outputs_and_poll( run = client.beta.threads.runs.submit_tool_outputs_and_poll(
thread_id=run.thread_id, thread_id=run.thread_id, run_id=run.id, tool_outputs=tool_outputs
run_id=run.id,
tool_outputs=tool_outputs
) )
logger.info("Tool outputs submitted successfully.") logger.info("Tool outputs submitted successfully.")
except Exception as e: except Exception as e:
logger.error("Failed to submit tool outputs:", e) logger.error("Failed to submit tool outputs:", e)
else: else:
logger.warning("No tool outputs to submit.") logger.warning("No tool outputs to submit.")
# ----- Run Handlers ----- # ----- Run Handlers -----
async def handle_run(run: Run) -> bool: async def handle_run(run: Run) -> bool:
@@ -126,13 +149,10 @@ async def handle_run(run: Run) -> bool:
Returns: Returns:
bool: Whether or not the run completed successfully. bool: Whether or not the run completed successfully.
""" """
while True: while True:
run = client.beta.threads.runs.retrieve( run = client.beta.threads.runs.retrieve(run.id, thread_id=run.thread_id)
run.id,
thread_id=run.thread_id
)
match run.status: match run.status:
case "completed": case "completed":
return True return True
case "failed": case "failed":
logger.error(f"Run {run.id} failed.") logger.error(f"Run {run.id} failed.")
@@ -148,7 +168,9 @@ async def handle_run(run: Run) -> bool:
await asyncio.sleep(1) await asyncio.sleep(1)
async def run(messages: list[dict], instructions: str = instructions, user_id : int = None) -> str: async def run(
messages: list[dict], instructions: str = instructions, user_id: int = None
) -> str:
""" """
Runs the AI with the given messages and instructions. Runs the AI with the given messages and instructions.
@@ -160,17 +182,15 @@ async def run(messages: list[dict], instructions: str = instructions, user_id :
Returns: Returns:
str: The response from the AI. 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)}") 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( run = client.beta.threads.create_and_run(
assistant_id=mr_jacobs.id, assistant_id=mr_jacobs.id,
instructions=instructions, instructions=instructions,
thread={ thread={"messages": messages},
"messages": messages metadata={"user_id": str(user_id if user_id else -1)},
}, )
metadata={
"user_id": str(user_id if user_id else -1)
}
)
response = await run_message(run) response = await run_message(run)
return response return response
@@ -193,4 +213,6 @@ async def run_message(run) -> str:
if msg_ob.id == thread_messages.first_id: if msg_ob.id == thread_messages.first_id:
response = msg_ob.content[0].text.value response = msg_ob.content[0].text.value
return response return response
logger.critical(f"Couldn't find the msg that matched with the first message ID:\nThread Messages List:\n{thread_messages}") logger.critical(
f"Couldn't find the msg that matched with the first message ID:\nThread Messages List:\n{thread_messages}"
)

39
app.py
View File

@@ -10,31 +10,54 @@ from discord_logic import bot, setup_discord_bot, send_quote, after_class
# Load environment variables # Load environment variables
load_dotenv() load_dotenv()
DISCORD_TOKEN = os.getenv('DISCORD_TOKEN') DISCORD_TOKEN = os.getenv("DISCORD_TOKEN")
# Logging. # Logging.
logging.basicConfig( logging.basicConfig(
filename=f"./logs/jacobs.log", filename=f"./logs/jacobs.log",
filemode="at+", filemode="at+",
level=logging.DEBUG if os.getenv('DEBUG') == "True" else logging.INFO, level=logging.DEBUG if os.getenv("DEBUG") == "True" else logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
) )
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
setup_discord_bot(bot) setup_discord_bot(bot)
@bot.event @bot.event
async def on_ready() -> None: async def on_ready() -> None:
"""Event handler for when the bot is initialized.""" """Event handler for when the bot is initialized."""
logger.info(f"{bot.user.name} has connected to Discord and is ready.") logger.info(f"{bot.user.name} has connected to Discord and is ready.")
print(f"{bot.user.name} is ready.") print(f"{bot.user.name} is ready.")
# After successful initialization, schedule tasks. # After successful initialization, schedule tasks.
task_scheduler = BackgroundScheduler(timezone=pytz.timezone('EST')) 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(
task_scheduler.add_job(lambda: bot.loop.create_task(send_quote()), 'cron', day_of_week='mon-fri', hour='8-14', minute="*/20", jitter=180) lambda: bot.loop.create_task(send_quote()),
task_scheduler.add_job(lambda: bot.loop.create_task(after_class()), 'cron', day_of_week='mon-fri', hour=10, minute=55) "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,
)
# I'm commenting this out since it doesn't really make sense to have active senior year.
# task_scheduler.add_job(
# lambda: bot.loop.create_task(after_class()),
# "cron",
# day_of_week="mon-fri",
# hour=10,
# minute=55,
# )
task_scheduler.start() task_scheduler.start()
logger.info("Presumably successfully initialized, starting bot.") logger.info("Presumably successfully initialized, starting bot.")
bot.run(DISCORD_TOKEN) bot.run(DISCORD_TOKEN)

View File

@@ -20,8 +20,9 @@ GENERAL_ID: int = 1339601047294840876
bot = Bot(intents=discord.Intents.all()) bot = Bot(intents=discord.Intents.all())
# ----- Discord Helper Functions ----- # ----- Discord Helper Functions -----
async def get_channel(channel_id: int) -> Channel: async def get_channel(channel_id: int = GENERAL_ID) -> Channel:
""" """
Tries to get a cached channel object, if it fails it will send an API request to retrieve a new one. Tries to get a cached channel object, if it fails it will send an API request to retrieve a new one.
@@ -37,43 +38,60 @@ async def get_channel(channel_id: int) -> Channel:
logger.debug(f"Attempting to get channel with ID {channel_id}") logger.debug(f"Attempting to get channel with ID {channel_id}")
channel = bot.get_channel(channel_id) channel = bot.get_channel(channel_id)
if not channel: if not channel:
logger.debug(f"Channel with ID {channel_id} not found in cache, fetching from API.") logger.debug(
f"Channel with ID {channel_id} not found in cache, fetching from API."
)
channel = await bot.fetch_channel(channel_id) channel = await bot.fetch_channel(channel_id)
if not channel: if not channel:
logger.critical(f"Could not fetch channel with channel_id {channel_id}, fetch_channel returned None.") logger.critical(
f"Could not fetch channel with channel_id {channel_id}, fetch_channel returned None."
)
return None return None
logger.info(f"Successfully retrieved {channel_id}, #{channel.name}") logger.info(f"Successfully retrieved {channel_id}, #{channel.name}")
return channel return channel
async def get_message(message: Message | int, channel_id: int = None, attempts: int = 3) -> Message:
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. Retrieves a message object either from cache or by fetching it from the channel.
Args: Args:
message (Message | int): The message object or message ID to retrieve. 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. 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. attempts (int, optional): The number of attempts to retrieve the message. Defaults to 3.
Returns: Returns:
Message: The retrieved message object. Message: The retrieved message object.
""" """
logger.debug(f"Attempting to get message with ID {message if isinstance(message, int) else message.id}") logger.debug(
f"Attempting to get message with ID {message if isinstance(message, int) else message.id}"
)
for attempt in range(attempts): for attempt in range(attempts):
if isinstance(message, int): if isinstance(message, int):
got_message: Message = bot.get_message(message) got_message: Message = bot.get_message(message)
if not got_message and channel_id: if not got_message and channel_id:
logger.debug(f"Message with ID {message} not found in cache, fetching from channel ID {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) channel: discord.abc.GuildChannel = await get_channel(channel_id)
got_message: Message = await channel.fetch_message(message) got_message: Message = await channel.fetch_message(message)
elif not got_message: elif not got_message:
logger.error(f"Message with ID {message} not found in cache and no channel ID provided.") logger.error(
f"Message with ID {message} not found in cache and no channel ID provided."
)
elif isinstance(message, Message): elif isinstance(message, Message):
got_message: Message = bot.get_message(message.id) got_message: Message = bot.get_message(message.id)
if not got_message: if not got_message:
logger.debug(f"Message with ID {message.id} not found in cache, fetching from channel.") 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) got_message: Message = await message.channel.fetch_message(message.id)
else: else:
logger.error(f"Couldn't retrieve message:\nMessage: {message}\nChannel ID: {channel_id}") logger.error(
f"Couldn't retrieve message:\nMessage: {message}\nChannel ID: {channel_id}"
)
return None return None
if got_message: if got_message:
@@ -86,6 +104,7 @@ async def get_message(message: Message | int, channel_id: int = None, attempts:
logger.error(f"Failed to retrieve message after {attempts} attempts.") logger.error(f"Failed to retrieve message after {attempts} attempts.")
return None return None
async def send_message(message: str, channel: int | Channel = GENERAL_ID) -> Message: 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. Has the bot send a message to a specified channel. If no channel is specified, the bot sends it to general.
@@ -103,13 +122,18 @@ async def send_message(message: str, channel: int | Channel = GENERAL_ID) -> Mes
if isinstance(channel, Channel): if isinstance(channel, Channel):
sent_message: Message = await channel.send(message) sent_message: Message = await channel.send(message)
if isinstance(sent_message, Message): if isinstance(sent_message, Message):
logger.info(f"Message '{message}' successfully sent to {sent_message.channel}") logger.info(
f"Message '{message}' successfully sent to {sent_message.channel}"
)
return sent_message return sent_message
else: else:
logger.error(f"Message likely wasn't successfully sent, as channel.send did not return a Message.") logger.error(
f"Message likely wasn't successfully sent, as channel.send did not return a Message."
)
elif isinstance(channel, None): elif isinstance(channel, None):
logger.error(f"Message couldn't be sent as a channel wasn't found/messagable.") 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): async def edit_message(message: Message | int, channel: int = None):
""" """
Edits a message in a channel. Edits a message in a channel.
@@ -118,7 +142,9 @@ async def edit_message(message: Message | int, channel: int = None):
message (Message | int): The message to edit. message (Message | int): The message to edit.
channel (int, optional): The channel ID of the message. Defaults to None. 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}") logger.debug(
f"Attempting to edit message {message if isinstance(message, int) else message.id}"
)
if isinstance(message, int): if isinstance(message, int):
message = await get_message(message, channel) message = await get_message(message, channel)
if isinstance(message, Message): if isinstance(message, Message):
@@ -127,6 +153,7 @@ async def edit_message(message: Message | int, channel: int = None):
else: else:
logger.error(f"Message couldn't be edited as it was not found.") logger.error(f"Message couldn't be edited as it was not found.")
# ----- Message Utility Functions ----- # ----- Message Utility Functions -----
async def msg_is_reply(message: Message) -> tuple[bool, Message]: async def msg_is_reply(message: Message) -> tuple[bool, Message]:
""" """
@@ -141,16 +168,19 @@ async def msg_is_reply(message: Message) -> tuple[bool, Message]:
logger.debug(f"Checking if message ID {message.id} is a reply.") logger.debug(f"Checking if message ID {message.id} is a reply.")
if message.reference is not None: if message.reference is not None:
replied_msg = message.reference.cached_message replied_msg = message.reference.cached_message
if replied_msg is None: if replied_msg is None:
logger.debug(f"Replied message not found in cache, fetching from channel ID {message.reference.channel_id}.") 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) channel = await get_channel(message.reference.channel_id)
replied_msg = await channel.fetch_message(message.reference.message_id) replied_msg = await channel.fetch_message(message.reference.message_id)
if isinstance(replied_msg, Message): if isinstance(replied_msg, Message):
return True, replied_msg return True, replied_msg
return False, None return False, None
async def get_reply_chain(msg: Message) -> list[Message]: async def get_reply_chain(msg: Message) -> list[Message]:
""" """
Retrieves the chain of replies for a given message. Retrieves the chain of replies for a given message.
@@ -172,6 +202,7 @@ async def get_reply_chain(msg: Message) -> list[Message]:
i += 1 i += 1
return reply_chain return reply_chain
async def gen_message_list(discord_messages: list[Message]) -> list[dict]: async def gen_message_list(discord_messages: list[Message]) -> list[dict]:
""" """
Generates a list of message dictionaries from a list of Discord messages. Generates a list of message dictionaries from a list of Discord messages.
@@ -186,7 +217,7 @@ async def gen_message_list(discord_messages: list[Message]) -> list[dict]:
messages = [] messages = []
if not isinstance(discord_messages, list): if not isinstance(discord_messages, list):
discord_messages = [discord_messages] discord_messages = [discord_messages]
for message in discord_messages: for message in discord_messages:
students_mentioned = None students_mentioned = None
if isinstance(message, Message): if isinstance(message, Message):
@@ -196,37 +227,45 @@ async def gen_message_list(discord_messages: list[Message]) -> list[dict]:
"author": message.author, "author": message.author,
"mentions": message.mentions, "mentions": message.mentions,
"created_at": message.created_at, "created_at": message.created_at,
"jump_url": message.jump_url "jump_url": message.jump_url,
} }
role = "assistant" if message_copy["author"].id == bot.user.id else "user" role = "assistant" if message_copy["author"].id == bot.user.id else "user"
students_mentioned = None students_mentioned = None
if message_copy["mentions"] and role == "user": if message_copy["mentions"] and role == "user":
students_mentioned = [student.id for student in message_copy["mentions"] if student.id in STUDENT_IDS] students_mentioned = [
mentions = re.findall(r'<@\d{17,19}>', message_copy["content"]) 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: for mention in mentions:
mention_id = mention.replace("<@", "").replace(">", "") mention_id = mention.replace("<@", "").replace(">", "")
student_info = ID_TO_NAME.get(int(mention_id)) student_info = ID_TO_NAME.get(int(mention_id))
if student_info: if student_info:
message_copy["content"] = message_copy["content"].replace(mention, student_info.split("\n")[0]) message_copy["content"] = message_copy["content"].replace(
message_copy["content"] += '\n{' mention, student_info.split("\n")[0]
)
message_copy["content"] += "\n{"
if students_mentioned: if students_mentioned:
message_copy["content"] += f"\n\nInfo on students mentioned (DO NOT REPEAT!): " message_copy["content"] += (
f"\n\nInfo on students mentioned (DO NOT REPEAT!): "
)
for student in students_mentioned: for student in students_mentioned:
message_copy["content"] += f"\n{STUDENTS.get(student)}" 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)'}" + '}' 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"]: + "}"
)
if not message_copy["content"]:
debug = True debug = True
logger.error(f"No message content for message {message_copy['id']}.") logger.error(f"No message content for message {message_copy['id']}.")
fetched_message = await get_message(message_copy["id"]) fetched_message = await get_message(message_copy["id"])
message_copy["content"] = fetched_message.content message_copy["content"] = fetched_message.content
thread_message = { thread_message = {"role": role, "content": message_copy["content"]}
"role": role, debug = True # Delete after fix
"content": message_copy["content"]
}
debug = True # Delete after fix
if debug: if debug:
debug_dump = "Messages Dump:\n" debug_dump = "Messages Dump:\n"
for msg_copy in discord_messages: for msg_copy in discord_messages:
@@ -237,14 +276,16 @@ async def gen_message_list(discord_messages: list[Message]) -> list[dict]:
Message Author: {msg_copy.author} Message Author: {msg_copy.author}
Message Content: {msg_copy.content} Message Content: {msg_copy.content}
Link: {msg_copy.jump_url} Link: {msg_copy.jump_url}
""" """
logger.debug(debug_dump) logger.debug(debug_dump)
messages.append(thread_message) messages.append(thread_message)
else: else:
logger.warning(f"Argument {message} is not of type Message and will be skipped.") logger.warning(
f"Argument {message} is not of type Message and will be skipped."
)
return messages return messages
# ----- AI Interaction Functions ----- # ----- AI Interaction Functions -----
async def send_quote(quote: str = None) -> None: async def send_quote(quote: str = None) -> None:
""" """
@@ -254,10 +295,18 @@ async def send_quote(quote: str = None) -> None:
quote (str, optional): The quote to send. If not provided, a random quote is selected. quote (str, optional): The quote to send. If not provided, a random quote is selected.
""" """
if not quote: if not quote:
# If the quote is not defined, it's likely this is a scheduled messages
# Therefore, we are going to return early if the bot is the last person
# who sent a message to prevent a lot of uneccassary messages in general
# when there is no conversation
if get_channel().last_message.message.author.id == bot.user.id:
return
# Since no quote is defined, we are getting a random one.
quote = select_quote() quote = select_quote()
logger.info(f"Sending quote '{quote}' in #general...") logger.info(f"Sending quote '{quote}' in #general...")
await send_message(quote) await send_message(quote)
async def after_class(student: int = None) -> None: async def after_class(student: int = None) -> None:
""" """
Sends a message to a student to see the bot after class. Sends a message to a student to see the bot after class.
@@ -269,7 +318,8 @@ async def after_class(student: int = None) -> None:
student = select_student() student = select_student()
logger.info(f"Sending mention to see {student} after class to #general...") logger.info(f"Sending mention to see {student} after class to #general...")
await send_message(f"Come see me after class <@{student}>") await send_message(f"Come see me after class <@{student}>")
async def has_essay(message: Message) -> bool: async def has_essay(message: Message) -> bool:
if message.author.id in ASSIGNED_ESSAY: if message.author.id in ASSIGNED_ESSAY:
async with message.channel.typing(): async with message.channel.typing():
@@ -277,14 +327,15 @@ async def has_essay(message: Message) -> bool:
message_list = await gen_message_list(message) message_list = await gen_message_list(message)
response = await run( response = await run(
message_list, 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\".", 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 message.author.id,
) )
await message.reply(response) await message.reply(response)
return True return True
else: else:
return False return False
# ----- Discord Commands & Event Handlers ----- # ----- Discord Commands & Event Handlers -----
def setup_discord_bot(bot: Bot) -> None: def setup_discord_bot(bot: Bot) -> None:
""" """
@@ -293,6 +344,7 @@ def setup_discord_bot(bot: Bot) -> None:
Args: Args:
bot (Bot): The Discord bot instance. bot (Bot): The Discord bot instance.
""" """
@bot.slash_command(description="Talk to Mr. Jacobs!!!") @bot.slash_command(description="Talk to Mr. Jacobs!!!")
async def message(ctx: discord.ApplicationContext, text: str) -> None: async def message(ctx: discord.ApplicationContext, text: str) -> None:
""" """
@@ -308,9 +360,13 @@ def setup_discord_bot(bot: Bot) -> None:
thread_messages = [{"role": "user", "content": text}] thread_messages = [{"role": "user", "content": text}]
response = await run(thread_messages, instructions, ctx.author.id) response = await run(thread_messages, instructions, ctx.author.id)
await ctx.respond(content=response) await ctx.respond(content=response)
@bot.slash_command(name="clear_essay", description="Clear the essay assigned to a student") @bot.slash_command(
async def clear_essay_command(ctx: discord.ApplicationContext, student: str) -> None: 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. Slash command to clear the essay assigned to a student.
@@ -332,9 +388,14 @@ def setup_discord_bot(bot: Bot) -> None:
return return
ASSIGNED_ESSAY.pop(student, None) ASSIGNED_ESSAY.pop(student, None)
await ctx.respond(f"Cleared essay for <@{student}>") 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") @bot.slash_command(
async def get_essay_command(ctx: discord.ApplicationContext, student: str = None) -> None: 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: if student:
try: try:
student = int(student) student = int(student)
@@ -345,14 +406,14 @@ def setup_discord_bot(bot: Bot) -> None:
await ctx.respond(get_essay(student)) await ctx.respond(get_essay(student))
else: else:
await ctx.respond(get_essay(), ephemeral=True) await ctx.respond(get_essay(), ephemeral=True)
@bot.slash_command(name="assign_essay", description="Assign an essay to a student") @bot.slash_command(name="assign_essay", description="Assign an essay to a student")
@commands.has_permissions(moderate_members=True) @commands.has_permissions(moderate_members=True)
async def assign_essay_command( async def assign_essay_command(
ctx: discord.ApplicationContext, ctx: discord.ApplicationContext,
student: str, student: str,
timeout: int | None = 0, timeout: int | None = 0,
topic: str | None = None topic: str | None = None,
) -> None: ) -> None:
""" """
Slash command to assign an essay to a student. Slash command to assign an essay to a student.
@@ -374,7 +435,9 @@ def setup_discord_bot(bot: Bot) -> None:
return return
if timeout <= 0: if timeout <= 0:
timeout = None timeout = None
logging.info(f"Assigning essay to student {student} with timeout {timeout} and topic {topic}") logging.info(
f"Assigning essay to student {student} with timeout {timeout} and topic {topic}"
)
if student not in STUDENT_IDS: if student not in STUDENT_IDS:
await ctx.respond("Invalid student ID.", ephemeral=True) await ctx.respond("Invalid student ID.", ephemeral=True)
return return
@@ -383,18 +446,30 @@ def setup_discord_bot(bot: Bot) -> None:
if timeout and topic: if timeout and topic:
timeout_until = discord.utils.utcnow() + timedelta(seconds=timeout) timeout_until = discord.utils.utcnow() + timedelta(seconds=timeout)
try: try:
member: discord.Member = await ctx.interaction.guild.fetch_member(student) member: discord.Member = await ctx.interaction.guild.fetch_member(
await member.timeout(until=timeout_until, reason=f"Assigned essay: {topic}") student
await ctx.respond(f'{member.mention} has been timed out for {timeout}.', ephemeral=True) )
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: except discord.Forbidden:
await ctx.respond(f'Failed to timeout {member.mention}. Missing permissions.', ephemeral=True) await ctx.respond(
f"Failed to timeout {member.mention}. Missing permissions.",
ephemeral=True,
)
except discord.HTTPException as e: except discord.HTTPException as e:
await ctx.respond(f'Failed to timeout {member.mention}. {e.text}', ephemeral=True) await ctx.respond(
f"Failed to timeout {member.mention}. {e.text}", ephemeral=True
)
return return
@bot.slash_command( @bot.slash_command(
name="rps", name="rps",
description="Play \"Rock, Paper, Scissors, Essay\" with Mr. Jacobs!!!", description='Play "Rock, Paper, Scissors, Essay" with Mr. Jacobs!!!',
) )
async def rps_essay( async def rps_essay(
ctx: discord.ApplicationContext, ctx: discord.ApplicationContext,
@@ -403,13 +478,13 @@ def setup_discord_bot(bot: Bot) -> None:
choices=[ choices=[
discord.OptionChoice("Rock"), discord.OptionChoice("Rock"),
discord.OptionChoice("Paper"), discord.OptionChoice("Paper"),
discord.OptionChoice("Scissors") discord.OptionChoice("Scissors"),
] ],
) ),
) -> None: ) -> None:
""" """
Play Rock Paper Scissors with Mr. Jacobs. Play Rock Paper Scissors with Mr. Jacobs.
Args: Args:
ctx: Application command context ctx: Application command context
choice: Your selection for the game choice: Your selection for the game
@@ -417,11 +492,11 @@ def setup_discord_bot(bot: Bot) -> None:
logging.info(f"{ctx.author} chose {choice}") logging.info(f"{ctx.author} chose {choice}")
outcomes = {"Rock": "Paper", "Paper": "Scissors", "Scissors": "Rock"} outcomes = {"Rock": "Paper", "Paper": "Scissors", "Scissors": "Rock"}
await ctx.respond( await ctx.respond(
f"I choose {outcomes[choice]}, you owe me a {randint(1,15)} page essay " 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, " f"on {assign_essay(ctx.author.id)}. MLA format, double spaced, "
"12pt Times New Roman!" "12pt Times New Roman!"
) )
@bot.event @bot.event
async def on_message(message: Message) -> None: async def on_message(message: Message) -> None:
""" """

View File

@@ -4,7 +4,7 @@ import datetime
from students import STUDENT_IDS from students import STUDENT_IDS
QUOTES = { QUOTES = {
"NORMAL" : ( "NORMAL": (
"Oh boy, we are about to wake up!", "Oh boy, we are about to wake up!",
"All right 👏 🫰", "All right 👏 🫰",
"How we doing, Paul?", "How we doing, Paul?",
@@ -27,24 +27,23 @@ QUOTES = {
"*Spins lanyard*", "*Spins lanyard*",
"Come on guys, you should know this already!", "Come on guys, you should know this already!",
"Let me just say this...", "Let me just say this...",
"Not trying to be mean or anything...but..." "Not trying to be mean or anything...but...",
), ),
"RARE" : ( "RARE": (
"Play stupid games, win big prizes! 🤑🤑", "Play stupid games, win big prizes! 🤑🤑",
"Oooooo raahahah!", "Oooooo raahahah!",
"It's cherry-pickin' time, y'all!", "It's cherry-pickin' time, y'all!",
"What does the fox say?" "What does the fox say?",
), ),
"MYTHIC" : ( "MYTHIC": (
"I'm proud of you.", "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." "I have somewhere to be at 9:30, so you guys will have a sub.",
), ),
"LEGENDARY": ( "LEGENDARY": (),
)
} }
def select_quote() -> str: def select_quote() -> str:
rarity = random.randint(0, 99) rarity = random.randint(0, 99)
if rarity < 1: if rarity < 1:
@@ -55,11 +54,16 @@ def select_quote() -> str:
quote = random.choice(QUOTES.get("NORMAL")) quote = random.choice(QUOTES.get("NORMAL"))
# Append log to a file (quotes.log) # Append log to a file (quotes.log)
with open("./logs/quotes.log", "at+") as log_file: 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") log_file.write(
f"At {datetime.datetime.now()}, rarity was rolled as {rarity} and selected: {quote}\n"
)
return quote return quote
def select_student() -> int: def select_student() -> int:
student = random.choice(STUDENT_IDS) student = random.choice(STUDENT_IDS)
with open("./logs/quotes.log", "at+") as log_file: with open("./logs/quotes.log", "at+") as log_file:
log_file.write(f"At {datetime.datetime.now()}, wanted to see {student} after class.\n") log_file.write(
f"At {datetime.datetime.now()}, wanted to see {student} after class.\n"
)
return student return student

View File

@@ -5,9 +5,7 @@ import pickle
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# Redacted # Redacted
ID_TO_NAME: dict[int, str] = { ID_TO_NAME: dict[int, str] = {}
}
# Redacted # Redacted
STUDENT_IDS: set = () STUDENT_IDS: set = ()
@@ -28,24 +26,27 @@ finally:
# Redacted # Redacted
STUDENTS: dict[int, str] = {} STUDENTS: dict[int, str] = {}
ESSAY_TOPICS = ("why to not throw rocks during a fire drill", ESSAY_TOPICS = (
"how to sit in a chair properly", "why to not throw rocks during a fire drill",
"how to keep your hands to yourself", "how to sit in a chair properly",
"how to be on time", "how to keep your hands to yourself",
"how to take accountability", "how to be on time",
"why you shouldn't be wrong (be right!!!)", "how to take accountability",
"why picking cherries is healthy for mental health", "why you shouldn't be wrong (be right!!!)",
"why losing is bad, actually", "why picking cherries is healthy for mental health",
"why you should be responsable when the bell rings and GET OUT because i'm HUNGRY", "why losing is bad, actually",
"why you shouldn't hitlerpost in the public discord chat", "why you should be responsable when the bell rings and GET OUT because i'm HUNGRY",
"why having your professionalism packet is essential for your future career", "why you shouldn't hitlerpost in the public discord chat",
"why playing rock-paper-scissors over text is very productive", "why having your professionalism packet is essential for your future career",
"why steak is the best food for my lunch break", "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: def get_essay_topic() -> str:
return random_choice(ESSAY_TOPICS) return random_choice(ESSAY_TOPICS)
def assign_essay(student_id: int, essay: str = get_essay_topic()) -> str: def assign_essay(student_id: int, essay: str = get_essay_topic()) -> str:
"""Assigns a student an essay """Assigns a student an essay
@@ -66,6 +67,7 @@ def assign_essay(student_id: int, essay: str = get_essay_topic()) -> str:
else: else:
raise ValueError(f"Student ID {student_id} is not a valid student ID.") raise ValueError(f"Student ID {student_id} is not a valid student ID.")
def get_essay(student_id: int = None) -> str: def get_essay(student_id: int = None) -> str:
"""Gets the assigned essay for a student """Gets the assigned essay for a student
@@ -86,7 +88,7 @@ def get_essay(student_id: int = None) -> str:
for essay in ASSIGNED_ESSAY: for essay in ASSIGNED_ESSAY:
essays += f"<@{essay}>: {ASSIGNED_ESSAY[essay]}\n" essays += f"<@{essay}>: {ASSIGNED_ESSAY[essay]}\n"
return essays if essays else "No essays assigned." return essays if essays else "No essays assigned."
def clear_essay(student_id): def clear_essay(student_id):
"""Clears an assigned essay from a student """Clears an assigned essay from a student
@@ -102,7 +104,8 @@ def clear_essay(student_id):
pickle.dump(ASSIGNED_ESSAY, open("./data/assigned_essay.pkl", "wb")) pickle.dump(ASSIGNED_ESSAY, open("./data/assigned_essay.pkl", "wb"))
else: else:
raise ValueError(f"Student ID {student_id} is not a valid student ID.") raise ValueError(f"Student ID {student_id} is not a valid student ID.")
if __name__ == "__main__": if __name__ == "__main__":
print(get_essay()) print(get_essay())