phowell
/
canvasapp
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

catchup

This commit is contained in:
Peter Howell 2025-07-24 10:53:43 -07:00
parent 68e48ec6ac
commit 0bddc4a3e4
8 changed files with 398 additions and 59 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

37
courses.py
View File

@ -247,7 +247,7 @@ def users_in_by_depts_live(depts=[], termid='181'):
print("Getting enrollments for %s" % c['course_code']) print("Getting enrollments for %s" % c['course_code'])
if d in courses_by_by_dept: courses_by_by_dept[d].append(c) if d in courses_by_by_dept: courses_by_by_dept[d].append(c)
else: courses_by_by_dept[d] = [ c, ] else: courses_by_by_dept[d] = [ c, ]
for u in course_enrollment_with_faculty(c['id'],1).values(): for u in course_enrollment_with_faculty(c['id'],0).values():
#if u['type'] != "StudentEnrollment": continue #if u['type'] != "StudentEnrollment": continue
if not (d in students_by_by_dept): if not (d in students_by_by_dept):
students_by_by_dept[d] = set() students_by_by_dept[d] = set()
@ -829,9 +829,10 @@ def course_term_summary_3():
# check number of students and publish state of all shells in a term # check number of students and publish state of all shells in a term
def all_semester_course_sanity_check(): def all_semester_course_sanity_check():
outputfile = 'cache/courses_checker.csv' term = "su25"
t = 287 target_start = "6-14"
term = "sp25" outputfile = f'cache/courses_checker_{term}.csv'
t = 288
c = getCoursesInTerm(t,1,0) c = getCoursesInTerm(t,1,0)
sched1 = requests.get(f"http://gavilan.cc/schedule/{term}_sched_expanded.json").json() sched1 = requests.get(f"http://gavilan.cc/schedule/{term}_sched_expanded.json").json()
sched = { x['crn']: x for x in sched1 } sched = { x['crn']: x for x in sched1 }
@ -840,7 +841,7 @@ def all_semester_course_sanity_check():
#output.write( ",".join(['what','id','parent_course_id','sis_course_id','name']) + "\n" ) #output.write( ",".join(['what','id','parent_course_id','sis_course_id','name']) + "\n" )
output2 = codecs.open(outputfile,'w','utf-8') output2 = codecs.open(outputfile,'w','utf-8')
output2.write( ",".join(['id','sis_course_id','name','state','mode','startdate','students']) + "\n" ) output2.write( ",".join(['id','sis_course_id','name','state','mode','startdate','students']) + "\n" )
htmlout = codecs.open('cache/courses_checker.html','w','utf-8') htmlout = codecs.open(f'cache/courses_checker_{term}.html','w','utf-8')
htmlout.write('<html><body><table>\n') htmlout.write('<html><body><table>\n')
htmlout.write(f'<tr><td><b>Name</b></td><td><b>SIS ID</b></td><td><b>State</b></td><td><b>Mode</b></td><td><b>Start Date</b></td><td><b># Stu</b></td></tr>\n') htmlout.write(f'<tr><td><b>Name</b></td><td><b>SIS ID</b></td><td><b>State</b></td><td><b>Mode</b></td><td><b>Start Date</b></td><td><b># Stu</b></td></tr>\n')
html_sections = [] html_sections = []
@ -859,10 +860,11 @@ def all_semester_course_sanity_check():
ctype = sched[crn]['type'] ctype = sched[crn]['type']
cstart = sched[crn]['start'] cstart = sched[crn]['start']
ts = sched[crn]['act'] ts = sched[crn]['act']
teacher = sched[crn]['teacher']
info = [ 'course', course['id'], '', course['sis_course_id'], course['name'], course['workflow_state'], ts ] info = [ 'course', course['id'], '', course['sis_course_id'], course['name'], course['workflow_state'], ts ]
info = list(map(str,info)) info = list(map(str,info))
info2 = [ course['id'], course['sis_course_id'], course['name'], course['workflow_state'], ctype, cstart, ts ] info2 = [ course['id'], course['sis_course_id'], course['name'], course['workflow_state'], ctype, cstart, ts, teacher ]
info2 = list(map(str,info2)) info2 = list(map(str,info2))
output2.write( ",".join(info2) + "\n" ) output2.write( ",".join(info2) + "\n" )
output2.flush() output2.flush()
@ -870,8 +872,8 @@ def all_semester_course_sanity_check():
#output.write( ",".join(info) + "\n" ) #output.write( ",".join(info) + "\n" )
uu = f"https://ilearn.gavilan.edu/courses/{course['id']}" uu = f"https://ilearn.gavilan.edu/courses/{course['id']}"
if course["workflow_state"]=='unpublished' and ctype=='online' and cstart=="1-27": if course["workflow_state"]=='unpublished' and ctype=='online' and cstart==target_start:
html_sections.append(f'<!--{course["name"]}--><tr><td><a href="{uu}" target="_blank">{course["name"]}</a></td><td>{course["sis_course_id"]}</td><td>{course["workflow_state"]}</td><td>{ctype}</td><td>{cstart}</td><td>{ts}</td></tr>\n') html_sections.append(f'<!--{course["name"]}--><tr><td><a href="{uu}" target="_blank">{course["name"]}</a></td><td>{course["sis_course_id"]}</td><td>{course["workflow_state"]}</td><td>{ctype}</td><td>{cstart}</td><td>{ts}</td><td>{teacher}</td></tr>\n')
#uu = url + '/api/v1/courses/%s/sections' % str(course['id']) #uu = url + '/api/v1/courses/%s/sections' % str(course['id'])
#course['sections'] = fetch(uu) #course['sections'] = fetch(uu)
#s_info = [ [ 'section', y['id'], y['course_id'], y['sis_course_id'], y['name'], y['total_students'] ] for y in course['sections'] ] #s_info = [ [ 'section', y['id'], y['course_id'], y['sis_course_id'], y['name'], y['total_students'] ] for y in course['sections'] ]
@ -1047,7 +1049,7 @@ def enroll_id_list_to_shell(id_list, shell_id, v=0):
# multiple semesters # multiple semesters
def enroll_stem_students_live(): def enroll_stem_students_live():
semesters = [286,287] semesters = [288,289]
for S in semesters: for S in semesters:
enroll_stem_students_live_semester(S) enroll_stem_students_live_semester(S)
@ -1249,8 +1251,11 @@ def enroll_gott_workshops():
# date, title, shell_id # date, title, shell_id
#['2025-02-23 16:00:00', 'GOTT 6: Intro to Synchronous Teaching (Sync/Hyflex)', 21835], #['2025-02-23 16:00:00', 'GOTT 6: Intro to Synchronous Teaching (Sync/Hyflex)', 21835],
['2025-03-14 17:00:00', 'GOTT 5: The Essentials of Blended Learning (Hybrid) ', '21886'], #['2025-03-14 17:00:00', 'GOTT 5: The Essentials of Blended Learning (Hybrid) ', '21886'],
#['2025-02-23 16:00:00', 'GOTT 1: Intro to Teaching Online (2 week, async)', 21874] #['2025-02-23 16:00:00', 'GOTT 1: Intro to Teaching Online (2 week, async)', 21874]
['2025-05-26 17:00:00', 'GOTT 2: Introduction to Asynchronous Teaching and Learning', 23015],
['2025-06-01 17:00:00', 'GOTT 1: Intro to Teaching Online', 23083],
['2025-06-01 17:00:00', 'GOTT 4: Assessments in Digital Learning', 21898],
] ]
#print(json.dumps(signups,indent=4)) #print(json.dumps(signups,indent=4))
#print(json.dumps(by_email,indent=4)) #print(json.dumps(by_email,indent=4))
@ -1271,6 +1276,11 @@ def enroll_gott_workshops():
'lorrmay36@mac.com': 'llevy@gavilan.edu', 'lorrmay36@mac.com': 'llevy@gavilan.edu',
'gkalu1@gmail.com': 'gkalu@gavilan.edu', 'gkalu1@gmail.com': 'gkalu@gavilan.edu',
'rpotter@gav.edu': 'rpotter@gavilan.edu', 'rpotter@gav.edu': 'rpotter@gavilan.edu',
'ally162@qq.com': 'aao@gavilan.edu',
'davidamancio791@gmail.com': 'damancio@gavilan.edu',
'carissaamunoz83@gmail.com': 'amunoz@gavilan.edu',
'jasonwcpa@yahoo.com': 'jwolowitz@gavilan.edu',
} }
for each_workshop in workshop_ids: for each_workshop in workshop_ids:
@ -1792,6 +1802,9 @@ def teacher_to_many_shells():
import os, pickle import os, pickle
def create_sandboxes(): def create_sandboxes():
## TODO: read all student names and determine ahead of time if initials conflict. deal with them
courses_to_sandbox = [ #(20567, ' Sandbox GOTT1 SU24'), courses_to_sandbox = [ #(20567, ' Sandbox GOTT1 SU24'),
#(20575, ' Sandbox GOTT2 SU24'), #(20575, ' Sandbox GOTT2 SU24'),
#(20600, ' Sandbox GOTT4 SU24'), #(20600, ' Sandbox GOTT4 SU24'),
@ -1800,7 +1813,9 @@ def create_sandboxes():
#(20761, ' Sandbox GOTT1 FA24'), #(20761, ' Sandbox GOTT1 FA24'),
#(21770, ' Sandbox GOTT1 WI25'), #(21770, ' Sandbox GOTT1 WI25'),
#(21772, ' Sandbox GOTT2 WI25'), #(21772, ' Sandbox GOTT2 WI25'),
(21874, ' Sandbox GOTT1 SP25'), (23083, ' Sandbox GOTT1 SU25'),
(23015, ' Sandbox GOTT2 SU25'),
(21898, ' Sandbox GOTT4 SU25'),
] ]
filepath = 'cache/sandbox_courses.pkl' filepath = 'cache/sandbox_courses.pkl'

