diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index 99282179..e36e7c17 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -12,7 +12,7 @@ repos:
exclude: '^tests/testcontent/samples/'
- id: check-yaml
- id: check-added-large-files
- exclude: '^tests/(cassettes|testcontent/samples)/'
+ exclude: '^tests/(cassettes|testcontent/(samples|imscp))/'
- id: debug-statements
- id: end-of-file-fixer
exclude: '(^.+?\.json$|^tests/testcontent/samples/)'
diff --git a/ricecooker/classes/files.py b/ricecooker/classes/files.py
index 5dfc811c..65583217 100644
--- a/ricecooker/classes/files.py
+++ b/ricecooker/classes/files.py
@@ -70,7 +70,9 @@ def __init__(
duration=None,
original_filename=None,
filename=None,
+ **kwargs,
):
+ # Extra keys are tolerated so a pipeline file-metadata dict splats in.
self.preset = preset
self.set_language(language)
self.default_ext = default_ext or self.default_ext
diff --git a/ricecooker/classes/nodes.py b/ricecooker/classes/nodes.py
index 94c05c51..b51427fe 100644
--- a/ricecooker/classes/nodes.py
+++ b/ricecooker/classes/nodes.py
@@ -182,10 +182,48 @@ def __init__(
self.learner_needs = learner_needs or []
self.role = role
+ self._set_license_fields(license, copyright_holder, license_description)
+
+ # License data lives on the node's License object, so it cannot be setattr'd.
+ _LICENSE_METADATA_KEYS = ("license", "copyright_holder", "license_description")
+
+ def _set_license_fields(
+ self, license=None, copyright_holder=None, license_description=None
+ ):
+ """Set the node's License, filling unsupplied fields from its current one."""
+ current = self.license
+ if license is None:
+ license = current
+ if isinstance(license, License) and (copyright_holder or license_description):
+ # Rights without a license type refine the existing license.
+ license = license.license_id
+ if current is not None:
+ copyright_holder = copyright_holder or current.copyright_holder
+ license_description = license_description or current.description
self.set_license(
license, copyright_holder=copyright_holder, description=license_description
)
+ def set_metadata(self, metadata):
+ """Apply constructor-style metadata fields to an already-built node.
+
+ Only supplied keys are touched: ``extra_fields`` merges rather than
+ replaces, license fields go through ``set_license``.
+ """
+ metadata = dict(metadata)
+ self.extra_fields.update(metadata.pop("extra_fields", None) or {})
+ if "language" in metadata:
+ self.set_language(metadata.pop("language"))
+ license_fields = {
+ key: metadata.pop(key)
+ for key in self._LICENSE_METADATA_KEYS
+ if key in metadata
+ }
+ if license_fields:
+ self._set_license_fields(**license_fields)
+ for key, value in metadata.items():
+ setattr(self, key, value)
+
def set_language(self, language):
"""Set self.language to internal lang. repr. code from str or Language object."""
if isinstance(language, str):
@@ -700,6 +738,80 @@ class TreeNode(Node):
See Node for inherited attributes.
"""
+ # Content-node metadata keys describing a node's shape, not its own fields.
+ STRUCTURAL_METADATA_KEYS = frozenset({"children", "files", "kind"})
+
+ # Fields inherited from the subtree root when the metadata is silent.
+ INHERITED_METADATA_KEYS = ("language",)
+
+ @classmethod
+ def own_metadata_fields(cls, metadata):
+ """The node's own fields in ``metadata``, less the keys describing its shape."""
+ return {
+ key: value
+ for key, value in metadata.items()
+ if key not in cls.STRUCTURAL_METADATA_KEYS
+ }
+
+ @classmethod
+ def from_metadata(cls, metadata, **inherited):
+ """Build a node, and its descendants, from a content-node metadata dict.
+
+ ``kind`` picks the class: absent or ``topic`` gives a folder, anything else
+ a leaf. Every other non-structural key is a constructor argument.
+ """
+ node_class = (
+ TopicNode
+ if metadata.get("kind", content_kinds.TOPIC) == content_kinds.TOPIC
+ else ContentNode
+ )
+ kwargs = {
+ key: value
+ for key, value in inherited.items()
+ if key in node_class.INHERITED_METADATA_KEYS
+ }
+ kwargs.update(cls.own_metadata_fields(metadata))
+ node = node_class(**kwargs)
+ # kind is a class attribute, not a constructor argument.
+ node.kind = metadata.get("kind") or node_class.kind
+ node.add_metadata_content(metadata, inherited)
+ return node
+
+ def add_metadata_children(self, metadata, inherited):
+ """Attach the descendants ``metadata`` describes."""
+ for child_metadata in metadata.get("children") or []:
+ self.add_child(self.from_metadata(child_metadata, **inherited))
+
+ def add_metadata_content(self, metadata, inherited):
+ """Attach what ``metadata`` hangs off this node. A folder takes children."""
+ self.add_metadata_children(metadata, inherited)
+
+ def expand_metadata_tree(self, metadata):
+ """Become the folder of the subtree ``metadata`` describes.
+
+ A decomposer (IMSCP/SCORM) returns a tree of content-node metadata in place
+ of a file, so the node it was declared on holds the tree as descendants.
+ """
+ self.kind = content_kinds.TOPIC
+ self.set_metadata(self.own_metadata_fields(metadata))
+ self.add_metadata_children(
+ metadata, {"language": self.language, "license": self.license}
+ )
+ for child in self.children:
+ child.process_metadata_subtree()
+
+ def process_metadata_subtree(self):
+ """Process and validate a metadata-built node and its descendants, leaves first.
+
+ Descendants must process themselves: ``ChannelManager.process_tree``
+ snapshots the node list before calling ``process_files()``, so nodes added
+ during that call are never reached by its walk.
+ """
+ for child in self.children:
+ child.process_metadata_subtree()
+ self.process_files()
+ self.validate()
+
def get_domain_namespace(self):
if not self.domain_ns:
self.domain_ns = self.parent.get_domain_namespace()
@@ -886,6 +998,10 @@ def _validate_uri(self):
def _validate(self):
"""Validate the content node. Raises InvalidNodeException on failure; returns None."""
+ # A node expanded into a folder has no files, license or uri of its own.
+ if self.kind == content_kinds.TOPIC:
+ super(ContentNode, self)._validate()
+ return
self._validate_values(self.license is None, "ContentNode must have a license")
if self._files_processed:
self._validate_values(self.kind is None, "No kind has been set")
@@ -925,6 +1041,21 @@ def _validate(self):
self._validate_uri()
super(ContentNode, self)._validate()
+ # A leaf inherits the package's license as well as its language.
+ INHERITED_METADATA_KEYS = ("language", "license")
+
+ def _file_from_metadata(self, metadata):
+ """Build a File from pipeline file metadata, inheriting the node's language.
+
+ The pipeline's own language wins where it inferred one (e.g. a subtitle's).
+ """
+ return File(**{"language": self.language, **metadata})
+
+ def add_metadata_content(self, metadata, inherited):
+ """A leaf is backed by its own processed files, not by descendants."""
+ for file_metadata in metadata.get("files") or []:
+ self.add_file(self._file_from_metadata(file_metadata))
+
def _process_uri(self):
try:
file_metadata_list = self.pipeline.execute(
@@ -934,26 +1065,28 @@ def _process_uri(self):
config.LOGGER.error(f"Error processing path: {self.uri} with error: {e}")
return None
content_metadata = {}
+ file_metadata_dicts = []
for file_metadata in file_metadata_list:
metadata_dict = file_metadata.to_dict()
if "content_node_metadata" in metadata_dict:
content_metadata.update(metadata_dict.pop("content_node_metadata"))
- # Remove path from metadata_dict as it is not needed for the File object
- metadata_dict.pop("path", None)
- # Inherit the node's language unless the pipeline inferred one of its own
- # (e.g. a subtitle language extracted from the file itself).
- metadata_dict.setdefault("language", self.language)
- file_obj = File(**metadata_dict)
- self.add_file(file_obj)
- for key, value in content_metadata.items():
- if key == "extra_fields":
- self.extra_fields.update(value)
- else:
- if key == "kind" and self.kind is not None and self.kind != value:
- raise InvalidNodeException(
- "Inferred kind is different from content node class kind."
- )
- setattr(self, key, value)
+ file_metadata_dicts.append(metadata_dict)
+ # Children ⇒ this node is a decomposed subtree's folder; its files belong
+ # to the leaves. Test ``is not None``: a decomposer emits [] when every
+ # resource was rejected, and the extract stage may have overwritten kind.
+ if content_metadata.get("children") is not None:
+ self.expand_metadata_tree(content_metadata)
+ return
+ for metadata_dict in file_metadata_dicts:
+ self.add_file(self._file_from_metadata(metadata_dict))
+ if (
+ self.kind is not None
+ and content_metadata.get("kind", self.kind) != self.kind
+ ):
+ raise InvalidNodeException(
+ "Inferred kind is different from content node class kind."
+ )
+ self.set_metadata(content_metadata)
def process_files(self):
if self.uri:
diff --git a/ricecooker/utils/SCORM_metadata.py b/ricecooker/utils/SCORM_metadata.py
new file mode 100644
index 00000000..c1074888
--- /dev/null
+++ b/ricecooker/utils/SCORM_metadata.py
@@ -0,0 +1,261 @@
+"""Map the raw LOM metadata dict :mod:`ricecooker.utils.imscp` extracts onto
+le_utils content-node fields. Ported from ricecooker PR #468.
+"""
+
+import logging
+import re
+
+from le_utils.constants import licenses
+from le_utils.constants.labels import learning_activities
+from le_utils.constants.labels import needs
+from le_utils.constants.labels import resource_type
+
+from ricecooker.utils.youtube import get_language_with_alpha2_fallback
+
+LOGGER = logging.getLogger(__name__)
+
+# Node validation rejects longer tags, and LOM keywords are routinely whole
+# phrases, so over-long ones are dropped rather than fail the channel.
+MAX_TAG_LENGTH = 30
+
+
+# LOM educational learningResourceType -> (le_utils learning activity,
+# educator-focused le_utils resource type). One table so the two vocabularies
+# cannot drift apart as terms are added.
+_LEARNING_RESOURCE_TYPE_MAPPINGS = {
+ "exercise": (learning_activities.PRACTICE, resource_type.EXERCISE),
+ "simulation": (learning_activities.EXPLORE, resource_type.ACTIVITY),
+ "questionnaire": (learning_activities.PRACTICE, resource_type.ACTIVITY),
+ "diagram": (learning_activities.EXPLORE, resource_type.MEDIA),
+ "figure": (learning_activities.EXPLORE, resource_type.MEDIA),
+ "graph": (learning_activities.EXPLORE, resource_type.MEDIA),
+ "index": (learning_activities.READ, resource_type.GUIDE),
+ "slide": (learning_activities.READ, resource_type.LESSON),
+ "table": (learning_activities.READ, resource_type.TUTORIAL),
+ "narrative text": (learning_activities.READ, resource_type.TEXTBOOK),
+ "exam": (learning_activities.PRACTICE, resource_type.EXERCISE),
+ "experiment": (learning_activities.EXPLORE, resource_type.ACTIVITY),
+ "problem statement": (learning_activities.REFLECT, resource_type.ACTIVITY),
+ "self assessment": (learning_activities.REFLECT, resource_type.ACTIVITY),
+ "lecture": (learning_activities.WATCH, resource_type.LESSON),
+}
+
+
+def _text_list(value):
+ """Flatten a LOM field (a string, a list, or a list of lists) to its text values.
+
+ Non-text entries are dropped: an empty LOM element parses to ``None``, and a
+ repeated element parses to a nested list.
+ """
+ if isinstance(value, str):
+ return [value]
+ if not isinstance(value, (list, tuple)):
+ return []
+ texts = []
+ for item in value:
+ if isinstance(item, str):
+ texts.append(item)
+ elif isinstance(item, (list, tuple)):
+ texts.extend(t for t in item if isinstance(t, str))
+ return texts
+
+
+def _first_text(value):
+ """The first non-blank text value of a LOM field, or None when it has none."""
+ return next((t for t in _text_list(value) if t.strip()), None)
+
+
+def map_scorm_to_le_utils_activities(metadata_dict):
+ interactivity_type = _first_text(metadata_dict.get("interactivityType"))
+ interactivity_level = _first_text(metadata_dict.get("interactivityLevel"))
+ is_interactive = interactivity_type in (
+ "active",
+ "mixed",
+ ) or interactivity_level in ("medium", "high")
+
+ activities = []
+ for lrt in _text_list(metadata_dict.get("learningResourceType")):
+ mapping = _LEARNING_RESOURCE_TYPE_MAPPINGS.get(lrt)
+ activity = mapping[0] if mapping else None
+ # A non-interactive resource cannot be explored: read a simulation of
+ # one, watch the rest.
+ if activity == learning_activities.EXPLORE and not is_interactive:
+ activity = (
+ learning_activities.READ
+ if lrt == "simulation"
+ else learning_activities.WATCH
+ )
+ if activity and activity not in activities:
+ activities.append(activity)
+
+ return activities
+
+
+# LOM intendedEndUserRole -> educator resource type, when the role is an educator.
+SCORM_intended_role_to_resource_type_mapping = {
+ "teacher": resource_type.LESSON_PLAN,
+ "author": resource_type.GUIDE,
+ "manager": resource_type.GUIDE,
+}
+
+
+def map_scorm_to_educator_resource_types(metadata_dict):
+ lrt_types = {
+ lrt: mapping[1] for lrt, mapping in _LEARNING_RESOURCE_TYPE_MAPPINGS.items()
+ }
+ types = []
+ for key, mapping in (
+ ("learningResourceType", lrt_types),
+ ("intendedEndUserRole", SCORM_intended_role_to_resource_type_mapping),
+ ):
+ for text in _text_list(metadata_dict.get(key)):
+ mapped = mapping.get(text)
+ if mapped and mapped not in types:
+ types.append(mapped)
+
+ return types
+
+
+def infer_beginner_level_from_difficulty(metadata_dict):
+ if _first_text(metadata_dict.get("difficulty")) in ("very easy", "easy"):
+ return [needs.FOR_BEGINNERS]
+ return []
+
+
+def _vcard_field(vcard_text, field):
+ """The value of VCARD ``field`` (``FN``, ``ORG``) in ``vcard_text``, or None."""
+ match = re.search(r"^{}:(.+)$".format(field), vcard_text or "", re.MULTILINE)
+ return match.group(1).strip() if match else None
+
+
+# CC licence patterns, ordered most specific first to avoid partial matches.
+_CC_LICENSE_PATTERNS = [
+ ("Attribution-NonCommercial-ShareAlike", licenses.CC_BY_NC_SA),
+ ("Attribution-NonCommercial-NoDerivs", licenses.CC_BY_NC_ND),
+ ("Attribution-ShareAlike", licenses.CC_BY_SA),
+ ("Attribution-NoDerivs", licenses.CC_BY_ND),
+ ("Attribution-NonCommercial", licenses.CC_BY_NC),
+ ("Attribution", licenses.CC_BY),
+]
+
+
+def infer_license_from_rights(metadata_dict):
+ """Infer a ``(license_id, license_description)`` tuple from rights metadata.
+
+ Either or both may be ``None``.
+ """
+ description = _first_text(metadata_dict.get("rights_description"))
+ copyright_restrictions = _first_text(
+ metadata_dict.get("copyrightAndOtherRestrictions")
+ )
+
+ if description:
+ for pattern, license_id in _CC_LICENSE_PATTERNS:
+ if pattern in description:
+ return license_id, description
+
+ if copyright_restrictions == "no":
+ return licenses.PUBLIC_DOMAIN, description
+
+ return None, description
+
+
+# LOM contribute role -> (result field name, whether to prefer ORG over FN).
+_ROLE_TO_FIELD = {
+ "author": ("author", False),
+ "publisher": ("provider", True),
+ "content provider": ("copyright_holder", True),
+}
+
+
+def extract_lifecycle_contributors(metadata_dict):
+ """Extract author/provider/copyright_holder from lifeCycle contribute data."""
+ result = {}
+ contribute = metadata_dict.get("contribute")
+ if not contribute:
+ return result
+
+ if isinstance(contribute, dict):
+ contribute = [contribute]
+
+ for entry in contribute:
+ role_value = entry.get("role", {})
+ if isinstance(role_value, dict):
+ role_value = role_value.get("value", "")
+ role_value = _first_text(role_value)
+ entity = _first_text(entry.get("entity")) or ""
+
+ field_config = _ROLE_TO_FIELD.get(role_value)
+ if not field_config:
+ continue
+ field_name, prefer_org = field_config
+ name = (prefer_org and _vcard_field(entity, "ORG")) or _vcard_field(
+ entity, "FN"
+ )
+ if name:
+ result[field_name] = name
+
+ return result
+
+
+def _normalize_language(lang_code):
+ """Normalize a language code, returning None if unrecognized."""
+ language = get_language_with_alpha2_fallback(lang_code) if lang_code else None
+ return language.code if language else None
+
+
+def _normalize_keywords(keyword):
+ """Normalize the keyword field to a list of usable tags, or None if empty."""
+ tags = []
+ for keyword_text in _text_list(keyword):
+ if not keyword_text:
+ continue
+ if len(keyword_text) > MAX_TAG_LENGTH:
+ LOGGER.warning(
+ "SCORM: dropping keyword longer than %s characters: %s",
+ MAX_TAG_LENGTH,
+ keyword_text,
+ )
+ else:
+ tags.append(keyword_text)
+ return tags or None
+
+
+def _drop_unattributable_license(fields):
+ """Drop an inferred license requiring a copyright holder LOM did not name.
+
+ Applying it would fail node validation, so the chef's license is kept.
+ """
+ # Imported here: ricecooker.classes imports the pipeline, which imports this.
+ from ricecooker.classes.licenses import get_license
+
+ license_id = fields.get("license")
+ if not license_id or fields.get("copyright_holder"):
+ return
+ if get_license(license_id).require_copyright_holder:
+ LOGGER.warning(
+ "SCORM: ignoring inferred %s license, no copyright holder was named",
+ license_id,
+ )
+ del fields["license"]
+
+
+def metadata_dict_to_content_node_fields(metadata_dict):
+ """Convert a raw LOM metadata dict to ``ContentNodeMetadata`` fields, dropping empties."""
+ license_id, license_description = infer_license_from_rights(metadata_dict)
+ fields = {
+ # LOM may supply one per language; these fields are single-valued.
+ "title": _first_text(metadata_dict.get("title")),
+ "description": _first_text(metadata_dict.get("description")),
+ "language": _normalize_language(_first_text(metadata_dict.get("language"))),
+ "tags": _normalize_keywords(metadata_dict.get("keyword", [])),
+ "learning_activities": map_scorm_to_le_utils_activities(metadata_dict),
+ "resource_types": map_scorm_to_educator_resource_types(metadata_dict),
+ "learner_needs": infer_beginner_level_from_difficulty(metadata_dict),
+ "license": license_id,
+ "license_description": license_description,
+ **extract_lifecycle_contributors(metadata_dict),
+ }
+ fields = {key: value for key, value in fields.items() if value}
+ _drop_unattributable_license(fields)
+ return fields
diff --git a/ricecooker/utils/imscp.py b/ricecooker/utils/imscp.py
new file mode 100644
index 00000000..31ac7949
--- /dev/null
+++ b/ricecooker/utils/imscp.py
@@ -0,0 +1,457 @@
+"""Parse an extracted IMS Content Package (``imsmanifest.xml``) into a tree of dicts.
+
+Ported from ``learningequality/imscp`` ``core.py`` to stdlib
+:mod:`xml.etree.ElementTree`. Manifests declare varied default namespaces
+(``imscp_rootv1p1p2``, ``imscp_v1p1``), so every ``find``/``findall`` uses a
+``{*}`` wildcard rather than a fixed namespace map.
+"""
+
+import io
+import logging
+import os
+import posixpath
+import re
+import shutil
+from collections import deque
+from urllib.parse import unquote
+from xml.etree import ElementTree as ET
+
+import chardet
+
+from ricecooker.utils.references import DEFAULT_MAPPERS
+from ricecooker.utils.references import is_data_uri
+from ricecooker.utils.references import is_external_url
+from ricecooker.utils.SCORM_metadata import metadata_dict_to_content_node_fields
+
+LOGGER = logging.getLogger(__name__)
+
+XML_BASE = "{http://www.w3.org/XML/1998/namespace}base"
+XML_LANG = "{http://www.w3.org/XML/1998/namespace}lang"
+
+QTI_RESOURCE_TYPE_PREFIX = "imsqti_"
+
+# The IMS Content Package manifest, always at the root of the package.
+IMSCP_MANIFEST = "imsmanifest.xml"
+
+# LOM sections and the fields lifted out of each, keyed by LOM element name.
+LOM_METADATA_KEYS = {
+ "general": ["title", "description", "language", "keyword"],
+ "rights": ["cost", "copyrightAndOtherRestrictions", "description"],
+ "educational": [
+ "interactivityType",
+ "interactivityLevel",
+ "learningResourceType",
+ "intendedEndUserRole",
+ "difficulty",
+ ],
+ "lifeCycle": ["contribute"],
+}
+
+
+def is_qti_resource(resource_type):
+ """True when ``resource_type`` names a QTI resource (spec-defined ``imsqti_`` prefix)."""
+ return bool(resource_type) and resource_type.startswith(QTI_RESOURCE_TYPE_PREFIX)
+
+
+def parse_imscp_manifest(ims_dir):
+ """Parse ``imsmanifest.xml`` in ``ims_dir`` into the manifest tree.
+
+ Returns ``{"identifier", "title", "metadata", "children": [node, ...]}``
+ where each ``node`` is a topic (``{"source_id", "title", "children"}``) or a
+ webcontent leaf (``{"source_id", "title", "type", "index_file", "href",
+ "scormtype", "files"}``). ``files`` are archive-member paths relative to
+ ``ims_dir``.
+ """
+ root = _read_manifest(os.path.join(ims_dir, IMSCP_MANIFEST))
+
+ metadata = collect_metadata(root, ims_dir)
+
+ resources = {
+ r.get("identifier"): r for r in root.findall("{*}resources/{*}resource")
+ }
+
+ children = []
+ for org in root.findall("{*}organizations/{*}organization"):
+ node = _walk_items(org, ims_dir)
+ _collect_resources(node, resources)
+ children.append(flatten_single_child_topics(node))
+
+ return {
+ "identifier": root.get("identifier"),
+ "title": metadata.get("title"),
+ "metadata": metadata,
+ "children": children,
+ }
+
+
+def _read_manifest(manifest_path):
+ """Parse the manifest, falling back to detected encoding on a parse error."""
+ try:
+ return ET.parse(manifest_path).getroot()
+ except ET.ParseError:
+ # Some manifests declare UTF-8 but contain other-encoded bytes; detect the
+ # real encoding, decode, and re-parse from re-encoded UTF-8 bytes.
+ with open(manifest_path, "rb") as f:
+ data = f.read()
+ encoding = chardet.detect(data)["encoding"]
+ if encoding is None:
+ # Nothing to re-decode from; the manifest is simply not parseable.
+ raise
+ return ET.parse(io.BytesIO(data.decode(encoding).encode("utf-8"))).getroot()
+
+
+def _strip_ns(key):
+ """Strip a ``{namespace}`` prefix off an attribute key."""
+ return re.sub(r"^\{.*\}", "", key)
+
+
+def _element_text(elem):
+ """Concatenate all descendant text/tail (ignoring `` ``), stripped."""
+ if elem is None:
+ return ""
+ return "".join(elem.itertext()).strip()
+
+
+def contained_path(root, member):
+ """Resolve ``member`` under ``root``; return the path, or None if it escapes.
+
+ Manifest hrefs, file paths and metadata locations are all untrusted.
+ """
+ root_abs = os.path.abspath(root)
+ target = os.path.abspath(os.path.join(root_abs, member))
+ if target != root_abs and not target.startswith(root_abs + os.sep):
+ return None
+ return target
+
+
+def _lom_text(elem):
+ """The stripped text of a LOM element, or None when empty."""
+ return (
+ elem.text.strip()
+ if elem is not None and elem.text and elem.text.strip()
+ else None
+ )
+
+
+def _extract_lom_text(elem, preferred_language):
+ """Read text from a LOM field, handling its several shapes.
+
+ Handles ````/```` (returning a
+ preferred-language match, the single value, or a list), ``/``
+ pairs, and bare element text.
+ """
+ strings = elem.findall("{*}string") or elem.findall("{*}langstring")
+ if strings:
+ if preferred_language is not None:
+ for s in strings:
+ lang = s.get("language", "") or s.get(XML_LANG, "")
+ if lang.startswith(preferred_language):
+ return _lom_text(s)
+ if len(strings) == 1:
+ return _lom_text(strings[0])
+ return [_lom_text(s) for s in strings]
+
+ # A vocabulary term is ``[]term...``; recurse so the term
+ # text is read, not the whitespace around it.
+ value = elem.find("{*}value")
+ if value is not None:
+ return _extract_lom_text(value, preferred_language)
+
+ return _lom_text(elem)
+
+
+def _extract_contribute(contrib_elem):
+ """Extract a lifeCycle ```` entry as ``{"role", "entity"}``."""
+ result = {}
+ role = contrib_elem.find("{*}role")
+ if role is not None:
+ # The role vocabulary term sits in ```` (bare or langstring-wrapped).
+ role_value = _extract_lom_text(role, None)
+ if role_value:
+ result["role"] = {"value": role_value}
+ entity = contrib_elem.find("{*}entity")
+ if entity is not None and entity.text:
+ result["entity"] = entity.text
+ return result
+
+
+def _get_lom_section(metadata_elem, tag):
+ """The LOM ```` section, whether wrapped in ```` or bare."""
+ section = metadata_elem.find("{*}lom/{*}" + tag)
+ if section is not None:
+ return section
+ return metadata_elem.find("{*}" + tag)
+
+
+def _detect_language(metadata_elem):
+ """The preferred language declared in LOM ````."""
+ general = _get_lom_section(metadata_elem, "general")
+ if general is not None:
+ return _lom_text(general.find("{*}language"))
+ return None
+
+
+def _resolve_metadata_elem(elem, ims_dir):
+ """The ```` of ``elem``, following an external ``adlcp:location`` ref."""
+ metadata_elem = elem.find("{*}metadata")
+ if metadata_elem is None:
+ return None
+ location = metadata_elem.find("{*}location")
+ if location is not None and location.text:
+ ext_path = contained_path(ims_dir, location.text.strip())
+ if ext_path and os.path.isfile(ext_path):
+ try:
+ return ET.parse(ext_path).getroot()
+ except ET.ParseError:
+ LOGGER.warning(
+ "IMSCP: could not parse external metadata %s", location.text
+ )
+ return metadata_elem
+
+
+def _collect_field(section, field, preferred_language):
+ """The value of LOM ``field`` in ``section``: scalar when single, list when repeated."""
+ elems = section.findall("{*}" + field)
+ if not elems:
+ return None
+ if field == "contribute":
+ values = [_extract_contribute(e) for e in elems]
+ else:
+ values = [_extract_lom_text(e, preferred_language) for e in elems]
+ return values[0] if len(values) == 1 else values
+
+
+def collect_metadata(elem, ims_dir):
+ """Extract the raw LOM metadata dict from ``elem``'s ````.
+
+ Covers the sections named in :data:`LOM_METADATA_KEYS`; mapping onto
+ content-node fields is :mod:`ricecooker.utils.SCORM_metadata`'s job.
+ """
+ metadata_elem = _resolve_metadata_elem(elem, ims_dir)
+ if metadata_elem is None:
+ return {}
+
+ preferred_language = _detect_language(metadata_elem)
+
+ metadata = {}
+ for tag, fields in LOM_METADATA_KEYS.items():
+ section = _get_lom_section(metadata_elem, tag)
+ if section is None:
+ continue
+ for field in fields:
+ value = _collect_field(section, field, preferred_language)
+ if value is not None:
+ # Prefix rights fields so ``rights/description`` does not collide
+ # with ``general/description``.
+ key = "rights_" + field if tag == "rights" else field
+ metadata[key] = value
+ return metadata
+
+
+def _walk_items(elem, ims_dir):
+ """Build an item/topic dict from ``elem`` and recurse into child ````s."""
+ node = {_strip_ns(k): v for k, v in elem.attrib.items()}
+
+ title = _element_text(elem.find("{*}title"))
+ if title:
+ node["title"] = title
+
+ # A child element, not an attribute, so the attrib copy above misses it.
+ mastery = _element_text(elem.find("{*}masteryscore"))
+ if mastery:
+ node["masteryscore"] = mastery
+
+ metadata = collect_metadata(elem, ims_dir)
+ if metadata:
+ node["metadata"] = metadata
+
+ children = [_walk_items(item, ims_dir) for item in elem.findall("{*}item")]
+ if children:
+ node["children"] = children
+
+ return node
+
+
+def _collect_resources(item, resources, index=1):
+ """Resolve resource references onto leaf items; recurse into topics.
+
+ ``index`` is the item's 1-based sibling position, for the ``item{n}``
+ source_id fallback when its identifier is blank.
+ """
+ item["source_id"] = item.get("identifier") or "item{}".format(index)
+
+ children = item.get("children")
+ if children:
+ for child_index, child in enumerate(children, start=1):
+ _collect_resources(child, resources, child_index)
+ elif item.get("identifierref"):
+ resource = resources.get(item["identifierref"])
+ if resource is None:
+ LOGGER.warning(
+ "IMSCP: item %s references missing resource %s",
+ item["source_id"],
+ item["identifierref"],
+ )
+ return
+ # The item's own attributes win — a resource carries its own
+ # ``identifier``, which must not displace the item's identity.
+ for key, value in resource.attrib.items():
+ item.setdefault(_strip_ns(key), value)
+ resource_type = resource.get("type")
+ # QTI resources are rejected downstream, but deriving their files keeps
+ # the leaf self-describing. Unknown resource types are left as-is.
+ if resource_type == "webcontent" or is_qti_resource(resource_type):
+ href = resource.get("href")
+ if href:
+ # Must carry the same ``xml:base`` offset _derive_files applies
+ # to the members, or it resolves to no real extracted path.
+ item["index_file"] = (resource.get(XML_BASE) or "") + href
+ item["files"] = _derive_files(resource, resources)
+ item.setdefault("scormtype", None)
+
+
+def _derive_files(resource, resources, seen=None, visited=None):
+ """Own ```` members plus flattened ```` files, order-preserving."""
+ if seen is None:
+ seen = set()
+ # Track resources already on the dependency chain so a cyclic
+ # (A→B→A, possible in a malformed/untrusted manifest) cannot recurse forever.
+ if visited is None:
+ visited = set()
+ identifier = resource.get("identifier")
+ if identifier in visited:
+ return []
+ visited.add(identifier)
+
+ base = resource.get(XML_BASE) or ""
+ files = []
+ for fe in resource.findall("{*}file"):
+ href = fe.get("href")
+ if not href:
+ continue
+ path = base + href
+ if path not in seen:
+ seen.add(path)
+ files.append(path)
+
+ for dep in resource.findall("{*}dependency"):
+ dep_ref = dep.get("identifierref")
+ dep_resource = resources.get(dep_ref)
+ if dep_resource is None:
+ LOGGER.warning(
+ "IMSCP: resource %s depends on missing resource %s",
+ identifier,
+ dep_ref,
+ )
+ continue
+ files.extend(_derive_files(dep_resource, resources, seen, visited))
+
+ return files
+
+
+# Keys naming a node's identity/shape; LOM descriptive metadata must not
+# overwrite them.
+_NODE_IDENTITY_KEYS = frozenset({"source_id", "title", "kind", "children", "files"})
+
+
+def lom_content_fields(node_dict):
+ """Map a parsed node's raw LOM ``metadata`` to content-node fields."""
+ return metadata_dict_to_content_node_fields(node_dict.get("metadata") or {})
+
+
+def merge_lom_fields(built, fields):
+ """Copy non-identity LOM-derived ``fields`` onto a built tree dict."""
+ for key, value in fields.items():
+ if key not in _NODE_IDENTITY_KEYS:
+ built.setdefault(key, value)
+
+
+def flatten_single_child_topics(node):
+ """Collapse a topic whose only child is itself a topic into that child.
+
+ IMS packages routinely wrap the whole tree in an ```` holding one
+ content-root ````. Leaf-only topics are left untouched. From PR #468.
+ """
+ children = node.get("children")
+ if not children:
+ return node
+
+ node["children"] = [flatten_single_child_topics(child) for child in children]
+
+ if len(node["children"]) == 1 and node["children"][0].get("children"):
+ only_child = node["children"][0]
+ if not only_child.get("title"):
+ only_child["title"] = node.get("title")
+ if not only_child.get("metadata") and node.get("metadata"):
+ only_child["metadata"] = node["metadata"]
+ return only_child
+
+ return node
+
+
+class IMSCPPackage:
+ """An extracted package, staging each resource into its own directory.
+
+ A resource's ```` list is under-declared often enough that the assets
+ its members reference are staged too, bounded to files present in the package.
+ Navigation links are not followed, so a leaf never absorbs what it links to.
+ """
+
+ def __init__(self, directory):
+ self.directory = directory
+ # Shared assets are staged into many leaves; the package never changes.
+ self._references = {}
+
+ def stage(self, members, dest_dir):
+ """Copy ``members`` and their reference closure into ``dest_dir``, paths preserved."""
+ staged = set()
+ pending = deque()
+ for member in members:
+ self._stage_member(member, dest_dir, staged, pending)
+ while pending:
+ member, mapper = pending.popleft()
+ member_dir = posixpath.dirname(member)
+ for ref in self._member_references(member, mapper):
+ self._stage_member(
+ posixpath.join(member_dir, ref), dest_dir, staged, pending
+ )
+
+ def _stage_member(self, member, dest_dir, staged, pending):
+ member = posixpath.normpath(member.replace("\\", "/"))
+ if member in staged:
+ return
+ # Staging the manifest would make the leaf a package, decomposing forever.
+ if member == IMSCP_MANIFEST:
+ return
+ # Manifest paths are untrusted: reject a ``../`` escape either way.
+ src = contained_path(self.directory, member)
+ dst = contained_path(dest_dir, member)
+ if src is None or dst is None or not os.path.isfile(src):
+ return
+ os.makedirs(os.path.dirname(dst), exist_ok=True)
+ shutil.copyfile(src, dst)
+ staged.add(member)
+ mapper = next((m for m in DEFAULT_MAPPERS if m.handles(member)), None)
+ if mapper is not None:
+ pending.append((member, mapper))
+
+ def _member_references(self, member, mapper):
+ """The package-local paths an HTML/CSS ``member`` references."""
+ if member not in self._references:
+ self._references[member] = self._extract_references(member, mapper)
+ return self._references[member]
+
+ def _extract_references(self, member, mapper):
+ try:
+ with open(contained_path(self.directory, member), encoding="utf-8") as fh:
+ content = fh.read()
+ except (OSError, UnicodeDecodeError):
+ return []
+ refs = []
+ for ref in mapper.extract(content):
+ if is_external_url(ref) or is_data_uri(ref):
+ continue
+ ref = unquote(ref.split("#")[0].split("?")[0])
+ if ref:
+ refs.append(ref)
+ return refs
diff --git a/ricecooker/utils/pipeline/context.py b/ricecooker/utils/pipeline/context.py
index 386e177b..13d37cac 100644
--- a/ricecooker/utils/pipeline/context.py
+++ b/ricecooker/utils/pipeline/context.py
@@ -32,9 +32,13 @@ class ContentNodeMetadata:
accessibility_labels: Optional[list[str]] = None
learner_needs: Optional[list[str]] = None
role: Optional[str] = None
+ language: Optional[str] = None
+ tags: Optional[list[str]] = None
source_id: Optional[str] = None
kind: Optional[str] = None
extra_fields: Optional[dict] = None
+ children: Optional[list] = None
+ files: Optional[list] = None
def _recursive_update(target, source):
diff --git a/ricecooker/utils/pipeline/convert.py b/ricecooker/utils/pipeline/convert.py
index 53d8fb5f..5abe9381 100644
--- a/ricecooker/utils/pipeline/convert.py
+++ b/ricecooker/utils/pipeline/convert.py
@@ -21,6 +21,7 @@
import filetype
import html5lib
from html5lib.html5parser import ParseError
+from le_utils.constants import content_kinds
from le_utils.constants import file_formats
from le_utils.constants import format_presets
from PIL import Image
@@ -33,15 +34,28 @@
from ricecooker.utils.audio import AudioCompressionError
from ricecooker.utils.audio import compress_audio
from ricecooker.utils.caching import generate_key
+from ricecooker.utils.imscp import contained_path
+from ricecooker.utils.imscp import IMSCP_MANIFEST
+from ricecooker.utils.imscp import IMSCPPackage
+from ricecooker.utils.imscp import is_qti_resource
+from ricecooker.utils.imscp import lom_content_fields
+from ricecooker.utils.imscp import merge_lom_fields
+from ricecooker.utils.imscp import parse_imscp_manifest
from ricecooker.utils.paths import extract_path_ext
from ricecooker.utils.pipeline.context import ContentNodeMetadata
from ricecooker.utils.pipeline.context import ContextMetadata
from ricecooker.utils.pipeline.context import FileMetadata
+from ricecooker.utils.pipeline.exceptions import ExpectedFileException
from ricecooker.utils.pipeline.exceptions import InvalidFileException
from ricecooker.utils.references import DEFAULT_MAPPERS
from ricecooker.utils.references import ReferenceMapper
from ricecooker.utils.references import sanitize_style_css
from ricecooker.utils.references import strip_scripts
+from ricecooker.utils.references import strip_stylesheet_links
+from ricecooker.utils.scorm import boilerplate_script_members
+from ricecooker.utils.scorm import has_assessment_semantics
+from ricecooker.utils.scorm import single_media_member
+from ricecooker.utils.scorm import strip_scorm_boilerplate
from ricecooker.utils.subtitles import build_subtitle_converter_from_file
from ricecooker.utils.subtitles import InvalidSubtitleFormatError
from ricecooker.utils.subtitles import InvalidSubtitleLanguageError
@@ -71,21 +85,29 @@ class PandocConversionError(Exception):
"""Raised when pandoc fails to convert a source document."""
-def sanitize_kpub_directory(temp_dir):
- """Strip disallowed CSS and scripts from index.html in an extracted KPUB dir, in place."""
- index_path = os.path.join(temp_dir, "index.html")
- try:
- with open(index_path, encoding="utf-8") as fh:
- html = fh.read()
- except (OSError, UnicodeDecodeError):
- return
+def sanitize_kpub_html(html):
+ """Strip disallowed CSS and scripts from a KPUB entry document.
+
+ Returns ``(html, removed)`` — descriptors of what was stripped, empty if unchanged.
+ """
html, removed = sanitize_style_css(html, KPUB_STYLE_ALLOWLIST)
# Hand-authored KPUBs already reject scripts in validate_archive; strip_scripts
# is here for the pandoc path, whose --standalone template can inject an html5shiv.
html, script_removed = strip_scripts(html)
- removed += script_removed
+ return html, removed + script_removed
+
+
+def sanitize_kpub_directory(temp_dir, entry="index.html"):
+ """Sanitize a KPUB's entry document in place."""
+ entry_path = os.path.join(temp_dir, entry)
+ try:
+ with open(entry_path, encoding="utf-8") as fh:
+ html = fh.read()
+ except (OSError, UnicodeDecodeError):
+ return
+ html, removed = sanitize_kpub_html(html)
if removed:
- with open(index_path, "w", encoding="utf-8") as fh:
+ with open(entry_path, "w", encoding="utf-8") as fh:
fh.write(html)
LOGGER.info("KPUB sanitizer removed disallowed content: %s", ", ".join(removed))
@@ -245,11 +267,16 @@ def FILE_TYPE(self) -> str:
def validate_archive(self, path: str):
pass
- def pre_process(self, temp_dir):
+ def pre_process(self, temp_dir, entry):
"""Hook run on the extracted archive dir before reference resolution. Default no-op."""
pass
- def handle_file(self, path, audio_settings=None, video_settings=None):
+ def seal_ext(self, temp_dir, ext, entry=None):
+ """Extension the processed dir is sealed as. Override to re-classify the output."""
+ return ext
+
+ def _convert_archive(self, path, audio_settings, video_settings, entry=None):
+ """Validate, extract, process and seal the archive at ``path``."""
# Imported here rather than at module level: archive_assets depends on
# this package's exceptions, so a top-level import would be circular.
from ricecooker.utils.archive_assets import ArchiveProcessor
@@ -266,7 +293,7 @@ def handle_file(self, path, audio_settings=None, video_settings=None):
# pre_process runs before reference resolution: a url() inside a