Some cleanup etc

Author: mbertrand

dg_projects/learning_resources/learning_resources/assets/youtube_shorts.py

@@ -5,7 +5,6 @@
 import hashlib
 import json
 import logging
-import re
 import tempfile
 from pathlib import Path
 from tempfile import NamedTemporaryFile
@@ -29,32 +28,6 @@
 
 from learning_resources.resources.youtube_client import YouTubeClientFactory
 
-
-def _check_video_duration(video_data: dict[str, Any], max_duration: int = 60) -> bool:
-    """
-    Check if video duration is within acceptable limits for shorts.
-
-    YouTube shorts should typically be 60 seconds or less.
-    """
-    duration_str = video_data.get("contentDetails", {}).get("duration", "")
-    if not duration_str:
-        return False
-
-    # Parse ISO 8601 duration format (PT1M30S)
-    pattern = r"PT(?:(\d+)H)?(?:(\d+)M)?(?:(\d+)S)?"
-    match = re.match(pattern, duration_str)
-
-    if not match:
-        return False
-
-    hours = int(match.group(1) or 0)
-    minutes = int(match.group(2) or 0)
-    seconds = int(match.group(3) or 0)
-
-    total_seconds = hours * 3600 + minutes * 60 + seconds
-    return total_seconds <= max_duration
-
-
 log = logging.getLogger(__name__)
 
 # Dynamic partitioning for individual video processing
@@ -67,7 +40,6 @@ class YouTubeShortsConfig(Config):
     bucket_name: str = "ol-devops-sandbox"
     s3_prefix: str = "youtube_shorts"
     max_workers: int = 4
-    max_video_duration: int = 60  # seconds
     cached_metadata: str | None = None  # JSON string of video metadata from sensor
 
 
@@ -101,8 +73,6 @@ def _generate_video_version(video_data: dict[str, Any]) -> str:
         "id": video_data.get("id"),
         "etag": video_data.get("etag"),
         "published_at": video_data.get("snippet", {}).get("publishedAt"),
-        "title": video_data.get("snippet", {}).get("title"),
-        "view_count": video_data.get("statistics", {}).get("viewCount"),
     }
 
     version_string = json.dumps(version_fields, sort_keys=True)
@@ -111,7 +81,7 @@ def _generate_video_version(video_data: dict[str, Any]) -> str:
 
 def fetch_video_metadata(youtube_client, video_id):
     request = youtube_client.client.videos().list(
-        part="snippet,contentDetails,statistics", id=video_id
+        part="snippet,contentDetails", id=video_id
     )
     return request.execute()
 