193
degrees.py
View File

@ -10,6 +10,7 @@ import pandas as pd
debug_out = codecs.open("cache/degrees_debug.txt", "w", "utf-8") debug_out = codecs.open("cache/degrees_debug.txt", "w", "utf-8")
def d(s): def d(s):
global debug_out
#if type(s) == tuple or type(s) == list: #if type(s) == tuple or type(s) == list:
# debug_out.write(" ".join(str(s)) + "\n") # debug_out.write(" ".join(str(s)) + "\n")
debug_out.write(str(s) + "\n") debug_out.write(str(s) + "\n")
@ -112,6 +113,13 @@ class DSLTransformer(Transformer):
d("\ncourse_declaration") d("\ncourse_declaration")
d(name) d(name)
d(units) d(units)
units_value = float(units.children[0]) # Handle "3", "3.0", etc.
#prereq_list = [p.value for p in prereqs if p is not None]
course = Course(name.value)
course.units = units_value
#course.prereqs = prereq_list
self.courses[name.value] = course
return course
def take_all(self): def take_all(self):
d("\ntake_all") d("\ntake_all")
@ -141,7 +149,10 @@ class DSLTransformer(Transformer):
def course(self, c): def course(self, c):
d("\ncourse") d("\ncourse")
d(c) d(c)
return Course(c.value) name = c.value
if name not in self.courses:
self.courses[name] = Course(name)
return self.courses[name]
def course_or(self, *items): def course_or(self, *items):
d("\ncourse_or") d("\ncourse_or")
@ -330,32 +341,40 @@ rule_lookup = {
'Choose (one) or more', 'Choose (one) or more',
'Choose (one|two|three) of the classes listed', 'Choose (one|two|three) of the classes listed',
'Choose (one|two|three)', 'Choose (one|two|three)',
'Choose (4) classes (12 units) from the following list:',
'Choose ([\d\w]+) courses from', 'Choose ([\d\w]+) courses from',
'Choose (\w+) of the following', 'Choose (\w+) of the following',
'Please select (ONE) course from the options below:',
'LIST [AB]: Select (\d)', 'LIST [AB]: Select (\d)',
'Select (1)', 'Select (1)',
'Select (one) of the following REQUIRED CORE', 'Select (one) of the following REQUIRED CORE',
'SELECT (ONE|TWO) OF THE FOLLOWING', 'Select (one|two|three)', 'SELECT (ONE|TWO) OF THE FOLLOWING', 'Select (one|two|three)',
'Select (\d+) courses', ], 'Select (\d+) courses', ],
'take at least n units': [ '(\d+) units total', 'take at least n units': [ '(\d+) units total', '(\d+) Unit Minimum',
'Any combination totaling (\d+) units', 'Any combination totaling (\d+) units',
'Choose (eight|\d+) units', 'Choose (eight|\d+) units',
'Choose (\w+) units from classes listed', 'Choose (\w+) units from classes listed',
'Choose a minimum of ([\w\d]+) units from', 'Choose a minimum of ([\w\d]+) units from',
'Choose a minimum of (\d+) units', 'Choose a minimum of (\d+) units',
'Choose at least ([\d|\.]+) units',
'Choose eighteen \((18)\) units from the following list:',
'Choose any combination of courses for a minimum of ([\w\d]+) units\:?', 'Choose any combination of courses for a minimum of ([\w\d]+) units\:?',
'Choose any combination of the following courses for a minimum of (\d+) units:',
'Choose courses for at least ([\w\d]+) units', 'Choose courses for at least ([\w\d]+) units',
'LIST A \((\d+) units\)', 'LIST A \((\d+) units\)',
'LIST B \((\d+) units\)', 'LIST B \((\d+) units\)',
'LIST C \- Any course .*\((\d+) units\)', 'LIST C \- Any course .*\((\d+) units\)',
'Select (\d+) units', 'Select (\d+) units',
'Select any (\d+)\-\d+ units from the following',], 'Select any (\d+)\-\d+ units from the following',],
'electives': ['Electives', 'Recommended electives?:', ], 'electives': ['Electives', 'Recommended electives?:', 'Recommended Elective'],
'take_all': ['RN PROGRAM', 'REQUIRED CORE', 'CORE COURSES', 'ADDITIONAL REQUIREMENTS','REQUIREMENTS:', 'take_all': ['RN PROGRAM', 'REQUIRED CORE', 'CORE COURSES', 'CORE Courses', 'Core Major Courses', 'Core Courses :',
'Requirements', 'Core Requirements', 'ADDITIONAL REQUIREMENTS', 'REQUIREMENTS:', 'REQUIREMENTS', 'Required:',
'Requirements', 'Core Requirements', 'Required Courses',
'Required Core', 'REQUIRED', 'LVN PROGRAM', 'Student Teaching Practicum', '^LIST A:?$', '^LIST B:$', 'Required Core', 'REQUIRED', 'LVN PROGRAM', 'Student Teaching Practicum', '^LIST A:?$', '^LIST B:$',
'Complete the following courses:', 'Complete the following required courses for 6 units:', 'Complete all of the following courses:',
'Complete all of the following General Education (Cal-GETC) courses:',
'Program Requirements', 'Required Courses:', 'PROGRAM REQUIREMENTS (5 Units)', 'Program Requirements', 'Required Courses:', 'PROGRAM REQUIREMENTS (5 Units)',
'PROGRAM REQUIREMENTS (162 Hours)', 'PROGRAM REQUIREMENTS (162 Hours)', 'PROGRAM REQUIREMENTS (5 Units):', 'PROGRAM REQUIREMENTS (162 Hours):',
], ],
} }
@ -373,6 +392,7 @@ def lookup_rule(line):
pass pass
if verbose: print(f"line: {line} matched: {each} with {num}") if verbose: print(f"line: {line} matched: {each} with {num}")
return key,num return key,num
print(f"** failed to lookup rule: {line}")
return None,None return None,None
@ -394,23 +414,153 @@ def num_units(s):
def examine5(li,award, verbose=0):
summary = [x[0] for x in li]
if summary[1] in ['and','or'] and summary[3] in ['and','or']:
if verbose: print(" - ", summary)
return summary[1]
return False
def examine(li,award, verbose=0): def examine(li,award, verbose=0):
summary = [x[0] for x in li] summary = [x[0] for x in li]
if summary[1] in ['and','or']: if summary[1] in ['and','or']:
if verbose: print(" - ", summary) if verbose: print(" - ", summary)
return summary[1]
return False
# given list of courses, pbd like: [c1, c2, c3, or, c4, c5, c6, or, c7, or, c8, c9],
# return [c1, c2, [or, c3, c4], c5, [or, c6, c7, c8], c9]
# given [c1, 'or', c2, 'or', c3, 'or', c4, 'and', c5, 'and', c6]
# return ['or', c1, c2, c3, [ 'and', c4, c5, c6 ]]
def check_ands_ors_pbd(award, pbd, verbose=0): def check_ands_ors_pbd(award, pbd, verbose=0):
verbose = 0 if not pbd:
return []
i = 0
or_group = []
and_group = []
while i < len(pbd):
token = pbd[i][0]
if token == 'or':
i += 1 # skip 'or', next item is part of or_group
continue # will be handled by next loop iteration
elif token == 'and':
# Start or continue and_group
i += 1
if i < len(pbd):
and_group.append(pbd[i])
i += 1
else:
# If previous token was not 'and', treat as part of or_group
if i == 0 or pbd[i-1] != 'and':
or_group.append(token)
i += 1
result = ['or'] + or_group
if and_group:
result.append(['and'] + and_group)
return result
def check_ands_ors_pbd2(award, pbd, verbose=0):
summary = [x[0] for x in pbd]
#if verbose: print(" ", summary)
for i,x in enumerate(summary):
desc = ''
if x == 'course': desc = f"\t{pbd[i][2]['code']} {pbd[i][2]['name']}"
elif x == 'or': desc = ''
elif x == 'and': desc = ''
else: desc = pbd[i][2]
print(f"{x}\t{desc}")
result = []
i = 0
while i < len(pbd):
#if pbd[i][1]=='21' and pbd[i][2]=='Political Science':
# print('here')
if pbd[i][0] == 'or':
# special case: math 233 or higher
if i < len(pbd) and pbd[i+1][2] == 'higher':
result[-1][2]['name'] += "OR HIGHER"
i += 2
continue
# Take last item from result
left = result.pop()
group = ['or', left, pbd[i+1]]
i += 2
if i < len(pbd) and pbd[i][0] == 'or':
group.append(pbd[i+1])
i += 2
result.append(group)
elif pbd[i][0] == 'and':
# Take last item from result
left = result.pop()
group = ['and', left, pbd[i+1]]
i += 2
if i < len(pbd) and pbd[i][0] == 'and':
group.append(pbd[i+1])
i += 2
result.append(group)
else:
result.append(pbd[i])
i += 1
#print(result)
return result
def check_ands_ors_pbd1(award, pbd, verbose=0):
verbose = 1
return_list = []
if verbose: print(f"check_ands_ors_pbd({award}, ...)") if verbose: print(f"check_ands_ors_pbd({award}, ...)")
summary = [x[0] for x in pbd] summary = [x[0] for x in pbd]
if verbose: print(" ", summary) #if verbose: print(" ", summary)
for i,x in enumerate(summary):
desc = ''
if x == 'course': desc = f"\t{pbd[i][2]['code']} {pbd[i][2]['name']}"
elif x == 'or': desc = ''
else: desc = pbd[i][2]
print(f"{x}\t{desc}")
# iterate through in groups of 5, from 0/1/2/3/4 up to 6/7/8/9/10
# (for length 1. length n: n-2, n-1, n)
skip = 0
for i in range(len(pbd)-4):
if skip>0:
skip -= 1
continue
cmd = examine5(pbd[i:i+5], award, verbose)
if cmd:
skip = 4
return_list.append( [cmd, pbd[i], pbd[i+2], pbd[i+4]] )
else:
return_list.append( pbd[i] )
# iterate through in groups of 3, from 0/1/2 up to 3/4/5 # iterate through in groups of 3, from 0/1/2 up to 3/4/5
# (for length 6. length n: n-2, n-1, n) # (for length 6. length n: n-2, n-1, n)
skip = 0
for i in range(len(pbd)-2): for i in range(len(pbd)-2):
examine(pbd[i:i+3], award, verbose) if skip>0:
if verbose: print() skip -= 1
continue
cmd = examine(pbd[i:i+3], award, verbose)
if cmd:
skip = 2
return_list.append( [cmd, pbd[i], pbd[i+2]] )
else:
return_list.append( pbd[i] )
#if verbose: print(return_list)
return return_list
def build_program_rules(verbose=0): def build_program_rules(verbose=0):
@ -467,13 +617,21 @@ def build_program_rules(verbose=0):
# Each 'program block definition' # Each 'program block definition'
# 1st is dict with unit totals, rest are lists. # 1st is dict with unit totals, rest are lists.
check_ands_ors_pbd( award, sorted( r[k][1:], key=lambda x: float(x[1])) ) requirements_inorder = sorted( r[k][1:], key=lambda x: float(x[1]))
for each_r in sorted( r[k][1:], key=lambda x: float(x[1])): reqs_with_or_blocks = check_ands_ors_pbd( award, requirements_inorder )
for each_r in reqs_with_or_blocks:
if each_r[0] in ['and','or']: if each_r[0] == 'or':
#print(' ', each_r[0],each_r[1]) if not this_rule:
pass this_rule = "take_all from "
if isinstance(each_r, list): this_rule += ' or '.join( [x[2]['code'] for x in each_r[1:] ] ) # each_r[1][2]['code']} or {each_r[2][2]['code']}, "
if each_r[0] == 'and':
if not this_rule:
this_rule = "take_all from "
this_rule += f"{each_r[1][2]['code']} and {each_r[2][2]['code']}, "
elif isinstance(each_r, list):
#print(each_r) #print(each_r)
if each_r[0] == 'h3' or (each_r[0]=='noncourse' and is_noncourse_new_section(each_r[2])): if each_r[0] == 'h3' or (each_r[0]=='noncourse' and is_noncourse_new_section(each_r[2])):
# This is a rule title # This is a rule title
@ -533,6 +691,7 @@ def build_program_rules(verbose=0):
if ok: if ok:
okay.append(p) okay.append(p)
verbose = 1
if verbose: if verbose:
print("\n\n\n\nThese programs are okay:") print("\n\n\n\nThese programs are okay:")
for p in okay: for p in okay:

