Use a function to assign target path, simplifyt yt_dlp download

Author: mbertrand

dg_projects/learning_resources/learning_resources/assets/youtube_shorts.py

@@ -41,6 +41,11 @@ class YouTubeShortsConfig(Config):
     cached_metadata: str | None = None  # JSON string of video metadata from sensor
 
 
+def _assign_s3_path(video_id: str, extension: str) -> str:
+    """Assign S3 path for a YouTube video based on its ID and file extension."""
+    return f"{video_id}/{video_id}.{extension}"
+
+
 def _load_metadata_from_config(
     config: YouTubeShortsConfig,
 ) -> dict[str, Any] | None:
@@ -147,7 +152,7 @@ def youtube_video_metadata(
         with metadata_file.open("w", encoding="utf-8") as f:
             json.dump(video_data, f, indent=2, ensure_ascii=False)
 
-        target_path = f"youtube_shorts/{video_id}/{video_id}.json"
+        target_path = _assign_s3_path(video_id, "json")
 
         return Output(
             value=(metadata_file, target_path),
@@ -203,22 +208,10 @@ def youtube_video_content(
     output_file = Path(f"{video_id}.mp4")
 
     ydl_opts = {
-        "format": "best[height<=720]/best",
+        "format": "best",
         "outtmpl": str(output_file),
         "noplaylist": True,
         "extract_flat": False,
-        # Anti-bot measures
-        "http_headers": {
-            "User-Agent": (
-                "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
-            ),
-            "Accept": (
-                "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8"
-            ),
-            "Accept-Language": "en-us,en;q=0.5",
-            "Accept-Encoding": "gzip, deflate",
-            "Connection": "keep-alive",
-        },
     }
 
     try:
@@ -235,8 +228,7 @@ def youtube_video_content(
         )
 
         # Return tuple (local_path, s3_target_path) for s3file_io_manager to upload
-        # Similar to canvas asset pattern
-        target_path = f"youtube_shorts/{video_id}/{video_id}.mp4"
+        target_path = _assign_s3_path(video_id, "mp4")
 
         return Output(
             value=(output_file, target_path),
@@ -339,9 +331,7 @@ def youtube_video_thumbnail(
             content_length,
         )
 
-        # Return tuple (local_path, s3_target_path) for s3file_io_manager to upload
-        # Similar to canvas asset pattern
-        target_path = f"youtube_shorts/{video_id}/{video_id}.jpg"
+        target_path = _assign_s3_path(video_id, "jpg")
 
         return Output(
             value=(thumbnail_file, target_path),