XBlock asset support for new libraries experience (WIP)

cms/djangoapps/contentstore/asset_storage_handlers.py

@@ -467,6 +467,7 @@ def _get_assets_in_json_format(assets, course_key, assets_usage_locations_map):
             course_key,
             asset_file_size,
             usage_locations,
+            asset['import_path'],
         )
 
         assets_in_json_format.append(asset_in_json)
@@ -709,7 +710,7 @@ def _delete_thumbnail(thumbnail_location, course_key, asset_key):  # lint-amnest
 
 
 def get_asset_json(display_name, content_type, date, location, thumbnail_location,
-                   locked, course_key, file_size=None, usage=None):
+                   locked, course_key, file_size=None, usage=None, import_path=None):
     '''
     Helper method for formatting the asset information to send to client.
     '''
@@ -731,4 +732,6 @@ def get_asset_json(display_name, content_type, date, location, thumbnail_locatio
         'id': str(location),
         'file_size': file_size,
         'usage_locations': usage_locations,
+        'import_path': import_path,
+        'expanded_path': import_path if import_path else display_name,
     }

cms/djangoapps/contentstore/helpers.py

@@ -3,6 +3,7 @@
 """
 from __future__ import annotations
 import logging
+import pathlib
 import urllib
 from lxml import etree
 from mimetypes import guess_type
@@ -270,8 +271,8 @@ def import_staged_content_from_user_clipboard(parent_key: UsageKey, request) ->
     if not user_clipboard:
         # Clipboard is empty or expired/error/loading
         return None, StaticFileNotices()
+
     olx_str = content_staging_api.get_staged_content_olx(user_clipboard.content.id)
-    static_files = content_staging_api.get_staged_content_static_files(user_clipboard.content.id)
     node = etree.fromstring(olx_str)
     store = modulestore()
     with store.bulk_operations(parent_key.course_key):
@@ -287,16 +288,40 @@ def import_staged_content_from_user_clipboard(parent_key: UsageKey, request) ->
             copied_from_block=str(user_clipboard.source_usage_key),
             tags=user_clipboard.content.tags,
         )
-    # Now handle static files that need to go into Files & Uploads:
-    notices = _import_files_into_course(
-        course_key=parent_key.context_key,
-        staged_content_id=user_clipboard.content.id,
-        static_files=static_files,
-    )
+
+        # Now handle static files that need to go into Files & Uploads.
+        static_files = content_staging_api.get_staged_content_static_files(user_clipboard.content.id)
+        notices, substitutions = _import_files_into_course(
+            course_key=parent_key.context_key,
+            staged_content_id=user_clipboard.content.id,
+            static_files=static_files,
+            usage_key=new_xblock.scope_ids.usage_id,
+        )
+
+        if hasattr(new_xblock, 'data') and substitutions:
+            data_with_substitutions = new_xblock.data
+
+            print(f"Substitutions: {substitutions}")
+
+            for old_static_ref, new_static_ref in substitutions.items():
+                data_with_substitutions = data_with_substitutions.replace(
+                    old_static_ref,
+                    new_static_ref,
+                )
+
+            print(f"data_with_substitutions: {data_with_substitutions}")
+
+            new_xblock.data = data_with_substitutions
+            store.update_item(new_xblock, request.user.id)
 
     return new_xblock, notices
 
 
+def _generate_usage_key(block_type: str) -> UsageKey:
+
+    pass
+
+
 def _import_xml_node_to_parent(
     node,
     parent_xblock: XBlock,
@@ -415,7 +440,8 @@ def _import_files_into_course(
     course_key: CourseKey,
     staged_content_id: int,
     static_files: list[content_staging_api.StagedContentFileData],
-) -> StaticFileNotices:
+    usage_key: UsageKey,
+) -> (StaticFileNotices, dict[str, str]):
     """
     For the given staged static asset files (which are in "Staged Content" such as the user's clipbaord, but which
     need to end up in the course's Files & Uploads page), import them into the destination course, unless they already
@@ -427,66 +453,100 @@ def _import_files_into_course(
     conflicting_files = []
     # List of files that had an error (shouldn't happen unless we have some kind of bug)
     error_files = []
+
+    # Store a mapping of asset URLs that need to be modified for the destination
+    # assets. This is necessary when you take something from a library and paste
+    # it into a course, because we need to translate Component-local static
+    # assets and shove them into the Course's global Files & Uploads space in a
+    # nested directory structure.
+    substitutions = {}
     for file_data_obj in static_files:
-        if not isinstance(file_data_obj.source_key, AssetKey):
+#        if not isinstance(file_data_obj.source_key, AssetKey):
             # This static asset was managed by the XBlock and instead of being added to "Files & Uploads", it is stored
             # using some other system. We could make it available via runtime.resources_fs during XML parsing, but it's
             # not needed here.
-            continue
+#            continue
         # At this point, we know this is a "Files & Uploads" asset that we may need to copy into the course:
         try:
-            result = _import_file_into_course(course_key, staged_content_id, file_data_obj)
+            result, substitution_for_file = _import_file_into_course(
+                course_key,
+                staged_content_id,
+                file_data_obj,
+                usage_key,
+            )
             if result is True:
                 new_files.append(file_data_obj.filename)
+                substitutions.update(substitution_for_file)
             elif result is None:
                 pass  # This file already exists; no action needed.
             else:
                 conflicting_files.append(file_data_obj.filename)
         except Exception:  # lint-amnesty, pylint: disable=broad-except
             error_files.append(file_data_obj.filename)
             log.exception(f"Failed to import Files & Uploads file {file_data_obj.filename}")
-    return StaticFileNotices(
+    notices = StaticFileNotices(
         new_files=new_files,
         conflicting_files=conflicting_files,
         error_files=error_files,
     )
+    return notices, substitutions
 
 
 def _import_file_into_course(
     course_key: CourseKey,
     staged_content_id: int,
     file_data_obj: content_staging_api.StagedContentFileData,
-) -> bool | None:
+    usage_key: UsageKey,
+) -> (bool | None, dict):
     """
     Import a single staged static asset file into the course, unless it already exists.
     Returns True if it was imported, False if there's a conflict, or None if
     the file already existed (no action needed).
     """
-    filename = file_data_obj.filename
-    new_key = course_key.make_asset_key("asset", filename)
+    # If this came from a library (need to adjust this condition later)
+    clipboard_file_path = file_data_obj.filename
+    if clipboard_file_path.startswith('static/'):
+        file_path = clipboard_file_path.lstrip('static/')
+        import_path = f"components/{usage_key.block_type}/{usage_key.block_id}/{file_path}"
+        filename = pathlib.Path(file_path).name
+        new_key = course_key.make_asset_key("asset", import_path.replace("/", "_"))
+    else:
+        file_path = clipboard_file_path
+        import_path = None
+        filename = pathlib.Path(file_path).name
+        new_key = course_key.make_asset_key("asset", file_path.replace("/", "_"))
+
+    # Yeah, I'd prefer a different delimiter, but this is what we already do
+    # during file import.
     try:
         current_file = contentstore().find(new_key)
     except NotFoundError:
         current_file = None
     if not current_file:
         # This static asset should be imported into the new course:
         content_type = guess_type(filename)[0]
-        data = content_staging_api.get_staged_content_static_file_data(staged_content_id, filename)
+        data = content_staging_api.get_staged_content_static_file_data(staged_content_id, clipboard_file_path)
         if data is None:
             raise NotFoundError(file_data_obj.source_key)
-        content = StaticContent(new_key, name=filename, content_type=content_type, data=data)
+        content = StaticContent(
+            new_key,
+            name=filename,
+            content_type=content_type,
+            data=data,
+            import_path=import_path
+        )
         # If it's an image file, also generate the thumbnail:
         thumbnail_content, thumbnail_location = contentstore().generate_thumbnail(content)
         if thumbnail_content is not None:
             content.thumbnail_location = thumbnail_location
         contentstore().save(content)
-        return True
+        return True, {clipboard_file_path: f"static/{import_path}"}
     elif current_file.content_digest == file_data_obj.md5_hash:
         # The file already exists and matches exactly, so no action is needed
-        return None
+        return None, {}
     else:
         # There is a conflict with some other file that has the same name.
-        return False
+        return False, {}
 
 
 def is_item_in_course_tree(item):