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

fancy term cross checking. also semester course review v1

This commit is contained in:
Peter Howell 2025-08-06 15:46:23 -07:00
parent 06f7f31fc5
commit 962aed84d6
4 changed files with 254 additions and 97 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

158
courses.py
View File

@ -13,6 +13,7 @@ from schedules import get_semester_schedule
from localcache import course_quick_stats, get_courses_in_term_local, course_student_stats, all_sem_courses_teachers, full_reload
from localcache2 import db, users_new_this_semester, users_new_this_2x_semester, course_from_id, user_ids_in_shell
from collections import defaultdict
from semesters import find_term
@ -1452,58 +1453,98 @@ def course_search_by_sis():
# print(json.dumps(x, indent=2))
# run overview_start_dates to get most recent info
def set_custom_start_dates():
TERM = 288
SEM = "su25"
from datetime import datetime
term = find_term( input("term? (ex: fa25) ") )
if not term or (not 'canvas_term_id' in term) or (not 'code' in term):
print(f"Couldn't find term. Try updating the saved terms list.")
return
TERM = term['canvas_term_id']
SEM = term['code']
term_start_month = term['begin'].split('/')[0]
term_start_day = term['begin'].split('/')[1]
term_start_year = '20' + term['code'][2:4]
print(f"term begins on {term_start_month}/{term_start_day}")
filepath = f"cache/overview_semester_shells_{SEM}.csv"
if not os.path.exists(filepath):
print(f"file does not exist: {filepath}")
print("Run overview_start_dates first")
return
make_changes = 1
do_all = 0
get_fresh = 0
term_start_month = 6
term_start_day = 2
# just do certain ids in cache/changeme.txt
limit_to_specific_ids = 1
limit_to_specific_ids = 0
limit_to = [x.strip() for x in open('cache/changeme.txt','r').readlines()]
# get list of online course shells
if get_fresh:
print(f"Getting list of courses in {SEM}")
c = getCoursesInTerm(TERM,get_fresh,0)
codecs.open(f'cache/courses_in_term_{TERM}.json','w','utf-8').write(json.dumps(c,indent=2))
else:
c = json.loads( codecs.open(f'cache/courses_in_term_{TERM}.json','r','utf-8').read() )
def adjust_shell_startdate(row):
# Placeholder stub
pass
# dict to match section numbers between shells and schedule
crn_to_canvasid = {}
for C in c:
if 'sis_course_id' in C and C['sis_course_id']:
print( f"{C['name']} -> {C['sis_course_id'][7:13]}" )
crn_to_canvasid[C['sis_course_id'][7:13]] = str(C['id'])
else:
print( f"---NO CRN IN: {C['name']} -> {C}" )
# get course info from schedule
s = requests.get(f"http://gavilan.cc/schedule/{SEM}_sched_expanded.json").json()
for S in s:
# get dates
start = re.sub( r'\-','/', S['start']) + '/20' + SEM[2:4]
d_start = datetime.strptime(start,"%m/%d/%Y")
# try to find online shell matching this schedule entry
def parse_date(date_str):
if not date_str or date_str.lower() == 'none':
return None
try:
this_id = crn_to_canvasid[S['crn']]
return datetime.fromisoformat(date_str.replace("Z", "").replace("T", " "))
except ValueError:
return None
if limit_to_specific_ids and (not this_id in limit_to):
continue
except Exception as e:
print(f"DIDN'T FIND CRN - {start} {d_start} - {S['code']} {S['crn']} {S['name']}" )
with open(filepath, newline='', encoding='utf-8') as csvfile:
reader = csv.DictReader(csvfile)
for row in reader:
annotations = {}
# Skip shells with no sections
if int(row["shell_numsections"]) == 0:
continue
print(f" - {start} {d_start} - id: {this_id} - {S['code']} {S['crn']} {S['name']}" )
sched_start = parse_date(row["sched_start"])
shell_start = parse_date(row["shell_start"])
shortname = row["shell_shortname"]
num_sections = int(row["shell_numsections"])
# Check early/late start
if sched_start:
if (sched_start.month, sched_start.day) != (int(term_start_month), int(term_start_day)):
if (sched_start.month, sched_start.day) < (int(term_start_month), int(term_start_day)):
annotations["is_early_start"] = sched_start.date().isoformat()
else:
annotations["is_late_start"] = sched_start.date().isoformat()
# Check if shell_start is manually set
if shell_start:
annotations["shell_custom_start"] = shell_start.date().isoformat()
else:
if "is_early_start" in annotations or "is_late_start" in annotations:
adjust_shell_startdate(row)
# Check section numbers in shortname vs shell_numsections
if '/' in shortname:
parts = shortname.split()
section_part = parts[-1] # last part assumed to be section numbers
section_groups = section_part.split('/')
if len(section_groups) != num_sections:
annotations["shell_warn_crosslist_sections"] = section_part
# Print row and annotations if anything notable happened
if annotations:
print("Annotations:", annotations)
print("Row:", row)
print("-" * 80)
return
'''
# Do we adjust the start date? Only if it doesn't match term
if d_start.month == term_start_month and d_start.day == term_start_day:
print(" Ignoring, term start date" )
@ -1523,39 +1564,41 @@ def set_custom_start_dates():
u2 = f"https://gavilan.instructure.com:443/api/v1/courses/{this_id}"
r3 = requests.put(u2, headers=header, params=data)
print(" updated.. OK")
'''
def overview_start_dates():
TERM = 288
SEM = "su25"
term = find_term( input("term? (ex: fa25) ") )
get_fresh = 1
if not term or (not 'canvas_term_id' in term) or (not 'code' in term):
print(f"Couldn't find term.")
return
term_start_month = 6
term_start_day = 2
TERM = term['canvas_term_id']
SEM = term['code']
output = codecs.open(f"cache/overview_semester_shells_{SEM}.csv","w","utf-8")
get_fresh = 0
# get list of online course shells
if get_fresh:
print(f"Getting list of courses in {SEM}")
c = getCoursesInTerm(TERM,get_fresh,0)
codecs.open(f'cache/courses_in_term_{TERM}.json','w','utf-8').write(json.dumps(c,indent=2))
else:
c = json.loads( codecs.open(f'cache/courses_in_term_{TERM}.json','r','utf-8').read() )
# dict to match section numbers between shells and schedule
crn_to_canvasid = {}
for C in c:
if 'sis_course_id' in C and C['sis_course_id']:
print( f"{C['name']} -> {C['sis_course_id'][7:13]}" )
#print( f"{C['name']} -> {C['sis_course_id'][7:13]}" )
crn_to_canvasid[C['sis_course_id'][7:13]] = str(C['id'])
else:
print( f"---NO CRN IN: {C['name']} -> {C}" )
print(f"id,shell_shortname,sched_start,shell_start,shell_end,shell_restrict_view_dates,shell_restrict_view_dates,shell_state,shell_numstudents" )
header = f"id,shell_shortname,sched_start,shell_start,shell_end,shell_restrict_view_dates,shell_restrict_view_dates,shell_state,shell_numstudents,shell_numsections"
output.write(header + "\n")
print("\n\n" + header)
# get course info from schedule
s = requests.get(f"http://gavilan.cc/schedule/{SEM}_sched_expanded.json").json()
s = requests.get(f"https://gavilan.cc/schedule/{SEM}_sched_expanded.json").json()
for S in s:
# get dates
start = re.sub( r'\-','/', S['start']) + '/20' + SEM[2:4]
@ -1578,10 +1621,21 @@ def overview_start_dates():
if 'access_restricted_by_date' in this_course:
shell_restrict_view_dates = this_course['access_restricted_by_date']
shell_shortname = this_course['course_code']
shell_numstudents = '?' #this_course['total_students']
shell_state = this_course['workflow_state']
print(f"{this_id},{shell_shortname},{d_start},{shell_start},{shell_end},{shell_restrict_view_dates},{shell_restrict_view_dates},{shell_state},{shell_numstudents}" )
# get user count
ss = f"{url}/api/v1/courses/{this_id}/users"
enrollments = fetch(ss, params={"enrollment_type[]":"student"})
shell_numstudents = len(enrollments)
# cross-listed?
sec = f"{url}/api/v1/courses/{this_id}/sections"
sections = fetch(sec, params={"include[]":"total_students"})
shell_numsections = len(sections)
content = f"{this_id},{shell_shortname},{d_start},{shell_start},{shell_end},{shell_restrict_view_dates},{shell_restrict_view_dates},{shell_state},{shell_numstudents},{shell_numsections}"
output.write(content + "\n")
print(content)