@@ -184,13 +154,6 @@ def youtube_video_metadata(  # noqa: PLR0913
 
             video_data = videos[0]
 
-        # Check if this is actually a short (duration filter)
-        # Only check duration if we fetched from API (sensor data is pre-filtered)
-        if not from_sensor and not _check_video_duration(video_data, max_duration=60):
-            duration = video_data.get("contentDetails", {}).get("duration", "")
-            msg = f"Video {video_id} is not a short (duration: {duration})"
-            raise ValueError(msg)  # noqa: TRY301
-
         # Generate version for change tracking
         version = _generate_video_version(video_data)
 
@@ -200,13 +163,6 @@ def youtube_video_metadata(  # noqa: PLR0913
         # Convert video_data to JSON and upload to S3
         metadata_json = json.dumps(video_data, indent=2, ensure_ascii=False)
 
-        try:
-            # Check if metadata already exists
-            s3.get_client().head_object(Bucket=config.bucket_name, Key=s3_key)
-            log.info("Metadata %s already exists in S3, updating...", video_id)
-        except s3.get_client().exceptions.NoSuchKey:
-            log.info("Uploading new metadata for %s to S3", video_id)
-
         # Upload metadata to S3
         upload_to_s3(
             s3.get_client(),
@@ -266,24 +222,8 @@ def youtube_video_content(
     # Check if video already exists in S3
     s3_key = f"{config.s3_prefix}/videos/{video_id}.mp4"
 
-    try:
-        # Check if file exists
-        s3.get_client().head_object(Bucket=config.bucket_name, Key=s3_key)
-        log.info("Video %s already exists in S3, skipping download", video_id)
-
-        return Output(
-            value=s3_key,
-            metadata={
-                "video_id": video_id,
-                "s3_bucket": config.bucket_name,
-                "s3_key": s3_key,
-                "status": "already_exists",
-            },
-        )
-
-    except s3.get_client().exceptions.NoSuchKey:
-        # File doesn't exist, proceed with download
-        pass
+    # Generate version for change tracking
+    version = _generate_video_version({"id": video_id})
 
     # Download video using yt-dlp
     with tempfile.TemporaryDirectory() as temp_dir:
@@ -331,6 +271,7 @@ def youtube_video_content(
 
             return Output(
                 value=s3_key,
+                data_version=DataVersion(version),
                 metadata={
                     "video_id": video_id,
                     "s3_bucket": config.bucket_name,
@@ -398,6 +339,9 @@ def youtube_video_thumbnail(
 
         video_metadata = videos[0]
 
+    # Generate version for change tracking
+    version = _generate_video_version(video_metadata)
+
     thumbnails = video_metadata.get("snippet", {}).get("thumbnails", {})
 
     # Try to get high, medium, or default quality thumbnail
@@ -413,25 +357,6 @@ def youtube_video_thumbnail(
 
     s3_key = f"{config.s3_prefix}/thumbnails/{video_id}.jpg"
 
-    try:
-        # Check if thumbnail already exists
-        s3.get_client().head_object(Bucket=config.bucket_name, Key=s3_key)
-        log.info("Thumbnail %s already exists in S3, skipping download", video_id)
-
-        return Output(
-            value=s3_key,
-            metadata={
-                "video_id": video_id,
-                "s3_bucket": config.bucket_name,
-                "s3_key": s3_key,
-                "status": "already_exists",
-            },
-        )
-
-    except s3.get_client().exceptions.NoSuchKey:
-        # File doesn't exist, proceed with download
-        pass
-
     try:
         # Download thumbnail
         response = requests.get(thumbnail_url, timeout=30)
@@ -454,6 +379,7 @@ def youtube_video_thumbnail(
 
         return Output(
             value=s3_key,
+            data_version=DataVersion(version),
             metadata={
                 "video_id": video_id,
                 "s3_bucket": config.bucket_name,
@@ -470,17 +396,17 @@ def youtube_video_thumbnail(
 
 
 @asset(
-    key=AssetKey(["youtube_shorts", "webhook_notification"]),
+    key=AssetKey(["youtube_shorts", "video_webhook"]),
     group_name="youtube_shorts",
     description="Send webhook notification after metadata extraction.",
     code_version="youtube_webhook_v1",
     partitions_def=youtube_video_ids,
     ins={"video_metadata": AssetIn(key=AssetKey(["youtube_shorts", "video_metadata"]))},
 )
-def youtube_webhook_notification(
+def youtube_webhook(
     context: AssetExecutionContext,
     learn_api: MITLearnApiClient,
-    video_metadata: dict[str, Any],  # noqa: ARG001
+    video_metadata: dict[str, Any],
 ) -> Output[bool]:
     """
     Send a webhook notification for a processed YouTube video.
@@ -494,6 +420,10 @@ def youtube_webhook_notification(
         Boolean indicating success of the webhook notification
     """
     video_id = context.partition_key
+
+    # Generate version for change tracking
+    version = _generate_video_version(video_metadata)
+
     try:
         webhook_response = send_webhook(learn_api, video_id)
         log.info(
@@ -503,6 +433,7 @@ def youtube_webhook_notification(
         )
         return Output(
             value=True,
+            data_version=DataVersion(version),
             metadata={
                 "video_id": video_id,
                 "webhook_response": webhook_response,

dg_projects/learning_resources/learning_resources/definitions.py

@@ -26,6 +26,7 @@
     youtube_video_content,
     youtube_video_metadata,
     youtube_video_thumbnail,
+    youtube_webhook,  # Renamed youtube_webhook_notification to youtube_webhook
 )
 from learning_resources.resources.api_client_factory import ApiClientFactory
 from learning_resources.resources.youtube_client import YouTubeClientFactory
@@ -107,6 +108,7 @@
         youtube_video_metadata,
         youtube_video_content,
         youtube_video_thumbnail,
+        youtube_webhook,  # Updated reference to youtube_webhook
     ],
     jobs=[youtube_shorts_job],
     schedules=[extract_api_daily_schedule],

dg_projects/learning_resources/learning_resources/resources/youtube_config.py

@@ -29,32 +29,12 @@ def get_config(self) -> list[dict[str, Any]]:
         Returns:
             List of configuration items defining channels/playlists to process
         """
-        try:
-            response = requests.get(self.config_url, timeout=30)
-            response.raise_for_status()
-            config_data = yaml.safe_load(response.text)
-
-            # Handle both single item and list formats
-            if isinstance(config_data, list):
-                return config_data
-            else:
-                return [config_data]
-
-        except (requests.RequestException, yaml.YAMLError, KeyError) as e:
-            # Fallback configuration for development/testing - using real MIT
-            # Open Learning uploads
-            fallback_config = [
-                {
-                    "name": "MIT Open Learning",
-                    "type": "playlist",
-                    # MIT Open Learning uploads playlist
-                    "id": "UUSHN0QBfKk0ZSytyX_16M11fA",
-                }
-            ]
-            # Use warning level but don't fail - fallback should work for testing
-            log.warning(
-                "Failed to fetch config from %s, using fallback: %s",
-                self.config_url,
-                e,
-            )
-            return fallback_config
+        response = requests.get(self.config_url, timeout=30)
+        response.raise_for_status()
+        config_data = yaml.safe_load(response.text)
+
+        # Handle both single item and list formats
+        if isinstance(config_data, list):
+            return config_data
+        else:
+            return [config_data]