Canvas LMS script to automatically check the quiz audit logs and filter out students who interrupted the quiz too many times.

canvas-audit.py

"""MIT License

Copyright (c) 2021-2022 Mate Cserep, https://mcserep.web.elte.hu/

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
"""

### SETTINGS ###
# Base URL of your Canvas instance
base_url = 'https://canvas.elte.hu'
# Create a new token at the '<base_url>/profile/settings' page.
access_token = 'INSERT YOUR ACCESS TOKEN'
# Course and quiz ID. You can get it from the URL upon opening the quiz in your browser.
# The format will be: '<base_url>/courses/<course_id>/quizzes/<quiz_id>'
course_id = 12345
quiz_id = 12345
# Minimum threshold of quiz interruptions to report
threshold_interruptions = 10
# Minimum threshold of interrupted questions to report
threshold_questions = 3

### PROGRAM ###
import requests
import datetime

def audit_submission(submission_id):
    """Evaluates the audit log for a single submission."""

    interruptions_all = 0
    interrupted_questions = 0

    current_interruption_start = None
    current_question_interrupted = False

    page = 1
    while True:
        response = requests.get('{0}/api/v1/courses/{1}/quizzes/{2}/submissions/{3}/events'.format(base_url, course_id, quiz_id, submission_id), {
            'access_token': access_token,
            'per_page': 50,
            'page': page
        })
        response.raise_for_status()
        events = response.json().get('quiz_submission_events', [])
        page += 1

        if len(events) == 0:
            break

        for event in events:
            if event['event_type'] == 'page_blurred':
                current_interruption_start = datetime.datetime.strptime(event['created_at'], '%Y-%m-%dT%H:%M:%SZ')

            if event['event_type'] in ['page_focused', 'question_answered']:
                if current_interruption_start is not None:
                    current_interruption_end = datetime.datetime.strptime(event['created_at'], '%Y-%m-%dT%H:%M:%SZ')
                    current_interruption_length = (current_interruption_end - current_interruption_start).total_seconds()

                    if current_interruption_length > 0:
                        interruptions_all += 1
                        current_question_interrupted = True
                    current_interruption_start = None

            if event['event_type'] == 'question_answered':
                if current_question_interrupted:
                    interrupted_questions += 1
                    current_question_interrupted = False

    return interruptions_all, interrupted_questions

def get_user_name(user_id):
    """Retrieves the name of the specified user."""

    response = requests.get('{0}/api/v1/courses/{1}/users'.format(base_url, course_id), {
        'access_token': access_token,
        'search_term': user_id
    })
    response.raise_for_status()
    user = response.json()

    if len(user) == 0:
        return user_id
    else:
        return user[0]['name']

# main
page = 1
try:
    response = requests.get('{0}/api/v1/courses/{1}/quizzes/{2}'.format(base_url, course_id, quiz_id), {
        'access_token': access_token,
        'per_page': 50,
        'page': page
    })
    response.raise_for_status()
    quiz = response.json()
    measure_question_interruptions = quiz['one_question_at_a_time'] and quiz['cant_go_back']
    if not measure_question_interruptions:
        print("NOTE: NUMBER OF INTERRUPTED QUESTIONS WON'T BE SHOWN")
        print("Interrupted questions can only be counted for quizzes \n" +
              "where only a single question is shown at a time and \n" +
              "students can't go back to previous questions.")
        print()

    catch_count = 0
    print('Students above the thresholds:')
    while True:
        response = requests.get('{0}/api/v1/courses/{1}/quizzes/{2}/submissions'.format(base_url, course_id, quiz_id), {
            'access_token': access_token,
            'per_page': 50,
            'page': page
        })
        response.raise_for_status()
        submissions = response.json().get('quiz_submissions', [])
        page += 1

        if len(submissions) == 0:
            break

        for submission in submissions:
            interruptions_all, interrupted_questions = audit_submission(submission['id'])

            if interruptions_all >= threshold_interruptions or interrupted_questions >= threshold_questions:
                user_name = get_user_name(submission['user_id'])
                catch_count += 1

                if (measure_question_interruptions):
                    print('#{0} {1}: {2} total interruptions, {3} interrupted questions'.format(catch_count, user_name, interruptions_all, interrupted_questions))
                else:
                    print('#{0} {1}: {2} total interruptions'.format(catch_count, user_name, interruptions_all, interrupted_questions))

    if catch_count == 0:
        print('None.')

except requests.exceptions.RequestException as err:
    print('Canvas communication error occurred, HTTP response code: {0}'.format(err.response.status_code))
    print('Response: {0}'.format(err.response.text))