Assign urls to edx contentfiles when possible

Author: mbertrand

learning_resources/etl/utils.py

@@ -18,10 +18,12 @@
 from pathlib import Path
 from subprocess import check_call
 from tempfile import TemporaryDirectory
+from typing import Optional
 
 import boto3
 import rapidjson
 import requests
+from defusedxml.ElementTree import ParseError, parse
 from django.conf import settings
 from django.utils.dateparse import parse_duration
 from django.utils.text import slugify
@@ -341,7 +343,7 @@ def documents_from_olx(
                         "mime_type": mimetype,
                         "archive_checksum": archive_checksum,
                         "file_extension": extension_lower,
-                        "source_path": f"{path}/{filename}",
+                        "source_path": f"{path}/{filename.replace(' ', '_')}",
                     },
                 )
 
@@ -407,7 +409,150 @@ def text_from_sjson_content(content: str):
     return " ".join(data.get("text", []))
 
 
+def get_root_url_for_source(etl_source: str) -> tuple[str, str]:
+    """
+    Get the base URL and path for an ETL source
+
+    Args:
+        etl_source (str): The ETL source path
+
+    Returns:
+        tuple[str, str]: The base URL and path
+    """
+    mapping = {
+        ETLSource.mitxonline.value: "https://courses.mitxonline.mit.edu",
+        ETLSource.xpro.value: "https://courses.xpro.mit.edu",
+        ETLSource.mit_edx.value: "https://www.edx.org",
+        ETLSource.oll.value: "https://openlearninglibrary.mit.edu",
+    }
+    return mapping.get(etl_source)
+
+
+def is_valid_uuid(uuid_string: str) -> bool:
+    """
+    Check if a string is a valid UUID
+    """
+    try:
+        uuid.UUID(uuid_string)
+    except ValueError:
+        return False
+    return True
+
+
+def get_url_from_module_id(
+    olx_path: str,
+    module_id: str,
+    run: LearningResourceRun,
+    assets_metadata: Optional[dict] = None,
+    video_srt_metadata: Optional[dict] = None,
+) -> str:
+    """
+    Get the URL for a module based on its ID
+
+    Args:
+        module_id (str): The module ID
+        run (LearningResourceRun): The run associated with the module
+
+    Returns:
+        str: The URL for the module
+    """
+    if not module_id:
+        log.warning("Module ID is empty")
+        return None
+    root_url = get_root_url_for_source(run.learning_resource.etl_source)
+    with Path.open("video_metadata.json", "w") as f:
+        json.dump(video_srt_metadata, f, indent=2)
+    if module_id.startswith("asset"):
+        log.info("Getting URL for asset %s", module_id)
+        asset_meta = (
+            assets_metadata.get(Path(olx_path).parts[-1], {}) if assets_metadata else {}
+        )
+        video_meta = video_srt_metadata.get(module_id, {}) if video_srt_metadata else {}
+        if video_meta:
+            log.info("Found video metadata for %s", module_id)
+            return f"{root_url}/xblock/{video_meta}"
+        elif module_id.endswith(".srt"):
+            log.info("NO VIDEO METADATA FOR %s", module_id)
+        middle_path = asset_meta.get("custom_md5", "")
+        return f"{root_url}/{(middle_path + '/') if middle_path else ''}{module_id}"
+    elif module_id.startswith("block") and is_valid_uuid(module_id.split("@")[-1]):
+        return f"{root_url}/xblock/{module_id}"
+    else:
+        log.warning("Unknown module ID format: %s", module_id)
+        return None
+
+
+def get_assets_metadata(olx_path: str) -> dict:
+    """
+    Get metadata for assets in an OLX path
+
+    Args:
+        olx_path (str): The path to the OLX directory
+    """
+    try:
+        with Path.open(Path(olx_path, "policies/assets.json"), "rb") as f:
+            return json.loads(f.read())
+    except FileNotFoundError:
+        log.warning("Assets metadata file does not exist: %s", olx_path)
+
+
+def parse_video_transcripts_xml(
+    run: LearningResourceRun, xml_content: str, path: Path
+) -> dict:
+    """
+    Parse video XML content and create a mapping of
+    transcript edx_module_id to video edx_module_id
+    """
+    transcript_mapping = {}
+    try:
+        root = parse(xml_content)
+
+        # Get the video url_name from the root video element
+        video_url_name = root.get("url_name")
+        if not video_url_name:
+            log.warning("No url_name found in video XML")
+            return {}
+
+        # Find all transcript elements and extract their src attributes
+        for transcript in root.findall(".//transcript"):
+            transcript_src = transcript.get("src")
+            if transcript_src:
+                transcript_mapping[
+                    get_edx_module_id(f"static/{transcript_src}", run)
+                ] = get_edx_module_id(str(path), run)
+    except ParseError:
+        log.exception("Error parsing video XML for %s:  %s", run, path)
+    return transcript_mapping
+
+
+def get_video_metadata(olx_path: str, run: LearningResourceRun) -> dict:
+    """
+    Get metadata for video SRT files in an OLX path
+    """
+    video_transcript_mapping = {}
+    video_path = Path(olx_path, "video")
+    if not video_path.exists():
+        log.warning("No video directory found in OLX path: %s", olx_path)
+        return video_transcript_mapping
+    for root, _, files in os.walk(str(Path(olx_path, "video"))):
+        path = "/".join(root.split("/")[3:])
+        for filename in files:
+            log.info("Processing video file %s in %s", filename, path)
+            extension_lower = Path(filename).suffix.lower()
+            if extension_lower == ".xml":
+                with Path.open(Path(root, filename), "rb") as f:
+                    video_xml = f.read().decode("utf-8")
+
+                # Parse the XML and get transcript mappings
+                transcript_mapping = parse_video_transcripts_xml(run, video_xml, f)
+                video_transcript_mapping.update(transcript_mapping)
+
+    return video_transcript_mapping
+
+
 def _process_olx_path(olx_path: str, run: LearningResourceRun, *, overwrite):
