This is a complex paragraph with emphasis and a + link to example.
+
+ 
This is a True/False question?
+QTI 3 is a new version released in 2022.
+Which of the following features are new to QTI 3?
+Pick 3 choices.
+Read this postcard from your English pen-friend, Sam.
+++Here is a postcard of my town. Please send me
+
+a postcard from your town. What size is your Town?
+What is the nicest part of your town?
+Where do you go in the evenings?Sam
+
Identify the missing word in this famous quote from Shakespeare's Richard III.
+++Now is the winter of our discontent
+
+Made glorious summer by this sun of; +
+And all the clouds that lour'd upon our house
+In the deep bosom of the ocean buried.
Calculate the value of x when 2x + 5 = 90:
+'
+ )
+
+ param_element = Param(name="test_param", value="test_value")
+ self.assertEqual(
+ param_element.to_xml_string(),
+ '',
+ )
+
+ object_element = Object(children=["Test Object"], params=[param_element])
+ self.assertEqual(
+ object_element.to_xml_string(),
+ '',
+ )
+
+ picture_source_element = Source(srcset="test.jpg 2x")
+ self.assertEqual(
+ picture_source_element.to_xml_string(), 'Test Blockquote', + ) + + div_element = Div(children=["Test Div"]) + self.assertEqual(div_element.to_xml_string(), "
| Test Th | Test Td |
|---|---|
| Test Th | Test Td |
| Test Th | Test Td |
Test P
") + + span_element = Span(children=["Test Span"]) + self.assertEqual(span_element.to_xml_string(), "Test Span") + + h1_element = H1(children=["Test H1"]) + self.assertEqual(h1_element.to_xml_string(), "Test Pre") + + em_element = Em(children=["Test Em"]) + self.assertEqual(em_element.to_xml_string(), "Test Em") + + code_element = Code(children=["Test Code"]) + self.assertEqual(code_element.to_xml_string(), "
Test Code")
+
+ kbd_element = Kbd(children=["Test Kbd"])
+ self.assertEqual(kbd_element.to_xml_string(), "Test Kbd")
+
+ i_element = I(children=["Test I"])
+ self.assertEqual(i_element.to_xml_string(), "Test I")
+
+ dfn_element = Dfn(children=["Test Dfn"])
+ self.assertEqual(dfn_element.to_xml_string(), "Test Dfn")
+
+ abbr_element = Abbr(children=["Test Abbr"])
+ self.assertEqual(abbr_element.to_xml_string(), "Test Abbr")
+
+ strong_element = Strong(children=["Test Strong"])
+ self.assertEqual(strong_element.to_xml_string(), "Test Strong")
+
+ sup_element = Sup(children=["Test Sup"])
+ self.assertEqual(sup_element.to_xml_string(), "Test Sup")
+
+ sub_element = Sub(children=["Test Sub"])
+ self.assertEqual(sub_element.to_xml_string(), "Test Sub")
+
+ var_element = Var(children=["Test Var"])
+ self.assertEqual(var_element.to_xml_string(), "Test Var")
+
+ small_element = Small(children=["Test Small"])
+ self.assertEqual(small_element.to_xml_string(), "Test Small")
+
+ samp_element = Samp(children=["Test Samp"])
+ self.assertEqual(samp_element.to_xml_string(), "Test Samp")
+
+ b_element = B(children=["Test B"])
+ self.assertEqual(b_element.to_xml_string(), "Test B")
+
+ cite_element = Cite(children=["Test Cite"])
+ self.assertEqual(cite_element.to_xml_string(), "Test Cite")
+
+ q_element = Q(children=["Test Q"])
+ self.assertEqual(q_element.to_xml_string(), "Test Q") + + bdo_element = Bdo(dir=BdoDir.LTR, children=["Test Bdo"]) + self.assertEqual(bdo_element.to_xml_string(), 'Test Bdo') + + bdi_element = Bdi(children=["Test Bdi"]) + self.assertEqual(bdi_element.to_xml_string(), "Test Bdi") + + rt_element = Rt(children=["Test Rt"]) + self.assertEqual(rt_element.to_xml_string(), "") + + rp_element = Rp(text="(") + self.assertEqual(rp_element.to_xml_string(), "") + + ruby_element = Ruby(children=["Test Ruby"]) + self.assertEqual(ruby_element.to_xml_string(), "Test Ruby") + + +class TestHTMLStringIntegration(unittest.TestCase): + def test_complex_html_parsing(self): + complex_html = """ +
This is a complex paragraph with emphasis and a + link to example.
+
+ Final paragraph with
line break.
Hello world!
" + elements = HTMLElement.from_html_string(simple_html) + + self.assertEqual(len(elements), 1) + p = elements[0] + self.assertIsInstance(p, P) + self.assertEqual(len(p.children), 3) + + # Check strong element + strong = p.children[1] + self.assertIsInstance(strong, Strong) + self.assertEqual(strong.children[0].text, "world") + + def test_empty_and_self_closing_elements(self): + """Test parsing empty elements and self-closing tags""" + + html = """ + +
+ Test bold and italic text.
+
+ Test Item Body Content
What is 2+2?
+4
3
5
Select all prime numbers:
+2
3
4
5
What is the capital of France?
+What does this even mean?
+Solve the equation for x. What is the value of x?
+6
3
1.5
9
What is the answer to this question? " + '
\n" + ) + + self._assert_conversion(markdown_text, expected) + + def test_block_math(self): + """Test conversion of block math""" + + markdown_text = ( + "Here's an equation:\n\n$$E = mc^2$$\n\nThat's Einstein's formula." + ) + expected = ( + "Here's an equation:
\n" + '" + "That's Einstein's formula.
\n" + ) + + self._assert_conversion(markdown_text, expected) + + def test_multiline_block_math(self): + """ + Ensure a $$ … $$ block spanning multiple lines is converted to MathML + and the literal $$ delimiters are removed. This currently fails with + the buggy BLOCK_PATTERN because it stops after the first '$'. + """ + markdown_text = ( + "$$\n" + "\\begin{aligned}\n" + "a = b + c \\\\\n" + "$5 = d + e\n" + "\\end{aligned}\n" + "$$" + ) + expected = ( + '" + ) + + self._assert_conversion(markdown_text, expected) + + def test_inline_math_with_dollar_inside(self): + """ + Ensure a $$ … $$ inline that contains an internal '$' (e.g. inside + \\text{}) is parsed correctly. With the old BLOCK_PATTERN the first '$' + prematurely terminates the match so the delimiters remain. + """ + markdown_text = "Test this $$\\text{Cost = 1.00 $USD$}$$" + expected = ( + "Test this " + '
\n" + ) + + self._assert_conversion(markdown_text, expected) + + def test_multiple_math_expressions(self): + """Test multiple math expressions in one document""" + + markdown_text = "First: $$a + b$$, then $$c \\times d$$, finally $$e^f$$." + expected = ( + "First: " + '' + ", then " + '' + ", finally " + '' + ".
\n" + ) + + self._assert_conversion(markdown_text, expected) + + def test_mixed_inline_and_block(self): + """Test document with both inline and block math""" + + markdown_text = ( + "This is inline math: $$a = b$$\n\n" + "And this is block math:\n\n" + "$$\\sum_{i=1}^{n} x_i = y$$\n\n" + "Back to text with more inline: $$z^2$$" + ) + expected = ( + "This is inline math: " + '' + "
\n" + "And this is block math:
\n" + '" + "Back to text with more inline: " + '' + "
\n" + ) + + self._assert_conversion(markdown_text, expected) + + def test_no_math_content(self): + """Test that regular markdown without math still works""" + + markdown_text = "This is just *regular* markdown with **bold** text." + expected = "This is just regular markdown with bold text.
\n" + + self._assert_conversion(markdown_text, expected) + + def test_simple_inline_math(self): + """Test simple inline math expression""" + + markdown_text = "The variable $$x$$ is unknown." + expected = ( + "The variable " + '' + " is unknown.
\n" + ) + + self._assert_conversion(markdown_text, expected) + + def test_simple_block_math(self): + """Test simple block math expression""" + + markdown_text = "$$y = mx + b$$" + expected = ( + '" + ) + + self._assert_conversion(markdown_text, expected) + + +class TestTexMathPlugin(TexMathTestMixin, unittest.TestCase): + """Test direct markdown conversion: markdown → HTML+MathML""" + + def _assert_conversion(self, markdown_text: str, expected: str): + """Test direct markdown to HTML+MathML conversion""" + result = render_markdown(markdown_text) + self.assertEqual(result, expected) + + +class TestTexMathPluginRoundtrip(TexMathTestMixin, unittest.TestCase): + """Test full roundtrip: markdown → HTML+MathML → Pydantic → string""" + + maxDiff = None + + def _assert_conversion(self, markdown_text: str, expected: str): + """Test full roundtrip conversion via Pydantic objects""" + result = render_markdown(markdown_text) + + # Parse to Pydantic objects and back to string + parsed = ElementTreeBase.from_string(result) + roundtrip_result = ( + "".join(e.to_xml_string().strip() for e in parsed) + if isinstance(parsed, list) + else parsed.to_xml_string().strip() + ) + self.assertEqual( + roundtrip_result.replace("\n", "").strip(), + expected.replace("\n", "").strip(), + ) diff --git a/contentcuration/contentcuration/utils/assessment/__init__.py b/contentcuration/contentcuration/utils/assessment/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/contentcuration/contentcuration/utils/assessment/base.py b/contentcuration/contentcuration/utils/assessment/base.py new file mode 100644 index 0000000000..0f668920a0 --- /dev/null +++ b/contentcuration/contentcuration/utils/assessment/base.py @@ -0,0 +1,395 @@ +import hashlib +import json +import logging +import os +import re +import zipfile +from abc import ABC +from abc import abstractmethod +from io import BytesIO +from tempfile import NamedTemporaryFile +from tempfile import TemporaryDirectory + +from django.core.files import File +from django.core.files.storage import default_storage as storage +from le_utils.constants import exercises +from PIL import Image + +from contentcuration import models + + +image_pattern = rf"!\[(?:[^\]]*)]\(\${exercises.CONTENT_STORAGE_PLACEHOLDER}/([^\s)]+)(?:\s=([0-9\.]+)x([0-9\.]+))*[^)]*\)" + + +def resize_image(image_content, width, height): + try: + with Image.open(BytesIO(image_content)) as img: + original_format = img.format + img = img.resize((int(width), int(height)), Image.LANCZOS) + buffered = BytesIO() + img.save(buffered, format=original_format) + return buffered.getvalue() + except Exception as e: + logging.warning(f"Error resizing image: {str(e)}") + return None + + +def get_resized_image_checksum(image_content): + return hashlib.md5(image_content).hexdigest() + + +class ExerciseArchiveGenerator(ABC): + """ + Abstract base class for exercise zip generators. + Handles common functionality for creating exercise zip files for different formats. + """ + + ZIP_DATE_TIME = (2015, 10, 21, 7, 28, 0) + ZIP_COMPRESS_TYPE = zipfile.ZIP_DEFLATED + ZIP_COMMENT = "".encode() + + @property + @abstractmethod + def file_format(self): + pass + + @property + @abstractmethod + def preset(self): + pass + + @abstractmethod + def get_image_file_path(self): + """ + Abstract method to get the archive file path for storing assessment image files. + + Returns: + str: The file path for images in the exercise archive + """ + pass + + def get_image_ref_prefix(self): + """ + A value to insert in front of the image file path - this is needed for Perseus to properly + find all image file paths in the frontend. + """ + return "" + + @abstractmethod + def create_assessment_item(self, assessment_item, processed_data): + """ + Abstract method to create an assessment item from processed data. + Args: + assessment_item: The assessment item to process + processed_data: Data processed from the assessment item + Returns: + filepath: Path for the created assessment item file + file_content: Content of the assessment item file + """ + pass + + def __init__( + self, ccnode, exercise_data, channel_id, default_language, user_id=None + ): + """ + Initialize the exercise zip generator. + + Args: + ccnode: Content node containing exercise data + exercise_data: Data specific to the exercise format + user_id: Optional user ID for tracking who created the exercise + """ + self.ccnode = ccnode + self.exercise_data = exercise_data + self.channel_id = channel_id + self.default_language = default_language + self.user_id = user_id + self.resized_images_map = {} + self.assessment_items = [] + self.files_to_write = [] + self.tempdir = None + + def write_to_zipfile(self, zf, filepath, content): + """ + This method is a copy of the write_file_to_zip_with_neutral_metadata function from ricecooker. + The comment, date_time, and compress_type are parameterized to allow for Perseus to override them. + This can be updated in future when we have a good way to avoid rebuilding perseus files, unless needed. + """ + filepath = filepath.replace("\\", "/") + info = zipfile.ZipInfo(filepath, date_time=self.ZIP_DATE_TIME) + info.comment = self.ZIP_COMMENT + info.compress_type = self.ZIP_COMPRESS_TYPE + info.create_system = 0 + zf.writestr(info, content) + + def add_file_to_write(self, filepath, content): + if self.tempdir is None: + raise RuntimeError( + "Cannot add files to write before creating the temporary directory." + ) + full_path = os.path.join(self.tempdir, filepath) + if os.path.exists(full_path): + return + os.makedirs(os.path.dirname(full_path), exist_ok=True) + with open(full_path, "wb") as f: + f.write(content) + self.files_to_write.append(full_path) + + def _add_original_image(self, checksum, filename, new_file_path): + """Extract original image handling""" + with storage.open( + models.generate_object_storage_name(checksum, filename), "rb" + ) as imgfile: + original_content = imgfile.read() + self.add_file_to_write(os.path.join(new_file_path, filename), original_content) + + def _get_similar_image(self, filename, width, height): + if filename not in self.resized_images_map: + self.resized_images_map[filename] = {} + return None + if (width, height) in self.resized_images_map[filename]: + return self.resized_images_map[filename][(width, height)] + + for key, resized_image in self.resized_images_map[filename].items(): + if ( + abs(key[0] - width) / width < 0.01 + and abs(key[1] - height) / height < 0.01 + ): + return resized_image + + def _resize_image(self, checksum, ext, filename, width, height, new_file_path): + with storage.open( + models.generate_object_storage_name(checksum, filename), + "rb", + ) as imgfile: + original_content = imgfile.read() + + resized_content = resize_image(original_content, width, height) + + if not resized_content: + logging.warning(f"Failed to resize image {filename}. Using original image.") + return + resized_checksum = get_resized_image_checksum(resized_content) + + new_img_ref = f"{resized_checksum}{ext}" + self.resized_images_map[filename][(width, height)] = new_img_ref + self.add_file_to_write( + os.path.join(new_file_path, new_img_ref), resized_content + ) + return new_img_ref + + def _process_single_image( + self, filename, checksum, ext, width, height, new_file_path + ): + if width is None and height is None: + # No resizing needed, just add original + self._add_original_image(checksum, filename, new_file_path) + return filename + + # Try to get similar or create resized image + similar_image = self._get_similar_image(filename, width, height) + if similar_image: + return similar_image + + resized_image = self._resize_image( + checksum, ext, filename, width, height, new_file_path + ) + return resized_image or filename + + def _replace_filename_in_match( + self, content, img_match, old_filename, new_filename + ): + """Extract filename replacement logic""" + start, end = img_match.span() + old_match = content[start:end] + new_match = old_match.replace(old_filename, new_filename) + return content[:start] + new_match + content[end:] + + def _is_valid_image_filename(self, filename): + checksum, ext = os.path.splitext(filename) + + if not ext: + logging.warning( + "While publishing channel `{}` a filename with no extension was encountered: `{}`".format( + self.channel_id, filename + ) + ) + return False + + try: + int(checksum, 16) # Validate hex checksum + return True + except ValueError: + logging.warning( + "while publishing channel `{}` a filename with an improper checksum was encountered: `{}`".format( + self.channel_id, filename + ) + ) + if os.environ.get("BRANCH_ENVIRONMENT", "") != "master": + raise + return False + + def process_image_strings(self, content): + new_file_path = self.get_image_file_path() + new_image_path = f"{self.get_image_ref_prefix()}{new_file_path}" + image_list = [] + processed_files = [] + for img_match in re.finditer(image_pattern, content): + # Add any image files that haven't been written to the zipfile + filename = img_match.group(1) + width = float(img_match.group(2)) if img_match.group(2) else None + height = float(img_match.group(3)) if img_match.group(3) else None + checksum, ext = os.path.splitext(filename) + + if not self._is_valid_image_filename(filename): + continue + + if width == 0 or height == 0: + # Can't resize an image to 0 width or height, so just ignore. + continue + + processed_filename = self._process_single_image( + filename, checksum, ext, width, height, new_file_path + ) + processed_files.append( + (img_match, filename, processed_filename, width, height) + ) + + # Process matches in reverse order to avoid index mismatch when modifying content + for img_match, filename, processed_filename, width, height in reversed( + processed_files + ): + content = self._replace_filename_in_match( + content, img_match, filename, processed_filename + ) + if width is not None and height is not None: + image_list.append( + {"name": processed_filename, "width": width, "height": height} + ) + + content = content.replace( + f"${exercises.CONTENT_STORAGE_PLACEHOLDER}", new_image_path + ) + return content, image_list + + def _process_content(self, content): + """ + Process the content to handle images. + + Args: + content: The content string to process + + Returns: + tuple: Processed content and list of image data + """ + return self.process_image_strings(content) + + def _sort_by_order(self, items, item_type): + try: + return sorted(items, key=lambda x: x.get("order")) + except TypeError: + logging.error(f"Unable to sort {item_type}, leaving unsorted.") + return items + + def _process_answers(self, assessment_item): + answer_data = json.loads(assessment_item.answers) + processed_answers = [] + + for answer in answer_data: + if answer["answer"]: + if isinstance(answer["answer"], str): + (answer["answer"], answer_images,) = self._process_content( + answer["answer"], + ) + answer["images"] = answer_images + + processed_answers.append(answer) + + return self._sort_by_order(processed_answers, "answers") + + def _process_hints(self, assessment_item): + hint_data = json.loads(assessment_item.hints) + + for hint in hint_data: + hint["hint"], hint_images = self._process_content( + hint["hint"], + ) + hint["images"] = hint_images + + return self._sort_by_order(hint_data, "hints") + + def process_assessment_item(self, assessment_item): + # Process question + question, question_images = self._process_content( + assessment_item.question, + ) + + # Process answers and hints + processed_answers = self._process_answers(assessment_item) + processed_hints = self._process_hints(assessment_item) + + new_file_path = self.get_image_file_path() + new_image_path = f"{exercises.IMG_PLACEHOLDER}/{new_file_path}" + context = { + "question": question, + "question_images": question_images, + "answers": processed_answers, + "multiple_select": assessment_item.type == exercises.MULTIPLE_SELECTION, + "raw_data": assessment_item.raw_data.replace( + exercises.CONTENT_STORAGE_PLACEHOLDER, new_image_path + ), + "hints": processed_hints, + "randomize": assessment_item.randomize, + } + filepath, file_content = self.create_assessment_item(assessment_item, context) + self.add_file_to_write(filepath, file_content) + + def handle_before_assessment_items(self): + pass + + def handle_after_assessment_items(self): + pass + + def _create_zipfile(self): + filename = "{0}.{ext}".format(self.ccnode.title, ext=self.file_format) + with NamedTemporaryFile(suffix="zip") as tempf: + with zipfile.ZipFile(tempf.name, "w") as zf: + for file_path in self.files_to_write: + with open(file_path, "rb") as f: + self.write_to_zipfile( + zf, + os.path.relpath(file_path, self.tempdir), + f.read(), + ) + file_size = tempf.tell() + tempf.flush() + + self.ccnode.files.filter(preset_id=self.preset).delete() + + assessment_file_obj = models.File.objects.create( + file_on_disk=File(open(tempf.name, "rb"), name=filename), + contentnode=self.ccnode, + file_format_id=self.file_format, + preset_id=self.preset, + original_filename=filename, + file_size=file_size, + uploaded_by_id=self.user_id, + ) + logging.debug( + "Created exercise for {0} with checksum {1}".format( + self.ccnode.title, assessment_file_obj.checksum + ) + ) + + def create_exercise_archive(self): + with TemporaryDirectory() as tempdir: + self.tempdir = tempdir + self.handle_before_assessment_items() + for question in ( + self.ccnode.assessment_items.prefetch_related("files") + .all() + .order_by("order") + ): + self.process_assessment_item(question) + self.handle_after_assessment_items() + self._create_zipfile() diff --git a/contentcuration/contentcuration/utils/assessment/markdown.py b/contentcuration/contentcuration/utils/assessment/markdown.py new file mode 100644 index 0000000000..c34da5dee1 --- /dev/null +++ b/contentcuration/contentcuration/utils/assessment/markdown.py @@ -0,0 +1,135 @@ +import re +import xml.etree.ElementTree as ET + +from latex2mathml.converter import convert +from markdown_it import MarkdownIt +from markdown_it.renderer import RendererProtocol +from markdown_it.rules_block import StateBlock +from markdown_it.rules_inline import StateInline +from markdown_it.token import Token +from markdown_it.utils import EnvType +from markdown_it.utils import OptionsDict + +from contentcuration.utils.assessment.qti.mathml.core import Annotation +from contentcuration.utils.assessment.qti.mathml.core import Semantics + + +# Regex patterns for $$ delimited math +INLINE_PATTERN = re.compile(r"^\$\$([\s\S]+?)\$\$") +BLOCK_PATTERN = re.compile(r"^\$\$([\s\S]+?)\$\$", re.M) + + +def math_inline_func(state: StateInline, silent: bool) -> bool: + """Parse inline math: $$expression$$""" + if not state.src.startswith("$$", state.pos): + return False + + match = INLINE_PATTERN.match(state.src[state.pos :]) + if not match: + return False + + if not silent: + token = state.push("math_inline", "math", 0) + token.content = match.group(1) + token.markup = "$$" + + state.pos += match.end() + return True + + +def math_block_func( + state: StateBlock, begLine: int, endLine: int, silent: bool +) -> bool: + """Parse block math: $$expression$$""" + begin = state.bMarks[begLine] + state.tShift[begLine] + + if not state.src.startswith("$$", begin): + return False + + match = BLOCK_PATTERN.match(state.src[begin:]) + if not match: + return False + + if not silent: + token = state.push("math_block", "math", 0) + token.block = True + token.content = match.group(1) + token.markup = "$$" + + # Advance to next line after the math block + endpos = begin + match.end() - 1 + line = begLine + while line < endLine: + if endpos >= state.bMarks[line] and endpos <= state.eMarks[line]: + state.line = line + 1 + break + line += 1 + + return True + + +def _convert(latex, inline=True): + # Remove the namespace declaration for cleaner output + markup = convert(latex, display="inline" if inline else "block").replace( + ' xmlns="http://www.w3.org/1998/Math/MathML"', "" + ) + # By default latex2mathml encodes operators that don't need to be encoded + # so we parse it with ElementTree and turn it back into a string here for consistency. + math_element = ET.fromstring(markup) + + # Create LaTeX annotation + latex_annotation_element = Annotation( + encoding="application/x-tex", children=[latex] + ).to_element() + + semantics_element = Semantics().to_element() + for child in math_element: + math_element.remove(child) + semantics_element.append(child) + semantics_element.append(latex_annotation_element) + math_element.append(semantics_element) + + return ET.tostring(math_element, encoding="unicode") + + +def render_math_inline( + self: RendererProtocol, + tokens: list[Token], + idx: int, + options: OptionsDict, + env: EnvType, +) -> str: + """Render inline math to MathML""" + return _convert(tokens[idx].content) + + +def render_math_block( + self: RendererProtocol, + tokens: list[Token], + idx: int, + options: OptionsDict, + env: EnvType, +) -> str: + """Render block math to MathML""" + return _convert(tokens[idx].content, inline=False) + + +def texmath_to_mathml_plugin(md: MarkdownIt) -> None: + """Simple plugin for parsing TeX math with $$ delimiters. + + Converts inline and block math expressions to MathML using latex2mathml. + """ + # Register parsing rules + md.inline.ruler.before("escape", "math_inline", math_inline_func) + md.block.ruler.before("fence", "math_block", math_block_func) + + # Register renderers + md.add_render_rule("math_inline", render_math_inline) + md.add_render_rule("math_block", render_math_block) + + +md = MarkdownIt("gfm-like").disable("linkify").use(texmath_to_mathml_plugin) + + +def render_markdown(markdown): + return md.render(markdown) diff --git a/contentcuration/contentcuration/utils/assessment/perseus.py b/contentcuration/contentcuration/utils/assessment/perseus.py new file mode 100644 index 0000000000..7ba4e1ce6f --- /dev/null +++ b/contentcuration/contentcuration/utils/assessment/perseus.py @@ -0,0 +1,131 @@ +import json +import re +import zipfile + +from django.core.files.storage import default_storage as storage +from django.template.loader import render_to_string +from le_utils.constants import exercises +from le_utils.constants import file_formats +from le_utils.constants import format_presets + +from contentcuration import models +from contentcuration.utils.assessment.base import ExerciseArchiveGenerator +from contentcuration.utils.parser import extract_value + + +_DOUBLE_DOLLAR_RE = re.compile(r"\$\$(.+?)\$\$", flags=re.DOTALL) + + +class PerseusExerciseGenerator(ExerciseArchiveGenerator): + """ + Exercise zip generator for Perseus format exercises. + """ + + ZIP_DATE_TIME = (2013, 3, 14, 1, 59, 26) + ZIP_COMPRESS_TYPE = zipfile.ZIP_STORED + ZIP_COMMENT = "Perseus file generated during export process".encode() + + file_format = file_formats.PERSEUS + preset = format_presets.EXERCISE + + TEMPLATE_MAP = { + exercises.MULTIPLE_SELECTION: "perseus/multiple_selection.json", + exercises.SINGLE_SELECTION: "perseus/multiple_selection.json", + exercises.INPUT_QUESTION: "perseus/input_question.json", + exercises.PERSEUS_QUESTION: "perseus/perseus_question.json", + "true_false": "perseus/multiple_selection.json", + } + + def _write_raw_perseus_image_files(self, assessment_item): + # For raw perseus JSON questions, the files must be + # specified in advance. + + # Files have been prefetched when the assessment item was + # queried, so take advantage of that. + files = sorted(assessment_item.files.all(), key=lambda x: x.checksum) + image_files = filter( + lambda x: x.preset_id == format_presets.EXERCISE_IMAGE, files + ) + graphie_files = filter( + lambda x: x.preset_id == format_presets.EXERCISE_GRAPHIE, files + ) + images_path = self.get_image_file_path() + for image in image_files: + image_name = "{}/{}.{}".format( + images_path, image.checksum, image.file_format_id + ) + with storage.open( + models.generate_object_storage_name(image.checksum, str(image)), + "rb", + ) as content: + self.add_file_to_write(image_name, content.read()) + + for image in graphie_files: + svg_name = "{}/{}.svg".format(images_path, image.original_filename) + json_name = "{}/{}-data.json".format(images_path, image.original_filename) + with storage.open( + models.generate_object_storage_name(image.checksum, str(image)), + "rb", + ) as content: + content = content.read() + # in Python 3, delimiter needs to be in bytes format + content = content.split(exercises.GRAPHIE_DELIMITER.encode("ascii")) + if len(content) != 2: + raise ValueError( + f"Graphie file '{image.original_filename}' " + f"missing delimiter {exercises.GRAPHIE_DELIMITER!r}" + ) + self.add_file_to_write(svg_name, content[0]) + self.add_file_to_write(json_name, content[1]) + + def _process_formulas(self, content): + return _DOUBLE_DOLLAR_RE.sub(r"$\1$", content) + + def _process_content(self, content): + content = self._process_formulas(content) + return super()._process_content(content) + + def process_assessment_item(self, assessment_item): + if assessment_item.type == exercises.PERSEUS_QUESTION: + self._write_raw_perseus_image_files(assessment_item) + return super().process_assessment_item(assessment_item) + + def _process_input_answers(self, processed_data): + """Extract input answer processing logic""" + non_empty_answers = [] + for answer in processed_data["answers"]: + answer["answer"] = extract_value(answer["answer"]) + if answer["answer"] or answer["answer"] == 0: + non_empty_answers.append(answer) + + return {**processed_data, "answers": non_empty_answers} + + def create_assessment_item(self, assessment_item, processed_data): + template = self.TEMPLATE_MAP.get(assessment_item.type) + if not template: + raise TypeError( + f"Unrecognized question type on item {assessment_item.assessment_id}: {assessment_item.type}" + ) + + # Handle input question special case + if assessment_item.type == exercises.INPUT_QUESTION: + processed_data = self._process_input_answers(processed_data) + + filename = f"{assessment_item.assessment_id}.json" + content = render_to_string(template, processed_data).encode("utf-8", "ignore") + return filename, content + + def get_image_file_path(self): + return "images" + + def get_image_ref_prefix(self): + return f"${exercises.IMG_PLACEHOLDER}/" + + def handle_before_assessment_items(self): + exercise_context = { + "exercise": json.dumps(self.exercise_data, sort_keys=True, indent=4) + } + exercise_result = render_to_string( + "perseus/exercise.json", exercise_context + ).encode("utf-8") + self.add_file_to_write("exercise.json", exercise_result) diff --git a/contentcuration/contentcuration/utils/assessment/qti/__init__.py b/contentcuration/contentcuration/utils/assessment/qti/__init__.py new file mode 100644 index 0000000000..c8cb0afb95 --- /dev/null +++ b/contentcuration/contentcuration/utils/assessment/qti/__init__.py @@ -0,0 +1,6 @@ +from .base import ElementTreeBase + + +__all__ = [ + "ElementTreeBase", +] diff --git a/contentcuration/contentcuration/utils/assessment/qti/archive.py b/contentcuration/contentcuration/utils/assessment/qti/archive.py new file mode 100644 index 0000000000..4a29f20c84 --- /dev/null +++ b/contentcuration/contentcuration/utils/assessment/qti/archive.py @@ -0,0 +1,284 @@ +import base64 +from typing import Any +from typing import Dict +from typing import List +from typing import Tuple + +from le_utils.constants import exercises +from le_utils.constants import format_presets + +from contentcuration.utils.assessment.base import ExerciseArchiveGenerator +from contentcuration.utils.assessment.markdown import render_markdown +from contentcuration.utils.assessment.qti.assessment_item import AssessmentItem +from contentcuration.utils.assessment.qti.assessment_item import CorrectResponse +from contentcuration.utils.assessment.qti.assessment_item import ItemBody +from contentcuration.utils.assessment.qti.assessment_item import OutcomeDeclaration +from contentcuration.utils.assessment.qti.assessment_item import ResponseDeclaration +from contentcuration.utils.assessment.qti.assessment_item import ResponseProcessing +from contentcuration.utils.assessment.qti.assessment_item import Value +from contentcuration.utils.assessment.qti.base import ElementTreeBase +from contentcuration.utils.assessment.qti.constants import BaseType +from contentcuration.utils.assessment.qti.constants import Cardinality +from contentcuration.utils.assessment.qti.constants import Orientation +from contentcuration.utils.assessment.qti.constants import ResourceType +from contentcuration.utils.assessment.qti.constants import ShowHide +from contentcuration.utils.assessment.qti.html import Div +from contentcuration.utils.assessment.qti.html import FlowContentList +from contentcuration.utils.assessment.qti.html import P +from contentcuration.utils.assessment.qti.imsmanifest import File as ManifestFile +from contentcuration.utils.assessment.qti.imsmanifest import Manifest +from contentcuration.utils.assessment.qti.imsmanifest import Metadata +from contentcuration.utils.assessment.qti.imsmanifest import Resource +from contentcuration.utils.assessment.qti.imsmanifest import Resources +from contentcuration.utils.assessment.qti.interaction_types.simple import ( + ChoiceInteraction, +) +from contentcuration.utils.assessment.qti.interaction_types.simple import SimpleChoice +from contentcuration.utils.assessment.qti.interaction_types.text_based import ( + TextEntryInteraction, +) +from contentcuration.utils.assessment.qti.prompt import Prompt + + +choice_interactions = { + exercises.MULTIPLE_SELECTION, + exercises.SINGLE_SELECTION, + "true_false", +} +text_entry_interactions = {exercises.INPUT_QUESTION, exercises.FREE_RESPONSE} + + +def hex_to_qti_id(hex_string): + """ + Encode a 32 digit hex to a 22 character base64 encoded id and a K prefix. + """ + bytes_data = bytes.fromhex(hex_string) + return f"K{base64.urlsafe_b64encode(bytes_data).decode('ascii').rstrip('=')}" + + +class QTIExerciseGenerator(ExerciseArchiveGenerator): + """ + Exercise zip generator for QTI format exercises. + Creates IMS Content Package with QTI 3.0 assessment items. + """ + + file_format = "zip" + preset = format_presets.QTI_ZIP + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.qti_items = [] + + def get_image_file_path(self) -> str: + """Get the file path for QTI assessment items.""" + return "items/images" + + def _create_html_content_from_text(self, text: str) -> FlowContentList: + """Convert text content to QTI HTML flow content.""" + if not text.strip(): + return [] + markup = render_markdown(text) + return ElementTreeBase.from_string(markup) + + def _create_choice_interaction_and_response( + self, processed_data: Dict[str, Any] + ) -> Tuple[ChoiceInteraction, ResponseDeclaration]: + """Create a QTI choice interaction for multiple choice questions.""" + + prompt = Prompt( + children=self._create_html_content_from_text(processed_data["question"]) + ) + + choices = [] + correct_values = [] + for i, answer in enumerate(processed_data.get("answers", [])): + choice_id = f"choice_{i}" + choice_content = self._create_html_content_from_text( + answer.get("answer", "") + ) + + choice = SimpleChoice( + identifier=choice_id, + children=choice_content, + show_hide=ShowHide.SHOW, + fixed=False, + ) + choices.append(choice) + + if answer.get("correct", False): + correct_values.append(Value(value=choice_id)) + + response_declaration = ResponseDeclaration( + identifier="RESPONSE", + cardinality=Cardinality.MULTIPLE + if processed_data["multiple_select"] + else Cardinality.SINGLE, + base_type=BaseType.IDENTIFIER, + correct_response=CorrectResponse(value=correct_values) + if correct_values + else None, + ) + + interaction = ChoiceInteraction( + response_identifier="RESPONSE", + prompt=prompt, + answers=choices, + shuffle=processed_data.get("randomize", False), + max_choices=len(choices) if processed_data["multiple_select"] else 1, + min_choices=0, + orientation=Orientation.VERTICAL, + ) + return interaction, response_declaration + + def _create_text_entry_interaction_and_response( + self, processed_data: Dict[str, Any] + ) -> Tuple[Div, ResponseDeclaration]: + prompt = self._create_html_content_from_text(processed_data["question"]) + interaction_element = TextEntryInteraction( + response_identifier="RESPONSE", + expected_length=50, # Default expected length + placeholder_text="Enter your answer here", + ) + # Text entry interaction is an inline element, so wrap it in a paragraph tag. + interaction_element = P(children=[interaction_element]) + # prompt is already a list of elements, so just append the interaction to it. + prompt.append(interaction_element) + interaction = Div(children=prompt) + + correct_values = [ + Value(value=answer["answer"]) + for answer in processed_data["answers"] + if answer["correct"] + ] + + response_declaration = ResponseDeclaration( + identifier="RESPONSE", + cardinality=Cardinality.MULTIPLE + if processed_data["multiple_select"] + else Cardinality.SINGLE, + base_type=BaseType.STRING, + correct_response=CorrectResponse(value=correct_values) + if correct_values + else None, + ) + return interaction, response_declaration + + def _qti_item_filepath(self, assessment_id): + return f"items/{assessment_id}.xml" + + def create_assessment_item( + self, assessment_item, processed_data: Dict[str, Any] + ) -> tuple[str, bytes]: + """Create QTI assessment item XML.""" + + # Skip Perseus questions as they can't be easily converted + if assessment_item.type == exercises.PERSEUS_QUESTION: + raise ValueError( + f"Perseus questions are not supported in QTI format: {assessment_item.assessment_id}" + ) + + if assessment_item.type in choice_interactions: + ( + interaction, + response_declaration, + ) = self._create_choice_interaction_and_response(processed_data) + elif assessment_item.type in text_entry_interactions: + ( + interaction, + response_declaration, + ) = self._create_text_entry_interaction_and_response(processed_data) + else: + raise ValueError(f"Unsupported question type: {assessment_item.type}") + + # Create item body with the interaction + item_body = ItemBody(children=[interaction]) + + # Create outcome declaration + outcome_declaration = OutcomeDeclaration( + identifier="SCORE", cardinality=Cardinality.SINGLE, base_type=BaseType.FLOAT + ) + + # Create response processing + response_processing = ResponseProcessing( + template="https://purl.imsglobal.org/spec/qti/v3p0/rptemplates/match_correct" + ) + + language = ( + self.ccnode.language.lang_code + if self.ccnode.language + else self.default_language + ) + + qti_item_id = hex_to_qti_id(assessment_item.assessment_id) + + # Create the assessment item + qti_item = AssessmentItem( + identifier=qti_item_id, + title=f"{self.ccnode.title} {len(self.qti_items) + 1}", + language=language, + adaptive=False, + time_dependent=False, + response_declaration=[response_declaration], + outcome_declaration=[outcome_declaration], + item_body=item_body, + response_processing=response_processing, + ) + + # Store for manifest creation + self.qti_items.append(qti_item) + + # Generate XML content + xml_content = qti_item.to_xml_string() + + # Add XML declaration and format nicely + full_xml = f'\n{xml_content}' + + filename = self._qti_item_filepath(qti_item_id) + return filename, full_xml.encode("utf-8") + + def _create_manifest_resources(self) -> List[Resource]: + """Create manifest resources for all QTI items.""" + resources = [] + + for qti_item in self.qti_items: + # Get file dependencies (images, etc.) + file_dependencies = qti_item.get_file_dependencies() + + # Create file entries + qti_item_filepath = self._qti_item_filepath(qti_item.identifier) + files = [ManifestFile(href=qti_item_filepath)] + for dep in file_dependencies: + files.append(ManifestFile(href=dep)) + + resource = Resource( + identifier=qti_item.identifier, + type_=ResourceType.ASSESSMENT_ITEM.value, + href=qti_item_filepath, + files=files, + ) + resources.append(resource) + + return resources + + def _create_imsmanifest(self) -> str: + # Create resources + resources = self._create_manifest_resources() + + # Create manifest + manifest = Manifest( + identifier=hex_to_qti_id(self.ccnode.content_id), + version="1.0", + metadata=Metadata(schema="QTI Package", schemaversion="3.0.0"), + resources=Resources(resources=resources), + ) + + xml_content = manifest.to_xml_string() + return f'\n{xml_content}' + + def handle_after_assessment_items(self): + # Create and write the IMS manifest + manifest_xml = self._create_imsmanifest() + self.add_file_to_write("imsmanifest.xml", manifest_xml.encode("utf-8")) + # Sort all paths to parallel the predictable zip generation logic in ricecooker + # and the Kolibri Studio frontend. + self.files_to_write = sorted(self.files_to_write) diff --git a/contentcuration/contentcuration/utils/assessment/qti/assessment_item.py b/contentcuration/contentcuration/utils/assessment/qti/assessment_item.py new file mode 100644 index 0000000000..830044ae79 --- /dev/null +++ b/contentcuration/contentcuration/utils/assessment/qti/assessment_item.py @@ -0,0 +1,237 @@ +from typing import Annotated +from typing import List +from typing import Optional +from typing import Union + +from annotated_types import Len +from pydantic import AnyUrl +from pydantic import Field +from pydantic import model_validator +from pydantic import PositiveInt + +from contentcuration.utils.assessment.qti.base import BaseSequence +from contentcuration.utils.assessment.qti.base import QTIBase +from contentcuration.utils.assessment.qti.base import TextType +from contentcuration.utils.assessment.qti.constants import BaseType +from contentcuration.utils.assessment.qti.constants import Cardinality +from contentcuration.utils.assessment.qti.constants import ExternalScored +from contentcuration.utils.assessment.qti.constants import View +from contentcuration.utils.assessment.qti.fields import BCP47Language +from contentcuration.utils.assessment.qti.fields import LocalHrefPath +from contentcuration.utils.assessment.qti.fields import QTIIdentifier +from contentcuration.utils.assessment.qti.html import BlockContentElement +from contentcuration.utils.assessment.qti.interaction_types.base import BlockInteraction + + +class Value(QTIBase): + """ + Represents a single value within a default value, correct response, + or other value container. + + For record values, both the field-identifier and base-type attributes + are required to identify which field of the record this value belongs to + and what type that field is. + + For non-record values (single, multiple, ordered cardinality), these + attributes are optional and typically not needed as the base-type is + determined by the parent variable declaration. + """ + + value: TextType # The actual value content + field_identifier: Optional[QTIIdentifier] = None # Required only for record values + base_type: Optional[BaseType] = None # Required only for record values + + +ValueType = Annotated[List[Value], Len(min_length=1)] + + +class CorrectResponse(QTIBase): + """Defines the correct response for the interaction.""" + + value: ValueType = Field(default_factory=list) + + +class DefaultValue(QTIBase): + """ + Defines the default value for a variable. Contains one or more + value elements depending on the cardinality of the variable. + """ + + value: ValueType = Field(default_factory=list) + # Human readable interpretation of the default value + interpretation: Optional[str] = None + + +def _validate_value(self, attribute_name="default_value"): + attr_value = getattr(self, attribute_name) + if attr_value is not None: + if self.cardinality == Cardinality.SINGLE: + # Single cardinality should have exactly one value + if len(attr_value.value) > 1: + raise ValueError( + f"Single cardinality cannot have multiple {attribute_name.replace('_', ' ')}s" + ) + elif self.cardinality == Cardinality.RECORD: + # Record cardinality requires field identifiers + for value in attr_value.value: + if not value.field_identifier: + raise ValueError( + f"Record cardinality requires field_identifier in {attribute_name.replace('_', ' ')}" + ) + if not value.base_type: + raise ValueError( + f"Record cardinality requires base_type in {attribute_name.replace('_', ' ')}" + ) + + +class OutcomeDeclaration(QTIBase): + """ + QTI outcome declaration defines an outcome variable, which represents the + result of response processing. Outcomes are typically scores but can also + be other results such as feedback identifiers or completion status. + """ + + identifier: QTIIdentifier + cardinality: Cardinality = Cardinality.SINGLE + base_type: Optional[BaseType] = None + view: Optional[View] = None + interpretation: Optional[AnyUrl] = None + long_interpretation: Optional[str] = None + normal_maximum: Optional[PositiveInt] = None + normal_minimum: Optional[float] = None + mastery_value: Optional[float] = None + external_scored: Optional[ExternalScored] = None + variable_identifier_ref: Optional[str] = None + default_value: Optional[DefaultValue] = None + + @model_validator(mode="after") + def validate_cardinality_compatibility(self): + _validate_value(self) + return self + + +class ItemBody(QTIBase, BaseSequence): + """Contains the content of the assessment item""" + + children: List[Union[BlockInteraction, BlockContentElement]] = Field( + default_factory=list + ) + + +class ContextDeclaration(QTIBase): + """ + QTI context declaration defines a 'contextual' variable with global scope to + an assessment item. Context variables provide contextual information to + template processing and response processing, such as candidate information, + test information, and environment information. + """ + + identifier: QTIIdentifier + cardinality: Cardinality + base_type: Optional[BaseType] = None + default_value: Optional[DefaultValue] = None + + @model_validator(mode="after") + def validate_cardinality_compatibility(self): + _validate_value(self) + return self + + +class MapEntry(QTIBase): + """Entry in a mapping that maps a specific value to a score""" + + # Key (usually an identifier) + map_key: str + # Value to map + mapped_value: float + # Whether string comparison is case sensitive + case_sensitive: bool = False + + +class Mapping(QTIBase): + """Maps response values to scores for partial credit scoring""" + + map_entries: List[MapEntry] = Field(default_factory=list) + # Score for responses not explicitly mapped + default_value: float = 0.0 + # Lower bound for mapping results + lower_bound: Optional[float] = None + # Upper bound for mapping results + upper_bound: Optional[float] = None + + +class AreaMapEntry(QTIBase): + """Entry in an area mapping that maps a specific area to a score""" + + # Shape of the area (rect, circle, poly, default) + shape: str + # Coordinates defining the area + coords: str + # Score for responses in this area + mapped_value: float + + +class AreaMapping(QTIBase): + """Maps areas to scores for graphical interactions""" + + area_map_entries: List[AreaMapEntry] = Field(default_factory=list) + # Score for responses not in any defined area + default_value: float = 0.0 + # Lower bound for mapping results + lower_bound: Optional[float] = None + # Upper bound for mapping results + upper_bound: Optional[float] = None + + +class ResponseDeclaration(QTIBase): + """ + QTI response declaration defines a response variable and optionally its + correct response value and/or mapping. Response variables capture candidate + interactions with the assessment item's interactions and are used in response + processing to determine outcomes. + """ + + identifier: QTIIdentifier + cardinality: Cardinality + base_type: BaseType + correct_response: Optional[CorrectResponse] = None + mapping: Optional[Mapping] = None + area_mapping: Optional[AreaMapping] = None + + @model_validator(mode="after") + def validate_cardinality_compatibility(self): + _validate_value(self, "correct_response") + return self + + +class ResponseProcessing(QTIBase): + """Represents response processing rules or template reference""" + + # URI reference to a response processing template + template: Optional[AnyUrl] = None + # Optional URL that resolves to the template - we additionally enforce that this be local + # although this is not required by the QTI spec + template_location: Optional[LocalHrefPath] = None + # rules deliberately not implemented yet + + +class AssessmentItem(QTIBase): + """Represents a QTI assessment item""" + + xmlns: str = "http://www.imsglobal.org/xsd/imsqtiasi_v3p0" + xmlns__xsi: str = "http://www.w3.org/2001/XMLSchema-instance" + xsi__schemaLocation: str = "http://www.imsglobal.org/xsd/imsqtiasi_v3p0 https://purl.imsglobal.org/spec/qti/v3p0/schema/xsd/imsqti_asiv3p0p1_v1p0.xsd" + identifier: QTIIdentifier + title: str + label: Optional[str] = None + adaptive: bool = False + time_dependent: Optional[bool] = None + language: BCP47Language + tool_name: str = "kolibri" + tool_version: str = "0.1" + + context_declaration: List[ContextDeclaration] = Field(default_factory=list) + response_declaration: List[ResponseDeclaration] = Field(default_factory=list) + outcome_declaration: List[OutcomeDeclaration] = Field(default_factory=list) + item_body: Optional[ItemBody] = None + response_processing: Optional[ResponseProcessing] = None diff --git a/contentcuration/contentcuration/utils/assessment/qti/base.py b/contentcuration/contentcuration/utils/assessment/qti/base.py new file mode 100644 index 0000000000..5467654a6b --- /dev/null +++ b/contentcuration/contentcuration/utils/assessment/qti/base.py @@ -0,0 +1,274 @@ +import re +import xml.etree.ElementTree as ET +from abc import ABC +from enum import Enum +from functools import partial +from typing import Annotated +from typing import List +from typing import Optional +from typing import Set +from typing import Type +from typing import Union + +from pydantic import BaseModel +from pydantic import BeforeValidator +from pydantic import ConfigDict +from pydantic import PrivateAttr + +from .constants import Dir +from .fields import entry_pattern as srcset_entry_pattern +from contentcuration.utils.assessment.qti.fields import BCP47Language + + +class TextNode(BaseModel): + """Class to represent text nodes within XML elements""" + + text: str + + +class XMLElement(BaseModel, ABC): + """Base class for XML elements""" + + # Pydantic configuration + model_config = ConfigDict( + # Prevent extra fields + extra="forbid", + validate_assignment=True, + # Prevent mutations to ensure immutability + frozen=True, + ) + + # Private attributes (not included in Pydantic fields) + _file_dependencies: Set[str] = PrivateAttr(default_factory=set) + _element: ET.Element = PrivateAttr(default=None) + + @classmethod + def element_name(cls): + return cls.__name__.lower() + + def to_element(self) -> ET.Element: # noqa: C901 + if self._element: + return self._element + + element = ET.Element(self.element_name()) + + self._file_dependencies = set() + + # Add attributes based on pydantic fields + for field_name in self.__class__.model_fields: + + value = getattr(self, field_name) + + # Skip None values + if value is None: + continue + + if isinstance(value, (XMLElement, TextNode)): + value = [value] + + if isinstance(value, list): + if all(isinstance(item, (XMLElement, TextNode)) for item in value): + for item in value: + if isinstance(item, XMLElement): + child_elements = item.to_element() + if not isinstance(child_elements, list): + child_elements = [child_elements] + for child_element in child_elements: + element.append(child_element) + self._file_dependencies |= item._file_dependencies + else: + current_children = list(element) + if current_children: + current_children[-1].tail = ( + current_children[-1].tail or "" + ) + item.text + else: + element.text = (element.text or "") + item.text + + continue + raise ValueError( + "List types should only contain XMLElement or TextNodes" + ) + + elif isinstance(value, bool): + value = str(value).lower() + + elif isinstance(value, Enum): + # Handle enum values + value = value.value + + # Some attribute names are reserved Python keywords or Python builtins + # to allow this, we allow a trailing underscore which we strip here. + # All attributes use kebab-case, which we can't easily use as field names + # so we encode them as snake_case and convert to kebab-case here. + # Some attributes also include : which we encode as double underscore. + attr_name = field_name.rstrip("_").replace("__", ":").replace("_", "-") + + # Set the attribute + element.set(attr_name, str(value)) + + if attr_name == "src" or attr_name == "href": + self._file_dependencies.add(value) + elif attr_name == "srcset": + entries = re.findall(srcset_entry_pattern, value) + for entry in entries: + # Each entry is a tuple of (url, descriptors) + url = entry[0].strip() + self._file_dependencies.add(url) + + self._element = element + + return self._element + + def to_xml_string(self) -> str: + """Convert to XML string""" + element = self.to_element() + return ET.tostring(element, encoding="unicode") + + def get_file_dependencies(self) -> List[str]: + # Ensure the element has been processed so that the file dependencies are collected. + self.to_element() + return list(self._file_dependencies) + + +class QTIBase(XMLElement): + """ + A base class to allow us to conventionally generate element names from class names for QTI elements. + """ + + @classmethod + def element_name(cls): + # Convert PascalCase to kebab-case + name = re.sub(r"(?<=[a-z])(?=[A-Z])", "-", cls.__name__) + return f"qti-{name.lower()}" + + +def coerce_str_to_model(element_type, value: Union[str, XMLElement]) -> XMLElement: + """Convert string to element_type if needed""" + if isinstance(value, str): + return element_type(text=value) + return value + + +def generate_coerced_string_type(element_type): + return Annotated[ + element_type, BeforeValidator(partial(coerce_str_to_model, element_type)) + ] + + +TextType = generate_coerced_string_type(TextNode) + + +class BaseSequence(XMLElement): + id_: Optional[str] = None + class_: Optional[str] = None + lang: Optional[BCP47Language] = None + # We explicitly do not set the deprecated language value. + label: Optional[str] = None + # We explicitly do not set the base value. + dir_: Optional[Dir] = None + + +# Pydantic's BaseModel Metaclass is only importable from an internal module, +# so we inspect the BaseSequence class to get its metaclass. +BaseSequenceMetaclass = type(BaseSequence) + + +class RegistryMeta(BaseSequenceMetaclass): + """Generic metaclass that creates separate registries for each subclass""" + + def __new__(mcs, name, bases, attrs): + cls = super().__new__(mcs, name, bases, attrs) + + # Each metaclass gets its own registry + if not hasattr(mcs, "_registry"): + mcs._registry = {} + + element_name = cls.element_name() + if element_name in mcs._registry and mcs._registry[element_name] is not cls: + raise ValueError( + f"Element name '{element_name}' already registered in {mcs.__name__}" + ) + mcs._registry[element_name] = cls + + return cls + + @classmethod + def _ensure_registry_complete(cls): + """Ensure all HTML and MathML classes are registered""" + if not hasattr(cls, "_registry_initialized"): + # Import modules to trigger registration + from contentcuration.utils.assessment.qti import html, mathml # noqa: F401 + + cls._registry_initialized = True + + @classmethod + def get_class_for_tag(cls, tag_name: str) -> Optional[Type]: + """Get the registered class for a given tag name""" + cls._ensure_registry_complete() + return getattr(cls, "_registry", {}).get(tag_name) + + +class ElementTreeBase(BaseSequence, metaclass=RegistryMeta): + @classmethod + def from_element(cls, element: ET.Element) -> "ElementTreeBase": + # Get the appropriate class for this tag + target_class = type(cls).get_class_for_tag(element.tag) + if target_class is None: + raise ValueError(f"No registered class found for tag: {element.tag}") + + # Convert attributes to field data - Pydantic will handle type coercion + field_data = {} + for attr_name, attr_value in element.attrib.items(): + field_name = cls._attr_name_to_field_name(attr_name) + field_data[field_name] = attr_value + + # Convert children and text + children = cls._extract_children(element) + if children: + field_data["children"] = children + + return target_class(**field_data) + + @classmethod + def _attr_name_to_field_name(cls, attr_name: str) -> str: + """Convert attribute name to Python field name""" + # kebab-case -> snake_case, : -> __ + field_name = attr_name.replace(":", "__").replace("-", "_") + + # Add trailing underscore for Python keywords + if field_name in {"class", "for", "type", "id", "dir"}: + field_name += "_" + + return field_name + + @classmethod + def _extract_children( + cls, element: ET.Element + ) -> List[Union["ElementTreeBase", TextNode]]: + """Extract child elements and text nodes from XML element""" + children = [] + + # Add initial text if present + if element.text and element.text.strip(): + children.append(TextNode(text=element.text)) + + # Process child elements + for child_elem in element: + children.append(cls.from_element(child_elem)) + # Add tail text after child element + if child_elem.tail and child_elem.tail.strip(): + children.append(TextNode(text=child_elem.tail)) + + return children + + @classmethod + def from_string(cls, string: str) -> List["ElementTreeBase"]: + """Parse markup string and return list of ElementTreeBase instances""" + try: + # Wrap in a root element to handle multiple top-level elements + wrapped_markup = f"