diff --git a/contentcuration/contentcuration/tests/test_exportchannel.py b/contentcuration/contentcuration/tests/test_exportchannel.py index 8fd3b03e7a..c054a104c8 100644 --- a/contentcuration/contentcuration/tests/test_exportchannel.py +++ b/contentcuration/contentcuration/tests/test_exportchannel.py @@ -4,6 +4,7 @@ import string import tempfile import uuid +import zipfile from unittest import mock import pytest @@ -35,6 +36,7 @@ from .testdata import slideshow from .testdata import thumbnail_bytes from .testdata import tree +from .utils.qti.test_validation import _item_xml from .utils.qti.test_validation import VALID_CHOICE_ITEM from .utils.restricted_filesystemstorage import RestrictedFileSystemStorage from contentcuration import models as cc @@ -56,6 +58,24 @@ pytestmark = pytest.mark.django_db +# A schema-valid native QTI item whose single interaction (order) Perseus cannot +# express, so a node containing it must publish QTI only. +UNSUPPORTED_QTI_ITEM = _item_xml( + "item_unsupported", + "Unsupported Item", + '' + "" + "choice_0" + "choice_1" + "" + "", + '' + "Put these in order." + 'First' + 'Second' + "", +) + def description(): return "".join(random.sample(string.printable, 20)) @@ -282,6 +302,29 @@ def setUp(self): randomize=False, ) + # Native QTI item whose interaction Perseus cannot express -> QTI only + native_qti_unsupported_exercise = create_node( + { + "kind_id": "exercise", + "title": "Native QTI Unsupported Exercise", + "extra_fields": qti_extra_fields, + } + ) + native_qti_unsupported_exercise.complete = True + native_qti_unsupported_exercise.parent = current_exercise.parent + native_qti_unsupported_exercise.save() + cc.AssessmentItem.objects.create( + contentnode=native_qti_unsupported_exercise, + assessment_id=uuid.uuid4().hex, + type=exercises.QTI, + question="", + answers="[]", + hints="[]", + raw_data=UNSUPPORTED_QTI_ITEM, + order=1, + randomize=False, + ) + # Only legacy structured-field items, no perseus_question -> must now route to QTI (was Perseus) legacy_no_perseus_exercise = create_node( { @@ -779,8 +822,36 @@ def test_qti_exercise_generates_qti_archive(self): "QTI file should be a zip archive", ) - def test_native_qti_item_routes_to_qti_packaging(self): + def test_native_qti_choice_item_publishes_both_archives(self): + node = cc.ContentNode.objects.get(title="Native QTI Exercise") + self.assertTrue(node.files.filter(preset_id=format_presets.QTI_ZIP).exists()) + self.assertTrue(node.files.filter(preset_id=format_presets.EXERCISE).exists()) + + def test_native_qti_perseus_ids_match_assessment_metadata(self): + """The derived Perseus item JSON filenames must equal the ids recorded + in the published node's ``AssessmentMetaData.assessment_item_ids`` (the + QTI manifest ``K``-ids), so older Kolibri resolves the derived items.""" node = cc.ContentNode.objects.get(title="Native QTI Exercise") + exercise_file = node.files.get(preset_id=format_presets.EXERCISE) + with exercise_file.file_on_disk.open("rb") as file_handle: + item_stems = { + name[: -len(".json")] + for name in zipfile.ZipFile(file_handle).namelist() + if name.endswith(".json") and name != "exercise.json" + } + + published_node = kolibri_models.ContentNode.objects.get( + title="Native QTI Exercise" + ) + assessment_item_ids = set( + published_node.assessmentmetadata.first().assessment_item_ids + ) + + self.assertTrue(item_stems) + self.assertEqual(item_stems, assessment_item_ids) + + def test_native_qti_unsupported_interaction_publishes_qti_only(self): + node = cc.ContentNode.objects.get(title="Native QTI Unsupported Exercise") self.assertTrue(node.files.filter(preset_id=format_presets.QTI_ZIP).exists()) self.assertFalse(node.files.filter(preset_id=format_presets.EXERCISE).exists()) diff --git a/contentcuration/contentcuration/tests/utils/qti/test_html_to_markdown.py b/contentcuration/contentcuration/tests/utils/qti/test_html_to_markdown.py new file mode 100644 index 0000000000..bd2e4002e2 --- /dev/null +++ b/contentcuration/contentcuration/tests/utils/qti/test_html_to_markdown.py @@ -0,0 +1,144 @@ +from le_utils.constants import exercises + +from contentcuration.utils.assessment.markdown import render_markdown +from contentcuration.utils.assessment.qti.html_to_markdown import html_to_markdown +from contentcuration.utils.assessment.qti.ingest import ( + strip_content_storage_placeholder, +) +from contentcuration.utils.assessment.qti.validation import parse_qti_xml + + +def _elements(fragment): + doc = parse_qti_xml("
{}
".format(fragment).encode("utf-8")) + return list(doc.getroot()) + + +def _markdown_from_html(html): + return html_to_markdown(_elements(html)) + + +def test_plain_paragraph(): + assert html_to_markdown(_elements("