90
localcache2.py
View File

@ -11,6 +11,7 @@ from os.path import exists, getmtime
#from pipelines import sync_non_interactive, url, header, gp, dean #from pipelines import sync_non_interactive, url, header, gp, dean
from tabulate import tabulate from tabulate import tabulate
from semesters import short_to_sis
from canvas_secrets import postgres_database, postgres_password, postgres_port, postgres_user, postgres_host from canvas_secrets import postgres_database, postgres_password, postgres_port, postgres_user, postgres_host
######### #########
@ -448,6 +449,94 @@ def courses_to_sched():
conn.close() conn.close()
# Populate schedule table and correlate to courses table
def refresh_semester_schedule_db(term="fa25"):
# TODO: fix units when they are variable... change to float in between range. round to 0.5 unit.
EXECUTE = 1
last_time = 0
sis_code = short_to_sis(term)
conn,cur = db()
drop_query = f"DELETE FROM canvas.schedule WHERE sem='{sis_code}'"
print(f"executing: {drop_query}")
cur.execute(drop_query)
# get all ilearn courses
query = """SELECT c.id, c.workflow_state, c.sis_source_id, c.course_code, c.enrollment_term_id, t.name
FROM canvas.courses c
JOIN canvas.enrollment_terms t ON c.enrollment_term_id = t.id
ORDER BY c.sis_source_id, c.course_code;"""
cur.execute(query)
conn.commit()
sis_to_sched = {}
for row in cur.fetchall():
sis_source_id = row[2] # c.sis_source_id
sis_to_sched.setdefault(sis_source_id, []).append(row)
vals_cache = []
i = 0
print(sis_code)
sched = 0
try:
sched = requests.get(f"http://gavilan.cc/schedule/{term}_sched_expanded.json").json()
except Exception as e:
print(e)
query = "INSERT INTO canvas.schedule (canvascourse, crn, code, units, teacher, start,\"end\", type, loc, site, partofday, cap, act, sem) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s);"
if sched:
for c in sched:
try:
pod = ''
if 'partofday' in c: pod = c['partofday']
#print(c['cred'])
cred_match = re.search(r'(\d+\.\d+)\-(\d+\.\d+)',c['cred'])
if cred_match:
#print(f"matched range: {cred_match.groups}")
cred_start = float(cred_match.group(1))
cred_end = float(cred_match.group(2))
mid = float(int( (cred_end-cred_start)/2 + cred_start ))
c['cred'] = str(mid)
#print(f"middle cred is {c['cred']}")
full_sis_code = sis_code+'-'+c['crn']
if full_sis_code in sis_to_sched:
print(c['cred'])
q = [sis_to_sched[full_sis_code][0][0], c['crn'], c['code'], c['cred'], c['teacher'], c['start'], c['end'], c['type'], c['loc'], c['site'], pod, int(c['cap']), int(c['act']), sis_code]
vals_cache.append( q ) # [ str(x) for x in q ] )
#print(f"{i}: {q}")
i += 1
if i % 100 == 0:
if EXECUTE:
cur.executemany(query, vals_cache)
conn.commit()
vals_cache = []
t = time.process_time()
delta = t - last_time
last_time = t
print(f"Loop {i} - committed to db in %0.3fs. " % delta, flush=True)
else:
print(f"{full_sis_code} not in canvas courses.")
except Exception as e:
print(e)
if EXECUTE:
cur.executemany(query, vals_cache)
conn.commit()
cur.close()
conn.close()
def student_count(courseid): def student_count(courseid):
conn,cursor = db() conn,cursor = db()
q = f"""SELECT COUNT(u.id) AS student_count FROM canvas.courses AS c q = f"""SELECT COUNT(u.id) AS student_count FROM canvas.courses AS c
@ -498,6 +587,7 @@ if __name__ == "__main__":
4: ['new students this semester', users_new_this_semester], 4: ['new students this semester', users_new_this_semester],
5: ['all semester courses + teachers', all_sem_courses_teachers], 5: ['all semester courses + teachers', all_sem_courses_teachers],
6: ['Populate schedule table and correlate to courses table', courses_to_sched], 6: ['Populate schedule table and correlate to courses table', courses_to_sched],
7: ['refresh db schedule 1 semester', refresh_semester_schedule_db],
} }

