"""Crowdmark-specific routines for printing examinations.
Routines for generating the API request to Crowdmark and handling the
responses.
"""
import datetime
import gzip
import json
from io import StringIO
from itertools import groupby
from operator import attrgetter
from subprocess import check_output
import urllib.request, urllib.error
import urllib3
from ll.xist.ns import html
from uw.web.wsgi.delegate import delegate_get_post
from uw.web.wsgi.form import use_form_param
from uw.web.wsgi.function import return_html
from uw.web.wsgi.status import HTTPNotFound, HTTPForbidden, HTTPBadRequest
from uw.local.config import read_config
from ...util.identity import use_remote_identity
api_key_config = read_config ()
page_sizes = {
'LTR': 'letter',
'LGL': 'legal',
}
[docs]def get_course_info (cursor, exam):
course_title = cursor.execute_required_value ("select format ('%%s (%%s)', admin_description, uw_term_format (term_id)) from teaching_admin, uw_term where (term_id, admin_id) = (%(term_id)s, %(admin_id)s)", term_id=exam.term_id, admin_id=exam.admin_id)
result = {'name': course_title}
learn_id = cursor.execute_optional_value ("select learn_id from learn_admin_term where (term_id, admin_id) = (%(term_id)s, %(admin_id)s)", term_id=exam.term_id, admin_id=exam.admin_id)
if learn_id is not None:
result['lms_id'] = learn_id
return {'type': 'course', 'attributes': result}
[docs]def get_user (cursor, person_id):
identity = cursor.execute_required_tuple ("select * from person_identity_complete where person_id = %(person_id)s", person_id=person_id)
return {
'name': ' '.join ([identity.givennames, identity.surname]),
'email': '%s@uwaterloo.ca' % identity.userid,
}
[docs]def get_staff_list (cursor, exam):
staff = cursor.execute_tuples ("select * from exam_exam_crowdmark_staff where (term_id, admin_id) = (%(term_id)s, %(admin_id)s) and person_id <> %(owner_person_id)s", term_id=exam.term_id, admin_id=exam.admin_id, owner_person_id=exam.exam_author_person_id)
result = []
for s in staff:
person = get_user (cursor, s.person_id)
person['role'] = s.person_role
result.append ({
'type': 'teammate',
'attributes': person,
})
return result
[docs]def get_candidates (cursor, exam):
instances = cursor.execute_tuples ("select exam_sequence - first_sequence as booklet_index, userid from exam_exam_student_sitting natural join exam_exam natural join person_identity_complete natural left join exam_exam_instance where exam_id = %(exam_id)s order by userid", exam_id=exam.exam_id)
instance_attributes = cursor.execute_tuples ("with t as (select * from exam_exam natural join division_division_exam_cover_display natural join exam_exam_student_sitting natural join division_division natural join division_student where exam_id = %(exam_id)s) select userid, division_description, division_value from t natural join person_identity_complete order by userid", exam_id=exam.exam_id)
instance_attributes = dict ((userid, list (rows)) for userid, rows in groupby (instance_attributes, attrgetter ('userid')))
result = []
for instance in instances:
instance_json = {
'type': 'enrollment',
'relationships': {
'student': {
'data': {
'type': 'user',
'attributes': {
'email': instance.userid + '@uwaterloo.ca',
},
},
},
},
}
attributes = {}
if instance.booklet_index is not None:
attributes['match_with_booklet_at_index'] = instance.booklet_index
if instance.userid in instance_attributes:
attributes['meta'] = dict ((r.division_description, r.division_value)
for r in instance_attributes[instance.userid])
if attributes:
instance_json['attributes'] = attributes
result.append (instance_json)
return result
[docs]def create_crowdmark_json (cursor, exam, now):
master_summary = cursor.execute_required_tuple ("select same_agg (size_code) as size_code, same_agg (master_pages_max) as master_pages_max from exam_exam_master_plus where (exam_id, type_code) = (%(exam_id)s, 'A') group by exam_id, type_code", exam_id=exam.exam_id)
booklet_count = cursor.execute_required_value ("select count(*) from exam_exam_instance where exam_id = %(exam_id)s", exam_id=exam.exam_id)
booklet_count = booklet_count * 11 // 10 + 20
candidate_json = get_candidates (cursor, exam)
result = {'data': {
'type': 'assessment',
'attributes': {
'title': ' '.join ([exam.term_time, str (exam.term_year), exam.full_title]),
'pages_count': master_summary.master_pages_max,
'booklets_count': max (booklet_count, len (candidate_json)),
'kind': 'administered',
'page_format': page_sizes[master_summary.size_code],
'is_double_sided': exam.master_duplex,
'omr_pages_count': exam.mc_pages,
},
'relationships': {
'course': {
'data': get_course_info (cursor, exam),
},
'owner': {
'data': {
'type': 'user',
'attributes': get_user (cursor, exam.exam_author_person_id),
},
},
'teammates': {
'data': get_staff_list (cursor, exam),
},
'enrollments': {
'data': candidate_json,
},
'webhooks': {
'data': [{
'type': 'webhook',
'attributes': {
'event': 'booklets_ready',
'url': 'https://odyssey.uwaterloo.ca/teaching/api/crowdmark-create/%s' % now.strftime ('%Y%m%d%H%M%S%f'),
'format': 'urlencoded',
'method': 'post',
'username': '_instruct_api_crowdmark',
'password': api_key_config.get ('crowdmark', 'response_password'),
},
}],
},
},
}}
return result
[docs]def issue_request (cursor, exam, request_json, now):
request = urllib.request.Request (api_key_config.get ('crowdmark', 'request_url'))
request.add_header ("Content-Type", "application/vnd.api+json")
request.data = (json.dumps (request_json).encode ())
request.add_header ("Authorization", "Token token=%s" % api_key_config.get ('crowdmark', 'request_token'))
try:
response = urllib.request.urlopen (request)
success = True
except urllib.error.HTTPError as x:
response = x
success = False
status = response.getcode ()
result = response.read ()
cursor.execute_none ("update crowdmark_create_request set response_status = %(status)s, response_body = %(response)s where request_time = %(time)s and response_status is null", status=status, response=result, time=now)
cursor.connection.commit ()
if success and status == 201:
# Success - record slug in DB
result_json = json.loads (result)
cursor.execute_none ("update exam_exam_crowdmark set crowdmark_exam_code = %(code)s where exam_id = %(exam_id)s", exam_id=exam.exam_id, code=result_json['data']['attributes']['slug'])
else:
print(result)
raise response
[docs]def crowdmark_exam (cursor, exam):
if not exam.is_crowdmark:
raise ValueError ('Assessment not configured for Crowdmark')
now = cursor.execute_required_value ("insert into crowdmark_create_request (exam_id) values (%(exam_id)s) returning request_time", exam_id=exam.exam_id)
cursor.connection.commit ()
request_json = create_crowdmark_json (cursor, exam, now)
cursor.execute_none ("update crowdmark_create_request set request_json = %(json)s where request_time = %(time)s and request_json is null", json=json.dumps (request_json), time=now)
cursor.connection.commit ()
issue_request (cursor, exam, request_json, now)
return request_json
[docs]def crowdmark_process_pdf (cursor, exam_id):
"""Pre-process the raw Crowdmark QR codes PDF for the specified examination.
:param cursor: DB cursor.
:param exam_id: The exam_id of the examination to process.
Obtain the raw QR codes PDF from
exam_exam_crowdmark.crowdmark_qrcodes_pdf_raw, process it through the
standard pre-processing, and store the result in
exam_exam_crowdmark.crowdmark_qrcodes_pdf.
At present there is no pre-processing - the saved version of the PDF is
identical to the raw version.
"""
pdf_raw, pdf_old = cursor.execute_required_tuple ("select crowdmark_qrcodes_pdf_raw, crowdmark_qrcodes_pdf from exam_exam_crowdmark where exam_id = %(exam_id)s for no key update nowait", exam_id=exam_id)
if pdf_raw is None:
raise SystemError (["QR codes raw PDF is NULL"])
if pdf_old is not None:
raise SystemError (["QR codes old PDF is not NULL"])
status, result = True, pdf_raw
if not status:
raise SystemError (result)
cursor.execute_none ("update exam_exam_crowdmark set crowdmark_qrcodes_pdf = %(pdf)s where exam_id = %(exam_id)s", exam_id=exam_id, pdf=result)
cursor.connection.commit ()
[docs]def crowdmark_save_raw_pdf (cursor, pdf_url, exam_id):
"""Download and save QR codes PDF.
:param cursor: DB cursor.
:param pdf_url: The URL from which to download the PDF.
:param exam_id: The exam_id of the examination to process.
Download the QR codes PDF from the specified URL and save it, then call
:func:`crowdmark_process_pdf` to embed fonts and saved the embedded version.
The following is the original code but on our server urllib2 doesn't work
with the server so we are using curl instead:
request = urllib2.Request (pdf_url)
request.add_header ("Accept-Encoding", 'gzip')
response = urllib2.urlopen (request)
status = response.getcode ()
result = response.read ()
if response.info ().get ('Content-Encoding') == 'gzip':
result = gzip.GzipFile (fileobj=StringIO (result)).read ()
"""
result = check_output (['curl', '--compressed', '-sS', pdf_url])
cursor.execute_none ("update exam_exam_crowdmark set crowdmark_qrcodes_pdf_raw = %(pdf)s where exam_id = %(exam_id)s", pdf=result, exam_id=exam_id)
cursor.connection.commit ()
crowdmark_process_pdf (cursor, exam_id)
@use_remote_identity
@use_form_param
@return_html
def crowdmark_post_handler (cursor, form, request_time, remote_identity):
"""Crowdmark callback URL handler.
Receives the QR codes PDF URL from Crowdmark, then retrieves the PDF and
stores it for later use by the printing process. The URL of the PDF is
expected to be provided in the data[booklets_url] form value. Which
examination is involved is indirecty inferred from the exact request time
encoded in the URL by looking up the request time in the
crowdmark_create_request table.
"""
if remote_identity.userid != '_instruct_api_crowdmark':
raise HTTPForbidden ()
try:
request_time = datetime.datetime.strptime (request_time, '%Y%m%d%H%M%S%f')
except ValueError:
request_time = None
pdf_url = form.optional_field_value ("data[booklets_url]")
response_time = cursor.execute_required_value ("insert into crowdmark_qr_response (request_time, pdf_url) values (%(request_time)s, %(pdf_url)s) returning response_time", request_time=request_time, pdf_url=pdf_url)
# Ensure response sent by Crowdmark is saved to database
cursor.connection.commit ()
if request_time is None:
raise HTTPNotFound ()
if pdf_url is None:
raise HTTPBadRequest ()
exam = cursor.execute_optional_tuple ("select ee.* from crowdmark_create_request join exam_exam ee using (exam_id) where request_time = %(request_time)s", request_time=request_time)
if exam is None:
raise HTTPNotFound ()
# Pre-process file to embed fonts
# Don't need to commit because crowdmark_save_raw_pdf() commits
crowdmark_save_raw_pdf (cursor, pdf_url, exam.exam_id)
# Quick hack to send Crowdmark jobs for printing right away
from .exam_print import submit_exam_print_job
job_id, output_dir = submit_exam_print_job (cursor, exam)
return 'Downloaded PDF', []
crowdmark_handler = delegate_get_post (None, crowdmark_post_handler)
[docs]def crowdmark_upload_pdf (cursor, exam, master_pdf):
"""Upload the given PDF to Crowdmark.
:param cursor: DB connection cursor.
:param exam: A database row representing an examination.
:param bytes master_pdf: The PDF to upload.
"""
url = "%s/%s/template" % (api_key_config.get ('crowdmark', 'request_url'), exam.crowdmark_exam_code)
fields = {
# None or '' filename breaks Crowdmark
'template[file]': (' ', master_pdf, 'application/pdf')
}
headers = {
'Authorization': "Token token=%s" % api_key_config.get ('crowdmark', 'request_token')
}
response = urllib3.PoolManager ().request ('PUT', url, fields=fields, headers=headers)
success = 200 <= response.status < 300
return success, response