"""Process to import examination times.
Defines a parser for the examination time upload file generated by Cyon, and
a procedure to process the time and duration updates.
"""
from datetime import datetime, timedelta
from uw.stringtools import split_upload_text, parse_tabular_file
from uw.sql.wrap import ConnectionPool, open_psycopg2_db_service_connection
from uw.web.html.format import format_datetime
from uw.local import termtools
from ..webui.ui import format_duration
from .cursor import Cursor
[docs]def parse_date (value):
return datetime.strptime (value, '%m/%d/%Y').date ()
[docs]def parse_time (value):
return datetime.strptime (value, '%H:%M').time ()
[docs]def parse_duration (value):
value = int (value)
if value == 60:
return None
else:
return timedelta (seconds=value)
columns = [
('Host Key', 'exam_id', int),
('Exam Start Date', 'exam_date', parse_date),
('Exam Start Time', 'exam_time', parse_time),
('Duration', 'duration', parse_duration),
]
[docs]def parse_exam_times_upload (contents):
grid = split_upload_text (contents)
return parse_tabular_file (columns, grid, 1, 'line')
[docs]def process_exam_times_upload (cursor, term, rows):
"""Process Cyon output lines.
:param cursor: DB connection cursor.
:param term: Object representing a UW term.
:param list rows: The update requests parsed from the input file.
Each row is a namedtuple consisting of (exam_id, exam_date, exam_time,
duration) produced by parsing the Cyon output file. Each row should
correspond to a single examination, which will have its start time and
duration updated accordingly.
"""
warnings = []
for r in rows:
e = cursor.execute_optional_tuple ("select primary_start_time, exam_duration, exam_exam_full_title (exam_id) as exam_title from exam_exam where (term_id, exam_id) = (%(term_id)s, %(exam_id)s)", term_id=term.code (), exam_id=r.exam_id)
if e is None:
warnings.append ((r.line, 'No exam found with exam_id %d' % r.exam_id))
continue
if r.exam_date is None != r.exam_time is None:
warnings.append ((r.line, 'Only one of start date/time found'))
continue
if r.exam_date is not None:
start_time = datetime.combine (r.exam_date, r.exam_time)
if start_time != e.primary_start_time:
if e.primary_start_time is not None:
warnings.append ((r.line, 'Updated start time for exam_id %d (%s) from %s to %s' % (r.exam_id, e.exam_title, format_datetime (e.primary_start_time), format_datetime (start_time))))
cursor.execute_optional_value ("select exam_update_primary_sitting (exam_id, exam_require_exam_sitting (exam_id, schedule_admin_id, %(start_time)s)) from exam_exam where exam_id = %(exam_id)s", start_time=start_time, exam_id=r.exam_id)
if r.duration is not None:
if r.duration != e.exam_duration:
if e.exam_duration is not None:
warnings.append ((r.line, 'Updated duration for exam_id %d (%s) from %s to %s' % (r.exam_id, e.exam_title, format_duration (e.exam_duration), format_duration (r.duration))))
cursor.execute_none ("update exam_exam set exam_duration = %(exam_duration)s where exam_id = %(exam_id)s", exam_duration=r.duration, exam_id=r.exam_id)
return warnings