import csv import os import re from glob import glob from pathlib import Path from typing import cast from canvasapi import Canvas as PyCanvas from canvasapi.assignment import Assignment from canvasapi.user import User from patoolib import extract_archive from patoolib.util import PatoolError from joint_teapot.config import settings from joint_teapot.utils.logger import logger from joint_teapot.utils.main import first, percentile class Canvas: def __init__( self, domain_name: str = "", suffix: str = "", access_token: str = "", # nosec course_id: int = 0, grade_filename: str = "GRADE.txt", ): domain_name = domain_name or settings.canvas_domain_name suffix = suffix or settings.canvas_suffix access_token = access_token or settings.canvas_access_token course_id = course_id or settings.canvas_course_id self.canvas = PyCanvas(f"https://{domain_name}{suffix}", access_token) self.course = self.canvas.get_course(course_id) logger.info(f"Canvas course loaded. {self.course}") # types = ["student", "observer"] types = ["student"] def patch_student(student: User) -> User: student.name = re.sub( r"[^\x00-\x7F]+", "", student.name ).strip() # We only care english name student.sis_id = student.login_id student.login_id = student.email.split("@")[0] return student self.students = [ patch_student(student) for student in self.course.get_users(enrollment_type=types) ] for attr in ["login_id", "name"]: if not hasattr(self.students[0], attr): raise Exception( f"Unable to gather students' {attr}, please contact the Canvas site admin" ) logger.debug("Canvas students loaded") self.assignments = self.course.get_assignments() logger.debug("Canvas assignments loaded") self.groups = self.course.get_groups() logger.debug("Canvas groups loaded") self.grade_filename = grade_filename logger.debug("Canvas initialized") def export_students_to_csv(self, filename: Path) -> None: with open(filename, mode="w", newline="") as file: writer = csv.writer(file) for student in self.students: writer.writerow([student.name, student.sis_id, student.login_id]) logger.info(f"Students exported to {filename}") def prepare_assignment_dir( self, dir_or_zip_file: str, create_grade_file: bool = True ) -> None: if os.path.isdir(dir_or_zip_file): assignments_dir = dir_or_zip_file else: assignments_dir = os.path.splitext(dir_or_zip_file)[0] if os.path.exists(assignments_dir): logger.error( f"{assignments_dir} exists, can not unzip submissions file" ) return extract_archive(dir_or_zip_file, outdir=assignments_dir, verbosity=-1) login_ids = {stu.id: stu.login_id for stu in self.students} for v in login_ids.values(): new_path = os.path.join(assignments_dir, v) if not os.path.exists(new_path): os.mkdir(new_path) if create_grade_file: grade_file_path = os.path.join(new_path, self.grade_filename) if not os.path.exists(grade_file_path): open(grade_file_path, mode="w") late_students = set() error_students = set() submitted_ids = set() for path in glob(os.path.join(assignments_dir, "*")): try: filename = os.path.basename(path) if "_" not in filename: continue segments = filename.split("_") if segments[1] == "late": file_id = int(segments[2]) else: file_id = int(segments[1]) login_id = login_ids[file_id] except Exception: logger.error(f"Error on parsing path: {path}") continue student = first(self.students, lambda x: x.login_id == login_id) target_dir = os.path.join(assignments_dir, login_id) if segments[1] == "late": # TODO: check the delay time of late submission if create_grade_file: grade_file_path = os.path.join(target_dir, self.grade_filename) if os.path.exists(grade_file_path): open(grade_file_path, mode="a").write("LATE SUBMISSION\n") late_students.add(student) try: extract_archive(path, outdir=target_dir, verbosity=-1) logger.info(f"Extract succeed: {student}") os.remove(path) except PatoolError as e: if not str(e).startswith("unknown archive format"): logger.exception(f"Extract failed: {student}") error_students.add(student) os.rename(path, os.path.join(target_dir, filename)) submitted_ids.add(login_id) if login_ids: no_submission_students = [ first(self.students, lambda x: x.login_id == login_id) for login_id in set(login_ids.values()) - submitted_ids ] if no_submission_students: tmp = ", ".join([str(student) for student in no_submission_students]) logger.info(f"No submission student(s): {tmp}") if late_students: tmp = ", ".join([str(student) for student in late_students]) logger.info(f"Late student(s): {tmp}") if error_students: tmp = ", ".join([str(student) for student in error_students]) logger.info(f"Extract error student(s): {tmp}") def upload_assignment_grades( self, assignments_dir: str, assignment_name: str ) -> None: assignment = first(self.assignments, lambda x: x.name == assignment_name) if assignment is None: logger.info(f"Canvas assignment {assignment_name} not found") return assignment = cast(Assignment, assignment) submission_dict = {} float_grades = [] is_float_grades = True for submission in assignment.get_submissions(): student = first(self.students, lambda x: x.id == submission.user_id) if student is None: continue grade_file_path = os.path.join( assignments_dir, student.login_id, self.grade_filename ) try: grade, *comments = list(open(grade_file_path)) grade = grade.strip() try: float_grades.append(float(grade)) except ValueError: is_float_grades = False data = { "submission": {"posted_grade": grade}, "comment": {"text_comment": "".join(comments)}, } submission_dict[(student, submission)] = data comment_no_newline = ( data["comment"]["text_comment"].strip().replace("\n", " ") ) logger.info( f"Grade file parsed for {assignment} {student}: " f"grade: {data['submission']['posted_grade']}, " f'comment: "{comment_no_newline}"' ) except Exception: logger.error(f"Can not parse grade file {grade_file_path}") return for (student, submission), data in submission_dict.items(): logger.info( f"Uploading grade for {assignment} {student}: {data.__repr__()}" ) submission.edit(**data) if is_float_grades and float_grades: summary = [ min(float_grades), percentile(float_grades, 0.25), percentile(float_grades, 0.5), percentile(float_grades, 0.75), max(float_grades), ] average_grade = sum(float_grades) / len(float_grades) logger.info( f"Grades summary: " f"Min: {summary[0]:.2f}, " f"Q1: {summary[1]:.2f}, " f"Q2: {summary[2]:.2f}, " f"Q3: {summary[3]:.2f}, " f"Max: {summary[4]:.2f}, " f"Average: {average_grade:.2f}" ) logger.info(f"Canvas assginemnt {assignment} grades upload succeed") if __name__ == "__main__": canvas = Canvas()