Conditional publishing of QTI archives instead of perseus.

Author: rtibbles

contentcuration/contentcuration/tests/test_exportchannel.py

@@ -15,6 +15,7 @@
 from kolibri_content.router import get_active_content_database
 from kolibri_content.router import set_active_content_database
 from le_utils.constants import exercises
+from le_utils.constants import format_presets
 from le_utils.constants.labels import accessibility_categories
 from le_utils.constants.labels import learning_activities
 from le_utils.constants.labels import levels
@@ -33,6 +34,7 @@
 from .testdata import tree
 from contentcuration import models as cc
 from contentcuration.models import CustomTaskMetadata
+from contentcuration.utils.assessment.qti.archive import hex_to_qti_id
 from contentcuration.utils.celery.tasks import generate_task_signature
 from contentcuration.utils.publish import ChannelIncompleteError
 from contentcuration.utils.publish import convert_channel_thumbnail
@@ -209,6 +211,48 @@ def setUp(self):
             ai.contentnode = legacy_exercise
             ai.save()
 
+        # Add an exercise with free response question to test QTI generation
+        qti_extra_fields = {
+            "options": {
+                "completion_criteria": {
+                    "model": "mastery",
+                    "threshold": {
+                        "m": 1,
+                        "n": 2,
+                        "mastery_model": exercises.M_OF_N,
+                    },
+                }
+            }
+        }
+        qti_exercise = create_node(
+            {
+                "kind_id": "exercise",
+                "title": "QTI Free Response Exercise",
+                "extra_fields": qti_extra_fields,
+            }
+        )
+        qti_exercise.complete = True
+        qti_exercise.parent = current_exercise.parent
+        qti_exercise.save()
+
+        # Create a free response assessment item
+        cc.AssessmentItem.objects.create(
+            contentnode=qti_exercise,
+            assessment_id=uuid.uuid4().hex,
+            type=exercises.FREE_RESPONSE,
+            question="What is the capital of France?",
+            answers=json.dumps([{"answer": "Paris", "correct": True}]),
+            hints=json.dumps([]),
+            raw_data="{}",
+            order=4,
+            randomize=False,
+        )
+
+        for ai in current_exercise.assessment_items.all()[:2]:
+            ai.id = None
+            ai.contentnode = qti_exercise
+            ai.save()
+
         first_topic = self.content_channel.main_tree.get_descendants().first()
 
         # Add a publishable topic to ensure it does not inherit but that its children do
@@ -558,6 +602,46 @@ def test_publish_no_modify_legacy_exercise_extra_fields(self):
             {"mastery_model": exercises.M_OF_N, "randomize": True, "m": 1, "n": 2},
         )
 
+    def test_qti_exercise_generates_qti_archive(self):
+        """Test that exercises with free response questions generate QTI archive files."""
+        qti_exercise = cc.ContentNode.objects.get(title="QTI Free Response Exercise")
+
+        # Check that a QTI archive file was created
+        qti_files = qti_exercise.files.filter(preset_id=format_presets.QTI_ZIP)
+        self.assertEqual(
+            qti_files.count(),
+            1,
+            "QTI exercise should have exactly one QTI archive file",
+        )
+
+        qti_file = qti_files.first()
+        self.assertIsNotNone(
+            qti_file.file_on_disk, "QTI file should have file_on_disk content"
+        )
+        self.assertTrue(
+            qti_file.original_filename.endswith(".zip"),
+            "QTI file should be a zip archive",
+        )
+
+    def test_qti_archive_contains_manifest_and_assessment_ids(self):
+
+        published_qti_exercise = kolibri_models.ContentNode.objects.get(
+            title="QTI Free Response Exercise"
+        )
+        assessment_ids = (
+            published_qti_exercise.assessmentmetadata.first().assessment_item_ids
+        )
+
+        # Should have exactly one assessment ID corresponding to our free response question
+        self.assertEqual(
+            len(assessment_ids), 3, "Should have exactly three assessment IDs"
+        )
+
+        # The assessment ID should match the one from our assessment item
+        qti_exercise = cc.ContentNode.objects.get(title="QTI Free Response Exercise")
+        for i, ai in enumerate(qti_exercise.assessment_items.order_by("order")):
+            self.assertEqual(assessment_ids[i], hex_to_qti_id(ai.assessment_id))
+
 
 class EmptyChannelTestCase(StudioTestCase):
     @classmethod

contentcuration/contentcuration/utils/assessment/qti/imsmanifest.py

@@ -1,12 +1,16 @@
+import re
+import zipfile
 from typing import Annotated
 from typing import List
 from typing import Optional
