feat: enhance YouTube video processing with improved error handling and logging

src/video_processing_tool.py

@@ -8,14 +8,20 @@ import re
 import shutil
 import time
 from smolagents.tools import Tool
+import logging
+
+# Set up logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
 
 class VideoProcessingTool(Tool):
     """
     Analyzes video content, extracting information such as frames, audio, or metadata.
     Useful for tasks like video summarization, frame extraction, transcript analysis, or content analysis.
+    Has limitations with YouTube content due to platform restrictions.
     """
     name = "video_processor"
-    description = "Analyzes video content from a file path or YouTube URL. Can extract frames, detect objects, get transcripts, and provide video metadata."
+    description = "Analyzes video content from a file path or YouTube URL. Can extract frames, detect objects, get transcripts, and provide video metadata. Note: Has limitations with YouTube content due to platform restrictions."
     inputs = {
         "file_path": {"type": "string", "description": "Path to the video file or YouTube URL.", "nullable": True},
         "task": {"type": "string", "description": "Specific task to perform (e.g., 'extract_frames', 'get_transcript', 'detect_objects', 'get_metadata').", "nullable": True},
@@ -75,24 +81,80 @@ class VideoProcessingTool(Tool):
         if task_parameters is None:
             task_parameters = {}
 
+        # Check for YouTube URL and provide appropriate warnings
         is_youtube_url = file_path and ("youtube.com/" in file_path or "youtu.be/" in file_path)
         video_source_path = file_path
 
+        # Special case for YouTube - check for likely restrictions before attempting download
         if is_youtube_url:
+            # For transcript tasks, try direct API first without downloading
+            if task == "get_transcript":
+                transcript_result = self.get_youtube_transcript(file_path)
+                if not transcript_result.get("error"):
+                    return transcript_result
+                
+                # If transcript API fails with certain errors, provide more helpful response
+                error_msg = transcript_result.get("error", "")
+                if "Transcripts are disabled" in error_msg:
+                    return {
+                        "error": "This YouTube video has disabled transcripts. Consider these alternatives:",
+                        "alternatives": [
+                            "Please provide a different video with transcripts enabled",
+                            "Upload a local video file that you have permission to use",
+                            "Provide a text summary of the video content manually"
+                        ]
+                    }
+
+            # For other tasks that require downloading
+            logger.info(f"YouTube URL detected: {file_path}. Attempting to access content...")
+            
+            # Try to get metadata about the video before downloading (title, etc.)
+            try:
+                with yt_dlp.YoutubeDL({'quiet': True, 'no_warnings': True}) as ydl:
+                    info = ydl.extract_info(file_path, download=False)
+                    video_title = info.get('title', 'Unknown')
+                    logger.info(f"Video title: {video_title}")
+            except Exception as e:
+                # YouTube is likely blocking access
+                error_text = str(e).lower()
+                if any(term in error_text for term in ["forbidden", "403", "blocked", "bot", "captcha", "cookie"]):
+                    return {
+                        "error": "YouTube access restricted. This agent cannot access this content due to platform restrictions.",
+                        "alternatives": [
+                            "Please upload a local video file instead",
+                            "For transcripts, try providing a text summary manually",
+                            "For visual analysis, consider uploading screenshots from the video"
+                        ]
+                    }
+                return {"error": f"Failed to access video info: {str(e)}"}
+            
+            # Proceed with download attempt but with better handling
             download_resolution = task_parameters.get("resolution", "360p")
             download_result = self.download_video(file_path, resolution=download_resolution)
+            
             if download_result.get("error"):
+                error_text = download_result.get("error", "").lower()
+                if any(term in error_text for term in ["forbidden", "403", "blocked", "bot", "captcha", "cookie"]):
+                    return {
+                        "error": "YouTube download restricted. This agent cannot download this content due to platform restrictions.",
+                        "alternatives": [
+                            "Please upload a local video file instead",
+                            "For transcripts, try obtaining them separately or summarizing manually",
+                            "For visual analysis, consider uploading key frames as images"
+                        ]
+                    }
                 return download_result
+                
             video_source_path = download_result.get("file_path")
             if not video_source_path or not os.path.exists(video_source_path):
-                 return {"error": f"Failed to download or locate video from URL: {file_path}"}
+                return {"error": f"Failed to download or locate video from URL: {file_path}"}
 
         elif file_path and not os.path.exists(file_path):
             return {"error": f"Video file not found: {file_path}"}
         elif not file_path and task not in ['get_transcript']: # transcript can work with URL directly
-             return {"error": "File path is required for this task."}
-
+            return {"error": "File path is required for this task."}
 
+        # Execute the appropriate task based on the request
         if task == "get_metadata":
             return self.get_video_metadata(video_source_path)
         elif task == "extract_frames":
@@ -108,7 +170,6 @@ class VideoProcessingTool(Tool):
             confidence_threshold = task_parameters.get("confidence_threshold", 0.5)
             frames_to_process = task_parameters.get("frames_to_process", 5) # Process N frames
             return self.detect_objects_in_video(video_source_path, confidence_threshold=confidence_threshold, num_frames_to_sample=frames_to_process)
-        # Add more tasks as needed, e.g., extract_audio
         else:
             return {"error": f"Unsupported task: {task}"}
 
@@ -120,7 +181,7 @@ class VideoProcessingTool(Tool):
         return None
 
     def download_video(self, youtube_url, resolution="360p"):
-        """Download YouTube video for processing."""
+        """Download YouTube video for processing with improved error handling."""
         video_id = self._extract_video_id(youtube_url)
         if not video_id:
             return {"error": "Invalid YouTube URL or could not extract video ID."}
@@ -132,6 +193,7 @@ class VideoProcessingTool(Tool):
             return {"success": True, "file_path": output_file_path, "message": "Video already downloaded."}
 
         try:
+            # First try with default options
             ydl_opts = {
                 'format': f'bestvideo[height<={resolution[:-1]}][ext=mp4]+bestaudio[ext=m4a]/best[ext=mp4]/best',
                 'outtmpl': output_file_path,
@@ -139,11 +201,15 @@ class VideoProcessingTool(Tool):
                 'quiet': True,
                 'no_warnings': True,
             }
+            
+            logger.info(f"Attempting to download YouTube video {video_id} at {resolution}...")
+            
             with yt_dlp.YoutubeDL(ydl_opts) as ydl:
                 ydl.download([youtube_url])
             
             if not os.path.exists(output_file_path): # Check if download actually created the file
-                 # Fallback for some formats if mp4 direct is not available
+                # Fallback for some formats if mp4 direct is not available
+                logger.info("Primary download method failed, trying alternative format...")
                 ydl_opts['format'] = f'best[height<={resolution[:-1]}]' # more generic
                 with yt_dlp.YoutubeDL(ydl_opts) as ydl:
                     info_dict = ydl.extract_info(youtube_url, download=True)
@@ -152,12 +218,10 @@ class VideoProcessingTool(Tool):
                     if downloaded_files:
                         actual_file_path = os.path.join(self.temp_dir, downloaded_files[0])
                         if actual_file_path != output_file_path and actual_file_path.endswith(('.mkv', '.webm', '.flv')):
-                            # Minimal conversion to mp4 if needed, or just use the downloaded format if cv2 supports it
-                            # For simplicity, we'll assume cv2 can handle common formats or user ensures mp4 compatible download
-                            output_file_path = actual_file_path # Use the actual downloaded file
-                        elif not actual_file_path.endswith('.mp4'): # if it's not mp4 and not handled above
-                             return {"error": f"Downloaded video is not in a directly usable format: {downloaded_files[0]}"}
-
+                            # Use the actual downloaded file
+                            output_file_path = actual_file_path
+                        elif not actual_file_path.endswith('.mp4'):
+                            return {"error": f"Downloaded video is not in a directly usable format: {downloaded_files[0]}"}
 
             if os.path.exists(output_file_path):
                 return {"success": True, "file_path": output_file_path}
@@ -165,7 +229,14 @@ class VideoProcessingTool(Tool):
                 return {"error": "Video download failed, file not found after attempt."}
 
         except yt_dlp.utils.DownloadError as e:
-            return {"error": f"yt-dlp download error: {str(e)}"}
+            error_msg = str(e)
+            if "Sign in to confirm your age" in error_msg:
+                return {"error": "Age-restricted video. Cannot download due to platform restrictions."}
+            elif "This video is private" in error_msg:
+                return {"error": "This video is private and cannot be accessed."}
+            elif any(term in error_msg.lower() for term in ["captcha", "bot", "cookie", "forbidden"]):
+                return {"error": f"YouTube access restricted due to bot detection. Consider uploading a local video file instead."}
+            return {"error": f"yt-dlp download error: {error_msg}"}
         except Exception as e:
             return {"error": f"Failed to download video: {str(e)}"}