Hello world

")) == "Hello world" + + +def test_two_paragraphs_join_with_blank_line(): + result = html_to_markdown(_elements("

First

Second

")) + assert result == "First\n\nSecond" + + +def test_mathml_annotation_becomes_double_dollar_latex(): + fragment = ( + "

x" + 'x^2' + "

" + ) + assert "$$x^2$$" in html_to_markdown(_elements(fragment)) + + +def test_image_gets_content_storage_prefix(): + result = html_to_markdown(_elements('

d

')) + expected = "![d]({})".format(exercises.CONTENT_STORAGE_FORMAT.format("abc123.png")) + assert expected in result + + +def test_interaction_is_dropped_from_prompt(): + fragment = "

Fill in

" + assert html_to_markdown(_elements(fragment)) == "Fill in" + + +def test_empty_input_returns_empty_string(): + assert html_to_markdown(_elements("")) == "" + assert html_to_markdown(_elements("

")) == "" + + +def test_headings(): + assert _markdown_from_html("

One

") == "# One" + assert _markdown_from_html("

Three

") == "### Three" + + +def test_inline_emphasis_styles(): + assert _markdown_from_html("

b

") == "**b**" + assert _markdown_from_html("

i

") == "*i*" + assert _markdown_from_html("

gone

") == "~~gone~~" + assert _markdown_from_html("

x = 1

") == "`x = 1`" + + +def test_link(): + result = _markdown_from_html('

text

') + assert result == "[text](https://example.com)" + + +def test_unordered_list_with_nesting(): + html = "" + assert _markdown_from_html(html) == "- one\n - a\n - b\n- two" + + +def test_ordered_list(): + html = "
  1. first
  2. second
" + assert _markdown_from_html(html) == "1. first\n2. second" + + +def test_blockquote(): + assert _markdown_from_html("

quoted

") == "> quoted" + + +def test_fenced_code_block_with_language(): + html = '
x = 1\n
' + assert _markdown_from_html(html) == "```python\nx = 1\n```" + + +def test_horizontal_rule(): + assert _markdown_from_html("

a

b