+    assets_metadata = get_assets_metadata(olx_path)
+    video_srt_metadata = get_video_metadata(olx_path, run)
     for document, metadata in documents_from_olx(olx_path):
         source_path = metadata.get("source_path")
         edx_module_id = get_edx_module_id(source_path, run)
@@ -465,6 +610,9 @@ def _process_olx_path(olx_path: str, run: LearningResourceRun, *, overwrite):
                 "file_extension": file_extension,
                 "source_path": source_path,
                 "edx_module_id": edx_module_id,
+                "url": get_url_from_module_id(
+                    source_path, edx_module_id, run, assets_metadata, video_srt_metadata
+                ),
                 **content_dict,
             }
         )
@@ -741,7 +889,7 @@ def parse_certification(offeror, runs_data):
     )
 
 
-def iso8601_duration(duration_str: str) -> str or None:
+def iso8601_duration(duration_str: str) -> str | None:
     """
     Parse the duration from a string and return it in ISO-8601 format
 
@@ -821,7 +969,7 @@ def calculate_weeks(num: int, from_unit: str) -> int:
     return num
 
 
-def transform_interval(interval_txt: str) -> str or None:
+def transform_interval(interval_txt: str) -> str | None:
     """
     Transform any interval units to standard English units
     Only languages currently supported are English and Spanish

learning_resources_search/indexing_api.py

@@ -398,6 +398,7 @@ def index_run_content_files(run_id, index_types):
         index_types (string): one of the values IndexestoUpdate. Whether the default
             index, the reindexing index or both need to be updated
     """
+    return
     run = LearningResourceRun.objects.get(pk=run_id)
     content_file_ids = run.content_files.filter(published=True).values_list(
         "id", flat=True

vector_search/tasks.py

@@ -52,6 +52,7 @@ def generate_embeddings(ids, resource_type, overwrite):
         resource_type (string): resource_type value for the learning resource objects
 
     """
+    return None
     try:
         with wrap_retry_exception(*SEARCH_CONN_EXCEPTIONS):
             embed_learning_resources(ids, resource_type, overwrite)