canvas: citation urls for html content (#2521)

shanbady · web-flow · commit b5fcde369443 · 2025-09-19T12:44:55.000-04:00
* updates to support urls for non-file items

* adding canvas type

* working urls for assignments and pages

* adding title to contentfile and test for urls

* update test

* titles for embedded pages
diff --git a/learning_resources/etl/canvas.py b/learning_resources/etl/canvas.py
@@ -110,10 +110,14 @@ def _get_url_config(bucket, export_tempdir: str, url_config_file: str) -> dict:
     bucket.download_file(url_config_file, url_config_path)
     url_config = {}
     with Path.open(url_config_path, "rb") as f:
-        for url_item in json.loads(f.read().decode("utf-8")).get("course_files", []):
+        url_json = json.loads(f.read().decode("utf-8"))
+        for url_item in url_json.get("course_files", []):
             url_key = url_item["file_path"]
             url_key = unquote_plus(url_key.lstrip(url_key.split("/")[0]))
             url_config[url_key] = url_item["url"]
+        for url_item in url_json.get("assignments", []) + url_json.get("pages", []):
+            url_key = url_item.get("name", url_item.get("title"))
+            url_config[url_key] = url_item.get("html_url")
     return url_config
 
 
@@ -199,22 +203,23 @@ def transform_canvas_content_files(
     """
     basedir = course_zipfile.name.split(".")[0]
     zipfile_path = course_zipfile.absolute()
-    # grab published module and file items
-    published_items = [
-        Path(item["path"]).resolve()
-        for item in parse_module_meta(zipfile_path)["active"]
-    ] + [
-        Path(item["path"]).resolve()
-        for item in parse_files_meta(zipfile_path)["active"]
-    ]
-    for item in parse_web_content(zipfile_path)["active"]:
-        published_items.append(Path(item["path"]).resolve())
-        published_items.extend(
-            [
-                Path(embedded_file).resolve()
-                for embedded_file in item.get("embedded_files", [])
-            ]
-        )
+    all_published_items = (
+        parse_module_meta(zipfile_path)["active"]
+        + parse_files_meta(zipfile_path)["active"]
+        + parse_web_content(zipfile_path)["active"]
+    )
+    published_items = {}
+    for item in all_published_items:
+        path = Path(item["path"]).resolve()
+        published_items[path] = item
+        for embedded_file in item.get("embedded_files", []):
+            embedded_path = Path(embedded_file).resolve()
+            if embedded_path in all_published_items:
+                continue
+            published_items[embedded_path] = {
+                "path": embedded_path,
+                "title": "",
+            }
 
     def _generate_content():
         """Inner generator for yielding content data"""
@@ -223,7 +228,8 @@ def _generate_content():
             zipfile.ZipFile(zipfile_path, "r") as course_archive,
         ):
             for member in course_archive.infolist():
-                if Path(member.filename).resolve() in published_items:
+                member_path = Path(member.filename).resolve()
+                if member_path in published_items:
                     course_archive.extract(member, path=olx_path)
                     log.debug("processing active file %s", member.filename)
                 else:
@@ -233,7 +239,14 @@ def _generate_content():
                 url_path = content_data["source_path"].lstrip(
                     content_data["source_path"].split("/")[0]
                 )
-                content_url = url_config.get(url_path, "")
+                item_meta = published_items.get(
+                    Path(content_data["source_path"]).resolve(), {}
+                )
+
+                content_url = url_config.get(url_path) or url_config.get(
+                    item_meta.get("title")
+                )
+                content_data["content_title"] = item_meta.get("title")
                 if content_url:
                     content_data["url"] = content_url
                 yield content_data
@@ -561,6 +574,19 @@ def _title_from_html(html: str) -> str:
     return title.get_text().strip() if title else ""
 
 
+def _title_from_assignment_settings(xml_string: str) -> str:
+    """
+    Extract the title from assignment_settings.xml
+    """
+    try:
+        root = ElementTree.fromstring(xml_string)
+    except Exception:
+        log.exception("Error parsing XML: %s", sys.stderr)
+        return ""
+    title_elem = root.find("cccv1p0:title", NAMESPACES)
+    return title_elem.text.strip() if title_elem is not None and title_elem.text else ""
+
+
 def parse_web_content(course_archive_path: str) -> dict:
     """
     Parse html pages and assignments and return publish/active status of resources
@@ -588,9 +614,12 @@ def parse_web_content(course_archive_path: str) -> dict:
                 if assignment_settings:
                     xml_content = course_archive.read(assignment_settings)
                     workflow_state = _workflow_state_from_xml(xml_content)
+                    title = _title_from_assignment_settings(xml_content)
+                    canvas_type = "assignment"
                 else:
                     workflow_state = _workflow_state_from_html(html_content)
-                title = _title_from_html(html_content)
+                    title = _title_from_html(html_content)
+                    canvas_type = "page"
 
                 lom_elem = (
                     resource_map_item.get("metadata", {})
@@ -606,6 +635,7 @@ def parse_web_content(course_archive_path: str) -> dict:
                         {
                             "title": title,
                             "path": file_path,
+                            "canvas_type": canvas_type,
                             "embedded_files": embedded_files,
                         }
                     )
@@ -614,6 +644,7 @@ def parse_web_content(course_archive_path: str) -> dict:
                         {
                             "title": title,
                             "path": file_path,
+                            "canvas_type": canvas_type,
                             "embedded_files": embedded_files,
                         }
                     )
@@ -635,7 +666,7 @@ def extract_resources_by_identifierref(manifest_xml: str) -> dict:
         title = (
             item.find("imscp:title", NAMESPACES).text
             if item.find("imscp:title", NAMESPACES) is not None
-            else "No Title"
+            else ""
         )
         resource = root.find(
             f'.//imscp:resource[@identifier="{identifierref}"]', NAMESPACES
diff --git a/learning_resources/etl/canvas_test.py b/learning_resources/etl/canvas_test.py
@@ -195,7 +195,7 @@ def test_parse_module_meta_returns_active_and_unpublished(tmp_path):
     """
     module_xml = b"""<?xml version="1.0" encoding="UTF-8"?>
     <modules xmlns="http://canvas.instructure.com/xsd/cccv1p0">
-      <module>
+      <module identifier="g31fce32d15019702eb75d5664798a877">
         <title>Module 1</title>
         <items>
           <item>
@@ -981,3 +981,101 @@ def test_embedded_files_from_html(tmp_path, mocker):
     )
 
     assert any(file["source_path"] == "web_resources/html_page.html" for file in files)
+
+
+def test_get_url_config_assignments_and_pages(mocker, tmp_path):
+    """
+    Test that _get_url_config correctly maps assignments and pages to URLs using their titles.
+    """
+    hmtl_page_title = "html page"
+
+    url_config = {
+        hmtl_page_title: "https://example.com/htmlpage",
+        "/file1.html": "https://example.com/file1",
+    }
+
+    run = LearningResourceRunFactory.create()
+
+    html_content = f"""
+    <html>
+    <head><meta name="workflow_state" content="active"/><title>{hmtl_page_title}</title></head>
+        <body>
+
+        </body>
+    </html>
+    """
+    module_xml = b"""<?xml version="1.0" encoding="UTF-8"?>
+    <modules xmlns="http://canvas.instructure.com/xsd/cccv1p0"
+    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+    >
+      <module>
+        <title>Module 1</title>
+        <items>
+          <item>
+            <workflow_state>active</workflow_state>
+            <title>Item 1</title>
+            <hidden>false</hidden>
+             <locked>false</locked>
+            <identifierref>RES1</identifierref>
+            <content_type>resource</content_type>
+          </item>
+        </items>
+      </module>
+    </modules>
+    """
+    manifest_xml = b"""<?xml version="1.0" encoding="UTF-8"?>
+    <manifest xmlns="http://www.imsglobal.org/xsd/imsccv1p1/imscp_v1p1">
+      <resources>
+      <resource identifier="RES1" identifierref="RES1" type="webcontent" href="web_resources/file1.html">
+          <file href="web_resources/file1.html"/>
+        </resource>
+        <resource identifier="RES2" type="webcontent" href="web_resources/file2.html">
+          <file href="web_resources/file2.html"/>
+        </resource>
+        <resource identifier="RES3" type="webcontent" href="web_resources/html_page.html">
+          <file href="web_resources/html_page.html"/>
+        </resource>
+      </resources>
+      <organizations>
+        <organization>
+          <item identifierref="RES1">
+            <title>module 1</title>
+          </item>
+
+        </organization>
+      </organizations>
+    </manifest>
+    """
+    zip_path = make_canvas_zip_with_module_meta(tmp_path, module_xml, manifest_xml)
+    with zipfile.ZipFile(zip_path, "a") as zf:
+        zf.writestr("web_resources/file1.html", "content of file1")
+        zf.writestr("web_resources/file2.html", "content of file2")
+        zf.writestr("web_resources/html_page.html", html_content)
+    mocker.patch(
+        "learning_resources.etl.utils.extract_text_metadata",
+        return_value={"content": "test"},
+    )
+    mocker.patch("learning_resources.etl.canvas.bulk_resources_unpublished_actions")
+    results = list(
+        transform_canvas_content_files(
+            Path(zip_path),
+            run=run,
+            url_config=url_config,
+            overwrite=False,
+        )
+    )
+    results = list(results)
+    list(
+        zip(
+            [result["content_title"] for result in results],
+            [result["url"] for result in results],
+        )
+    )
+    results = dict(
+        zip(
+            [result["content_title"] for result in results],
+            [result["url"] for result in results],
+        )
+    )
+    assert results["html page"] == "https://example.com/htmlpage"
+    assert results["Item 1"] == "https://example.com/file1"