+from xml.etree import ElementTree as ET
 
 from pydantic import Field
 
 from contentcuration.utils.assessment.qti.base import generate_coerced_string_type
 from contentcuration.utils.assessment.qti.base import TextType
 from contentcuration.utils.assessment.qti.base import XMLElement
+from contentcuration.utils.assessment.qti.constants import ResourceType
 
 
 IMSCPIdentifier = Annotated[
@@ -100,3 +104,74 @@ class Manifest(XMLElement):
     organizations: Organizations = Field(default_factory=Organizations)
     resources: Resources = Field(default_factory=Resources)
     manifests: List["Manifest"] = Field(default_factory=list)
+
+
+def _get_item_ids_from_assessment_test(zip_file, test_href):
+    """Extract assessment item identifiers from an assessment test file."""
+    try:
+        with zip_file.open(test_href) as test_file:
+            test_content = test_file.read()
+            test_root = ET.fromstring(test_content)
+
+            # Look for both item references and inline items
+            qti_ns = {"qti": "http://www.imsglobal.org/xsd/imsqti_v3p0"}
+            item_refs = test_root.findall(".//qti:qti-assessment-item-ref", qti_ns)
+            # TODO: Add handling for assessment sections and assessment section refs.
+
+            all_items = list(item_refs)
+
+            return [
+                item.get("identifier") for item in all_items if item.get("identifier")
+            ]
+    except (KeyError, ET.ParseError):
+        return []
+
+
+namespace_re = re.compile("\\{([^}]+)\\}")
+
+
+def get_assessment_ids_from_manifest(zip_file_handle):
+    try:
+        with zipfile.ZipFile(zip_file_handle, "r") as zip_file:
+
+            # Read and parse the manifest
+            with zip_file.open("imsmanifest.xml") as manifest_file:
+                manifest_content = manifest_file.read()
+
+            # Parse the XML
+            root = ET.fromstring(manifest_content)
+
+            namespace = namespace_re.search(root.tag).group(1)
+
+            # Define namespace map for IMS Content Packaging
+            namespaces = {"imscp": namespace}
+
+            # Find all resources
+            resources = root.findall(".//imscp:resource", namespaces)
+
+            assessment_ids = []
+
+            # First, collect direct assessment item resources
+            for resource in resources:
+                resource_type = resource.get("type", "")
+                resource_identifier = resource.get("identifier")
+                if (
+                    resource_type == ResourceType.ASSESSMENT_ITEM.value
+                    and resource_identifier
+                ):
+                    assessment_ids.append(resource_identifier)
+
+                if resource_type == ResourceType.ASSESSMENT_TEST.value:
+                    assessment_ids.extend(
+                        _get_item_ids_from_assessment_test(
+                            zip_file, resource.get("href")
+                        )
+                    )
+
+            return assessment_ids
+    except ET.ParseError:
+        raise ValueError("Invalid XML in manifest")
+    except zipfile.BadZipFile:
+        raise ValueError("File is not a valid zip archive")
+    except KeyError:
+        raise ValueError("No IMS Manifest found in zip file")

contentcuration/contentcuration/utils/publish.py

@@ -44,6 +44,10 @@
 from contentcuration import models as ccmodels
 from contentcuration.decorators import delay_user_storage_calculation
 from contentcuration.utils.assessment.perseus import PerseusExerciseGenerator
+from contentcuration.utils.assessment.qti.archive import QTIExerciseGenerator
+from contentcuration.utils.assessment.qti.imsmanifest import (
+    get_assessment_ids_from_manifest,
+)
 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
@@ -319,20 +323,43 @@ def recurse_nodes(self, node, inherited_fields):  # noqa C901
             )
 
             if node.kind_id == content_kinds.EXERCISE:
-                exercise_data = process_assessment_metadata(node, kolibrinode)
+                exercise_data = process_assessment_metadata(node)
+                any_free_response = any(
+                    t == exercises.FREE_RESPONSE
+                    for t in exercise_data["assessment_mapping"].values()
+                )
+                generator_class = (
+                    QTIExerciseGenerator
+                    if any_free_response
+                    else PerseusExerciseGenerator
+                )
+
+                # If this exercise previously had a file generated by a different
+                # generator, make sure we clean it up here.
+                stale_presets = {
+                    PerseusExerciseGenerator.preset,
+                    QTIExerciseGenerator.preset,
+                } - {generator_class.preset}
+
+                # Remove archives produced by the previously-used generator
+                node.files.filter(preset_id__in=stale_presets).delete()
+
                 if (
                     self.force_exercises
                     or node.changed
-                    or not node.files.filter(preset_id=format_presets.EXERCISE).exists()
+                    or not node.files.filter(preset_id=generator_class.preset).exists()
                 ):
