get_new_quiz_responses.ipynb

{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "PDT3kFJmokal"
      },
      "source": [
        "# Retrieve student responses to New Quizzes\n",
        "\n",
        "Author: Matthew Ford <mford@cornell.edu>,\n",
        "Postdoctoral Associate in Mechanical & Aerospace Engineering,\n",
        "Cornell Active Learning Initiative\n",
        "\n",
        "The New Quizzes feature in Canvas is still under development (and has been foreeeeever), and lacks basic functionality to export a report of student responses. New Quizzes is implemented as an external tool, so results are not available through the documented Canvas API.\n",
        "\n",
        "However, New Quizzes apparently uses a RESTful API very similar to the standard Canvas API, and can be reverse-engineered by monitoring requests made by your browser when viewing New Quiz results. __This script uses a combination of the standard Canvas API and the \"secret\" New Quiz API to retrieve student responses automatically__.\n",
        "\n",
        "This script produces 4 output files:\n",
        "\n",
        "- `course_<num>_students.csv`: Table of students with user_id, net_id, and name.\n",
        "- `quiz_<num>_submissions.csv`: Table of quiz submissions, one row per submission. Students are identified by user_id.\n",
        "- `quiz_<num>_responses.csv`: Table of student _responses_ to quiz items. Data is in \"long format,\" i.e. one row per item. Data can easily be pivoted to obtain a table of student responses in which each column represents a different quiz item. Responses are identified by a unique response_id.\n",
        "- `quiz_<num>_item_options.csv`: Table of answer options for each multiple-choice item, including item text response_id and response text.\n",
        "\n",
        "These tables can be easily pivoted, joined, and manipulated to get data into whatever convenient format. For example:\n",
        "\n",
        "`quiz_x_responses.csv >> LEFT JOIN with quiz_x_item_options.csv ON item_id >> PIVOT item_text to columns`\n",
        "\n",
        "would give you table with one row per submission and one column per quiz item with the student response _text_ in each cell."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "rzmox9Nnokaq"
      },
      "outputs": [],
      "source": [
        "!pip install canvasapi\n",
        "from canvasapi import Canvas\n",
        "import csv\n",
        "import re\n",
        "from urllib.parse import urlparse, parse_qs\n",
        "import requests\n",
        "import json\n",
        "\n",
        "from IPython.display import JSON"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "pZflNJBUokas"
      },
      "source": [
        "## configuration json"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "FQlGhhTY3Rnq"
      },
      "outputs": [],
      "source": [
        "# configuration json\n",
        "# you can get the value by clicking on the assignent title inside Canvas, and retrieve the assignment_id value from the following url pattern:\n",
        "# https://umich.instructure.com/courses/<course_id>/assignments/<assignment_id>/edit\n",
        "config = {\n",
        "  \"API_TOKEN\": \"\",\n",
        "  \"COURSE_ID\": course_id,\n",
        "  \"NEWQUIZ_ID\": quiz_id,\n",
        "}"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "# read in configuration\n",
        "api_token = config[\"API_TOKEN\"]\n",
        "course_id = config[\"COURSE_ID\"]\n",
        "newquiz_id = config[\"NEWQUIZ_ID\"]"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "LgReW1YVokas"
      },
      "outputs": [],
      "source": [
        "# get course and quiz objects\n",
        "canvas = Canvas('https://umich.instructure.com', api_token)\n",
        "course = canvas.get_course(course_id)\n",
        "newquiz = course.get_assignment(newquiz_id)\n",
        "# Uncomment the line below to use the assignment_id as the quiz name.\n",
        "quiz_name = 'quiz_{:d}'.format(int(newquiz_id))\n",
        "\n",
        "print(quiz_name)\n",
        "print(course.name)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Lr4M0Ncpokax"
      },
      "source": [
        "## Get quiz information"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "i5gqCb7Bokay"
      },
      "outputs": [],
      "source": [
        "canvas = Canvas('https://umich.instructure.com', api_token)\n",
        "course = canvas.get_course(course_id)\n",
        "quiz = course.get_assignment(newquiz_id)\n",
        "\n",
        "quiz = course.get_assignment(newquiz_id)\n",
        "\n",
        "# Uncomment the code below for the course in which the quizzes were all named \"Quiz <X> - ...\"\n",
        "# quiz_num = int(re.search('Quiz ([0-9]+)', quiz.name).group(1))\n",
        "# quiz_name = 'quiz_{:02d}'.format(quiz_num)\n",
        "\n",
        "# Uncomment the line below to use the assignment_id as the quiz name.\n",
        "quiz_name = 'quiz_{:d}'.format(int(newquiz_id))\n",
        "\n",
        "print(quiz_name)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "fSnwLAN8okay"
      },
      "source": [
        "## Download quiz submission metadata\n",
        "\n",
        "Creates a file `quiz_<newquiz_id>_submissions.csv` with the following columns:\n",
        "\n",
        "quiz_id | student_id | submission_id | attempt | submission_time | participant_session_id | quiz_session_id\n",
        "\n",
        "**NOTE:** Check the output file `quiz_<newquiz_id>_submissions.csv` in the current folder"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "bNQRqPNUokaz",
        "scrolled": true
      },
      "outputs": [],
      "source": [
        "with open('{:s}_submissions.csv'.format(quiz_name), 'w', newline='') as csv_file,\\\n",
        "     open('{:s}_log.txt'.format(quiz_name), 'w') as log:\n",
        "\n",
        "    log.write('Reading quiz submissions...\\n')\n",
        "\n",
        "    fout = csv.DictWriter(csv_file,\n",
        "                          fieldnames=['user_id', 'assignment_id',\n",
        "                                      'submission_id', 'attempt', 'submission_time',\n",
        "                                      'participant_session_id', 'quiz_session_id'])\n",
        "    fout.writeheader()\n",
        "\n",
        "    n_subs = 0\n",
        "    n_missing = 0\n",
        "    for i, s in enumerate(quiz.get_submissions()):\n",
        "\n",
        "        if not s.missing:\n",
        "\n",
        "            try:\n",
        "                #print(s)\n",
        "                #print(s.external_tool_url)\n",
        "                quiz_session = parse_qs(urlparse(s.external_tool_url).query)\n",
        "\n",
        "                fout.writerow({\n",
        "                    'user_id': s.user_id,\n",
        "                    'assignment_id': s.assignment_id,\n",
        "                    'submission_id': s.id,\n",
        "                    'attempt': s.attempt,\n",
        "                    'submission_time': s.submitted_at_date,\n",
        "                    'participant_session_id': quiz_session['participant_session_id'][0],\n",
        "                    'quiz_session_id': quiz_session['quiz_session_id'][0]})\n",
        "                n_subs += 1\n",
        "            except Exception as e:\n",
        "                #print(e)\n",
        "                msg = 'ERROR processing submission for User ID: {:d}, submission ID: {:d}'\\\n",
        "                    .format(s.user_id, s.id)\n",
        "                print(msg)\n",
        "                log.write(msg + '\\n')\n",
        "        else:\n",
        "            msg = 'Missing submission for User ID: {:d}, submission ID: {:d}'.format(s.user_id, s.id)\n",
        "            print(msg)\n",
        "            log.write(msg + '\\n')\n",
        "            n_missing += 1\n",
        "\n",
        "    print('Wrote submission metadata for {:d} submissions'.format(n_subs))\n",
        "    print('Missing {:d} submissions'.format(n_missing))\n",
        "\n",
        "    log.write('Wrote submission metadata for {:d} submissions\\n'.format(n_subs))\n",
        "    log.write('Missing {:d} submissions\\n\\n'.format(n_missing))"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "L1U9vrI-okaz"
      },
      "source": [
        "## Obtain a \"Bearer\" authorization token\n",
        "\n",
        "- Run the code snipped below and click on the URL that appears (in Google Chrome)\n",
        "- Open the Developer Tools, go to the Network tab, and refresh the page\n",
        "- Click on \"XHR\" to see only AJAX requests\n",
        "- Find the request called \"grade\" and locate the \"authorization\" field in Request Headers \"https://umich.quiz-lti-iad-prod.instructure.com/api/participant_sessions/(session_id)/grade\"\n",
        "- Copy the long string after \"Bearer \" and save to a file called \"BEARER.txt\" in the same directory as this file\n",
        "\n",
        "The bearer token expires in 60 minutes, so make sure you complete the rest of the code within that time."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "rYAErggDoka0"
      },
      "outputs": [],
      "source": [
        "# Get first valid submission\n",
        "for s in quiz.get_submissions():\n",
        "    print(s)\n",
        "    if (not s.missing) and (s.submitted_at is not None) and hasattr(s, 'external_tool_url'):\n",
        "        break\n",
        "\n",
        "# Get submission information\n",
        "if hasattr(s, \"external_tool_url\"):\n",
        "  print(s)\n",
        "  quiz_session = parse_qs(urlparse(s.external_tool_url).query)\n",
        "\n",
        "  # Get submission result URL\n",
        "  url = urlparse(s.preview_url)\n",
        "  print(url.scheme + \"://\" + url.netloc + url.path)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "CaNA87k_6GBh"
      },
      "outputs": [],
      "source": [
        "# Get bearer token (get from Chrome Developer Mode - good for 60 minutes)\n",
        "bearer_token = \"\""
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "cgrTtlQ6oka0"
      },
      "source": [
        "## Download all submission responses\n",
        "\n",
        "Iterates over all submissions and creates a table with the following columns:\n",
        "\n",
        "user_id | assignment_id | submission_id | status | item_id | item_order | item_choice_choice | correct\n",
        "\n",
        "This code retrieves the item responses for all the submissions listed in `quiz_<newquiz_id>_submissions.csv`. If you only want to retrieve a subset of responses (maybe you had a few failures and only want to re-retrieve the ones that failed), just delete the rows for which you don't want to retrieve responses.\n",
        "\n",
        "**NOTE**: checkout file `quiz_<newquiz_id>_submissions.csv`"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "CWnu1Yxeoka0"
      },
      "outputs": [],
      "source": [
        "# Set verbose = True to print information about every submission as it is retrieved.\n",
        "# Information will be written to the log file regardless of this setting\n",
        "verbose = False\n",
        "auth_url =         'https://umich.quiz-lti-iad-prod.instructure.com/api/participant_sessions/{:s}/grade'\n",
        "quiz_session_url = 'https://umich.quiz-api-iad-prod.instructure.com/api/quiz_sessions/{:s}'\n",
        "results_url =      'https://umich.quiz-api-iad-prod.instructure.com/api/quiz_sessions/{:s}/results/{:s}/session_item_results'\n",
        "\n",
        "\n",
        "with open('{:s}_responses.csv'.format(quiz_name), 'w', newline='') as out_file,\\\n",
        "     open('{:s}_log.txt'.format(quiz_name), 'a+') as log,\\\n",
        "     open('{:s}_submissions.csv'.format(quiz_name), 'r') as subs_file:\n",
        "\n",
        "    log.write('Reading quiz responses...\\n')\n",
        "\n",
        "    n_subs = sum(1 for row in subs_file) - 1\n",
        "    n_err = 0\n",
        "\n",
        "    subs_file.seek(0)\n",
        "    submissions = csv.DictReader(subs_file)\n",
        "\n",
        "    fout = csv.DictWriter(out_file,\n",
        "                          fieldnames=['user_id', 'assignment_id', 'submission_id', 'status',\n",
        "                                      'item_id', 'item_order', 'item_choice_id', 'correct'])\n",
        "    fout.writeheader()\n",
        "\n",
        "    for s_i, s in enumerate(submissions):\n",
        "\n",
        "        msg1 = 'User ID {:6d}, Submission ID {:7d} ...'.format(int(s['user_id']), int(s['submission_id']))\n",
        "        log.write(msg1)\n",
        "        if verbose:\n",
        "            print(msg1, end='')\n",
        "        else:\n",
        "            print('{:3.0%} ... {:d} errors'.format((s_i+1) / n_subs, n_err), end='\\r', flush=True)\n",
        "\n",
        "        # Get an authorization token for this grading session\n",
        "        r_auth = requests.get(url=auth_url.format(s['participant_session_id']),\n",
        "                              headers={'authorization': 'Bearer {:s}'.format(bearer_token)})\n",
        "\n",
        "        if r_auth.status_code == requests.codes.ok and 'token' in r_auth.json():\n",
        "            auth_token = r_auth.json()['token']\n",
        "        else:\n",
        "            n_err += 1\n",
        "            log.write('\\n *** FAILED TO GET AUTH TOKEN\\n')\n",
        "            log.write(' {:d}\\n'.format(r_auth.status_code))\n",
        "            log.write(' ' + r_auth.text + '\\n\\n')\n",
        "\n",
        "            if verbose:\n",
        "                print('\\n *** FAILED TO GET AUTH TOKEN')\n",
        "\n",
        "            fout.writerow({'user_id': s['user_id'], 'assignment_id': s['assignment_id'],\n",
        "                           'submission_id': s['submission_id'], 'status': 'FAILED'})\n",
        "            continue\n",
        "\n",
        "        # Get quiz submission details\n",
        "        r_sub = requests.get(url=quiz_session_url.format(s['quiz_session_id']),\n",
        "                             headers={'authorization': auth_token})\n",
        "        print(quiz_session_url.format(s['quiz_session_id']))\n",
        "\n",
        "        if r_sub.status_code == requests.codes.ok and 'authoritative_result' in r_sub.json():\n",
        "            result_id = r_sub.json()['authoritative_result']['id']\n",
        "        else:\n",
        "            n_err += 1\n",
        "            log.write('\\n *** FAILED TO GET SUBMISSION INFORMATION\\n')\n",
        "            log.write(' {:d}\\n'.format(r_sub.status_code))\n",
        "            log.write(' ' + r_sub.text + '\\n\\n')\n",
        "\n",
        "            if verbose:\n",
        "                print('\\n *** FAILED TO GET SUBMISSION INFORMATION')\n",
        "\n",
        "            fout.writerow({'user_id': s['user_id'], 'assignment_id': s['assignment_id'],\n",
        "                           'submission_id': s['submission_id'], 'status': 'FAILED'})\n",
        "            continue\n",
        "\n",
        "        # Get student responses\n",
        "        r_res = requests.get(url=results_url.format(s['quiz_session_id'], result_id),\n",
        "                             headers={'authorization': auth_token})\n",
        "\n",
        "        if r_res.status_code == requests.codes.ok:\n",
        "            # Loop over quiz items\n",
        "            #for i, item in enumerate(r_res.json()):\n",
        "            for item in r_res.json():\n",
        "              # Loop over item choices to determine which was chosen by student\n",
        "              selected_choice = ''\n",
        "\n",
        "              if type(item['scored_data']['value']) is dict:\n",
        "                # for multiple choices type\n",
        "                item_choice_selected = ''\n",
        "                for item_choice in item['scored_data']['value']:\n",
        "                    if item['scored_data']['value'][item_choice]['user_responded'] == True:\n",
        "                        item_choice_selected = item_choice\n",
        "\n",
        "                fout.writerow({'user_id': s['user_id'],\n",
        "                                'assignment_id': s['assignment_id'],\n",
        "                                'submission_id': s['submission_id'],\n",
        "                                'status': 'OK',\n",
        "                                'item_id': item['item_id'],\n",
        "                                'item_order': i+1,\n",
        "                                'item_choice_id': item_choice_selected,\n",
        "                                'correct': item['scored_data']['correct']})\n",
        "              else:\n",
        "                # for essay type\n",
        "                item_value = item['scored_data']['value']\n",
        "                item_scored_data = item['scored_data']\n",
        "\n",
        "                fout.writerow({'user_id': s['user_id'],\n",
        "                'assignment_id': s['assignment_id'],\n",
        "                'submission_id': s['submission_id'],\n",
        "                'status': 'OK',\n",
        "                'item_id': item['item_id'],\n",
        "                'item_order': i+1,\n",
        "                'item_choice_id': item_value,\n",
        "                'correct': \"N/A\"})\n",
        "            log.write(' OK\\n')\n",
        "\n",
        "            if verbose:\n",
        "                print(' OK')\n",
        "\n",
        "        else:\n",
        "            n_err += 1\n",
        "            log.write('\\n *** FAILED TO GET RESULT INFORMATION\\n')\n",
        "            log.write(' {:d}\\n'.format(r_res.status_code))\n",
        "            log.write(' ' + r_res.text + '\\n\\n')\n",
        "\n",
        "            if verbose:\n",
        "                print('\\n *** FAILED TO GET RESULT INFORMATION')\n",
        "\n",
        "            fout.writerow({'user_id': s['user_id'], 'assignment_id': s['assignment_id'],\n",
        "                           'submission_id': s['submission_id'], 'status': 'FAILED'})\n",
        "            continue\n",
        "\n",
        "print('Completed with {:d} errors.'.format(n_err))"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ZU534wdQoka1"
      },
      "source": [
        "## Get answer options\n",
        "\n",
        "Get all the possible answer options for each question. Creates a file `quiz_<assignment_id>_item_options.csv` with the following columns:\n",
        "\n",
        "assignment_id | item_id | item_name | item_text | item_choice_id | correct | item_choice_text\n",
        "\n",
        "**NOTE:** Check the output file `quiz_<newquiz_id>_item_options.csv` in the current folder"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "bIIBm244oka1",
        "scrolled": true
      },
      "outputs": [],
      "source": [
        "# Get first valid submission\n",
        "count = 0\n",
        "for s in quiz.get_submissions():\n",
        "    count +=1\n",
        "\n",
        "    if (not s.missing) and (s.submitted_at is not None) and hasattr(s, 'external_tool_url'):\n",
        "      print(s)\n",
        "      break\n",
        "\n",
        "try:\n",
        "    quiz_session = parse_qs(urlparse(s.external_tool_url).query)\n",
        "\n",
        "    # Request to determine the quiz_session_id (not needed anymore)\n",
        "    #url = 'https://umich.quiz-lti-iad-prod.instructure.com/api/participant_sessions/{:s}/grade'\\\n",
        "    #    .format(quiz_session['participant_session_id'][0])\n",
        "    url = 'https://umich.quiz-lti-iad-prod.instructure.com/api/participant_sessions/{:s}/grade'\\\n",
        "        .format(quiz_session['participant_session_id'][0])\n",
        "\n",
        "    r_auth = requests.get(url, headers={'authorization': 'Bearer {:s}'.format(bearer_token)})\n",
        "\n",
        "    if r_auth.status_code == requests.codes.ok and 'token' in r_auth.json():\n",
        "        auth_token = r_auth.json()['token']\n",
        "    else:\n",
        "        print('Failed to get auth token')\n",
        "        print(r_auth)\n",
        "        r_auth.raise_for_status()\n",
        "\n",
        "\n",
        "    session_items_url = 'https://umich.quiz-api-iad-prod.instructure.com/api/quiz_sessions/{:s}/session_items'\n",
        "    print(session_items_url)\n",
        "\n",
        "    r_session = requests.get(url=session_items_url.format(quiz_session['quiz_session_id'][0]),\n",
        "                            headers={'authorization': auth_token})\n",
        "    output_file_name = '{:s}_item_options.csv'.format(quiz_name)\n",
        "    print(f'output file name: {output_file_name}')\n",
        "    with open(output_file_name, 'w', newline='') as out_file:\n",
        "\n",
        "        fout = csv.DictWriter(out_file,\n",
        "                                fieldnames=['assignment_id', 'item_id', 'item_name', 'item_text',\n",
        "                                            'item_choice_id', 'correct', 'item_choice_text'])\n",
        "        fout.writeheader()\n",
        "\n",
        "        for item in r_session.json():\n",
        "            #print(json.dumps(item['item']['interaction_data'], indent=2))\n",
        "            #print(item['item']['interaction_data'])\n",
        "            if 'choices' in item['item']['interaction_data']:\n",
        "                # for multiple choice questions\n",
        "                for choice in item['item']['interaction_data']['choices']:\n",
        "                    fout.writerow({\n",
        "                        'assignment_id': newquiz_id,\n",
        "                        'item_id': item['item']['id'],\n",
        "                        'item_name': item['item']['title'],\n",
        "                        'item_text': item['item']['item_body'],\n",
        "                        'item_choice_id': choice['id'],\n",
        "                        'item_choice_text': choice['item_body'],\n",
        "                        'correct': choice['id'] == item['scoring_data']['value']\n",
        "                    })\n",
        "            else:\n",
        "                # for essay type\n",
        "                fout.writerow({\n",
        "                        'assignment_id': newquiz_id,\n",
        "                        'item_id': item['item']['id'],\n",
        "                        'item_name': item['item']['title'],\n",
        "                        'item_text': item['item']['item_body'],\n",
        "                        'item_choice_id': \"essay type\",\n",
        "                        'item_choice_text': \"N/A\",\n",
        "                        'correct': \"N/A\"\n",
        "                    })\n",
        "except Exception as e:\n",
        "  #print(e)\n",
        "  msg = 'ERROR processing submission for User ID: {:d}, submission ID: {:d}'\\\n",
        "      .format(s.user_id, s.id)\n",
        "  print(msg)\n",
        "  log.write(msg + '\\n')\n",
        "\n",
        "print('Done')"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Get student enrollment data"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "# get student enrollments\n",
        "output_file_name = 'course_{:d}_users.csv'.format(course.id)\n",
        "print(f'output file name: {output_file_name}')\n",
        "with open(output_file_name, 'w', newline='') as out_file:\n",
        "\n",
        "    fout = csv.DictWriter(out_file,\n",
        "                          fieldnames=['user_id', 'user_name'])\n",
        "    fout.writeheader()\n",
        "    users = course.get_users(params = {})\n",
        "    for user in users:\n",
        "      print(f'{user.name} {user.id}')\n",
        "      fout.writerow({\n",
        "          'user_id': user.id,\n",
        "          'user_name': user.name\n",
        "      })\n",
        "\n",
        "print('Done')"
      ]
    }
  ],
  "metadata": {
    "colab": {
      "provenance": []
    },
    "kernelspec": {
      "display_name": "Python 3",
      "language": "python",
      "name": "python3"
    },
    "language_info": {
      "codemirror_mode": {
        "name": "ipython",
        "version": 3
      },
      "file_extension": ".py",
      "mimetype": "text/x-python",
      "name": "python",
      "nbconvert_exporter": "python",
      "pygments_lexer": "ipython3",
      "version": "3.10.0"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}