More tweaks

Author: mbertrand

dg_projects/learning_resources/learning_resources/assets/youtube_shorts.py

@@ -53,7 +53,7 @@ def _load_metadata_from_config(
         return json.loads(config.cached_metadata)
     except (json.JSONDecodeError, TypeError):
         log.warning(
-            "Failed to parse metadata from run_config, will fetch from API",
+            "Failed to parse metadata from run_config",
             exc_info=True,
         )
         return None
@@ -84,24 +84,11 @@ def fetch_video_metadata(youtube_client, video_id):
     return request.execute()
 
 
-def upload_to_s3(s3_client, bucket, key, body, content_type):
-    s3_client.put_object(
-        Bucket=bucket,
-        Key=key,
-        Body=body,
-        ContentType=content_type,
-    )
-
-
-def send_webhook(learn_api: ApiClientFactory, video_id: str) -> dict[str, Any]:
-    return learn_api.client.notify_shorts_processed(video_id)
-
-
 @asset(
     key=AssetKey(["youtube_shorts", "video_metadata"]),
     group_name="youtube_shorts",
     description="Extract YouTube video metadata from API and upload to S3",
-    code_version="youtube_video_metadata_v3",
+    code_version="youtube_video_metadata_v1",
     partitions_def=youtube_video_ids,
     automation_condition=upstream_or_code_changes(),
     io_manager_key="s3file_io_manager",
@@ -176,14 +163,13 @@ def youtube_video_metadata(
     except Exception:
         log.exception("Failed to extract metadata for video: %s", video_id)
         raise
-        raise
 
 
 @asset(
     key=AssetKey(["youtube_shorts", "video_content"]),
     group_name="youtube_shorts",
     description="Download and upload YouTube video content to S3",
-    code_version="youtube_video_content_v3",
+    code_version="youtube_video_content_v1",
     partitions_def=youtube_video_ids,
     automation_condition=upstream_or_code_changes(),
     io_manager_key="s3file_io_manager",
@@ -205,10 +191,13 @@ def youtube_video_content(
     video_id = context.partition_key
     video_url = f"https://www.youtube.com/watch?v={video_id}"
 
-    log.info("Processing video content for: %s", video_id)
+    log.info("Processing video file for: %s", video_id)
+
+    # Load metadata from config if available for better version tracking
+    video_data = _load_metadata_from_config(config)
 
     # Generate version for change tracking
-    version = _generate_video_version({"id": video_id})
+    version = _generate_video_version(video_data or {"id": video_id})
 
     # Download video using yt-dlp to local file
     output_file = Path(f"{video_id}.mp4")
@@ -399,7 +388,7 @@ def youtube_webhook(
     video_id = context.partition_key
 
     try:
-        webhook_response = send_webhook(learn_api, video_id)
+        webhook_response = learn_api.client.notify_shorts_processed(video_id)
         log.info(
             "Successfully sent webhook notification for video %s: %s",
             video_id,

dg_projects/learning_resources/learning_resources/sensors/youtube_shorts.py

@@ -241,6 +241,9 @@ def create_run_requests(videos_to_process, metadata_by_video_id):
                 tags={"video_id": video_id},
                 run_config={
                     "ops": {
+                        "youtube_shorts__video_content": {
+                            "config": {"cached_metadata": metadata_json}
+                        },
                         "youtube_shorts__video_thumbnail": {
                             "config": {"cached_metadata": metadata_json}
                         },