") == "a\n\n---\n\nb" + + +def test_table(): + html = ( + "" + "
AB
12
" + ) + assert _markdown_from_html(html) == "| A | B |\n| --- | --- |\n| 1 | 2 |" + + +# A single canonical chunk exercising the full range of ``gfm-like`` formatting +# (plus ``$$…$$`` math and an image) that ``render_markdown`` accepts. The image +# carries the Perseus content-storage placeholder, exercising the one asymmetric +# transform: the forward ingest path strips the placeholder before building QTI +# HTML, and the img rule re-adds it on the way back. +CANONICAL_MARKDOWN = "\n\n".join( + [ + "# Heading level 1", + "## Heading level 2", + ( + "A paragraph with **bold**, *italic*, ~~strikethrough~~, `inline code`, " + "a [link](https://example.com), and math $$x^2 + y^2$$ inline." + ), + "![alt text]({})".format(exercises.CONTENT_STORAGE_FORMAT.format("abc123.png")), + "> A blockquote paragraph.", + "- First bullet\n- Second bullet\n - Nested bullet\n- Third bullet", + "1. First numbered\n2. Second numbered", + "```python\nx = 1\ny = 2\n```", + "| Column A | Column B |\n| --- | --- |\n| 1 | 2 |\n| 3 | 4 |", + "First line \nsecond line after a hard break.", + "---", + "$$a^2 + b^2 = c^2$$", + ] +) + + +def test_block_math_separates_from_following_block(): + """Display math is a top-level ```` sibling in the + forward HTML; it must keep the blank line before the next block rather than + gluing the following paragraph onto the math line.""" + md = "Given the equation:\n\n$$E = mc^2$$\n\nExplain what it means." + assert _markdown_from_html(render_markdown(md)) == md + + +def test_round_trips_losslessly_with_render_markdown(): + """markdown -> HTML -> markdown is identity through the real forward path. + + The forward transform mirrors the ingest pipeline: strip the content-storage + placeholder (``ingest.py``) before ``render_markdown`` builds the QTI HTML, + which is why images round-trip losslessly in production. + """ + html = render_markdown(strip_content_storage_placeholder(CANONICAL_MARKDOWN)) + assert _markdown_from_html(html) == CANONICAL_MARKDOWN diff --git a/contentcuration/contentcuration/tests/utils/qti/test_perseus_derive.py b/contentcuration/contentcuration/tests/utils/qti/test_perseus_derive.py new file mode 100644 index 0000000000..69dcca757d --- /dev/null +++ b/contentcuration/contentcuration/tests/utils/qti/test_perseus_derive.py @@ -0,0 +1,279 @@ +import json + +import pytest +from le_utils.constants import exercises + +from contentcuration.tests.utils.qti.test_validation import _item_xml +from contentcuration.utils.assessment.qti.perseus_derive import derive_perseus_item +from contentcuration.utils.assessment.qti.perseus_derive import is_perseus_derivable + + +class _Item: + """Minimal stand-in for a Django AssessmentItem.""" + + def __init__(self, raw_data, randomize=False, assessment_id="a" * 32): + self.raw_data = raw_data + self.randomize = randomize + self.assessment_id = assessment_id + + +def _choice_item( + cardinality, + correct_values, + choices, + prompt="Pick one.", + catalog="", + response_identifier="RESPONSE", + declaration_identifier=None, +): + correct = "".join("{}".format(v) for v in correct_values) + simple_choices = "".join( + '{}' + "".format(identifier, body) + for identifier, body in choices + ) + xml = _item_xml( + "item_choice", + "Choice Item", + '{}' + "".format( + declaration_identifier or response_identifier, cardinality, correct + ), + '' + "{}{}" + "".format(response_identifier, prompt, simple_choices), + ) + if catalog: + xml = xml.replace( + "{}".format(v) for v in correct_values) + return _item_xml( + "item_text", + "Text Item", + '{}' + "".format( + cardinality, correct + ), + '
{}

'.format(prompt), + ) + + +HINT_CATALOG = ( + '' + '' + "

First hint.

" + '' + "

Second hint.

" + "" +) + + +def test_single_choice_derivation(): + item = _Item( + _choice_item( + "single", + ["choice_0"], + [("choice_0", "Option A"), ("choice_1", "Option B")], + prompt="Select the correct answer.", + ) + ) + result = derive_perseus_item(item) + assert result.type == exercises.SINGLE_SELECTION + assert result.question == "Select the correct answer." + answers = json.loads(result.answers) + assert [a["answer"] for a in answers] == ["Option A", "Option B"] + assert [a["correct"] for a in answers] == [True, False] + assert [a["order"] for a in answers] == [0, 1] + + +def test_multiple_choice_derivation(): + item = _Item( + _choice_item( + "multiple", + ["choice_0", "choice_2"], + [ + ("choice_0", "A"), + ("choice_1", "B"), + ("choice_2", "C"), + ], + ) + ) + result = derive_perseus_item(item) + assert result.type == exercises.MULTIPLE_SELECTION + answers = json.loads(result.answers) + assert [a["correct"] for a in answers] == [True, False, True] + + +def test_text_input_derivation(): + item = _Item(_text_item("single", ["42"])) + result = derive_perseus_item(item) + assert result.type == exercises.INPUT_QUESTION + assert "What is 6 times 7?" in result.question + assert "qti-text-entry-interaction" not in result.question + answers = json.loads(result.answers) + assert answers == [{"answer": "42", "correct": True, "order": 0}] + + +def test_text_input_multiple_correct_values(): + item = _Item(_text_item("multiple", ["1", "2"])) + result = derive_perseus_item(item) + assert result.type == exercises.INPUT_QUESTION + answers = json.loads(result.answers) + assert [a["answer"] for a in answers] == ["1", "2"] + assert all(a["correct"] for a in answers) + + +def test_math_prompt_survives_into_question(): + prompt = ( + "

Solve x" + 'x^2' + "