1
pipelines.py
View File

@ -8,7 +8,6 @@ from datetime import timedelta
import datetime
#from collections import defaultdict
from semesters import short_to_long
from canvas_secrets import apiKey, apiSecret, FTP_SITE, FTP_USER, FTP_PW, url, domain, account_id, header, header_media, g_id, g_secret
from canvas_secrets import instructure_url, instructure_username, instructure_private_key

76
semesters.py
View File

@ -33,6 +33,18 @@ def short_to_long(s):
seasons = {'sp':'spring','su':'summer','fa':'fall','wi':'winter'}
return '20'+yr+seasons[season]
# from "Summer 2024" or "2024 Summer" to ("Summer", 2024)
def normalize(label):
"""Return (season, year) tuple, or None if not standard."""
m = re.search(r'(Fall|Summer|Spring|Winter)\s+(\d{4})', label, re.I)
if m:
return (m.group(1).title(), int(m.group(2)))
else:
m = re.search(r'(\d{4})\s+(Fall|Summer|Spring|Winter)', label, re.I)
if m:
return (m.group(2), int(m.group(1).title()))
return None
# from "Summer 2024" to 202450
def human_to_sis(semester):
try:
@ -125,6 +137,8 @@ begin = ['08/25','05/22','01/26','01/01', # not sure on fa26
canvas_label = []
sems_by_human_name = {}
startdate_by_code = dict(zip(code,begin))
for s in list(zip(standard,code,begin)):
season,year = s[0].split(' ')
@ -146,7 +160,69 @@ def dump():
print(json.dumps(sems_by_short_name,indent=2))
GET_FRESH_TERMS = 0
if (GET_FRESH_TERMS):
from pipelines import url, fetch_collapse
import codecs
canvas_terms = fetch_collapse(f'{url}/api/v1/accounts/1/terms', 'enrollment_terms')
ff = codecs.open('cache/courses/terms.txt', 'w', 'utf-8') # TODO unsafe overwrite
ff.write(json.dumps(canvas_terms, indent=2))
ff.close()
else:
canvas_terms = json.loads(open('cache/courses/terms.txt', 'r').read())
# Build main records from standard list
records = {}
for std, code_str, start in zip(standard, code, begin):
season, year = normalize(std)
records[(season, year)] = {
'standard': std,
'code': code_str,
'begin': start,
'canvas_term_id': None,
'canvas_name': None,
'canvas_start_at': None,
'canvas_end_at': None
}
# Attach canvas term ids where possible
for term in canvas_terms: # term_id, label
key = normalize(term['name'])
if key and key in records:
records[key]['canvas_term_id'] = term['id']
records[key]['canvas_name'] = term['name']
records[key]['canvas_start_at'] = term['start_at']
records[key]['canvas_end_at'] = term['end_at']
# Build alias lookup
lookup = {}
for key, rec in records.items():
season, year = key
aliases = [
rec['standard'],
f"{year} {season}",
f"{season} {year}",
rec['code'],
rec['canvas_name'] if rec['canvas_name'] else None,
str(rec['canvas_term_id'])
]
for alias in aliases:
if alias:
lookup[alias.lower()] = rec
def find_term(s):
return lookup.get(s.lower())
#print(find_term("2025 Fall"))
# {'standard': 'Fall 2025', 'code': 'fa25', 'begin': '08/25',
# 'canvas_term_id': 289, 'canvas_label': '2025 Fall'}
#print(find_term("fa25"))
# Same record as above
#print(find_term("Fall 2025"))
# Same record as above
def weeks_from_date():

64
users.py
View File

@ -2700,6 +2700,7 @@ def summarize_submissions(submissions):
},
"assignment": {
"id": assignment.get("id"),
"name": assignment.get("name"),
"excerpt": strip_html_and_truncate(assignment.get("description", "")),
"due_at": assignment.get("due_at"),
"is_quiz": assignment.get("is_quiz_assignment", False),
@ -2708,35 +2709,62 @@ def summarize_submissions(submissions):
})
return summary
def format_assingments_results_table(results):
from datetime import datetime
import pytz
def format_assignments_results_table(results):
def safe(val):
return str(val) if val is not None else "-"
def clip(text):
return (text[:40] + "...") if text and len(text) > 43 else (text or "")
def clip(text,length=40):
return (text[:length] + "...") if text and len(text) > length+3 else (text or "")
def to_pacific(iso):
if not iso:
return "-"
utc = datetime.strptime(iso, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=pytz.utc)
pacific = utc.astimezone(pytz.timezone("US/Pacific"))
return pacific.strftime("%Y-%m-%d %I:%M%p")
# Sort by assignment due date (missing dates go last)
def get_due_at(item):
dt = item["assignment"].get("due_at")
return datetime.max if not dt else datetime.strptime(dt, "%Y-%m-%dT%H:%M:%SZ")
results = sorted(results, key=get_due_at)
header = (
"| Assignment ID | Due Date | Points | Assignment Excerpt "
"| Submission ID | Grade | Submitted At | Late | Missing | Submission Excerpt |"
"| Type | Subm/Assmt ID | Due Date (PT) | Submitted (PT) | Grade/Points | Assignment Excerpt | Submission Excerpt |"
)
sep = (
"|---------------|---------------------|--------|-----------------------"
"|----------------|-----------|--------------------------|-------|---------|-----------------------|"
"|------------|---------------------|--------------------|----------------------|----------------------|-------------------------------|-------------------------------|"
)
rows = []
for item in results:
a = item['assignment']
s = item['submission']
a = item["assignment"]
s = item["submission"]
kind = "quiz" if a.get("is_quiz") else "assignment"
id_combo = f"{safe(s['id'])}/{safe(a['id'])}"
due_pt = to_pacific(a.get("due_at"))
submitted_pt = to_pacific(s.get("submitted_at"))
grade = safe(s.get("grade"))
points = safe(a.get("points_possible"))
flags = []
if s.get("late"):
flags.append("late")
if s.get("missing"):
flags.append("missing")
gradepoints = f"{grade}/{points}" + (" " + ",".join(flags) if flags else "")
row = (
f"| {safe(a['id']):<13} | {safe(a['due_at']):<19} | {safe(a['points_possible']):<6} | {clip(a['excerpt']):<23} "
f"| {safe(s['id']):<14} | {safe(s['grade']):<9} | {safe(s['submitted_at']):<24} | {safe(s['late']):<5} | {safe(s['missing']):<7} | {clip(s['excerpt']):<23} |"
f"| {kind:<10} | {id_combo:<19} | {due_pt:<18} | {submitted_pt:<20} | {gradepoints:<20} | {clip(a.get('name'),20) + ' - ' + clip(a.get('excerpt')):<49} | {clip(s.get('excerpt')):<29} |"
)
rows.append(row)
return '\n'.join([header, sep] + rows)
def user_course_enrollment(user_id, course_id):
user_url = f"{url}/api/v1/courses/{course_id}/enrollments"
myparams = {"user_id": user_id, "type[]": "StudentEnrollment", "state[]": ['active','invited','deleted','rejected','completed','inactive']}
@ -2750,18 +2778,18 @@ def get_student_course_assignments(student_id, course_id):
submissions_url = f"{url}/api/v1/courses/{course_id}/students/submissions"
submissions = fetch(submissions_url, params=submission_params)
summary = summarize_submissions(submissions)
fmt = format_assingments_results_table(summary)
fmt = format_assignments_results_table(summary)
return fmt
def testme():
course_id = 21186
student_id = 73180
course_id = 22054
student_id = 63638
x = get_student_course_assignments(student_id, course_id)
print(x)
# print(json.dumps(x,indent=2))
testme()
exit()
#testme()
#exit()
def get_student_course_grades(student_id, course_id):
results = {}
@ -2814,7 +2842,7 @@ def get_student_course_grades(student_id, course_id):
assignments = []
results = {
"course_code": course_name,
#"course_code": course_name,
"final_score": final_score,
"final_grade": final_grade,
"assignments": assignments