"""Strict migration of simple QTI 1.2 single-answer MC questions to YAML.
Unsupported QTI item structures are rejected rather than silently discarded.
"""
from __future__ import annotations
import csv
import json
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING
from zipfile import ZipFile
import yaml
from .archive_io import read_member, validate_archive
from .core import MIN_CHOICES, QTI, SAFE_ID, parse_bank, verify_qti
from .models import Bank
from .xml_io import parse_xml
from .yaml_data import YamlMapping, YamlValue, yaml_value
if TYPE_CHECKING:
from xml.etree.ElementTree import Element
@dataclass(frozen=True)
class _Presentation:
"""Hold extracted question text before bank schema validation.
Attributes
----------
stem
Text from the stem's mattext element, or an empty string when the
element has no text. Nonempty-content validation occurs later.
text_format
QTI MIME type: ``text/plain`` or ``text/html`` when constructed
by the presentation parser.
Notes
-----
Instances are frozen. Construction alone does not validate field values.
"""
stem: str
text_format: str
@dataclass(frozen=True)
class _Choices:
"""Hold ordered choices and translations of their source identifiers.
Attributes
----------
values
Mapping from generated labels A through Z to extracted choice text,
in source order. Values still require bank schema validation.
labels
Mapping from original QTI choice identifiers to generated labels,
used to translate the correct answer.
shuffle_choices
Whether the QTI render_choice element requests choice shuffling.
Notes
-----
Attribute assignment is frozen, but the contained dictionaries remain
mutable. Construction alone does not validate their contents.
"""
values: YamlMapping
labels: dict[str, str]
shuffle_choices: bool
[docs]
def import_qti(
path: str | Path,
course: str,
bank_id: str,
title: str,
metadata_csv: str | Path | None = None,
) -> Bank:
"""Import supported single-answer QTI 1.2 items as a validated bank.
Parameters
----------
path
Input ZIP containing imsmanifest.xml and its referenced QTI document.
course
Course identifier for the resulting bank.
bank_id
Bank identifier for the resulting bank; must satisfy the YAML schema.
title
Display title for the resulting bank.
metadata_csv
Optional UTF-8 CSV keyed by the case-sensitive ID column. Matching
rows provide authoring overrides. None means no CSV overrides.
Returns
-------
bank
Validated schema-version-1 bank with questions in document order.
Choices are relabeled A through Z in source order, the QTI shuffle
setting is retained per question, and legacy exam eligibility is
normalized to assessments.
Raises
------
ValueError
If archive resource limits or member-name uniqueness checks fail,
XML is unsafe or malformed, supported QTI structural checks fail,
metadata cannot be interpreted, or the resulting bank is invalid.
Final schema errors are prefixed with "Imported data needs attention".
OSError
If the archive or metadata CSV cannot be read.
UnicodeError
If the metadata CSV cannot be decoded as UTF-8.
csv.Error
If metadata CSV parsing fails.
zipfile.BadZipFile
If the input is not a readable ZIP archive.
Notes
-----
No files are written. Missing metadata uses field-specific defaults:
Unclassified for area/topic, Moderate for difficulty, unrestricted
eligibility, 2.5 minutes, 2 points, and unshuffled choices when QTI omits
the shuffle attribute. CSV overrides and QTI metadata are resolved before
schema validation; absent feedback receives an explicit placeholder.
""" # noqa: DOC503
report = verify_qti(path)
extras = _read_metadata_csv(metadata_csv)
with ZipFile(path) as archive:
members = validate_archive(archive)
assessment = members.get(report["assessment_file"])
if assessment is None:
raise ValueError(
"Verified QTI assessment member is no longer present"
)
root = parse_xml(read_member(archive, assessment))
questions: list[YamlValue] = [
_import_item(item, extras) for item in root.findall(f".//{{{QTI}}}item")
]
bank: YamlMapping = {
"schema_version": 1,
"course": course,
"bank_id": bank_id,
"title": title,
"questions": questions,
}
try:
return parse_bank(bank)
except ValueError as exc:
raise ValueError(f"Imported data needs attention:\n{exc}") from exc
def _read_metadata_csv(
metadata_csv: str | Path | None,
) -> dict[str, dict[str, str]]:
"""Read authoring overrides keyed by the CSV ID column.
Parameters
----------
metadata_csv
UTF-8 CSV path, optionally with a byte-order mark. None or an empty
path string returns an empty mapping without accessing a file.
Returns
-------
extras
Mapping from question ID to string-valued cells, retaining the
original column names and text. Rows without a nonempty ID are
skipped; later rows replace earlier rows with the same ID.
Raises
------
OSError
If the CSV cannot be opened or read.
UnicodeError
If its content cannot be decoded as UTF-8.
csv.Error
If CSV parsing fails.
ValueError
If the CSV lacks the required case-sensitive ID column or a row has
more cells than the header.
Notes
-----
Column names and IDs are case-sensitive and are not stripped.
Schema validation and metadata interpretation occur later.
""" # noqa: DOC503
extras: dict[str, dict[str, str]] = {}
if metadata_csv:
with open(metadata_csv, newline="", encoding="utf-8-sig") as inp:
reader = csv.DictReader(inp, strict=True)
if reader.fieldnames is None or "ID" not in reader.fieldnames:
raise ValueError(
"Metadata CSV must contain a case-sensitive ID column"
)
for row in reader:
if None in row:
raise ValueError(
f"Metadata CSV row {reader.line_num} has more cells "
"than its header"
)
if row.get("ID"):
extras[row["ID"]] = {
key: value
for key, value in row.items()
if isinstance(key, str) and isinstance(value, str)
}
return extras
def _read_item_metadata(item: Element) -> dict[str, str]:
"""Read metadata fields beneath one QTI item.
Parameters
----------
item
Parsed QTI item element to inspect without modification.
Returns
-------
metadata
Field labels mapped to entry text. Later fields replace earlier
values for the same label. Fields lacking a label or entry element
are skipped; present but empty elements retain empty strings.
"""
metadata: dict[str, str] = {}
for f in item.findall(f".//{{{QTI}}}qtimetadatafield"):
key = f.findtext(f"{{{QTI}}}fieldlabel")
value = f.findtext(f"{{{QTI}}}fieldentry")
if key is not None and value is not None:
metadata[key] = value
return metadata
def _parse_presentation(item: Element, ident: str) -> _Presentation:
"""Extract a single-answer presentation with a plain-text or HTML stem.
Parameters
----------
item
Parsed QTI item containing the presentation to inspect.
ident
Question identifier used in error messages.
Returns
-------
presentation
Extracted stem text and its QTI MIME type. Missing text becomes an
empty string; a missing texttype attribute defaults to text/plain.
Raises
------
ValueError
If there is not exactly one presentation response_lid with Single
cardinality, the stem mattext element is absent, or its texttype
is neither text/plain nor text/html.
Notes
-----
This checks supported presentation structure, not nonempty content or
HTML rendering. The XML item is not modified.
"""
response_lids = item.findall(
f"./{{{QTI}}}presentation/{{{QTI}}}response_lid"
)
if (
len(response_lids) != 1
or response_lids[0].get("rcardinality") != "Single"
):
raise ValueError(f"Unsupported response type in {ident}")
stemnode = item.find(
f"./{{{QTI}}}presentation/{{{QTI}}}material/{{{QTI}}}mattext"
)
if stemnode is None:
raise ValueError(f"Missing question stem in {ident}")
fmt = stemnode.get("texttype", "text/plain")
if fmt not in {"text/plain", "text/html"}:
raise ValueError(f"Unsupported mattext format {fmt} in {ident}")
return _Presentation(stemnode.text or "", fmt)
def _parse_choices(item: Element, ident: str, fmt: str) -> _Choices:
"""Relabel choices in source order and retain identifier translations.
Parameters
----------
item
Parsed QTI item containing presentation response labels.
ident
Question identifier used in error messages.
fmt
Expected QTI MIME type taken from the parsed stem. Each choice must
use the same type; omitted choice texttype defaults to text/plain.
Returns
-------
choices
Extracted choice text keyed by A through Z, a mapping from original
identifiers to those labels, and the effective QTI shuffle setting.
Missing text becomes an empty string, to be checked by subsequent
bank validation.
Raises
------
ValueError
If render_choice has an invalid shuffle value, there are fewer than
two or more than 26 choices, a choice lacks a mattext element or
identifier, or its text format differs from fmt.
Notes
-----
The source item is unchanged. Duplicate source identifiers are not
rejected here; later occurrences replace earlier translation entries.
Empty or duplicate choice text is checked by bank schema validation.
"""
render = item.find(
f"./{{{QTI}}}presentation/{{{QTI}}}response_lid/{{{QTI}}}render_choice"
)
if render is None:
raise ValueError(f"Unsupported number of choices in {ident}")
shuffle = render.get("shuffle", "No")
if shuffle not in {"Yes", "No"}:
raise ValueError(f"Unsupported shuffle value in {ident}: {shuffle}")
labels = render.findall(f"{{{QTI}}}response_label")
keys = "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
if len(labels) < MIN_CHOICES or len(labels) > len(keys):
raise ValueError(f"Unsupported number of choices in {ident}")
choices: YamlMapping = {}
mapping: dict[str, str] = {}
for n, lab in enumerate(labels):
mat = lab.find(f"./{{{QTI}}}material/{{{QTI}}}mattext")
if mat is None:
raise ValueError(f"Missing answer text in {ident}")
if mat.get("texttype", "text/plain") != fmt:
raise ValueError(f"Inconsistent response text formats in {ident}")
choices[keys[n]] = mat.text or ""
label_id = lab.get("ident")
if not label_id:
raise ValueError(f"Missing choice identifier in {ident}")
mapping[label_id] = keys[n]
return _Choices(choices, mapping, shuffle == "Yes")
def _parse_correct_answer(
item: Element, ident: str, mapping: dict[str, str]
) -> str:
"""Translate the choice ID from one supported full-credit condition.
Parameters
----------
item
Parsed QTI item containing response-processing conditions.
ident
Question identifier used in error messages.
mapping
Original choice identifiers mapped to generated YAML labels.
Returns
-------
answer
Generated label for the unique matching correct-answer condition.
Raises
------
ValueError
If there is not exactly one condition with a varequal text value and
a setvar text value of 100, or its choice ID is absent from mapping.
Notes
-----
Score text is stripped before comparison with the literal "100".
Other score representations, such as "100.0", do not match. This helper
extracts the supported scoring pattern; it is not a general QTI scorer.
"""
conditions = item.findall(
f"./{{{QTI}}}resprocessing/{{{QTI}}}respcondition"
)
matches: list[str] = []
for condition in conditions:
eq = condition.find(f"./{{{QTI}}}conditionvar/{{{QTI}}}varequal")
score = condition.find(f"./{{{QTI}}}setvar")
if (
eq is not None
and eq.text is not None
and score is not None
and (score.text or "").strip() == "100"
):
matches.append(eq.text)
if len(matches) != 1 or matches[0] not in mapping:
raise ValueError(f"Cannot determine unique answer for {ident}")
return mapping[matches[0]]
def _parse_explanation(item: Element, ident: str) -> str:
"""Extract feedback or describe its absence explicitly.
Parameters
----------
item
Parsed QTI item containing itemfeedback/flow_mat/material/mattext.
ident
Question identifier used in malformed-feedback diagnostics.
Returns
-------
explanation
Text from the first matching feedback element. If QTI contains no
feedback container, an explicit nonempty placeholder states that the
imported package supplied no explanation. An empty feedback text
value returns an empty string for bank validation to reject.
Raises
------
ValueError
If an itemfeedback container exists without the supported nested
mattext structure.
"""
feedback = item.find(
f"./{{{QTI}}}itemfeedback/{{{QTI}}}flow_mat/{{{QTI}}}"
f"material/{{{QTI}}}mattext"
)
if feedback is None:
if item.find(f"./{{{QTI}}}itemfeedback") is not None:
raise ValueError(f"Missing explanation in {ident}")
return "No explanation was provided in the imported QTI package."
return feedback.text or ""
def _normalize_question_id(ident: str) -> str:
"""Make a digit-leading QTI identifier valid for the YAML schema.
Parameters
----------
ident
Nonempty source identifier from the QTI item.
Returns
-------
question_id
Original identifier when already valid, or the identifier prefixed
with ``q_`` when its only schema conflict is a nonletter first
character.
Raises
------
ValueError
If characters after the optional prefix remain outside the YAML
identifier grammar.
Notes
-----
Canvas commonly emits hexadecimal item identifiers beginning with a
digit. Prefixing preserves every source character and is deterministic.
Final bank validation detects any collision with an existing identifier.
"""
if SAFE_ID.fullmatch(ident):
return ident
normalized = f"q_{ident}"
if SAFE_ID.fullmatch(normalized):
return normalized
raise ValueError(f"Unsupported QTI item identifier {ident!r}")
def _resolve_eligibility(
e: dict[str, str], metadata: dict[str, str]
) -> YamlMapping:
"""Resolve assessment eligibility using CSV overrides before QTI fields.
Parameters
----------
e
String-valued CSV cells for this question, or an empty mapping.
metadata
QTI field labels mapped to their entry text.
Returns
-------
eligibility
YAML mapping containing either assessments decoded from JSON or a
legacy exam string. Values remain unvalidated until bank parsing,
which also normalizes legacy eligibility.
Raises
------
ValueError
If selected JSON is invalid or cannot be represented as supported
YAML data. Malformed JSON raises the JSONDecodeError subclass.
Notes
-----
Precedence is nonempty CSV Assessments, nonempty CSV Suggested exam,
QTI assessments when present, then QTI suggested_exam or Either.
Empty CSV strings supply no override. A JSON empty list explicitly
requests unrestricted eligibility. Neither input mapping is modified.
""" # noqa: DOC502
if e.get("Assessments"):
return {"assessments": yaml_value(json.loads(e["Assessments"]))}
elif e.get("Suggested exam"):
return {"exam": e["Suggested exam"]}
elif "assessments" in metadata:
return {"assessments": yaml_value(json.loads(metadata["assessments"]))}
else:
return {"exam": metadata.get("suggested_exam") or "Either"}
def _resolve_authoring_metadata(
item: Element, ident: str, e: dict[str, str], metadata: dict[str, str]
) -> YamlMapping:
"""Resolve authoring values, eligibility, and field-specific defaults.
Parameters
----------
item
QTI item supplying its title attribute.
ident
Question identifier used as the fallback title and to remove a
leading identifier followed by a space from an existing title.
e
String-valued CSV cells for this question.
metadata
QTI metadata field labels and entry text.
Returns
-------
fields
Unvalidated YAML mapping with title, area, topic, difficulty,
eligibility, minutes, and points. Integral point values become
integers; other point values remain floating-point numbers.
Raises
------
ValueError
If numeric text cannot be converted or eligibility decoding fails.
Notes
-----
Area uses CSV Area, QTI area, QTI topic, then Unclassified. Topic uses
CSV Topic, QTI topic, then Unclassified. Difficulty uses CSV Difficulty,
QTI difficulty, then Moderate; Easy becomes Basic and High becomes
Challenging. Minutes use CSV Suggested minutes or 2.5; points use QTI
points_possible or 2. Empty strings trigger these fallbacks.
Eligibility follows _resolve_eligibility. Neither input mapping nor the
item is modified. Schema constraints, including positivity and finiteness,
are checked later.
""" # noqa: DOC502
difficulty = e.get("Difficulty") or metadata.get("difficulty") or "Moderate"
difficulty = {"Easy": "Basic", "High": "Challenging"}.get(
difficulty, difficulty
)
pts = float(metadata.get("points_possible") or 2)
minutes = float(e.get("Suggested minutes") or 2.5)
raw_title = item.get("title") or ident
if raw_title.startswith(ident + " "):
raw_title = raw_title[len(ident) + 1 :]
eligibility = _resolve_eligibility(e, metadata)
return {
"title": raw_title,
"area": e.get("Area")
or metadata.get("area")
or metadata.get("topic")
or "Unclassified",
"topic": e.get("Topic") or metadata.get("topic") or "Unclassified",
"difficulty": difficulty,
**eligibility,
"minutes": minutes,
"points": int(pts) if pts.is_integer() else pts,
}
def _import_item(
item: Element, extras: dict[str, dict[str, str]]
) -> YamlMapping:
"""Extract one question before final bank schema validation.
Parameters
----------
item
Parsed QTI item from an archive that has passed structural
verification.
extras
CSV overrides keyed by original question ID. A missing ID entry
supplies no overrides for that question.
Returns
-------
question
Unvalidated YAML mapping containing normalized identity, question
content, relabeled choices, effective shuffling, translated answer,
feedback, and resolved authoring fields. QTI text/plain and text/html
become plain and html respectively.
Raises
------
ValueError
If the item lacks an identifier, a supported content or scoring
check fails, or authoring metadata cannot be interpreted.
Notes
-----
Checks run in presentation, choices, scoring, then feedback order before
authoring metadata is resolved. XML and overrides are unchanged.
The result must pass parse_bank before it is treated as a valid Question.
"""
ident = item.get("ident")
if not ident:
raise ValueError("QTI item missing identifier")
metadata = _read_item_metadata(item)
presentation = _parse_presentation(item, ident)
choices = _parse_choices(item, ident, presentation.text_format)
answer = _parse_correct_answer(item, ident, choices.labels)
explanation = _parse_explanation(item, ident)
authoring = _resolve_authoring_metadata(
item, ident, extras.get(ident, {}), metadata
)
return {
"id": _normalize_question_id(ident),
**authoring,
"format": "html"
if presentation.text_format == "text/html"
else "plain",
"stem": presentation.stem,
"choices": choices.values,
"shuffle_choices": choices.shuffle_choices,
"answer": answer,
"explanation": explanation,
}
[docs]
def save_bank(bank: object, out: str | Path) -> Path:
"""Serialize YAML without schema validation and return the output path.
Parameters
----------
bank
Bank or draft data representable by yaml.safe_dump. Schema validation
is not performed; call parse_bank first when compliance is required.
out
Destination YAML path. Parent directories are created as needed,
and any existing file is overwritten.
Returns
-------
path
Destination as a Path, without resolving it to an absolute path.
Raises
------
OSError
If directory creation or writing fails.
UnicodeError
If output cannot be encoded as UTF-8.
yaml.YAMLError
If data cannot be represented as safe YAML.
Notes
-----
Output preserves mapping insertion order, Unicode text, and supported
optional metadata. YAML comments and original source formatting are not
retained. Input data is not modified.
""" # noqa: DOC502
out = Path(out)
out.parent.mkdir(parents=True, exist_ok=True)
out.write_text(
yaml.safe_dump(bank, allow_unicode=True, sort_keys=False, width=110),
encoding="utf-8",
)
return out