" + ) + item = _Item(_text_item("single", ["4"], prompt=prompt)) + result = derive_perseus_item(item) + assert "$$x^2$$" in result.question + + +def test_hint_derivation(): + item = _Item( + _choice_item( + "single", + ["choice_0"], + [("choice_0", "A"), ("choice_1", "B")], + catalog=HINT_CATALOG, + ) + ) + result = derive_perseus_item(item) + hints = json.loads(result.hints) + assert [h["hint"] for h in hints] == ["First hint.", "Second hint."] + assert [h["order"] for h in hints] == [0, 1] + + +def test_derived_fields_carry_item_metadata(): + item = _Item( + _choice_item("single", ["choice_0"], [("choice_0", "A"), ("choice_1", "B")]), + randomize=True, + assessment_id="b" * 32, + ) + result = derive_perseus_item(item) + assert result.randomize is True + # The proxy's id is the QTI item's root identifier (not the Django + # assessment_id), so the derived Perseus item JSON filename matches the id + # the QTI manifest records in the node's assessment metadata. + assert result.assessment_id == "item_choice" + assert result.raw_data == "{}" + assert json.loads(result.hints) == [] + + +def test_custom_response_identifier_derives_correct_answers(): + """The response declaration is keyed off the interaction's own + response-identifier, not a hardcoded ``RESPONSE``.""" + item = _Item( + _choice_item( + "single", + ["choice_0"], + [("choice_0", "A"), ("choice_1", "B")], + response_identifier="RESPONSE_1", + ) + ) + result = derive_perseus_item(item) + assert result.type == exercises.SINGLE_SELECTION + answers = json.loads(result.answers) + assert [a["correct"] for a in answers] == [True, False] + + +def test_response_identifier_mismatch_degrades_to_qti_only(): + """An interaction whose response-identifier resolves to no declaration is + not derivable, so the node degrades to QTI-only rather than silently + deriving zero correct answers.""" + raw_data = _choice_item( + "single", + ["choice_0"], + [("choice_0", "A"), ("choice_1", "B")], + response_identifier="RESPONSE_A", + declaration_identifier="RESPONSE_B", + ) + assert derive_perseus_item(_Item(raw_data)) is None + assert is_perseus_derivable(raw_data) is False + + +ORDER_INTERACTION_BODY = ( + '' + "Order the steps." + 'First' + "" + 'Second' + "" + "" +) + +ORDER_ITEM = _item_xml( + "item_order", + "Order Item", + '' + "step1step2" + "", + ORDER_INTERACTION_BODY, +) + +TWO_INTERACTION_ITEM = _item_xml( + "item_two", + "Two Interaction Item", + 'choice_0' + "", + "
" + '' + 'A' + "" + '

