"""YAML I/O, validation, and deterministic Canvas QTI 1.2 generation.
Supports single-answer multiple-choice questions with two or more choices.
Canvas import acceptance must still be checked in the institution's course.
"""
from __future__ import annotations
import csv
import hashlib
import html
import io
import json
import math
import re
from collections import Counter
from collections.abc import Hashable, Sequence
from copy import deepcopy
from pathlib import Path
from typing import cast
# Standard-library XML generation only; external parsing uses xml_io.
from xml.etree import ElementTree as ET # ruff: ignore[suspicious-xml-etree-import]
from zipfile import ZIP_DEFLATED, ZipFile, ZipInfo
import yaml
from .archive_io import read_member, validate_archive
from .assessments import question_assessments
from .models import Bank, BuildReport, QTIReport, Question
from .xml_io import parse_xml
from .yaml_data import YamlMapping, YamlValue, yaml_value
QTI = "http://www.imsglobal.org/xsd/ims_qtiasiv1p2"
IMS = "http://www.imsglobal.org/xsd/imscp_v1p1"
IMS_CC = "http://www.imsglobal.org/xsd/imsccv1p1/imscp_v1p1"
CANVAS_CC = "http://canvas.instructure.com/xsd/cccv1p0"
XSI = "http://www.w3.org/2001/XMLSchema-instance"
CANVAS_ASSESSMENT_TYPE = "imsqti_xmlv1p2/imscc_xmlv1p1/assessment"
ET.register_namespace("", QTI)
ET.register_namespace("xsi", XSI)
SAFE_ID = re.compile(r"^[A-Za-z][A-Za-z0-9_.-]*$")
DIFFICULTIES = {"Basic", "Moderate", "Challenging"}
MIN_CHOICES = 2
class _BankLoader(yaml.SafeLoader):
"""Load safe YAML while rejecting ambiguous mapping definitions.
Parameters
----------
stream
YAML text, bytes, or a readable stream accepted by SafeLoader.
Notes
-----
Inherits SafeLoader's supported constructors and overrides mapping
construction to reject duplicate keys, merge keys, and unhashable keys.
This loader does not validate the question-bank schema. load_yaml handles
loader cleanup and converts YAML parser errors to ValueError.
"""
def construct_mapping(
self, node: yaml.Node, deep: bool = False
) -> dict[Hashable, object]:
"""Construct a YAML mapping after checking keys for ambiguity.
Parameters
----------
node
Mapping node supplied by the YAML constructor.
deep
Whether to request deep construction of child values. Defaults to
False and is forwarded to the underlying constructor.
Returns
-------
mapping
Constructed dictionary, after key checks and SafeLoader conversion.
Raises
------
yaml.constructor.ConstructorError
If node is not a mapping, a merge key is present, a key is
unhashable, or a key compares equal to one already seen.
yaml.YAMLError
If constructing a key or value fails.
Notes
-----
Keys are checked before values can overwrite earlier entries.
Diagnostics include YAML source marks. Scalar-only normalization of
the resulting data is performed separately by yaml_value.
""" # noqa: DOC503
if not isinstance(node, yaml.MappingNode):
raise yaml.constructor.ConstructorError(
"while loading a bank",
node.start_mark,
"expected a mapping node",
node.start_mark,
)
seen: set[object] = set()
for key_node, _ in node.value:
if key_node.tag == "tag:yaml.org,2002:merge":
raise yaml.constructor.ConstructorError(
"while loading a bank",
node.start_mark,
"YAML merge keys (<<) are not supported; write fields "
"explicitly",
key_node.start_mark,
)
# PyYAML's stubs omit the node parameter type; key_node is a Node.
key: object = self.construct_object( # pyright: ignore[reportUnknownMemberType]
key_node, deep=deep
)
try:
duplicate = key in seen
seen.add(key)
except TypeError as exc:
raise yaml.constructor.ConstructorError(
"while loading a bank",
node.start_mark,
"mapping keys must be scalar values",
key_node.start_mark,
) from exc
if duplicate:
raise yaml.constructor.ConstructorError(
"while loading a bank",
node.start_mark,
f"duplicate key {key!r}",
key_node.start_mark,
)
return super().construct_mapping(node, deep=deep)
[docs]
def load_yaml(path: str | Path) -> YamlMapping:
"""Load one UTF-8 YAML mapping without validating its bank schema.
Parameters
----------
path
Source file containing exactly one YAML document.
Returns
-------
data
Normalized mapping, including supported optional authoring metadata.
Use parse_bank to validate it or edit it as an incomplete draft.
Raises
------
ValueError
If YAML is malformed, uses unsupported constructs, has duplicate or
merge keys, contains cycles, or does not produce a top-level mapping.
Parser errors include the source path and available source marks.
OSError
If the source file cannot be read.
UnicodeError
If the source file cannot be decoded as UTF-8.
Notes
-----
Uses a SafeLoader subclass and copies supported containers through
yaml_value. Shared containers become independent copies. The source
file is not changed, and legacy eligibility is not migrated here.
""" # noqa: DOC503
path = Path(path)
try:
# _BankLoader subclasses SafeLoader and adds stricter key checks.
# Ruff does not recognize custom SafeLoader subclasses as safe.
obj: object = yaml.load(
path.read_text(encoding="utf-8"),
Loader=_BankLoader, # ruff: ignore[unsafe-yaml-load]
)
except yaml.YAMLError as exc:
raise ValueError(f"Invalid YAML in {path}: {exc}") from exc
data = yaml_value(obj)
if not isinstance(data, dict):
# Invalid file contents use the CLI's existing ValueError contract.
raise ValueError( # ruff: ignore[type-check-without-type-error]
f"Bank source must be a YAML mapping: {path}"
)
return data
[docs]
def load_bank(path: str | Path) -> Bank:
"""Load a UTF-8 YAML file and construct a validated bank.
Parameters
----------
path
Source question-bank YAML file.
Returns
-------
bank
Validated bank with authoring metadata retained and legacy assessment
eligibility normalized. Omitted format remains omitted.
Raises
------
ValueError
If loading or schema validation fails. Schema errors are accumulated
in the validation diagnostic.
OSError
If the source file cannot be read.
UnicodeError
If the source file cannot be decoded as UTF-8.
Notes
-----
The file is not modified. Use load_yaml when loading incomplete drafts
that do not yet satisfy the bank schema.
""" # noqa: DOC502
return parse_bank(load_yaml(path))
[docs]
def validate(bank: object) -> list[str]:
"""Collect schema diagnostics for external Python data.
Parameters
----------
bank
Candidate bank data, including values not yet normalized as YAML.
Returns
-------
errors
Empty list for a valid bank; otherwise ordered diagnostic strings.
Normalization failures and non-mapping inputs return a single error.
Notes
-----
Normalizes supported YAML values before checking the schema. Expected
input-validation failures are returned rather than raised. Input data
is unchanged; successful validation does not replace it with a
normalized Bank or certify Canvas rendering.
"""
try:
data = yaml_value(bank)
except ValueError as exc:
return [str(exc)]
if not isinstance(data, dict):
return ["Bank source must be a YAML mapping."]
_, errors = _parse_bank(data)
return errors
def _parse_bank(data: YamlMapping) -> tuple[Bank | None, list[str]]:
"""Collect bank and question diagnostics before constructing a bank.
Parameters
----------
data
Normalized YAML mapping to check against the bank schema.
Returns
-------
bank
Constructed bank when every check succeeds; otherwise None.
errors
Ordered bank-field and question diagnostics, empty on success.
Notes
-----
A missing, empty, or non-list questions field prevents question
validation. Otherwise all question entries are checked. Input data
is not modified, but this helper does not detach all nested metadata;
parse_bank supplies the deep-copy boundary.
"""
errors = _validate_bank_fields(data)
items = data.get("questions")
if not isinstance(items, list) or not items:
return None, [*errors, "questions must be a nonempty list"]
questions, question_errors = _parse_questions(items)
errors.extend(question_errors)
if errors:
return None, errors
return _construct_bank(data, questions), []
def _validate_bank_fields(data: YamlMapping) -> list[str]:
"""Check top-level bank metadata without modifying the input.
Parameters
----------
data
Normalized mapping whose keys and bank metadata are checked.
Returns
-------
errors
Diagnostics for non-string field names, a version other than integer
1, missing or blank course/bank_id/title, invalid bank ID syntax, and
a nonboolean choice-shuffling default.
Notes
-----
Questions are validated separately. Multiple diagnostics can describe
one field, and additional string-keyed authoring metadata is allowed.
"""
errors: list[str] = []
if any(not isinstance(key, str) for key in data):
errors.append("Bank field names must be strings")
version = data.get("schema_version")
if type(version) is not int or version != 1:
errors.append("schema_version must be integer 1")
for field in ("course", "bank_id", "title"):
text = data.get(field)
if not isinstance(text, str) or not text.strip():
errors.append(f"{field} must be a nonempty string")
bank_id = data.get("bank_id")
if not isinstance(bank_id, str) or not SAFE_ID.fullmatch(bank_id):
errors.append("bank_id must match [A-Za-z][A-Za-z0-9_.-]*")
if (
"default_shuffle_choices" in data
and type(data["default_shuffle_choices"]) is not bool
):
errors.append("default_shuffle_choices must be a boolean")
return errors
def _parse_questions(
items: list[YamlValue],
) -> tuple[list[Question], list[str]]:
"""Validate question entries while tracking bank-wide duplicates.
Parameters
----------
items
Question entries in source order, not yet checked as mappings.
Returns
-------
questions
Constructed questions whose own checks passed, in source order.
errors
All entry diagnostics in source order. Question references include
the one-based position and supplied ID when available.
Notes
-----
Checks run in identity, fields, choices, then duplicate-stem order.
Valid identifiers and string stems are tracked even when other fields
are invalid. Callers must reject the whole bank if errors is nonempty.
The input list and mappings are not modified.
"""
errors: list[str] = []
ids: set[str] = set()
norm_prompts: set[str] = set()
questions: list[Question] = []
for no, entry in enumerate(items, 1):
supplied_id = (
str(entry.get("id", "?")) if isinstance(entry, dict) else "?"
)
ref = f"question #{no} ({supplied_id})"
if not isinstance(entry, dict):
errors.append(f"{ref}: entry must be a mapping")
continue
question_errors = _validate_question_identity(entry, ref, ids)
question_errors.extend(_validate_question_fields(entry, ref))
question_errors.extend(_validate_choices(entry, ref))
question_errors.extend(_check_stem_uniqueness(entry, ref, norm_prompts))
if not question_errors:
questions.append(_construct_question(entry))
errors.extend(question_errors)
return questions, errors
def _validate_question_identity(
q: YamlMapping, ref: str, ids: set[str]
) -> list[str]:
"""Check question keys and identifiers, recording valid IDs.
Parameters
----------
q
Question mapping to inspect without modification.
ref
Question reference prefixed to each diagnostic.
ids
Previously encountered valid IDs. A new valid ID is added in place.
Returns
-------
errors
Diagnostics for non-string keys, invalid IDs, or repeated IDs.
Notes
-----
ID syntax must match SAFE_ID. IDs are recorded independently of other
question fields so later duplicates remain detectable.
"""
errors: list[str] = []
if any(not isinstance(key, str) for key in q):
errors.append(f"{ref}: question field names must be strings")
qid = q.get("id")
if not isinstance(qid, str) or not SAFE_ID.fullmatch(qid):
errors.append(f"{ref}: invalid id")
elif qid in ids:
errors.append(f"{ref}: duplicate id {qid}")
else:
ids.add(qid)
return errors
def _validate_question_fields(q: YamlMapping, ref: str) -> list[str]:
"""Check question content, eligibility, and grading fields.
Parameters
----------
q
Question mapping to inspect without modification.
ref
Question reference prefixed to each diagnostic.
Returns
-------
errors
Diagnostics for required text, difficulty, assessment eligibility,
positive finite points/minutes, optional rendering format, and an
optional boolean choice-shuffling override.
Notes
-----
Identity, choices, and cross-question uniqueness are checked elsewhere.
Eligibility errors are collected from question_assessments. Boolean
numeric values are rejected, and omitted format is treated as plain.
"""
errors: list[str] = []
for key in ("title", "stem", "topic", "area", "explanation"):
text = q.get(key)
if not isinstance(text, str) or not text.strip():
errors.append(f"{ref}: {key} must be nonempty text")
difficulty = q.get("difficulty")
if not isinstance(difficulty, str) or difficulty not in DIFFICULTIES:
errors.append(
f"{ref}: difficulty must be one of {sorted(DIFFICULTIES)}"
)
try:
question_assessments(q)
except ValueError as exc:
errors.append(f"{ref}: {exc}")
for field in ("points", "minutes"):
value = q.get(field)
if (
isinstance(value, bool)
or not isinstance(value, (int, float))
or value <= 0
or (isinstance(value, float) and not math.isfinite(value))
):
errors.append(f"{ref}: {field} must be a finite positive number")
fmt = q.get("format", "plain")
if not isinstance(fmt, str) or fmt not in {"plain", "html"}:
errors.append(f"{ref}: format must be plain or html")
if "shuffle_choices" in q and type(q["shuffle_choices"]) is not bool:
errors.append(f"{ref}: shuffle_choices must be a boolean")
return errors
def _validate_choices(q: YamlMapping, ref: str) -> list[str]:
"""Check choice labels, text, and correct-answer membership.
Parameters
----------
q
Question mapping containing choices and answer.
ref
Question reference prefixed to each diagnostic.
Returns
-------
errors
Ordered diagnostics for the choice mapping, labels, content,
duplicate text, and answer membership.
Notes
-----
If choices is not a mapping with at least two entries, only that
structural error is returned. Otherwise text must be nonblank and
unique after trimming outer whitespace, with case preserved. Labels
must match SAFE_ID and answer must select one label. Input is unchanged.
"""
errors: list[str] = []
choices = q.get("choices")
options = choices if isinstance(choices, dict) else {}
if len(options) < MIN_CHOICES:
errors.append(
f"{ref}: choices must be a mapping with at least two entries"
)
else:
labels = list(options)
if any(
not isinstance(label, str) or not SAFE_ID.fullmatch(label)
for label in labels
):
errors.append(f"{ref}: choice keys must be valid identifiers")
if any(
not isinstance(x, str) or not x.strip() for x in options.values()
):
errors.append(f"{ref}: choices must contain nonempty text")
texts = [v.strip() for v in options.values() if isinstance(v, str)]
if len(set(texts)) != len(texts):
errors.append(f"{ref}: duplicate answer texts")
answer = q.get("answer")
if not isinstance(answer, str) or answer not in options:
errors.append(f"{ref}: answer must select one of the choice keys")
return errors
def _check_stem_uniqueness(
q: YamlMapping, ref: str, norm_prompts: set[str]
) -> list[str]:
"""Record string stems and detect normalized duplicates.
Parameters
----------
q
Question mapping whose stem is inspected.
ref
Question reference prefixed to a duplicate diagnostic.
norm_prompts
Previously normalized stems. The current string stem is added
in place, even if other question checks have failed.
Returns
-------
errors
One diagnostic for a duplicate string stem, otherwise an empty list.
Notes
-----
Normalization lowercases text and collapses whitespace. Non-string
stems are ignored here; field validation handles their errors. The
question mapping and its original stem text are unchanged.
"""
errors: list[str] = []
stem = q.get("stem")
if isinstance(stem, str):
norm = " ".join(stem.lower().split())
if norm in norm_prompts:
errors.append(f"{ref}: duplicate question stem")
norm_prompts.add(norm)
return errors
def _construct_bank(data: YamlMapping, questions: list[Question]) -> Bank:
"""Assemble checked bank fields and retain authoring extensions.
Parameters
----------
data
Mapping whose bank fields have already passed validation.
questions
Validated questions to install as the bank's questions list.
Returns
-------
bank
New dictionary with schema version 1, checked identity fields, the
supplied questions list, an optional choice-shuffling default, and
additional string-keyed metadata.
Raises
------
KeyError
If a required identity field is missing.
TypeError
If a required identity field is not text.
Notes
-----
This is a construction helper, not full validation. The questions list
and extension values are retained by reference. parse_bank provides
the input-copy boundary before construction.
""" # noqa: DOC502
bank = Bank(
# Schema validation above accepts only integer 1.
schema_version=1,
course=_text(data, "course"),
bank_id=_text(data, "bank_id"),
title=_text(data, "title"),
questions=questions,
)
# TypedDict describes known fields, not authoring extensions. Copy only
# unknown keys using the ordinary dict API, never replacing checked fields.
dict[str, object].update(
bank,
{
key: value
for key, value in data.items()
if isinstance(key, str) and key not in bank
},
)
return bank
def _text(data: YamlMapping, key: str) -> str:
"""Read a string field without coercion.
Parameters
----------
data
Mapping containing the field.
key
Field name to read; the key must exist.
Returns
-------
value
Original string value.
Raises
------
KeyError
If the key is absent.
TypeError
If the field value is not a string.
Notes
-----
This checks representation only, not whether text is blank.
""" # noqa: DOC503
value = data[key]
if not isinstance(value, str):
raise TypeError(f"{key} must be text")
return value
def _number(data: YamlMapping, key: str) -> float:
"""Read an integer or float without accepting booleans.
Parameters
----------
data
Mapping containing the field.
key
Numeric field name to read; the key must exist.
Returns
-------
value
Original integer or floating-point value, without conversion.
Raises
------
KeyError
If the key is absent.
TypeError
If the field value is a boolean or is neither an integer nor a float.
Notes
-----
Positivity and finiteness are checked before construction, not here.
""" # noqa: DOC503
value = data[key]
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise TypeError(f"{key} must be a number, not a boolean")
return value
def _construct_question(data: YamlMapping) -> Question:
"""Assemble a question from fields that have passed validation.
Parameters
----------
data
Question mapping already checked for schema constraints.
Returns
-------
question
New question dictionary with copied choices, normalized assessments,
optional format and choice-shuffling override when supplied, and
additional authoring metadata. The legacy exam key is omitted.
Raises
------
KeyError
If a required construction field is absent.
TypeError
If choices or required text/numeric fields have invalid types.
ValueError
If assessment eligibility cannot be normalized.
Notes
-----
Does not repeat all schema checks. Extension values are retained by
reference; parse_bank deep-copies input before calling this helper.
""" # noqa: DOC503
raw_choices = data["choices"]
if not isinstance(raw_choices, dict):
raise TypeError("choices must be a mapping")
choices: dict[str, str] = {}
for label, text in raw_choices.items():
if not isinstance(label, str) or not isinstance(text, str):
raise TypeError("choice labels and text must be strings")
choices[label] = text
question = Question(
id=_text(data, "id"),
title=_text(data, "title"),
area=_text(data, "area"),
topic=_text(data, "topic"),
difficulty=_text(data, "difficulty"),
assessments=question_assessments(data),
minutes=_number(data, "minutes"),
points=_number(data, "points"),
stem=_text(data, "stem"),
choices=choices,
answer=_text(data, "answer"),
explanation=_text(data, "explanation"),
)
if "format" in data:
question["format"] = _text(data, "format")
dict[str, object].update(
question,
{
key: value
for key, value in data.items()
if isinstance(key, str) and key not in question and key != "exam"
},
)
return question
[docs]
def parse_bank(data: YamlMapping) -> Bank:
"""Construct a validated bank from normalized YAML data.
Parameters
----------
data
Normalized YAML mapping to validate and copy.
Returns
-------
bank
Validated schema-version-1 bank independent of the input, retaining
additional authoring metadata and original question/choice order.
Raises
------
ValueError
If schema checks fail. The message begins with Validation failed
and includes the accumulated diagnostics.
Notes
-----
Deep-copies the input before validation and construction. Legacy exam
values become assessments; omitted assessments become an unrestricted
empty list. Omitted format remains omitted. Revalidate after editing
the result; type annotations alone cannot enforce schema constraints.
"""
bank, errors = _parse_bank(deepcopy(data))
if bank is None:
raise ValueError("Validation failed:\n" + "\n".join(errors))
return bank
[docs]
def require_valid_bank(bank: object) -> Bank:
"""Normalize external Python data and construct a validated bank.
Parameters
----------
bank
Candidate bank data to normalize through yaml_value.
Returns
-------
validated
Validated bank copied from the input, with supported metadata kept.
Raises
------
ValueError
If data contains unsupported YAML values or cycles, is not a mapping,
or fails bank schema validation.
Notes
-----
Compatibility entry point for arbitrary Python data. Prefer load_bank
for files or parse_bank for normalized YAML mappings. Input is unchanged.
"""
data = yaml_value(bank)
if not isinstance(data, dict):
# File contents use the public ValueError diagnostic contract.
raise ValueError("Bank source must be a YAML mapping.") # ruff: ignore[type-check-without-type-error]
return parse_bank(data)
def _element(
parent: ET.Element, name: str, text: object = None, **attrs: object
) -> ET.Element:
"""Append a QTI element, converting content and attributes to strings.
Parameters
----------
parent
Element to which the new child is appended.
name
Local tag name, qualified with the QTI namespace.
text
Optional content converted with str. None leaves element text unset.
**attrs
Attribute names and values; values are converted with str.
Returns
-------
node
New child element, already attached to parent.
Notes
-----
Mutates parent. It does not parse markup supplied as text.
"""
node = ET.SubElement(
parent, f"{{{QTI}}}{name}", {k: str(v) for k, v in attrs.items()}
)
if text is not None:
node.text = str(text)
return node
def _material(parent: ET.Element, text: str, html: bool = False) -> ET.Element:
"""Append a material/mattext pair with the requested content format.
Parameters
----------
parent
Element to receive a new material child.
text
Text assigned to the nested mattext element.
html
Whether texttype is text/html. False uses text/plain.
Returns
-------
node
Nested mattext element, not the enclosing material element.
Notes
-----
Mutates parent. HTML is stored as text for XML serialization; this
helper does not parse, sanitize, or render it.
"""
material = _element(parent, "material")
return _element(
material,
"mattext",
text,
texttype="text/html" if html else "text/plain",
)
[docs]
def question_to_xml(
q: Question, *, default_shuffle_choices: bool = False
) -> ET.Element:
"""Encode a validated question with single-answer scoring and feedback.
Parameters
----------
q
Validated question with ordered choices and a correct-answer label.
default_shuffle_choices
Bank-level shuffling behavior used when the question has no override.
Defaults to False.
Returns
-------
item
New QTI item element containing metadata, presentation, scoring,
and explanation feedback for correct and incorrect responses.
Notes
-----
Choice IDs combine the question ID and choice label. The render_choice
element receives ``shuffle="Yes"`` when the effective setting is true;
the attribute is omitted for the QTI default of no shuffling. Correct
answers score 100, while Canvas's general-feedback condition precedes the
correct-answer condition. Grading weight is stored separately in
points_possible metadata. Assessment eligibility is a JSON array. Omitted
format means plain text. Input is unchanged and is not revalidated;
minutes and additional authoring metadata are not exported.
"""
qid = q["id"]
item = ET.Element(f"{{{QTI}}}item", ident=qid, title=f"{qid} {q['title']}")
meta = _element(_element(item, "itemmetadata"), "qtimetadata")
answer_ids = [f"{qid}_{key}" for key in q["choices"]]
for key, value in [
("question_type", "multiple_choice_question"),
("points_possible", q["points"]),
("original_answer_ids", ",".join(answer_ids)),
("assessment_question_identifierref", qid),
("topic", q["topic"]),
("area", q["area"]),
("difficulty", q["difficulty"]),
("assessments", json.dumps(q.get("assessments", []))),
]:
field = _element(meta, "qtimetadatafield")
_element(field, "fieldlabel", key)
_element(field, "fieldentry", value)
presentation = _element(item, "presentation")
html = q.get("format", "plain") == "html"
_material(presentation, q["stem"], html)
resp = _element(
presentation, "response_lid", ident="response1", rcardinality="Single"
)
shuffle = q.get("shuffle_choices", default_shuffle_choices)
render = (
_element(resp, "render_choice", shuffle="Yes")
if shuffle
else _element(resp, "render_choice")
)
for key, choice in q["choices"].items():
_material(
_element(render, "response_label", ident=f"{qid}_{key}"),
choice,
html,
)
res = _element(item, "resprocessing")
_element(
_element(res, "outcomes"),
"decvar",
varname="SCORE",
vartype="Decimal",
minvalue="0",
maxvalue="100",
)
feedback = _element(res, "respcondition", **{"continue": "Yes"})
_element(_element(feedback, "conditionvar"), "other")
_element(
feedback,
"displayfeedback",
feedbacktype="Response",
linkrefid="general_fb",
)
correct = _element(res, "respcondition", **{"continue": "No"})
_element(
_element(correct, "conditionvar"),
"varequal",
f"{qid}_{q['answer']}",
respident="response1",
)
_element(correct, "setvar", "100", varname="SCORE", action="Set")
_material(
_element(
_element(item, "itemfeedback", ident="general_fb"), "flow_mat"
),
q["explanation"],
html,
)
return item
[docs]
def qti_xml(bank: Bank, questions: Sequence[Question] | None = None) -> bytes:
"""Serialize an assessment as UTF-8 XML in the QTI 1.2 namespace.
Parameters
----------
bank
Validated bank supplying assessment identity and title.
questions
Optional question sequence replacing the bank's full question list.
None uses all bank questions in their existing order.
Returns
-------
payload
XML bytes including the declaration, assessment, and root section.
Notes
-----
Inputs are not revalidated or modified. An empty supplied sequence
produces an empty section; create_qti rejects empty selections. Registers
QTI as ElementTree's default namespace before serialization.
"""
ET.register_namespace("", QTI)
questions = bank["questions"] if questions is None else questions
root = ET.Element(
f"{{{QTI}}}questestinterop",
{
f"{{{XSI}}}schemaLocation": (
f"{QTI} http://www.imsglobal.org/xsd/ims_qtiasiv1p2p1.xsd"
)
},
)
assessment = _element(
root, "assessment", ident=bank["bank_id"], title=bank["title"]
)
qti_metadata = _element(assessment, "qtimetadata")
metadata = _element(qti_metadata, "qtimetadatafield")
_element(metadata, "fieldlabel", "cc_maxattempts")
_element(metadata, "fieldentry", "1")
for label, entry in [
("qmd_assessmenttype", "Examination"),
("qmd_scoretype", "Percentage"),
]:
field = _element(qti_metadata, "qtimetadatafield")
_element(field, "fieldlabel", label)
_element(field, "fieldentry", entry)
section = _element(
assessment, "section", ident="root_section", title=bank["title"]
)
default_shuffle = bank.get("default_shuffle_choices", False)
for q in questions:
section.append(
question_to_xml(q, default_shuffle_choices=default_shuffle)
)
# ElementTree stubs return Any here; explicit UTF-8 guarantees bytes.
return cast(
bytes, ET.tostring(root, encoding="utf-8", xml_declaration=True)
)
[docs]
def manifest_xml(bank: Bank) -> bytes:
"""Serialize a Canvas Common Cartridge assessment manifest.
Parameters
----------
bank
Validated bank supplying the bank ID for manifest/resource IDs.
Returns
-------
payload
UTF-8 XML bytes with a declaration, one Canvas QTI assessment
resource, and its associated Canvas metadata resource.
Notes
-----
Does not write files or modify the bank. Registers the Common Cartridge
content-packaging namespace as ElementTree's default namespace.
"""
ET.register_namespace("", IMS_CC)
root = ET.Element(
f"{{{IMS_CC}}}manifest",
{
"identifier": bank["bank_id"] + "_manifest",
f"{{{XSI}}}schemaLocation": (
f"{IMS_CC} http://www.imsglobal.org/profile/cc/ccv1p1/"
"ccv1p1_imscp_v1p2_v1p0.xsd"
),
},
)
metadata = ET.SubElement(root, f"{{{IMS_CC}}}metadata")
ET.SubElement(metadata, f"{{{IMS_CC}}}schema").text = "IMS Common Cartridge"
ET.SubElement(metadata, f"{{{IMS_CC}}}schemaversion").text = "1.1.0"
ET.SubElement(root, f"{{{IMS_CC}}}organizations")
resources = ET.SubElement(root, f"{{{IMS_CC}}}resources")
assessment_path = f"{bank['bank_id']}/assessment_qti.xml"
metadata_path = f"{bank['bank_id']}/assessment_meta.xml"
res = ET.SubElement(
resources,
f"{{{IMS_CC}}}resource",
identifier=bank["bank_id"],
type=CANVAS_ASSESSMENT_TYPE,
)
ET.SubElement(res, f"{{{IMS_CC}}}file", href=assessment_path)
ET.SubElement(
res,
f"{{{IMS_CC}}}dependency",
identifierref=bank["bank_id"] + "_metadata",
)
meta_res = ET.SubElement(
resources,
f"{{{IMS_CC}}}resource",
identifier=bank["bank_id"] + "_metadata",
type="associatedcontent/imscc_xmlv1p1/learning-application-resource",
href=metadata_path,
)
ET.SubElement(meta_res, f"{{{IMS_CC}}}file", href=metadata_path)
# ElementTree stubs return Any here; explicit UTF-8 guarantees bytes.
return cast(
bytes, ET.tostring(root, encoding="utf-8", xml_declaration=True)
)
def _zip_write(zf: ZipFile, name: str, payload: bytes) -> None:
"""Write a ZIP member with fixed timestamp and permission metadata.
Parameters
----------
zf
Open ZIP archive accepting writes.
name
Member name to store inside the archive.
payload
Bytes to store using DEFLATE compression.
Returns
-------
None
Return after writing the member.
Raises
------
OSError
If writing to the underlying archive fails.
ValueError
If the archive is closed or not open for writing.
Notes
-----
Uses a 1980-01-01 timestamp and permission bits 0644 to avoid
source-file metadata affecting reproducibility. The archive remains open.
""" # noqa: DOC502
info = ZipInfo(name, date_time=(1980, 1, 1, 0, 0, 0))
info.compress_type = ZIP_DEFLATED
info.external_attr = 0o644 << 16
zf.writestr(info, payload)
[docs]
def create_qti(
bank: Bank,
dest: str | Path,
questions: Sequence[Question] | None = None,
) -> Path:
"""Write a QTI ZIP with reproducible member metadata.
Parameters
----------
bank
Validated bank supplying assessment identity and default questions.
dest
Destination ZIP path. Parent directories are created as needed,
and existing output is overwritten.
questions
Optional ordered sequence replacing the bank's questions. None uses
all bank questions; an empty sequence is rejected.
Returns
-------
path
Destination as a Path, without resolving it to an absolute path.
Raises
------
ValueError
If the selected question sequence is empty.
OSError
If directory creation or archive writing fails.
Notes
-----
Writes a Canvas Common Cartridge manifest, assessment, and associated
metadata document with fixed timestamps and permissions. Inputs must
already be validated; no schema validation or post-write verification
occurs here. The input bank is unchanged.
""" # noqa: DOC503
questions = bank["questions"] if questions is None else questions
if not questions:
raise ValueError("Cannot create a package without questions")
dest = Path(dest)
dest.parent.mkdir(parents=True, exist_ok=True)
directory = bank["bank_id"]
with ZipFile(dest, "w") as archive:
_zip_write(archive, "imsmanifest.xml", manifest_xml(bank))
_zip_write(
archive,
f"{directory}/assessment_qti.xml",
qti_xml(bank, questions),
)
_zip_write(
archive,
f"{directory}/assessment_meta.xml",
assessment_meta_xml(bank, questions),
)
return dest
def _manifest_resource_href(
resource: ET.Element, *, allow_file_href: bool
) -> str | None:
"""Return a resource's direct href or its first file href.
Parameters
----------
resource
Manifest resource element from an IMS or Common Cartridge manifest.
allow_file_href
Whether a child ``file/@href`` may replace an absent direct href.
Returns
-------
href
Resource href, or ``None`` when neither supported location supplies
one.
Notes
-----
Canvas Common Cartridge exports place the wrapper assessment path on a
child ``file`` element, while simple QTI packages use ``resource/@href``.
"""
href = resource.get("href")
if href or not allow_file_href:
return href
file_node = resource.find("./{*}file")
return None if file_node is None else file_node.get("href")
def _validated_manifest_href(
resource: ET.Element,
members: dict[str, ZipInfo],
description: str,
*,
allow_file_href: bool = False,
) -> str:
"""Return a safe archive member path declared by a resource.
Parameters
----------
resource
Manifest resource containing a direct or child-file href.
members
Validated archive members keyed by their exact ZIP names.
description
Resource description included in a rejected-path diagnostic.
allow_file_href
Whether a child ``file/@href`` may replace an absent direct href.
Returns
-------
href
Relative path that names an existing unique archive member.
Raises
------
ValueError
If the href is absent, absolute, parent-relative, or missing from the
validated archive.
"""
href = _manifest_resource_href(resource, allow_file_href=allow_file_href)
if (
not href
or href not in members
or href.startswith("/")
or ".." in Path(href).parts
):
raise ValueError(f"{description} href missing or invalid")
return href
def _resolve_qti_member(
archive: ZipFile,
members: dict[str, ZipInfo],
manifest_root: ET.Element,
) -> tuple[str, ET.Element]:
"""Resolve a direct QTI assessment or Canvas-referenced object bank.
Parameters
----------
archive
Open archive containing the already parsed manifest.
members
Validated archive members keyed by exact ZIP names.
manifest_root
Parsed IMS or Common Cartridge manifest document.
Returns
-------
assessment_file
Archive member containing the question items.
assessment_root
Parsed QTI root from that member.
Raises
------
ValueError
If the manifest does not contain exactly one QTI resource, its href
is invalid, or a Canvas source-bank reference is absent or ambiguous.
Notes
-----
Simple packages store items in the QTI resource itself. Canvas item-bank
exports instead store a wrapper assessment there; its sourcebank_ref
identifies an associated manifest resource containing the actual items.
"""
supported_manifest_roots = {
f"{{{IMS}}}manifest",
f"{{{IMS_CC}}}manifest",
}
if manifest_root.tag not in supported_manifest_roots:
raise ValueError("Incorrect IMS manifest namespace or root")
resources = manifest_root.findall(".//{*}resource")
qti_resources = [
resource
for resource in resources
if resource.get("type") in {"imsqti_xmlv1p2", CANVAS_ASSESSMENT_TYPE}
]
if len(qti_resources) != 1:
raise ValueError("Missing/incorrect QTI 1.2 assessment resource")
common_cartridge = manifest_root.tag == f"{{{IMS_CC}}}manifest"
qti_href = _validated_manifest_href(
qti_resources[0],
members,
"Manifest QTI",
allow_file_href=common_cartridge,
)
qti_root = parse_xml(read_member(archive, members[qti_href]))
source_refs = {
(node.text or "").strip()
for node in qti_root.findall(f".//{{{QTI}}}sourcebank_ref")
if (node.text or "").strip()
}
if not source_refs:
return qti_href, qti_root
if len(source_refs) != 1:
raise ValueError("Canvas QTI source bank reference is ambiguous")
source_ref = next(iter(source_refs))
matches = [
resource
for resource in resources
if resource.get("identifier") == source_ref
]
if len(matches) != 1:
raise ValueError("Canvas QTI source bank reference is missing")
bank_href = _validated_manifest_href(
matches[0],
members,
"Manifest source bank",
allow_file_href=common_cartridge,
)
bank_root = parse_xml(read_member(archive, members[bank_href]))
return bank_href, bank_root
[docs]
def verify_qti(path: str | Path) -> QTIReport:
"""Check supported archive structure, identifiers, and scoring shape.
Parameters
----------
path
QTI ZIP archive to read without extracting files.
Returns
-------
report
Question count, identifiers in document order, and the verified
assessment member path from the manifest.
Raises
------
ValueError
If archive resource limits or member-name uniqueness checks fail,
XML is unsafe or malformed, the manifest/resource reference is
invalid, the QTI root is incorrect, or item checks fail.
OSError
If the archive cannot be read.
zipfile.BadZipFile
If the input is not a readable ZIP archive.
Notes
-----
Applies the shared QTI archive limits before reading XML. Requires unique
nonempty item IDs, at least two response labels, one varequal referencing
a choice ID, one or two setvar elements per item, and any supplied
render_choice shuffle value to be Yes or No. Resolves either a direct QTI
resource or one Canvas Common Cartridge source-bank reference and rejects
absent, absolute, or parent-relative member paths. XML parsing forbids DTDs
and entities. This is not complete QTI validation: score values, supported
presentation content, bank schema constraints, and Canvas rendering require
other checks.
""" # noqa: DOC503
with ZipFile(path) as zf:
members = validate_archive(zf)
manifest = members.get("imsmanifest.xml")
if manifest is None:
raise ValueError("QTI package missing root imsmanifest.xml")
meta = parse_xml(read_member(zf, manifest))
href, root = _resolve_qti_member(zf, members, meta)
if root.tag != f"{{{QTI}}}questestinterop":
raise ValueError("Incorrect QTI 1.2 namespace or root")
items = root.findall(f".//{{{QTI}}}item")
if not items:
raise ValueError("No questions found")
ids: set[str] = set()
ordered_ids: list[str] = []
for item in items:
iid = item.get("ident")
if not iid:
raise ValueError("QTI item missing identifier")
if iid in ids:
raise ValueError(f"Duplicate QTI identifier {iid}")
ids.add(iid)
ordered_ids.append(iid)
renders = item.findall(f".//{{{QTI}}}render_choice")
if any(
render.get("shuffle", "No") not in {"Yes", "No"}
for render in renders
):
raise ValueError(f"Unsupported shuffle value for {iid}")
lids = item.findall(f".//{{{QTI}}}response_label")
if len(lids) < MIN_CHOICES:
raise ValueError(f"Incomplete response choices for {iid}")
val = item.findall(
f".//{{{QTI}}}respcondition/{{{QTI}}}conditionvar/{{{QTI}}}varequal"
)
if len(val) != 1 or val[0].text not in {x.get("ident") for x in lids}:
raise ValueError(f"Missing or invalid correct-answer key for {iid}")
score_rules = item.findall(f".//{{{QTI}}}respcondition/{{{QTI}}}setvar")
if len(score_rules) not in {1, MIN_CHOICES}:
raise ValueError(f"Unexpected scoring rules for {iid}")
return {
"count": len(items),
"ids": ordered_ids,
"assessment_file": href,
}
[docs]
def instructor_markdown(bank: Bank) -> str:
"""Render instructor review text including answers and explanations.
Parameters
----------
bank
Validated bank supplying identity, question content, and metadata.
Returns
-------
markdown
Markdown string with an instructor-only notice, ordered questions
and choices, metadata summaries, correct answers, and explanations.
Unrestricted assessment eligibility is displayed as Any.
Notes
-----
Authored text is inserted without escaping or HTML conversion. The
result contains answer keys. No files are written and input is unchanged.
"""
out = [
f"# {bank['course']} — {bank['title']}",
"",
"**INSTRUCTOR ONLY: contains answer keys.**",
"",
]
for q in bank["questions"]:
out.extend([
f"## {q['id']}: {q['title']}",
"",
(
f"Area: {q['area']} | Topic: {q['topic']} | Difficulty: "
f"{q['difficulty']} | Assessments: "
f"{', '.join(q.get('assessments', [])) or 'Any'} | Time: "
f"{q['minutes']} min"
),
"",
q["stem"],
"",
])
for k, v in q["choices"].items():
out.append(f"- **{k}.** {v}")
out.extend(["", f"**Answer: {q['answer']}.** {q['explanation']}", ""])
return "\n".join(out) + "\n"
[docs]
def question_catalog_csv(bank: Bank) -> str:
"""Render an ordered CSV catalog including instructor answers.
Parameters
----------
bank
Validated bank containing the questions to list.
Returns
-------
catalog
CSV text with a header and one row per question in bank order.
Columns are id, title, area, topic, difficulty, assessments, minutes,
points, answer, stem, and explanation.
Notes
-----
Assessment lists are encoded as JSON arrays. Choices, rendering format,
and additional authoring metadata are omitted. CSV quoting is handled by
csv.DictWriter. No files are written and the input is unchanged.
"""
fields = [
"id",
"title",
"area",
"topic",
"difficulty",
"assessments",
"minutes",
"points",
"answer",
"stem",
"explanation",
]
stream = io.StringIO(newline="")
writer = csv.DictWriter(stream, fieldnames=fields)
writer.writeheader()
for q in bank["questions"]:
row = {f: q.get(f, "") for f in fields}
row["assessments"] = json.dumps(q.get("assessments", []))
writer.writerow(row)
return stream.getvalue()
[docs]
def build(bank: object, output_dir: str | Path) -> tuple[Path, BuildReport]:
"""Validate a bank and write its QTI archive and instructor artifacts.
Parameters
----------
bank
Candidate Python bank data, normalized and validated before writing.
output_dir
Destination directory, created as needed. Matching output filenames
are overwritten.
Returns
-------
archive
Path to the generated {course}_{bank_id}_QTI12.zip archive.
report
Structural verification results, bank identity, difficulty counts,
and the SHA-256 digest of the generated archive bytes.
Raises
------
ValueError
If bank validation or generated-archive verification fails.
OSError
If creating directories, writing artifacts, or reading output fails.
UnicodeError
If text artifacts cannot be encoded as UTF-8.
csv.Error
If rendering the question catalog fails.
zipfile.BadZipFile
If the generated archive cannot be read during verification.
Notes
-----
Bank validation completes before output is created. Writes the QTI ZIP,
instructor_review.md, question_catalog.csv, and validation.json.
The bank is unchanged. Later I/O or verification failures may leave
partial output; generated files are not rolled back.
""" # noqa: DOC502
bank = require_valid_bank(bank)
output_dir = Path(output_dir)
output_dir.mkdir(parents=True, exist_ok=True)
name = f"{bank['course']}_{bank['bank_id']}_QTI12.zip"
path = create_qti(bank, output_dir / name)
checked = verify_qti(path)
(output_dir / "instructor_review.md").write_text(
instructor_markdown(bank), encoding="utf-8"
)
(output_dir / "question_catalog.csv").write_text(
question_catalog_csv(bank), encoding="utf-8"
)
report: BuildReport = {
**checked,
"course": bank["course"],
"bank_id": bank["bank_id"],
"difficulty": dict(Counter(q["difficulty"] for q in bank["questions"])),
"sha256": hashlib.sha256(path.read_bytes()).hexdigest(),
}
(output_dir / "validation.json").write_text(
json.dumps(report, indent=2) + "\n", encoding="utf-8"
)
return path, report