Page MenuHomedesp's stash
Files F368790

bot.py
No OneTemporary
Actions

bot.py
View Options

from bs4 import *
import discord, requests
from discord.ext import commands, tasks
from discord import ui
from collections import defaultdict
import json, logging, asyncio, random, re
#for getting members from to dm, otherwise use intents.members
GUILD = int(open('guild.txt').read())
intents = discord.Intents.default()
intents.message_content = True
#intents.members = True #makes things way too slow so we bind to guild instead
bot = commands.Bot(command_prefix='.pinger ', intents=intents)
logger = logging.getLogger('discord').getChild('ubcpinger')
logger.setLevel(logging.DEBUG)
# helper for easier access of data in json
class coursedict(defaultdict):
#json automatically turns int to strings when in keys, so we normalize it ourselves too
def __delitem__(self, __key) -> None:
return super().__delitem__(str(__key)) if isinstance(__key, int) else super().__delitem__(__key)
def __setitem__(self, __key, __value):
return super().__setitem__(str(__key), __value) if isinstance(__key, int) else super().__setitem__(__key, __value)
#TODO seems like this change made it so that the data that should be empty is now '' in json? maybe its only after merge
def __getitem__(self, __key):
return super().__getitem__(str(__key)) if isinstance(__key, int) else super().__getitem__(__key)
def __contains__(self, __key) -> bool:
return super().__contains__(str(__key)) if isinstance(__key, int) else super().__contains__(__key)
__getattr__ = __getitem__
__setattr__ = __setitem__
__delattr__ = __delitem__
#recursively convert all dicts/list into defattrdicts
def convert(self, data):
iter = data.items() if isinstance(data, dict) else enumerate(data)
for key, val in iter:
if type(val) in [dict, list]:
if type(val) == dict:
data[key] = coursedict(val)
self.convert(data[key])
def __init__(self, data: dict):
super().__init__(lambda: '', data)
self.convert(self)
def __str__(self) -> str:
if self.subj:
s = f'{self.subj} {self.crsno}'
if self.section:
s += f' (section mode)'
else:
#funny ljust needed in actv so strip is needed to revert it on show
s += f', search:' + ' '.join([f'{k}={str(v).strip()}' for k,v in self if v and k not in ['metadata', 'subj', 'crsno']])
return s
else:
return dict.__repr__(self)
#remove the default factory print
__repr__ = dict.__repr__
#implicitly allow iter on items()
def __iter__(self):
return self.items().__iter__()
def get_data():
with open('courses.json') as f:
data = f.read()
return coursedict(json.loads(data)) if data else {}
def set_data(obj):
with open('courses.json', 'w') as data:
data.write(json.dumps(obj, indent=2))
def try_ping(pinger: coursedict):
optional = lambda fmtstr, *data: fmtstr.format(*data) if all(d for d in data) else ''
#must include user agent, accept, content type (if post), otherwise ssc will complain
#if no pname+tname specified, session cookie is needed to avoid "Please wait while your request is processed - this may take up to a minute to complete."
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/114.0",
"Accept": "text/html",
}
if pinger.section: #section mode
doc = requests.get('https://courses.students.ubc.ca/cs/courseschedule?pname=subjarea&tname=subj-course'
f'&dept={pinger.subj}'
f'&course={pinger.crsno}'
+ optional('&sessyr={}&sesscd={}', pinger.sessyr, pinger.sesscd), headers=headers).text
else: #search mode
reqdata = dict(pinger) #make a copy and remove metadata
del reqdata['metadata'], reqdata['sessyr'], reqdata['sesscd']
reqdata['scrsz'] = 100 #change pagnation size to 100 instead of 20 to include the most results possible
#no need to explicitly set content type if we set data= alr since requests does that for us
doc = requests.post('https://courses.students.ubc.ca/cs/courseschedule?pname=subjarea&tname=sectsearch'
+ optional('&sessyr={}&sesscd={}', pinger.sessyr, pinger.sesscd), #turns out this has to be a search param not a post data
data=reqdata,
headers=headers).text
#they both use the same system to display sections
soup = BeautifulSoup(doc, 'html.parser')
ret = {}
#remove thead, since we cant search directly with tbody as its doesnt necessarily exist in the doc
thead = soup.select_one('.section-summary thead')
if thead:
thead.clear()
#table likely doesnt exist if thead doesnt, but in any case continue to the for loop since that wont run if its empty anyways
for child in soup.select('.section-summary tr'):
section = child.select_one("td:nth-of-type(2) a")
#some sections have multiple timespans, so ignore the second part of that which has no section and status elements (see CPEN 491F 001 for example)
if not section:
continue
status = child.select_one("td:nth-of-type(1)").text
ret[section.text] = 'https://courses.students.ubc.ca' + section['href'], (status if status != '\xa0' and status.strip() else 'PING')
return ret
#pinger.metadata.user is now a dict instead of a list of id and list - this allows us to fix double firing on merge,
#even though that also means on merge it will not be consistent with separate pingers that has overlapping sections
#as i believe most ppl wont want double firing at all (coz why its literally useless and more annoying)
#this change will allow the bot to be able to stop double firing, and have a single consistent list under ViewSectionList
#and also simplifies viewing (even though it complicates merging))
def merge_user_sections(olduserdict: coursedict, newuserdict: coursedict):
#in case of the same user pinging in both pingers, we need specific logic to merge
#exists in old: leave it be since olduserdict is kept
#exists in new: add to old
#exists in both: grab old list, add new list, remove duplicates
for id, sections in newuserdict:
#set is needed to remove duplicates
olduserdict[id] = list(set(olduserdict[id] + sections)) if id in olduserdict else sections
#done make this a dict instead so its unique? this would remove the need of a "add myself to existing courses" button
#it should be pretty simple to just match the basics but more optimization (like catching CPSC 110 101 == CPSC 110 1*) might be hard
#and its practically impossible to actually figure out whether a search includes a specific section programmatically
#actually it might be possible if i do a ping once and check if theres a strict superset/subset of the current object
#still gonna be hard to optimize for pingers that span across multiple pingers
def add_course(obj):
data = get_data()
#obj.metadata.courses contains the entire list of sections that can be pinged through this pinger
#whereas obj.metadata.user[0].list contains the wanted list of sections (same for search mode, likely smaller for section mode)
#ping, check against existing pingers' course list (in metadata), if matches merge
user_id = list(obj.metadata.user.keys())[0]
resp = try_ping(obj)
if resp:
obj.metadata.courses = list(resp.keys())
#populate user.list to signify which part of the pinger should trigger a ping
obj.metadata.user[user_id] = list(obj.metadata.courses)
if obj.section: #do filtering (only needed in section mode, search mode is already filtered by ssc)
check = re.compile(f'{obj.subj} {obj.crsno} {obj.section.replace("*", ".*")}')
obj.metadata.user[user_id] = [c for c in obj.metadata.user[user_id] if check.match(c)]
#after we sort through that section should not be bound to a specific section now since the data is in user.list
#we do need to set data though to allow other things (e.g. try_ping) to match on whether its in search or section mode
obj.section = True
courses = set(obj.metadata.user[user_id])
#now we have the proper name for the pinger
name = str(obj)
logger.info(f'Adding {name} to the pingers...')
merged = False
for pinger_name, pinger_obj in coursedict(data): #coursedict not only for easier editing but also for copying for iteration (otherwise "dict was modified during iteration" might pop up)
against = set(pinger_obj.metadata.courses)
#only check merging if they are the same session
if pinger_obj.sessyr == obj.sessyr and pinger_obj.sesscd == obj.sesscd:
#there should only be one of the cases below: either existing pinger(s) are subsets of the current one, or this pinger is a subset of an existing pinger
#otherwise it shouldve been optimized out since it is sequentially added and this optimzation runs every time
if courses.issuperset(against):
logger.info(f' Optimization: merging {pinger_name} into {name}')
merge_user_sections(obj.metadata.user, pinger_obj.metadata.user)
del data[pinger_name]
elif against.issuperset(courses):
#this should be a terminating action to avoid adding it into multiple pingers when one is already handling it
#(think of a case where existing courses are [1234, 1256] and our pinger is [12] - 2 pingers match, but only 1 is needed)
merge_user_sections(pinger_obj.metadata.user, obj.metadata.user)
logger.info(f' Optimization: merging {name} into {pinger_name}')
name = pinger_name
merged = True
break
if not merged:
data[name] = obj
set_data(data)
#returns the list of courses this pinger will ping, and the name (changed if merged)
return name, "\n".join(obj.metadata.user[user_id])
else:
raise ValueError('No sections found!')
#done add ppl to existing pingers? see above
#done optimization for when multiple people are looking at the same course but want different sections pinged (and not all of them)
#TODO currently the way courses in section mode are handled means that they basically dont get optimized into search modes
#it would be better if pinger.metadata.courses tracks the total set of courses its users are pinging instead,
#and then have the optimizer do special code on reading section mode to not match on the courses in the list, but on the course names
#aka if everything a user is pinging has the same prefix of CPSC 110 for example,
#then just throw it into CPSC 110 section mode regardless even if currently the CPSC 110 pinger.metadata.courses does not have it
#this means that the following has to be changed:
# - instead of obj.metadata.courses = list(resp.keys()), do obj.metadata.courses = <total of all user lists>
# - at the superset checks, add check on whether everything being added is in the same course, and if there is a course pinger in section mode for that already
def get_pingers_with_user(id):
ret = []
for name, pinger in get_data():
if id in pinger.metadata.user:
ret.append(name)
return ret
def opt(label, default=False):
return discord.SelectOption(label=label, default=default)
class AddCourseDuration(ui.Modal, title='Enter the time range to search for'):
duration = ui.TextInput(label='Duration', placeholder='class duration (from-to, HHMM), e.g. 1000-1300, 1000-, -1300', min_length=5, max_length=9, required=False)
async def on_submit(self, interaction: discord.Interaction):
#set default val for duration for next open
self.duration.default = self.duration.value
await interaction.response.defer()
class AddCourseSearch(discord.ui.View):
def __init__(self, base):
super().__init__()
#put in class to allow default changing on callback
#DO NOT PUT THESE GLOBALLY OR ELSE THE VALUES ARE SHARED WITH EVERY INSTANCE EVER
#type opts is way too long, so some of the little used ones are truncated (opt("Reserved Section"),opt("Optional Section"),opt("Independent Study")
#TODO reorder to make it easier to search for the main ones? rn its alph order
self.type_opts = [opt("Directed Studies"),opt("Discussion"),opt("Essay/Report"),opt("Exchange Program"),opt("Experiential"),opt("Field Trip"),opt("Flexible Learning"),opt("Lab-Seminar"),opt("Laboratory"),opt("Lecture"),opt("Lecture-Discussion"),opt("Lecture-Laboratory"),opt("Lecture-Seminar"),opt("Practicum"),opt("Problem Session"),opt("Project"),opt("Rehearsal"),opt("Research"),opt("Seminar"),opt("Studio"),opt("Thesis"),opt("Tutorial"),opt("Waiting List"),opt("Work Placement"),opt("Workshop")]
self.term_opts = [opt("Term 1"), opt("Term 2"), opt("Term 1-2")]
#TODO remove sunday and saturday? oh actually there are a lot of courses on those days huh
self.days_opts = [opt("Sunday"), opt("Monday"), opt("Tuesday"), opt("Wednesday"), opt("Thursday"), opt("Friday"), opt("Saturday")]
self.cred_opts = [opt(str(i+1)) for i in range(6)] #pretty sure only 6 is allowed
self.term_menu = ui.Select(placeholder='Term', options=self.term_opts, min_values=0, row=3, custom_id='ubcpinger-search-days-term-menu')
self.days_menu = ui.Select(placeholder='Days', options=self.days_opts, min_values=0, max_values=7, row=2, custom_id='ubcpinger-search-days-menu')
self.type_menu = ui.Select(placeholder='Type', options=self.type_opts, min_values=0, row=3, custom_id='ubcpinger-search-days-type-menu')
self.cred_menu = ui.Select(placeholder='Credits', options=self.cred_opts, min_values=0, row=3, custom_id='ubcpinger-search-days-cred-menu')
self.base = base
self.dur = AddCourseDuration()
async def change_menu(self, item: discord.ui.Select, interaction: discord.Interaction):
#remove opened menus
for i in self.children:
if isinstance(i, discord.ui.Select):
self.remove_item(i)
if item:
self.add_item(item)
await interaction.response.edit_message(view=self)
async def interaction_check(self, interaction: discord.Interaction):
#set default and close menu as response; we can extract the data later on done
if 'custom_id' in interaction.data and '-menu' in interaction.data['custom_id']:
#really hacky but im also too sleepy to actually figure this out lmao
for opt in getattr(self, interaction.data['custom_id'].split('-')[2] + '_opts'):
#seems like setting default doesnt really work for the single option dropdowns which kinda checks out but hmmm
if opt.label in interaction.data['values']:
opt.default = True
else:
#remember to reset too
opt.default = False
await self.change_menu(None, interaction)
return False
return True
@discord.ui.button(label='Days', custom_id='ubcpinger-search-days', style = discord.ButtonStyle.primary)
async def days(self, interaction: discord.Interaction, button: discord.ui.Button):
await self.change_menu(self.days_menu, interaction)
@discord.ui.button(label='Type', custom_id='ubcpinger-search-type', style = discord.ButtonStyle.primary)
async def type(self, interaction: discord.Interaction, button: discord.ui.Button):
await self.change_menu(self.type_menu, interaction)
@discord.ui.button(label='Term', custom_id='ubcpinger-search-term', style = discord.ButtonStyle.primary)
async def term(self, interaction: discord.Interaction, button: discord.ui.Button):
await self.change_menu(self.term_menu, interaction)
@discord.ui.button(label='Credits', custom_id='ubcpinger-search-cred', row=1, style = discord.ButtonStyle.primary)
async def cred(self, interaction: discord.Interaction, button: discord.ui.Button):
await self.change_menu(self.cred_menu, interaction)
@discord.ui.button(label='Done', custom_id='ubcpinger-search-done', row=1, style = discord.ButtonStyle.success)
async def done(self, interaction: discord.Interaction, button: discord.ui.Button):
try:
optional = lambda data: data.values[0] if data.values else ''
#most of these values are selected for us already so no need to parse
#since we set default on callback that is basically the canonical value of whether the day is selected
#also make it a dict for easier shoving in through kwargs in the pinger loop
days = {f'DAY{i+1}': True for i, o in enumerate(self.days_opts) if o.default}
stime, etime = self.dur.duration.value.split('-') if self.dur.duration.value else '', ''
obj = coursedict({
'subj': self.base.subj.value,
'crsno': self.base.course.value,
#note how theres no section
'credit': optional(self.cred_menu),
'stime': stime,
'etime': etime,
**days,
**self.base.sesobj,
'actv': optional(self.type_menu).ljust(50),
'term': optional(self.term_menu),
#we are currently not in coursedict yet so the int to str normalization doesnt apply here, apply ourselves
'metadata': {'user': {str(interaction.user.id): None}}
})
#delegate name setting to defattrdict formatting
name, added = add_course(obj)
#edit the menu instead of sending another ephemeral msg since we cant delete the old one
#TODO figure out a better way to print since ill be using that to show a dropdown menu for the removal ui too
await interaction.response.edit_message(content=f'The pinger `{name}` is now tracking the requested classes! This includes the following:\n```\n{added}\n```\nYou will now get pinged when it opens up.', view=None)
except Exception as e:
logger.error('AddCourseSearch fail:', exc_info=e)
await interaction.response.edit_message(content=f'Something went wrong (likely wrong params): {type(e)}\n{e}', view=None)
#at the end to wrap it around nicer
@discord.ui.button(label='Time range', custom_id='ubcpinger-search-duration', row=1, style = discord.ButtonStyle.primary)
async def duration(self, interaction: discord.Interaction, button: discord.ui.Button):
await interaction.response.send_modal(self.dur)
class AddCourseBase(ui.Modal, title='Add a course to ping'):
#cringe discord.py does some cringe things on reading global members but not the constructors before calling super.__init__ so put it here instead even though this means the values are shared across all instances
subj = ui.TextInput(label='Subject', placeholder='e.g. CPSC, CP*', min_length=1, max_length=4)
course = ui.TextInput(label='Course no.', placeholder='e.g. 110, 1*', min_length=1, max_length=3)
section = ui.TextInput(label='Section', placeholder='e.g. 101, T2*, 10*/201/T1D, leave blank to open search menu', min_length=1, max_length=3, required=False)
session = ui.TextInput(label='Session', placeholder='e.g. 2022S, 2023W, defaults to coming session', min_length=5, max_length=5, required=False)
async def on_submit(self, interaction: discord.Interaction):
try:
#response in this case is the same msg as the button, dont do anything to it
await interaction.response.defer()
#parse session
self.sesobj = {}
if self.session.value:
self.sesobj = {
'sessyr': self.session.value[:4],
'sesscd': self.session.value[-1],
}
if not self.section.value:
#discord doesnt allow us to send another modal right away? (yep check InteractionResponseType, 9 is modal and not in the list)
await interaction.followup.send('Configure the search parameters below:', view=AddCourseSearch(self), ephemeral=True)
else:
if '*' in self.subj.value or '*' in self.course.value:
await interaction.response.send_message('Wildcards for subject and courses are not allowed in section mode! Leave section blank to do a broad search.', ephemeral=True)
return
obj = coursedict({
'subj': self.subj.value,
'crsno': self.course.value,
#even though we dont actually use the wildcard we need to differentiate between no section and a fully wildcard section so dont remove yet
'section': self.section.value,
**self.sesobj,
#we are currently not in coursedict yet so the int to str normalization doesnt apply here, apply ourselves
'metadata': {'user': {str(interaction.user.id): None}}
})
name, added = add_course(obj)
await interaction.followup.send(f'The pinger `{name}` is now tracking the requested classes! This includes the following:\n```\n{added}\n```\nYou will now get pinged when it opens up.', ephemeral=True)
except Exception as e:
logger.error('AddCourseBase fail:', exc_info=e)
await interaction.followup.send(content=f'Something went wrong (likely wrong params): {type(e)}\n{e}', ephemeral=True)
#TODO dm option? or maybe just only allow dms (iirc the reason why i dont dm is coz theres no mechanism to dm multiple ppl or sth so i just send a single msg to ping everyone)
#common class for listing courses (used by remove pingers, and also pinger list)
class SingleMenu(ui.View):
def __init__(self, user, placeholder, list, menu_id, single=False):
if not list:
raise ValueError("Menu options cannot be empty")
super().__init__()
#to be consistent we should use self.user instead of interaction.user since they might be different (even though when ephemeral=True they should be the same)
self.user = user
self.menu = ui.Select(placeholder=placeholder, options=[opt(s) for s in list], min_values=1, max_values=len(list) if not single else 1, custom_id=menu_id)
self.add_item(self.menu)
class RemoveCourseList(SingleMenu):
#i love python asyncio : ) why the fuck
@classmethod
async def create(cls, user):
if not await bot.is_owner(user):
pingers = get_pingers_with_user(user.id)
else:
pingers = get_data().keys()
return cls(user, 'Pinger', pingers, 'ubcpinger-remove-menu')
async def interaction_check(self, interaction: discord.Interaction):
if 'custom_id' in interaction.data and interaction.data['custom_id'] == 'ubcpinger-remove-menu':
data = get_data()
for name in interaction.data['values']:
#if bot owner, on remove of pinger with own id added, remove normally; else remove the entire thing
if data[name].metadata.user[self.user.id]:
del data[name].metadata.user[self.user.id]
if not data[name].metadata.user:
logger.info(f"{data[name]} is now empty, removing...")
del data[name]
break
else:
if self.id == bot.owner_id:
logger.info(f"bot owner override on {data[name]}, removing...")
del data[name]
else:
logger.error(f'Something pretty wrong happened: {data} does not have {self.id} even though on setup it had')
set_data(data)
pinger_names = "\n".join(interaction.data['values'])
await interaction.response.edit_message(content=f"Successfully removed you from these pingers:\n```\n{pinger_names}\n```", view=None)
class ViewSectionList(SingleMenu):
def __init__(self, user, pinger_name):
#user list should exist or else view course list wouldnt even show
self.pinger_name = pinger_name
super().__init__(user, 'Section', get_data()[pinger_name].metadata.user[user.id], 'ubcpinger-section-menu')
async def interaction_check(self, interaction: discord.Interaction):
if 'custom_id' in interaction.data and interaction.data['custom_id'] == 'ubcpinger-section-menu':
#dont reuse data since state mightve changed
data = get_data()
for name in interaction.data['values']:
data[self.pinger_name].metadata.user[self.user.id].remove(name)
set_data(data)
values = "\n".join(interaction.data['values'])
await interaction.response.edit_message(content=f"Successfully removed your selected sections from the pinger `{self.pinger_name}` as follows:\n```\n{values}\n```\nYou will no longer get pinged even if the pinger sees the sections.", view=None)
class ViewCourseList(SingleMenu):
@classmethod
async def create(cls, user):
pingers = get_pingers_with_user(user.id)
return cls(user, 'Pinger', pingers, 'ubcpinger-course-menu', single=True)
async def interaction_check(self, interaction: discord.Interaction):
if 'custom_id' in interaction.data and interaction.data['custom_id'] == 'ubcpinger-course-menu':
#should always be 1 but just to be safe
if len(interaction.data['values']) > 0:
await interaction.response.edit_message(content=f"You are currently pinged for these sections under pinger `{interaction.data['values'][0]}` :\n"
"(**Click dismiss message to do nothing, otherwise select sections to remove; NOTE: cannot re-add without recreating the pinger!**)",
view=ViewSectionList(interaction.user, interaction.data['values'][0]))
class CourseButtons(discord.ui.View):
@discord.ui.button(label="Add pinger", style=discord.ButtonStyle.green, custom_id='ubcpinger-add')
async def add(self, interaction: discord.Interaction, button: discord.ui.Button):
await interaction.response.send_modal(AddCourseBase())
@discord.ui.button(label="Remove pinger", style=discord.ButtonStyle.red, custom_id='ubcpinger-remove')
async def remove(self, interaction: discord.Interaction, button: discord.ui.Button):
await interaction.response.defer()
try:
await interaction.followup.send("Choose a pinger to remove:", view=await RemoveCourseList.create(interaction.user), ephemeral=True)
except ValueError:
await interaction.followup.send("You currently have no pingers!", ephemeral=True)
@discord.ui.button(label="View pinged sections", style=discord.ButtonStyle.primary, custom_id='ubcpinger-view')
async def view(self, interaction: discord.Interaction, button: discord.ui.Button):
await interaction.response.defer()
try:
await interaction.followup.send("Choose a pinger to view sections", view=await ViewCourseList.create(interaction.user), ephemeral=True)
except ValueError:
await interaction.followup.send("You currently have no pingers!", ephemeral=True)
#done remove button (only allow ppl to remove their own, except if user id matches me so i can actually do cleanup and stuff without going into the server)
#TODO button for viewing what courses you get pinged for in a pinger? rn its not too descriptive since the optimizer makes it the broadest possible (now that ppl can remove sections the searches might also need to be refined the same way (aka dynamically resize on remove or sth; idk sounds kinda hard))
#TODO that button would probably also work as a filtering system for when you dont need that many courses anymore
@tasks.loop(seconds=1)
async def pinger():
try:
obj = get_data()
if not obj:
return
#obj is a mapping of pingers against its name
for pinger_name, pinger_obj in obj:
logger.debug(f'In {pinger_name}:')
courses = try_ping(pinger_obj)
for id, sections in pinger_obj.metadata.user:
for section in sections:
url, status = courses[section]
logger.debug(f' {section}: {status} ({id})')
if status == 'PING':
#create dm short circuits if dm is found
dm = await bot.get_guild(GUILD).get_member(id).create_dm()
#done button to ssc
view = ui.View()
view.add_item(discord.ui.Button(label='Get me to SSC', url=url, style=discord.ButtonStyle.primary))
await dm.send(f'## wake up `{section}` is currently open', view=view)
await asyncio.sleep(random.uniform(4, 7)) #total 4-7 secs
except Exception as e:
logger.error("Something went wrong during pinging:", exc_info=e)
#hope that 10s later something fixes itself
await asyncio.sleep(10)
@bot.event
async def on_ready():
logger.info(f'Logged on as {bot.user}')
buttons = CourseButtons(timeout=None)
bot.add_view(buttons)
pinger.start()
#touch if not exist
open('courses.json', 'a').close()
bot.run(open('token.txt').read())

File Metadata

Mime Type
text/x-python
Expires
Sat, Jul 5, 2:49 PM (23 h, 50 m)
Storage Engine
local-disk
Storage Format
Raw Data
Storage Handle
d3/00/e9ed2575235d6be1d16b83463dda

Event Timeline

Back to Main Site