' + "
", +) + +EXTENDED_TEXT_ITEM = _item_xml( + "item_extended", + "Extended Text Item", + 'whatever' + "", + 'Write an essay.' + "", +) + +MALFORMED_XML = "" + + +@pytest.mark.parametrize( + "raw_data", + [ + pytest.param(ORDER_ITEM, id="order_interaction"), + pytest.param(TWO_INTERACTION_ITEM, id="two_interactions"), + pytest.param(EXTENDED_TEXT_ITEM, id="extended_text"), + pytest.param(MALFORMED_XML, id="malformed_xml"), + ], +) +def test_not_derivable(raw_data): + assert derive_perseus_item(_Item(raw_data)) is None + assert is_perseus_derivable(raw_data) is False + + +def test_is_perseus_derivable_true_for_choice_and_text(): + choice = _choice_item( + "single", ["choice_0"], [("choice_0", "A"), ("choice_1", "B")] + ) + assert is_perseus_derivable(choice) is True + assert is_perseus_derivable(_text_item("single", ["42"])) is True diff --git a/contentcuration/contentcuration/tests/utils/test_exercise_creation.py b/contentcuration/contentcuration/tests/utils/test_exercise_creation.py index 97a00751b8..a3785b6507 100644 --- a/contentcuration/contentcuration/tests/utils/test_exercise_creation.py +++ b/contentcuration/contentcuration/tests/utils/test_exercise_creation.py @@ -26,6 +26,7 @@ from contentcuration.utils.assessment.perseus import PerseusExerciseGenerator from contentcuration.utils.assessment.qti.archive import hex_to_qti_id from contentcuration.utils.assessment.qti.archive import QTIExerciseGenerator +from contentcuration.utils.assessment.qti.validation import parse_qti_xml class TestPerseusExerciseCreation(StudioTestCase): @@ -1320,6 +1321,17 @@ def _create_qti_zip(self, exercise_data): ) return generator.create_exercise_archive() + def _create_perseus_zip(self, exercise_data): + """Create Perseus exercise zip using the generator""" + generator = PerseusExerciseGenerator( + self.exercise_node, + exercise_data, + self.channel.id, + "en-US", + user_id=self.user.id, + ) + return generator.create_exercise_archive() + def _validate_qti_zip_structure(self, exercise_file): """Helper to validate basic structure of the QTI Content Package""" # Use Django's storage backend to read the file @@ -1976,3 +1988,62 @@ def test_republish_replaces_stale_native_qti_archive(self): preset_id=format_presets.QTI_ZIP ).checksum self.assertNotEqual(first_checksum, second_checksum) + + def test_native_qti_perseus_derivation(self): + """A native QTI choice item is derived into a rendered Perseus item JSON.""" + catalog_info = ( + '' + '' + "

First hint.

" + "" + ) + raw_data = VALID_CHOICE_ITEM.replace( + "`` carries a bare +``.`` ``src``, which is re-prefixed with the Perseus content-storage +placeholder here (see ``test_html_to_markdown`` for the round-trip coverage). + +This module has no knowledge of QTI items — it operates on a sequence of sibling +lxml elements. +""" +import logging +from typing import Iterable + +from le_utils.constants import exercises +from lxml import etree + +logger = logging.getLogger(__name__) + +CONTENT_STORAGE_PREFIX = exercises.CONTENT_STORAGE_FORMAT.format("") + + +def _localname(el): + return etree.QName(el).localname + + +def _element_children(el): + """Child elements only — skips comments/processing instructions and text.""" + return [child for child in el if isinstance(child.tag, str)] + + +def _render_inline(el): + """Render an element's inline content: its text, children, and each child's tail.""" + parts = [] + if el.text: + parts.append(el.text) + for child in el: + parts.append(_render_element(child)) + if child.tail: + parts.append(child.tail) + return "".join(parts) + + +def _render_math(el): + # ``render_markdown`` emits display math as a top-level ```` sibling (not wrapped in a paragraph), so it must carry + # its own trailing block separator like the other block renderers; otherwise + # the following block is glued onto the math line. Inline math + # (``display="inline"``) stays inline within its paragraph. + suffix = "\n\n" if el.get("display") == "block" else "" + for annotation in el.iter(): + if ( + _localname(annotation) == "annotation" + and annotation.get("encoding") == "application/x-tex" + ): + return "$${}$${}".format(annotation.text or "", suffix) + logger.warning("MathML element without an application/x-tex annotation; dropping") + return "" + + +def _render_img(el): + src = el.get("src", "") + if not src: + return "" + return "![{}]({}{})".format(el.get("alt", ""), CONTENT_STORAGE_PREFIX, src) + + +def _render_heading(el): + level = int(_localname(el)[1]) + return "{} {}\n\n".format("#" * level, _render_inline(el).strip()) + + +def _render_code_block(el): + """``
…
`` → a fenced code block.""" + code = next((c for c in _element_children(el) if _localname(c) == "code"), None) + language = "" + text = "" + if code is not None: + text = code.text or "" + css_class = code.get("class", "") + if css_class.startswith("language-"): + language = css_class[len("language-") :] + if not text.endswith("\n"): + text += "\n" + return "```{}\n{}```\n\n".format(language, text) + + +def _render_blockquote(el): + inner = "".join(_render_element(child) for child in _element_children(el)).strip() + lines = [("> " + line) if line else ">" for line in inner.split("\n")] + return "\n".join(lines) + "\n\n" + + +def _list_lines(el): + """Flatten a ``
    ``/``
      `` into markdown lines, indenting nested lists.""" + ordered = _localname(el) == "ol" + lines = [] + items = (c for c in _element_children(el) if _localname(c) == "li") + for index, li in enumerate(items, start=1): + marker = "{}. ".format(index) if ordered else "- " + # Split the item's own inline content from any nested lists it contains. + inline_parts = [] + nested_lists = [] + if li.text: + inline_parts.append(li.text) + for child in li: + if _localname(child) in ("ul", "ol"): + nested_lists.append(child) + elif isinstance(child.tag, str): + inline_parts.append(_render_element(child)) + if child.tail: + inline_parts.append(child.tail) + lines.append(marker + "".join(inline_parts).strip()) + indent = " " * len(marker) + for nested_list in nested_lists: + lines.extend(indent + line for line in _list_lines(nested_list)) + return lines + + +def _render_list(el): + return "\n".join(_list_lines(el)) + "\n\n" + + +def _render_table(el): + rows = [] + for section in _element_children(el): + for tr in _element_children(section): + if _localname(tr) != "tr": + continue + rows.append( + [ + _render_inline(cell).strip() + for cell in _element_children(tr) + if _localname(cell) in ("th", "td") + ] + ) + if not rows: + return "" + width = max(len(row) for row in rows) + padded = [row + [""] * (width - len(row)) for row in rows] + out = ["| " + " | ".join(padded[0]) + " |"] + out.append("| " + " | ".join(["---"] * width) + " |") + for row in padded[1:]: + out.append("| " + " | ".join(row) + " |") + return "\n".join(out) + "\n\n" + + +# localname -> (prefix, suffix) for inline elements that wrap their content. +_INLINE_WRAPPERS = { + "strong": ("**", "**"), + "b": ("**", "**"), + "em": ("*", "*"), + "i": ("*", "*"), + "s": ("~~", "~~"), + "del": ("~~", "~~"), + "strike": ("~~", "~~"), +} + +# localname -> handler(el) -> markdown, for elements with dedicated rendering. +# ``render_markdown`` emits a hard break as ``
      \n``; the trailing newline +# lives in the element's tail, so two spaces here reconstruct the break. +_ELEMENT_RENDERERS = { + "math": _render_math, + "img": _render_img, + "a": lambda el: "[{}]({})".format(_render_inline(el), el.get("href", "")), + "code": lambda el: "`{}`".format(el.text or ""), + "br": lambda el: " ", + "hr": lambda el: "---\n\n", + "pre": _render_code_block, + "blockquote": _render_blockquote, + "ul": _render_list, + "ol": _render_list, + "table": _render_table, + "p": lambda el: "{}\n\n".format(_render_inline(el)), + "div": lambda el: "{}\n\n".format(_render_inline(el)), + **{"h{}".format(level): _render_heading for level in range(1, 7)}, +} + + +def _render_element(el): + if not isinstance(el.tag, str): + # Comment / processing instruction — nothing to render. + return "" + + localname = _localname(el) + + if localname.endswith("-interaction"): + return "" + if localname in _INLINE_WRAPPERS: + prefix, suffix = _INLINE_WRAPPERS[localname] + return "{}{}{}".format(prefix, _render_inline(el), suffix) + renderer = _ELEMENT_RENDERERS.get(localname) + if renderer is not None: + return renderer(el) + # Unknown element (e.g. inline HTML passthrough): unwrap, keeping its content. + return _render_inline(el) + + +def html_to_markdown(elements: Iterable[etree._Element]) -> str: + return "".join(_render_element(el) for el in elements).strip() diff --git a/contentcuration/contentcuration/utils/assessment/qti/perseus_derive.py b/contentcuration/contentcuration/utils/assessment/qti/perseus_derive.py new file mode 100644 index 0000000000..0f524f0420 --- /dev/null +++ b/contentcuration/contentcuration/utils/assessment/qti/perseus_derive.py @@ -0,0 +1,235 @@ +"""Native QTI item XML → structured legacy Perseus data. + +Reverse of the forward ``convert``/``ingest`` pipeline for the subset of QTI +interactions Perseus can express: single/multiple ``qti-choice-interaction`` +and inline ``qti-text-entry-interaction``. Everything else is *not* expressible +and derivation returns ``None`` so the node publishes QTI only. + +All parse/derive failures log + return ``None``/``False`` — a single malformed +item must never abort the channel publish. +""" +import json +import logging +from dataclasses import dataclass +from typing import List +from typing import Optional + +from le_utils.constants import exercises +from lxml import etree + +from contentcuration.utils.assessment.qti.catalog import KOLIBRI_HINT_SUPPORT +from contentcuration.utils.assessment.qti.html_to_markdown import html_to_markdown +from contentcuration.utils.assessment.qti.validation import parse_qti_xml + +logger = logging.getLogger(__name__) + + +@dataclass +class DerivedAssessmentItem: + """Legacy-shaped proxy consumed by ``PerseusExerciseGenerator``. + + Carries every field ``base.process_assessment_item`` and + ``perseus.create_assessment_item`` read off a Django ``AssessmentItem``. + """ + + type: str + question: str + answers: str + hints: str + assessment_id: str + randomize: bool = False + raw_data: str = "{}" + + +def _local(el) -> str: + return etree.QName(el).localname + + +def _first_descendant(root, localname): + for el in root.iter(): + if _local(el) == localname: + return el + return None + + +def _interaction_elements(item_body) -> List[etree._Element]: + return [el for el in item_body.iter() if _local(el).endswith("-interaction")] + + +def _response_declaration(root, identifier): + """The ``qti-response-declaration`` an interaction links to via its + ``response-identifier``. Keyed off the interaction's own identifier (not a + hardcoded ``RESPONSE``) so an item authored with a different identifier + resolves correctly, and a genuine mismatch resolves to ``None``. + """ + if identifier is None: + return None + for el in root.iter(): + if ( + _local(el) == "qti-response-declaration" + and el.get("identifier") == identifier + ): + return el + return None + + +def _derivable_interaction(root, item_body): + """Return ``(deriver, interaction, declaration)`` when the item is + Perseus-expressible, else ``None``. + + Requires exactly one interaction, that it be a ``qti-choice-interaction`` or + ``qti-text-entry-interaction``, *and* that its ``response-identifier`` + resolves to a response declaration; a mismatch yields ``None`` so the node + degrades to QTI-only rather than deriving zero correct answers. + """ + interactions = _interaction_elements(item_body) + if len(interactions) != 1: + return None + interaction = interactions[0] + deriver = _INTERACTION_DERIVERS.get(_local(interaction)) + if deriver is None: + return None + declaration = _response_declaration(root, interaction.get("response-identifier")) + if declaration is None: + return None + return deriver, interaction, declaration + + +def _correct_values(declaration) -> List[str]: + correct_responses = _children_by_localname(declaration, "qti-correct-response") + if not correct_responses: + return [] + return [ + value.text or "" + for value in _children_by_localname(correct_responses[0], "qti-value") + ] + + +def _children_by_localname(el, localname): + return [child for child in el if _local(child) == localname] + + +def _derive_hints(root) -> List[dict]: + hints = [] + order = 0 + for card in root.iter(): + if _local(card) != "qti-card" or card.get("support") != KOLIBRI_HINT_SUPPORT: + continue + html_content = _children_by_localname(card, "qti-html-content") + text = html_to_markdown(html_content[:1]) + if not text: + logger.warning("Skipping hint card with no derivable text") + continue + hints.append({"hint": text, "order": order}) + order += 1 + return hints + + +def _derive_choice(interaction, item_body, declaration): + prompt = _children_by_localname(interaction, "qti-prompt") + question = html_to_markdown(prompt[:1]) + correct = set(_correct_values(declaration)) + answers = [ + { + "answer": html_to_markdown([choice]), + "correct": choice.get("identifier") in correct, + "order": order, + } + for order, choice in enumerate( + _children_by_localname(interaction, "qti-simple-choice") + ) + ] + item_type = ( + exercises.MULTIPLE_SELECTION + if declaration.get("cardinality") == "multiple" + else exercises.SINGLE_SELECTION + ) + return item_type, question, answers + + +def _derive_text(interaction, item_body, declaration): + question = html_to_markdown([item_body]) + answers = [ + {"answer": value, "correct": True, "order": order} + for order, value in enumerate(_correct_values(declaration)) + ] + return exercises.INPUT_QUESTION, question, answers + + +# Interaction localname -> deriver(interaction, item_body, declaration). Defined +# below the derivers so the table can reference them directly. +_INTERACTION_DERIVERS = { + "qti-choice-interaction": _derive_choice, + "qti-text-entry-interaction": _derive_text, +} + + +def _parse(raw_data): + """Parse untrusted XML; return the item-body element or ``None``.""" + try: + root = parse_qti_xml(raw_data.encode("utf-8")).getroot() + except etree.XMLSyntaxError: + logger.warning("Unable to parse QTI item XML during Perseus derivation") + return None, None + item_body = _first_descendant(root, "qti-item-body") + return root, item_body + + +def is_perseus_derivable(raw_data: str) -> bool: + """True iff the item has exactly one Perseus-expressible interaction whose + response-identifier resolves to a response declaration.""" + root, item_body = _parse(raw_data) + if item_body is None: + return False + return _derivable_interaction(root, item_body) is not None + + +def derive_perseus_item(assessment_item) -> Optional[DerivedAssessmentItem]: + """Django ``AssessmentItem`` → ``DerivedAssessmentItem`` proxy, or ``None``. + + Returns ``None`` (with a warning) when the item is unparseable or its + interaction is not Perseus-expressible. + + The proxy's ``assessment_id`` is the QTI item's root ``identifier`` — the + same id the QTI archive records for the item in the manifest, and hence in + ``AssessmentMetaData.assessment_item_ids`` for these dual-published nodes — + so older Kolibri resolves the derived Perseus item JSON by that id. + """ + root, item_body = _parse(assessment_item.raw_data) + if item_body is None: + # A syntax error is already logged by _parse (root is None); this covers + # the "parsed, but no item body" case, which would otherwise be silent. + if root is not None: + logger.warning( + "QTI item %s has no item body; skipping derivation", + assessment_item.assessment_id, + ) + return None + + derivable = _derivable_interaction(root, item_body) + if derivable is None: + logger.warning( + "QTI item %s is not Perseus-expressible; skipping derivation", + assessment_item.assessment_id, + ) + return None + + identifier = root.get("identifier") + if not identifier: + logger.warning( + "QTI item %s is missing a root identifier; skipping derivation", + assessment_item.assessment_id, + ) + return None + + deriver, interaction, declaration = derivable + item_type, question, answers = deriver(interaction, item_body, declaration) + + return DerivedAssessmentItem( + type=item_type, + question=question, + answers=json.dumps(answers), + hints=json.dumps(_derive_hints(root)), + assessment_id=identifier, + randomize=assessment_item.randomize, + ) diff --git a/contentcuration/contentcuration/utils/publish.py b/contentcuration/contentcuration/utils/publish.py index c3db2ad4d9..eabc141ece 100644 --- a/contentcuration/contentcuration/utils/publish.py +++ b/contentcuration/contentcuration/utils/publish.py @@ -50,6 +50,7 @@ from contentcuration.utils.assessment.qti.imsmanifest import ( get_assessment_ids_from_manifest, ) +from contentcuration.utils.assessment.qti.perseus_derive import is_perseus_derivable from contentcuration.utils.cache import delete_public_channel_cache_keys from contentcuration.utils.files import create_thumbnail_from_base64 from contentcuration.utils.files import get_thumbnail_encoding @@ -249,6 +250,27 @@ def assign_license_to_contentcuration_nodes(channel, license): ] +# Legacy structured-field item types Perseus can render, sourced from the +# generator's own template map so the two never drift. ``perseus_question`` is +# excluded: it takes the dedicated Perseus branch, never the derivation gate. +PERSEUS_EXPRESSIBLE_LEGACY_TYPES = frozenset(PerseusExerciseGenerator.TEMPLATE_MAP) - { + exercises.PERSEUS_QUESTION +} + + +def _node_is_perseus_derivable(node): + """True iff the node has >=1 native QTI item and every item is Perseus-expressible.""" + has_native_qti = False + for item in node.assessment_items.all(): + if item.type == exercises.QTI: + has_native_qti = True + if not is_perseus_derivable(item.raw_data): + return False + elif item.type not in PERSEUS_EXPRESSIBLE_LEGACY_TYPES: + return False + return has_native_qti + + def has_assessments(node): """Check if a node should have its assessment items published. @@ -367,36 +389,40 @@ def recurse_nodes(self, node, inherited_fields): # noqa C901 t == exercises.PERSEUS_QUESTION for t in exercise_data["assessment_mapping"].values() ) - generator_class = ( - PerseusExerciseGenerator - if any_perseus_question - else QTIExerciseGenerator - ) - - # If this exercise previously had a file generated by a different - # generator, make sure we clean it up here. + if any_perseus_question: + generator_classes = [PerseusExerciseGenerator] + else: + generator_classes = [QTIExerciseGenerator] + # Also emit a Perseus archive when every item is a native QTI + # interaction Perseus can express, so older Kolibri renders it. + if _node_is_perseus_derivable(node): + generator_classes.append(PerseusExerciseGenerator) + + # If this exercise previously had files generated by generators no + # longer in use, make sure we clean them up here. + target_presets = {g.preset for g in generator_classes} stale_presets = { PerseusExerciseGenerator.preset, QTIExerciseGenerator.preset, - } - {generator_class.preset} - - # Remove archives produced by the previously-used generator + } - target_presets node.files.filter(preset_id__in=stale_presets).delete() - if ( - self.force_exercises - or node.changed - or not node.files.filter(preset_id=generator_class.preset).exists() - ): - - generator = generator_class( - node, - exercise_data, - self.channel_id, - self.default_language.lang_code, - user_id=self.user_id, - ) - generator.create_exercise_archive() + for generator_class in generator_classes: + if ( + self.force_exercises + or node.changed + or not node.files.filter( + preset_id=generator_class.preset + ).exists() + ): + generator = generator_class( + node, + exercise_data, + self.channel_id, + self.default_language.lang_code, + user_id=self.user_id, + ) + generator.create_exercise_archive() # Only create assessment metadata for exercises, not UNIT topics # UNIT topics store their assessment config in options/completion_criteria