-                    generator = PerseusExerciseGenerator(
+
+                    generator = generator_class(
                         node,
                         exercise_data,
                         self.channel_id,
                         self.default_language.lang_code,
                         user_id=self.user_id,
                     )
                     generator.create_exercise_archive()
+
+                create_kolibri_assessment_metadata(node, kolibrinode)
             elif node.kind_id == content_kinds.SLIDESHOW:
                 create_slideshow_manifest(node, user_id=self.user_id)
             elif node.kind_id == content_kinds.TOPIC:
@@ -625,11 +652,7 @@ def parse_assessment_metadata(ccnode):
     )
 
 
-def process_assessment_metadata(ccnode, kolibrinode):
-    # Get mastery model information, set to default if none provided
-    assessment_items = ccnode.assessment_items.all().order_by("order")
-    assessment_item_ids = [a.assessment_id for a in assessment_items]
-
+def _get_exercise_data_from_ccnode(ccnode, num_assessment_items):
     randomize, mastery_criteria = parse_assessment_metadata(ccnode)
 
     exercise_data = deepcopy(mastery_criteria)
@@ -638,14 +661,14 @@ def process_assessment_metadata(ccnode, kolibrinode):
     mastery_model = {"type": exercise_data_type or exercises.M_OF_N}
     if mastery_model["type"] == exercises.M_OF_N:
         mastery_model.update(
-            {"n": exercise_data.get("n") or min(5, assessment_items.count()) or 1}
+            {"n": exercise_data.get("n") or min(5, num_assessment_items) or 1}
         )
         mastery_model.update(
-            {"m": exercise_data.get("m") or min(5, assessment_items.count()) or 1}
+            {"m": exercise_data.get("m") or min(5, num_assessment_items) or 1}
         )
     elif mastery_model["type"] == exercises.DO_ALL:
         mastery_model.update(
-            {"n": assessment_items.count() or 1, "m": assessment_items.count() or 1}
+            {"n": num_assessment_items or 1, "m": num_assessment_items or 1}
         )
     elif mastery_model["type"] == exercises.NUM_CORRECT_IN_A_ROW_2:
         mastery_model.update({"n": 2, "m": 2})
@@ -655,6 +678,17 @@ def process_assessment_metadata(ccnode, kolibrinode):
         mastery_model.update({"n": 5, "m": 5})
     elif mastery_model["type"] == exercises.NUM_CORRECT_IN_A_ROW_10:
         mastery_model.update({"n": 10, "m": 10})
+    return randomize, exercise_data, mastery_model
+
+
+def process_assessment_metadata(ccnode):
+    # Get mastery model information, set to default if none provided
+    assessment_items = ccnode.assessment_items.all().order_by("order")
+    assessment_item_ids = [a.assessment_id for a in assessment_items]
+
+    randomize, exercise_data, mastery_model = _get_exercise_data_from_ccnode(
+        ccnode, len(assessment_item_ids)
+    )
 
     exercise_data.update(
         {
@@ -673,18 +707,31 @@ def process_assessment_metadata(ccnode, kolibrinode):
         }
     )
 
+    return exercise_data
+
+
+def create_kolibri_assessment_metadata(ccnode, kolibrinode):
+    assessment_items = ccnode.assessment_items.all().order_by("order")
+    assessment_item_ids = [a.assessment_id for a in assessment_items]
+    randomize, _, mastery_model = _get_exercise_data_from_ccnode(
+        ccnode, len(assessment_item_ids)
+    )
+    qti_file = ccnode.files.filter(preset_id=format_presets.QTI_ZIP).first()
+    if qti_file:
+        # Open the zip file from Django storage
+        with qti_file.file_on_disk.open("rb") as file_handle:
+            assessment_item_ids = get_assessment_ids_from_manifest(file_handle)
+
     kolibrimodels.AssessmentMetaData.objects.create(
         id=uuid.uuid4(),
         contentnode=kolibrinode,
         assessment_item_ids=assessment_item_ids,
-        number_of_assessments=assessment_items.count(),
+        number_of_assessments=len(assessment_item_ids),
         mastery_model=mastery_model,
         randomize=randomize,
         is_manipulable=ccnode.kind_id == content_kinds.EXERCISE,
     )
 
-    return exercise_data
-
 
 def map_prerequisites(root_node):