6
schedules.py
View File

@ -124,7 +124,7 @@ HTM,skill,cwp'''
areas_csv = '''Area,AreaCode,DeanCode,DeanName areas_csv = '''Area,AreaCode,DeanCode,DeanName
Nursing and Allied Health,nah,et,[Dean: Nursing/Allied Health] Nursing and Allied Health,nah,et,[Dean: Nursing/Allied Health]
Career Education and Workforce Pathways,cwp,ss,Vins Chacko Career Education and Workforce Pathways,cwp,vc,Vins Chacko
Arts Humanities and Social Sciences,ahss,nl,[Dean: AHSS] Arts Humanities and Social Sciences,ahss,nl,[Dean: AHSS]
Counseling,c,de,Diego Espinoza Counseling,c,de,Diego Espinoza
Student Support and Special Programs,sssp,de,Diego Espinoza Student Support and Special Programs,sssp,de,Diego Espinoza
@ -1084,6 +1084,8 @@ def find_goo_for_name(name_dict,name):
return '' return ''
# All teachers' teaching history # All teachers' teaching history
# need schedule files to exist and be current here:
# \canvasapp\cache\schedules
def teachers_history(): def teachers_history():
# Let's just make a giant dict of names we know about -> G numbers # Let's just make a giant dict of names we know about -> G numbers
@ -1129,8 +1131,6 @@ def teachers_history():
all_sections_by_goo = defaultdict(list) all_sections_by_goo = defaultdict(list)
#non_matches_file = codecs.open('cache/nonmatches.txt','w','utf-8')
# Define the regex pattern # Define the regex pattern
pattern = re.compile(r"(\w\w\d\d)_sched_expanded\.json") pattern = re.compile(r"(\w\w\d\d)_sched_expanded\.json")

1
semesters.py
View File

@ -24,6 +24,7 @@ def short_to_sis(s):
return "20" + s[2:5] + s_to_n[season] return "20" + s[2:5] + s_to_n[season]
# go from sp20 to 2020spring # go from sp20 to 2020spring
def short_to_long(s): def short_to_long(s):
parts = re.search(r'(\w\w)(\d\d)', s) parts = re.search(r'(\w\w)(\d\d)', s)

18
ssb.py
View File

@ -337,6 +337,9 @@ def scrape_schedule(short_sem, semester_label):
writepage(driver.page_source) writepage(driver.page_source)
print(driver.title) print(driver.title)
# switch to new tab
driver.switch_to.window(driver.window_handles[1])
driver.find_element(By.LINK_TEXT,"Registration").click() driver.find_element(By.LINK_TEXT,"Registration").click()
print('registration') print('registration')
driver.implicitly_wait(5) driver.implicitly_wait(5)
@ -421,6 +424,7 @@ def scrape_schedule(short_sem, semester_label):
except Exception as e: except Exception as e:
print("Couldn't Diff.") print("Couldn't Diff.")
print("Got an exception: ", e) print("Got an exception: ", e)
return 2
# fname = pathlib.Path('test.py') # fname = pathlib.Path('test.py')
# assert fname.exists(), f'No such file: {fname}' # check that the file exists # assert fname.exists(), f'No such file: {fname}' # check that the file exists
# print(fname.stat()) # print(fname.stat())
@ -438,14 +442,14 @@ def scrape_schedule(short_sem, semester_label):
except Exception as e: except Exception as e:
print("Got an exception: ", e) print("Got an exception: ", e)
#print("There was an error: " + e.args[0] + ". The line where the code failed was " + str(traceback.extract_stack())) #print("There was an error: " + e.args[0] + ". The line where the code failed was " + str(traceback.extract_stack()))
return 1
finally: finally:
driver.quit() driver.quit()
return 0
def expanded(as_dict, short_sem): def expanded(short_sem):
#as_dict = scrape_schedule()
course_to_gp, course_to_area, areacode_to_area, area_to_dean, course_to_dean, dean_code_to_name = schedules.campus_dept_hierarchy() course_to_gp, course_to_area, areacode_to_area, area_to_dean, course_to_dean, dean_code_to_name = schedules.campus_dept_hierarchy()
expanded = list_latestarts(short_sem) expanded = list_latestarts(short_sem)
@ -582,9 +586,11 @@ with open('cache/to_scrape.json', 'r') as f:
# Loop through each item and call the function # Loop through each item and call the function
for item in semesters: for item in semesters:
as_dict = scrape_schedule(item['short_sem'], item['sem']) result = scrape_schedule(item['short_sem'], item['sem'])
ex = expanded(as_dict, item['short_sem']) if result == 0:
ex = expanded(item['short_sem'])
print(f"Done with {item['sem']}. Sleeping 45 seconds.") print(f"Done with {item['sem']}. Sleeping 45 seconds.")
time.sleep(45) time.sleep(45)
else:
print(f"Stopped due to error: {result}")

26
tasks.py
View File

@ -20,7 +20,7 @@ from collections import defaultdict
from time import gmtime, strftime from time import gmtime, strftime
from time import mktime from time import mktime
from semesters import sem_to_short from semesters import human_to_short
from canvas_secrets import badgr_target, badgr_hd from canvas_secrets import badgr_target, badgr_hd
@ -167,6 +167,7 @@ def send_email(fullname, firstname, addr, subj, content):
def convert_to_pdf(name1, name2): def convert_to_pdf(name1, name2):
wd = 'C:\\Users\\peter\\Documents\\gavilan\\canvasapp\\' wd = 'C:\\Users\\peter\\Documents\\gavilan\\canvasapp\\'
wd = 'I:/canvasapp/' wd = 'I:/canvasapp/'
wd = 'C:/Users/phowell/source/repos/canvasapp/'
print( wd + name1 ) print( wd + name1 )
try: try:
word = win32.DispatchEx("Word.Application") word = win32.DispatchEx("Word.Application")
@ -182,16 +183,27 @@ def convert_to_pdf(name1, name2):
# Build (docx/pdf) certificates for gott graduates # Build (docx/pdf) certificates for gott graduates
def certificates_gott_build(): def certificates_gott_build():
course = "gott_1_su25"
coursedate = "Summer 2025"
certificate = "gott 1 template.docx"
#course = "gott_4_su25"
#certificate = "gott 4 template.docx"
i = 0 i = 0
for row in csv.reader( open('cache/completers_gott2_wi24.csv','r'), delimiter=','): for row in csv.reader( open(f'cache/completers_{course}.csv','r'), delimiter=','):
i += 1 i += 1
if i < 3: continue if i < 2: continue
print(row[0]) print(row[0])
try:
lname,fname = row[0].split(",") lname,fname = row[0].split(",")
name = fname.strip() + " " + lname.strip() name = fname.strip() + " " + lname.strip()
doc = DocxTemplate("cache/certificates/gott 2 template.docx") except Exception as e:
doc.render({ 'name' : row[0] }) name = row[0].strip()
fn = "cache/certificates/gott_2_wi24%s." % re.sub('\s', '_', name.lower()) doc = DocxTemplate(f"cache/certificates/{certificate}")
doc.render({ 'name' : name, 'coursedate': coursedate })
name_as_filename = re.sub('\s', '_', name.lower())
fn = f"cache/certificates/{course}_{name_as_filename}."
print(fn+'docx') print(fn+'docx')
doc.save(fn+'docx') doc.save(fn+'docx')
convert_to_pdf(fn+'docx', fn+'pdf') convert_to_pdf(fn+'docx', fn+'pdf')
@ -1309,7 +1321,7 @@ if __name__ == "__main__":
15: ['create a week calendar in word (general purpose)', word_calendar_v2], 15: ['create a week calendar in word (general purpose)', word_calendar_v2],
16: ['create GOTT certificates', certificates_gott_build], 16: ['create GOTT certificates', certificates_gott_build],
20: ['build_quiz', build_quiz], 20: ['build_quiz', build_quiz],
21: ['certificates_gott_build, certificates_gott_build'] #21: ['certificates_gott_build, certificates_gott_build']
} }
if len(sys.argv) > 1 and re.search(r'^\d+',sys.argv[1]): if len(sys.argv) > 1 and re.search(r'^\d+',sys.argv[1]):

76
users.py
View File

@ -1,4 +1,5 @@
from importlib import reload
import json, codecs, requests, re, pdb, csv, textdistance, collections import json, codecs, requests, re, pdb, csv, textdistance, collections
import sys, csv, string, funcy, math, shutil, os import sys, csv, string, funcy, math, shutil, os
import pytz, time import pytz, time
@ -20,6 +21,7 @@ from schedules import campus_dept_hierarchy
from util import dept_from_name, most_common_item from util import dept_from_name, most_common_item
from os.path import exists, getmtime from os.path import exists, getmtime
from localcache2 import refresh_semester_schedule_db
from canvas_secrets import url from canvas_secrets import url
from dateutil import parser from dateutil import parser
@ -2206,10 +2208,11 @@ def cross_ref_training():
wb = load_workbook("C:/Users/phowell/Downloads/GOTT_Completion_masterlist 2023 DEC.xlsx") wb = load_workbook("C:/Users/phowell/Downloads/GOTT_Completion_masterlist 2023 DEC.xlsx")
print(wb.sheetnames) print(wb.sheetnames)
term = "202570"
# Fetch from Canvas DB. Make sure its recently updated. # Fetch from Canvas DB. Make sure its recently updated.
# Also relies on schedule being in database. Run localcache2.courses_to_sched() # Also relies on schedule being in database. Run localcache2.courses_to_sched()
courses = all_2x_sem_courses_teachers('202550', '202570') # #courses = all_2x_sem_courses_teachers('202550', '202570') #
#courses = all_sem_courses_teachers('202470') courses = all_sem_courses_teachers(term)
# report for email # report for email
@ -2220,6 +2223,13 @@ def cross_ref_training():
ask = input('download new list of teachers? (y/n) ') ask = input('download new list of teachers? (y/n) ')
if ask.strip()=='y': RELOAD_TEACHERS = 1 if ask.strip()=='y': RELOAD_TEACHERS = 1
RELOAD_SCHEDULE = 0
ask2 = input('refresh schedule db? (y/n) ')
if ask2.strip()=='y': RELOAD_SCHEDULE = 1
if RELOAD_SCHEDULE:
refresh_semester_schedule_db(term)
if RELOAD_TEACHERS: if RELOAD_TEACHERS:
teacherRolesUpdateCache() teacherRolesUpdateCache()
@ -2257,7 +2267,9 @@ def cross_ref_training():
if record[0]=='Name': continue if record[0]=='Name': continue
print(f" {goo}") print(f" {goo}")
try: try:
if record[4]=="None": if len(record)<3:
by_goo[goo][course] = "ok"
elif record[4]=="None":
by_goo[goo][course] = "ok" by_goo[goo][course] = "ok"
else: else:
by_goo[goo][course] = record[4] # record the end date by_goo[goo][course] = record[4] # record the end date
@ -2275,11 +2287,11 @@ def cross_ref_training():
for c in courses: for c in courses:
print(c) print(c)
try: try:
goo = c[8] goo = c[6] ## is this changing? c[8]
crn = c[4] crn = c[8]
name = c[1] # full course name name = c[1] # full course name
teacher = c[6] # last, first teacher = c[4] # last, first
ctype = c[3] ctype = c[7]
dept1 = re.search(r'([A-Z]+)(\d+)',c[2].split(' ')[0]).group(1) dept1 = re.search(r'([A-Z]+)(\d+)',c[2].split(' ')[0]).group(1)
alldepts.add(dept1) alldepts.add(dept1)
d = list(c) d = list(c)
@ -2302,7 +2314,7 @@ def cross_ref_training():
deptfont = Font(bold=True) deptfont = Font(bold=True)
flagfont = PatternFill("solid", fgColor="00FFFFCC") flagfont = PatternFill("solid", fgColor="00FFFFCC")
for thedean in ['et','nl','ss','jn', 'de']: for thedean in ['et','nl','vc','jn', 'de']:
sheet.cell(row=r, column=1).value = dean_code_to_name[thedean] sheet.cell(row=r, column=1).value = dean_code_to_name[thedean]
sheet.cell(row=r, column=1).font = deptfont sheet.cell(row=r, column=1).font = deptfont
r += 2 r += 2
@ -2325,8 +2337,8 @@ def cross_ref_training():
waived = 0 waived = 0
sects = teachers[t] sects = teachers[t]
print(f"Sections for {t}: {sects}") print(f"Sections for {t}: {sects}")
goo = sects[0][8] goo = sects[0][6]
course_mode = sects[0][3] course_mode = sects[0][7]
print(t) print(t)
sheet.cell(row=r, column=1).value = f"{t}" sheet.cell(row=r, column=1).value = f"{t}"
sheet.cell(row=r, column=2).value = f"{goo}" sheet.cell(row=r, column=2).value = f"{goo}"
@ -2573,6 +2585,48 @@ def get_port_pages(id=0):
p = fetch(f"{url}/api/v1/eportfolios/{id}/pages") p = fetch(f"{url}/api/v1/eportfolios/{id}/pages")
print(json.dumps(p, indent=2)) print(json.dumps(p, indent=2))
def set_email_skip_confirm():
user_id = '76741'
old_email = 'scalhoun@gavilan.edu'
new_email = 'scalhoun@hartnell.edu'
create_url = url + f'/api/v1/users/{user_id}/communication_channels'
print(create_url)
list_channels = fetch(create_url)
print(json.dumps(list_channels,indent=2))
for ch in list_channels:
id = ch['id']
result = requests.delete(create_url+f"/{id}", headers=header)
print(f"deleting id {id}")
print(result.content)
print('after deleting:')
list_channels = fetch(create_url)
print(json.dumps(list_channels,indent=2))
print('\n\n')
body = {
'communication_channel[address]' : old_email,
'communication_channel[type]' : 'email',
'skip_confirmation': True
}
response = requests.post(create_url, headers=header, data = body)
print(response.json()) #confirm that the channel was created
body = {
'communication_channel[address]' : new_email,
'communication_channel[type]' : 'email',
'skip_confirmation': True
}
response = requests.post(create_url, headers=header, data = body)
print(response.json()) #confirm that the channel was created
print('after creating:')
list_channels = fetch(create_url)
print(json.dumps(list_channels,indent=2))
print('\n\n')
if __name__ == "__main__": if __name__ == "__main__":
@ -2607,6 +2661,8 @@ if __name__ == "__main__":
30: ['get portfolios for user id', get_portfolios], 30: ['get portfolios for user id', get_portfolios],
31: ['get portfolio pages for portfolio id', get_port_pages], 31: ['get portfolio pages for portfolio id', get_port_pages],
40: ['reset user email without confirmation', set_email_skip_confirm]
#3: ['Main index, 1 year, teachers and their classes', getAllTeachersInTerm], #3: ['Main index, 1 year, teachers and their classes', getAllTeachersInTerm],
#5: ['Match names in schedule & ilearn', match_usernames], #5: ['Match names in schedule & ilearn', match_usernames],
#6: ['Create Dept\'s ZTC list', create_ztc_list], #6: ['Create Dept\'s ZTC list', create_ztc_list],