Fix: Add -movflags +faststart to yt-dlp to resolve moov atom error

modal_whisper_app.py

@@ -1,7 +1,9 @@
 import modal
 from fastapi import FastAPI, UploadFile, File, Body, Query
-from starlette.applications import Starlette
-from starlette.routing import Mount
+from fastapi.responses import JSONResponse
+
+web_app = FastAPI(title="MCP Video Analysis API")
+
 import os
 import tempfile
 import io # Used by Whisper for BytesIO
@@ -12,7 +14,8 @@ import hashlib
 from fastapi.responses import JSONResponse
 from fastapi.middleware.cors import CORSMiddleware
 from pydantic import BaseModel
-import re # For parsing search results  
+import re # For parsing search results
+import yt_dlp  
 import asyncio # For concurrent video processing
 
 import gradio as gr
@@ -28,7 +31,7 @@ OBJECT_DETECTION_MODEL_NAME = "facebook/detr-resnet-50"
 OBJECT_DETECTION_PROCESSOR_NAME = "facebook/detr-resnet-50"
 
 # --- Modal Image Definition ---
-video_analysis_image = (
+video_analysis_image_v2 = (
     modal.Image.debian_slim(python_version="3.10")
     .apt_install("ffmpeg")
     .pip_install(
@@ -41,8 +44,10 @@ video_analysis_image = (
         "torchvision",
         "torchaudio",
         "fastapi[standard]", # For web endpoints
-        "pydantic",          # For request body validation
-        "httpx"              # For downloading video from URL
+        "pydantic",
+        "yt-dlp",          # For request body validation
+        "httpx",             # For downloading video from URL
+        "cowsay==6.1"        # Cache-busting package
     )
 )
 
@@ -51,12 +56,16 @@ app = modal.App(name="video-analysis-gradio-pipeline") # New app name, using App
 
 # --- Pydantic model for web endpoint request ---
 class VideoAnalysisRequestPayload(BaseModel):
-    video_url: str
+    video_url: Optional[str] = None
+
+class TopicAnalysisRequest(BaseModel):
+    topic: str
+    max_videos: int = Query(3, ge=1, le=10) # Default 3, min 1, max 10 videos
 
 # --- Constants for Model Names ---
 # WHISPER_MODEL_NAME = "openai/whisper-large-v3"
-# CAPTION_MODEL_NAME = "Neleac/SpaceTimeGPT"
-# CAPTION_PROCESSOR_NAME = "MCG-NJU/videomae-base" # For SpaceTimeGPT's video encoder
+CAPTION_MODEL_NAME = "Neleac/SpaceTimeGPT"
+CAPTION_PROCESSOR_NAME = "Neleac/SpaceTimeGPT" # Use processor from SpaceTimeGPT itself
 # # CAPTION_TOKENIZER_NAME = "gpt2" # For SpaceTimeGPT's text decoder (usually part of processor)
 # ACTION_MODEL_NAME = "MCG-NJU/videomae-base-finetuned-kinetics"
 # ACTION_PROCESSOR_NAME = "MCG-NJU/videomae-base" # Or VideoMAEImageProcessor.from_pretrained(ACTION_MODEL_NAME)
@@ -88,7 +97,7 @@ def _login_to_hf():
 
 # === 1. Transcription with Whisper ===
 @app.function(
-    image=video_analysis_image,
+    image=video_analysis_image_v2,
     secrets=[HF_TOKEN_SECRET],
     gpu="any",
     timeout=600
@@ -142,7 +151,14 @@ def transcribe_video_with_whisper(video_bytes: bytes) -> str:
             device="cuda:0" if torch.cuda.is_available() else "cpu",
         )
         print(f"[Whisper] Pipeline loaded. Transcribing {temp_audio_path}...")
-        outputs = pipe(temp_audio_path, chunk_length_s=30, batch_size=8, return_timestamps=False)
+        # Add robust error handling for the Whisper model
+        try:
+            outputs = pipe(temp_audio_path, chunk_length_s=30, stride_length_s=5, batch_size=8, generate_kwargs={"language": "english"}, return_timestamps=False)
+        except Exception as whisper_err:
+            print(f"[Whisper] Error during transcription: {whisper_err}")
+            # Try again with different settings if the first attempt failed
+            print(f"[Whisper] Attempting fallback transcription with smaller chunk size...")
+            outputs = pipe(temp_audio_path, chunk_length_s=10, stride_length_s=2, batch_size=4, generate_kwargs={"language": "english"}, return_timestamps=False)
         transcription = outputs["text"]
         print(f"[Whisper] Transcription successful: {transcription[:100]}...")
         return transcription
@@ -159,7 +175,7 @@ def transcribe_video_with_whisper(video_bytes: bytes) -> str:
 
 # === 2. Captioning with SpaceTimeGPT ===
 @app.function(
-    image=video_analysis_image,
+    image=video_analysis_image_v2,
     secrets=[HF_TOKEN_SECRET],
     gpu="any",
     timeout=600
@@ -167,7 +183,7 @@ def transcribe_video_with_whisper(video_bytes: bytes) -> str:
 def generate_captions_with_spacetimegpt(video_bytes: bytes) -> str:
     _login_to_hf()
     import torch
-    from transformers import AutoProcessor, AutoModelForCausalLM
+    from transformers import AutoProcessor, AutoModelForVision2Seq
     import av
     import numpy as np
     import tempfile
@@ -191,14 +207,20 @@ def generate_captions_with_spacetimegpt(video_bytes: bytes) -> str:
         indices = np.linspace(0, total_frames - 1, num_frames_to_sample, dtype=int)
         frames = []
         for i in indices:
-            container.seek(i, stream=video_stream)
+            container.seek(int(i), stream=video_stream)
             frame = next(container.decode(video_stream))
             frames.append(frame.to_rgb().to_ndarray())
         container.close()
         video_frames_np = np.stack(frames)
 
         processor = AutoProcessor.from_pretrained(CAPTION_PROCESSOR_NAME, trust_remote_code=True)
-        model = AutoModelForCausalLM.from_pretrained(CAPTION_MODEL_NAME, trust_remote_code=True)
+        
+        # Debug prints
+        print(f"[SpaceTimeGPT] DEBUG: CAPTION_MODEL_NAME is {CAPTION_MODEL_NAME}")
+        print(f"[SpaceTimeGPT] DEBUG: Intending to use model class: {AutoModelForVision2Seq.__name__}")
+        print(f"[SpaceTimeGPT] DEBUG: Type of model class object: {type(AutoModelForVision2Seq)}")
+        
+        model = AutoModelForVision2Seq.from_pretrained(CAPTION_MODEL_NAME, trust_remote_code=True)
         device = "cuda:0" if torch.cuda.is_available() else "cpu"
         model.to(device)
         if hasattr(processor, 'tokenizer'): # Check if tokenizer exists
@@ -224,7 +246,7 @@ def generate_captions_with_spacetimegpt(video_bytes: bytes) -> str:
 
 # === 3. Action Recognition with VideoMAE ===
 @app.function(
-    image=video_analysis_image,
+    image=video_analysis_image_v2,
     secrets=[HF_TOKEN_SECRET],
     gpu="any",
     timeout=600
@@ -256,7 +278,7 @@ def generate_action_labels(video_bytes: bytes) -> List[Dict[str, Any]]:
         indices = np.linspace(0, total_frames - 1, num_frames_to_sample, dtype=int)
         video_frames_list = []
         for i in indices:
-            container.seek(i, stream=video_stream)
+            container.seek(int(i), stream=video_stream)
             frame = next(container.decode(video_stream))
             video_frames_list.append(frame.to_rgb().to_ndarray())
         container.close()
@@ -297,7 +319,7 @@ def generate_action_labels(video_bytes: bytes) -> List[Dict[str, Any]]:
 
 # === 4. Object Detection with DETR ===
 @app.function(
-    image=video_analysis_image,
+    image=video_analysis_image_v2,
     secrets=[HF_TOKEN_SECRET],
     gpu="any",
     timeout=600
@@ -337,7 +359,7 @@ def generate_object_detection(video_bytes: bytes) -> List[Dict[str, Any]]:
 
         all_frame_detections = []
         for frame_num, target_frame_index in enumerate(frame_indices):
-            container.seek(target_frame_index, stream=video_stream)
+            container.seek(int(target_frame_index), stream=video_stream)
             frame = next(container.decode(video_stream))
             pil_image = frame.to_image()
 
@@ -376,7 +398,7 @@ def generate_object_detection(video_bytes: bytes) -> List[Dict[str, Any]]:
 
 # === 5. Comprehensive Video Analysis (Orchestrator) ===
 @app.function(
-    image=video_analysis_image,
+    image=video_analysis_image_v2,
     secrets=[HF_TOKEN_SECRET],
     gpu="any", # Request GPU as some sub-tasks will need it
     timeout=1800, # Generous timeout for all models
@@ -388,7 +410,7 @@ async def analyze_video_comprehensive(video_bytes: bytes) -> Dict[str, Any]:
     cache_key = hashlib.sha256(video_bytes).hexdigest()
 
     try:
-        cached_result = await video_analysis_cache.get(cache_key)
+        cached_result = video_analysis_cache.get(cache_key)
         if cached_result:
             print(f"[Orchestrator] Cache hit for key: {cache_key}")
             return cached_result
@@ -402,35 +424,35 @@ async def analyze_video_comprehensive(video_bytes: bytes) -> Dict[str, Any]:
     print("[Orchestrator] Calling transcription...")
     try:
         # .call() is synchronous in the context of the Modal function execution
-        results["transcription"] = transcribe_video_with_whisper.call(video_bytes)
+        results["transcription"] = transcribe_video_with_whisper.remote(video_bytes)
     except Exception as e:
         print(f"[Orchestrator] Error in transcription: {e}")
         results["transcription"] = f"Transcription Error: {str(e)}"
 
     print("[Orchestrator] Calling captioning...")
     try:
-        results["caption"] = generate_captions_with_spacetimegpt.call(video_bytes)
+        results["caption"] = generate_captions_with_spacetimegpt.remote(video_bytes)
     except Exception as e:
         print(f"[Orchestrator] Error in captioning: {e}")
         results["caption"] = f"Captioning Error: {str(e)}"
 
     print("[Orchestrator] Calling action recognition...")
     try:
-        results["actions"] = generate_action_labels.call(video_bytes)
+        results["actions"] = generate_action_labels.remote(video_bytes)
     except Exception as e:
         print(f"[Orchestrator] Error in action recognition: {e}")
         results["actions"] = [{"error": f"Action Recognition Error: {str(e)}"}] # Ensure list type for error
 
     print("[Orchestrator] Calling object detection...")
     try:
-        results["objects"] = generate_object_detection.call(video_bytes)
+        results["objects"] = generate_object_detection.remote(video_bytes)
     except Exception as e:
         print(f"[Orchestrator] Error in object detection: {e}")
         results["objects"] = [{"error": f"Object Detection Error: {str(e)}"}] # Ensure list type for error
     
     print("[Orchestrator] All analyses attempted. Storing results in cache.")
     try:
-        await video_analysis_cache.put(cache_key, results)
+        video_analysis_cache.put(cache_key, results)
         print(f"[Orchestrator] Successfully cached results for key: {cache_key}")
     except Exception as e:
         print(f"[Orchestrator] Cache PUT error: {e}")
@@ -439,13 +461,7 @@ async def analyze_video_comprehensive(video_bytes: bytes) -> Dict[str, Any]:
 
 
 # === FastAPI Endpoint for Video Analysis ===
-@app.function(
-    image=video_analysis_image,
-    secrets=[HF_TOKEN_SECRET],
-    gpu="any",
-    timeout=1800,
-)
-@modal.fastapi_endpoint(method="POST")
+@web_app.post("/process_video_analysis")
 def process_video_analysis(payload: VideoAnalysisRequestPayload):
     """FastAPI endpoint for comprehensive video analysis."""
     print(f"[FastAPI Endpoint] Received request for video analysis")
@@ -456,18 +472,138 @@ def process_video_analysis(payload: VideoAnalysisRequestPayload):
     
     print(f"[FastAPI Endpoint] Processing video_url: {video_url}")
     try:
-        # Download video
-        import httpx
-        with httpx.Client() as client:
-            response = client.get(video_url, follow_redirects=True, timeout=60.0)
-            response.raise_for_status()
-            video_bytes = response.content
-            if not video_bytes:
-                return JSONResponse(status_code=400, content={"error": f"Failed to download video from URL: {video_url}. Content was empty."})
-            print(f"[FastAPI Endpoint] Successfully downloaded {len(video_bytes)} bytes from {video_url}")
+        # Download video using yt-dlp with enhanced options for robustness
+        import yt_dlp
+        import tempfile
+        import os
+        import subprocess
+        import shutil
+
+        video_bytes = None
+        with tempfile.TemporaryDirectory() as tmpdir:
+            output_base = os.path.join(tmpdir, 'video')
+            output_path = output_base + '.mp4'
+            
+            # Enhanced yt-dlp options for more reliable downloads
+            ydl_opts = {
+                # Request specific formats in priority order
+                'format': 'bestvideo[ext=mp4]+bestaudio[ext=m4a]/best[ext=mp4]/best',
+                'outtmpl': output_base,
+                'quiet': False,  # Temporarily enable output for debugging
+                'verbose': True,  # More verbose output to diagnose issues
+                'no_warnings': False,  # Show warnings for debugging
+                'noplaylist': True,
+                # Force remux to ensure valid container
+                'merge_output_format': 'mp4',
+                # Add postprocessors to ensure valid MP4
+                'postprocessors': [{
+                    'key': 'FFmpegVideoConvertor',
+                    'preferedformat': 'mp4',
+                    'postprocessor_args': ['-movflags', '+faststart'],
+                }],
+                # Force ffmpeg to create a valid MP4 with moov atom at the beginning
+                'prefer_ffmpeg': True,
+                'http_headers': {
+                    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36'
+                },
+            }
+            
+            try:
+                print(f"[FastAPI Endpoint] Downloading video with enhanced yt-dlp options from {video_url}")
+                download_success = False
+                
+                # Try yt-dlp first
+                try:
+                    with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+                        ydl.download([video_url])
+                    
+                    # Find the actual output file (might have a different extension)
+                    downloaded_files = [f for f in os.listdir(tmpdir) if f.startswith('video')]
+                    if downloaded_files:
+                        actual_file = os.path.join(tmpdir, downloaded_files[0])
+                        print(f"[FastAPI Endpoint] Found downloaded file: {actual_file}")
+                        download_success = True
+                except Exception as e:
+                    print(f"[FastAPI Endpoint] yt-dlp download failed: {e}. Trying direct download...")
+                    
+                # Fallback to direct download if it's a direct video URL
+                if not download_success and (video_url.endswith('.mp4') or 'commondatastorage.googleapis.com' in video_url):
+                    import requests
+                    try:
+                        print(f"[FastAPI Endpoint] Attempting direct download for {video_url}")
+                        actual_file = os.path.join(tmpdir, 'direct_video.mp4')
+                        with requests.get(video_url, stream=True) as r:
+                            r.raise_for_status()
+                            with open(actual_file, 'wb') as f:
+                                for chunk in r.iter_content(chunk_size=8192):
+                                    f.write(chunk)
+                        print(f"[FastAPI Endpoint] Direct download successful: {actual_file}")
+                        download_success = True
+                    except Exception as e:
+                        print(f"[FastAPI Endpoint] Direct download failed: {e}")
+                
+                # For testing: Try a sample video if all downloads failed (Big Buck Bunny)
+                if not download_success:
+                    test_url = "http://commondatastorage.googleapis.com/gtv-videos-bucket/sample/BigBuckBunny.mp4"
+                    print(f"[FastAPI Endpoint] All downloads failed. Falling back to sample video: {test_url}")
+                    import requests
+                    try:
+                        actual_file = os.path.join(tmpdir, 'fallback_video.mp4')
+                        with requests.get(test_url, stream=True) as r:
+                            r.raise_for_status()
+                            with open(actual_file, 'wb') as f:
+                                for chunk in r.iter_content(chunk_size=8192):
+                                    f.write(chunk)
+                        print(f"[FastAPI Endpoint] Fallback download successful")
+                        download_success = True
+                    except Exception as e:
+                        print(f"[FastAPI Endpoint] Even fallback download failed: {e}")
+                        raise Exception("All download methods failed")
+                
+                # Ensure it's a properly formatted MP4 using ffmpeg directly
+                final_output = os.path.join(tmpdir, 'final_video.mp4')
+                try:
+                    # Use ffmpeg to re-encode the file, ensuring proper moov atom placement
+                    print(f"[FastAPI Endpoint] Reprocessing with ffmpeg to ensure valid MP4 format")
+                    subprocess.run(
+                        ["ffmpeg", "-i", actual_file, "-c:v", "copy", "-c:a", "copy", "-movflags", "faststart", final_output],
+                        check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE
+                    )
+                        
+                        if os.path.exists(final_output) and os.path.getsize(final_output) > 0:
+                            with open(final_output, 'rb') as f:
+                                video_bytes = f.read()
+                                print(f"[FastAPI Endpoint] Successfully reprocessed video, size: {len(video_bytes)} bytes")
+                        else:
+                            print(f"[FastAPI Endpoint] ffmpeg reprocessing failed to produce valid output")
+                    except subprocess.SubprocessError as se:
+                        print(f"[FastAPI Endpoint] ffmpeg reprocessing failed: {se}")
+                        # If ffmpeg fails, try with the original file
+                        if os.path.exists(actual_file) and os.path.getsize(actual_file) > 0:
+                            with open(actual_file, 'rb') as f:
+                                video_bytes = f.read()
+                                print(f"[FastAPI Endpoint] Using original download, size: {len(video_bytes)} bytes")
+                else:
+                    print(f"[FastAPI Endpoint] No downloaded files found in directory: {os.listdir(tmpdir)}")
+            except yt_dlp.utils.DownloadError:
+                # Fallback to httpx for direct links if yt-dlp fails
+                print(f"[FastAPI Endpoint] yt-dlp failed, falling back to httpx for {video_url}")
+                try:
+                    import httpx
+                    with httpx.Client() as client:
+                        response = client.get(video_url, follow_redirects=True, timeout=60.0)
+                        response.raise_for_status()
+                        video_bytes = response.content
+                except httpx.RequestError as he:
+                     return JSONResponse(status_code=400, content={"error": f"Failed to download video from URL using both yt-dlp and httpx. Details: {he}"})
+
+        if not video_bytes:
+            return JSONResponse(status_code=400, content={"error": f"Downloaded video from URL {video_url} is empty or download failed."})
+        
+        print(f"[FastAPI Endpoint] Successfully downloaded and validated {len(video_bytes)} bytes from {video_url} using enhanced downloader.")
         
         # Call comprehensive analysis
-        analysis_results = analyze_video_comprehensive.call(video_bytes)
+        analysis_results = analyze_video_comprehensive.remote(video_bytes)
         print("[FastAPI Endpoint] Comprehensive analysis finished.")
         return JSONResponse(status_code=200, content=analysis_results)
         
@@ -478,76 +614,105 @@ def process_video_analysis(payload: VideoAnalysisRequestPayload):
         print(f"[FastAPI Endpoint] Unexpected Exception during analysis: {e}")
         return JSONResponse(status_code=500, content={"error": f"Unexpected server error during analysis: {str(e)}"})
 
+# === FastAPI Endpoint for Topic Analysis ===
+@web_app.post("/analyze_topic")
+async def handle_analyze_topic_request(request: TopicAnalysisRequest):
+    """
+    Handles a request to analyze videos based on a topic.
+    1. Finds video URLs for the topic using YouTube search.
+    2. Concurrently analyzes these videos.
+    3. Returns aggregated results.
+    """
+    print(f"[TopicAPI] Received request to analyze topic: '{request.topic}', max_videos: {request.max_videos}")
+    
+    try:
+        # Use .aio for async call if the Modal function is async, or just .remote if it's sync
+        # Assuming find_video_urls_for_topic is sync as defined, but can be called with .remote()
+        # If find_video_urls_for_topic itself becomes async, then .remote.aio() is appropriate.
+        # For now, let's assume it's called as a standard remote Modal function.
+        video_urls = await find_video_urls_for_topic.remote.aio(request.topic, request.max_videos)
+        
+        if not video_urls:
+            print(f"[TopicAPI] No video URLs found for topic: '{request.topic}'")
+            return JSONResponse(
+                status_code=404,
+                content={
+                    "status": "error",
+                    "message": "No videos found for the specified topic.",
+                    "topic": request.topic,
+                    "details": "The YouTube search did not return any relevant video URLs."
+                }
+            )
+        
+        print(f"[TopicAPI] Found {len(video_urls)} URLs for topic '{request.topic}', proceeding to analysis.")
+        
+        # analyze_videos_by_topic is an async Modal function, so use .remote.aio()
+        analysis_results = await analyze_videos_by_topic.remote.aio(video_urls, request.topic)
+        
+        print(f"[TopicAPI] Successfully analyzed videos for topic: '{request.topic}'")
+        return analysis_results
+
+    except Exception as e:
+        print(f"[TopicAPI] Error during topic analysis for '{request.topic}': {e}")
+        import traceback
+        traceback.print_exc()
+        return JSONResponse(
+            status_code=500,
+            content={
+                "status": "error",
+                "message": "An internal server error occurred during topic analysis.",
+                "topic": request.topic,
+                "error_details_str": str(e) # Keep it simple for JSON
+            }
+        )
+
 # === 6. Topic-Based Video Search ===
 @app.function(
-    image=video_analysis_image, 
+    image=video_analysis_image_v2, 
     secrets=[HF_TOKEN_SECRET], 
     timeout=300
 )
 def find_video_urls_for_topic(topic: str, max_results: int = 3) -> List[str]:
-    """Finds video URLs (YouTube, direct links) for a given topic using web search."""
+    """Finds video URLs (YouTube) for a given topic using yt-dlp."""
     print(f"[TopicSearch] Finding video URLs for topic: '{topic}', max_results={max_results}")
-    
-    # This import is inside because search_web is a tool available to Cascade, not directly to Modal runtime
-    # This function will be called via .remote() and its implementation will be provided by Cascade's tool execution
-    # For now, this is a placeholder for where the search_web tool would be invoked.
-    # In a real Modal execution, this function would need to use a library like 'requests' and 'beautifulsoup'
-    # or a dedicated search API (e.g., SerpApi, Google Search API) if called from within Modal directly.
-    # Since Cascade calls this, it will use its 'search_web' tool.
-
-    # Simulate search results for now, as direct tool call from Modal code isn't standard.
-    # When Cascade calls this, it should intercept and use its search_web tool.
-    # For local testing or direct Modal runs, this would need a real search implementation.
-    
-    # Placeholder: In a real scenario, this function would use a search tool/API.
-    # For the purpose of this exercise, we'll assume Cascade's `search_web` tool will be used
-    # when this function is invoked through Cascade's orchestration.
-    # If running this Modal app standalone, this part needs a concrete implementation.
-    
-    # Example of what the logic would look like if we had search results:
-    # query = f"{topic} video youtube OR .mp4 OR .mov"
-    # search_results = [] # This would be populated by a search_web call
-    
-    # For demonstration, let's return some dummy URLs. Replace with actual search logic.
-    # print(f"[TopicSearch] This is a placeholder. Actual search via Cascade's 'search_web' tool is expected.")
-    # print(f"[TopicSearch] If running standalone, implement search logic here.")
-
-    # The actual implementation will be handled by Cascade's search_web tool call
-    # when this function is called via .remote() by another function that Cascade is orchestrating.
-    # This function definition serves as a Modal-compatible stub for Cascade's tool.
-    
-    # This function is more of a declaration for Cascade to use its tool.
-    # The actual search logic will be implicitly handled by Cascade's tool call mechanism
-    # when `find_video_urls_for_topic.remote()` is used in a subsequent step orchestrated by Cascade.
-    
-    # If this function were to be *truly* self-contained within Modal and callable independently 
-    # *without* Cascade's direct tool invocation, it would need its own HTTP client and parsing logic here.
-    # However, given the context of Cascade's operation, this stub is appropriate for Cascade to inject its tool usage.
-
-    # The `search_web` tool will be called by Cascade when it orchestrates the call to this function.
-    # So, this Python function in `modal_whisper_app.py` mostly defines the signature and intent.
-    # We will rely on Cascade to make the actual search_web call and provide the results back to the orchestrator.
-
-    # This function, when called by Cascade, will trigger a `search_web` tool call. 
-    # The tool call will be made by Cascade, not by the Modal runtime directly.
-    # For now, let's assume this function's body is a placeholder for that interaction.
-    # The key is that the *calling* function (e.g., analyze_videos_by_topic) will use .remote(),
-    # and Cascade will manage the search_web tool call.
-
-    # To make this runnable standalone (for testing Modal part without Cascade), one might add: 
-    # if modal.is_local(): 
-    #     # basic requests/bs4 search or return dummy data
-    #     pass 
-
-    # For the flow with Cascade, this function primarily serves as a named Modal function
-    # that Cascade understands it needs to provide search results for.
-    # The actual search logic is deferred to Cascade's tool execution. 
-    # We will return an empty list here, expecting Cascade to populate it via its mechanisms when called.
-    print(f"[TopicSearch] Function '{find_video_urls_for_topic.__name__}' called. Expecting Cascade to perform web search.")
-    # This is a conceptual placeholder. The actual search will be done by Cascade's tool.
-    # When `analyze_videos_by_topic` calls `find_video_urls_for_topic.remote()`, 
-    # Cascade will execute its `search_web` tool and the result will be used.
-    return [] # Placeholder: Cascade will provide actual URLs via its search_web tool.
+    video_urls = []
+    try:
+        # Add a common user-agent to avoid getting blocked
+        # Let yt-dlp find ffmpeg in the PATH instead of hardcoding it
+        ydl_opts = {
+            'quiet': True,
+            'extract_flat': 'discard_in_playlist',
+            'force_generic_extractor': False,
+            'default_search': f"ytsearch{max_results}",
+            'noplaylist': True,
+            'prefer_ffmpeg': True,
+            'http_headers': {
+                'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36'
+            }
+        }
+        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+            # extract_info with a search query like 'ytsearchN:query' returns a playlist dictionary
+            search_result = ydl.extract_info(topic, download=False)
+            if search_result and 'entries' in search_result:
+                for entry in search_result['entries']:
+                    # Ensure entry is a dictionary and has 'webpage_url'
+                    if isinstance(entry, dict) and entry.get('webpage_url'):
+                        video_urls.append(entry['webpage_url'])
+                        # yt-dlp search might return more than max_results, so we cap it here
+                        if len(video_urls) >= max_results:
+                            break 
+            # Sometimes a single video result might not be in 'entries'
+            elif isinstance(search_result, dict) and search_result.get('webpage_url'): 
+                video_urls.append(search_result['webpage_url'])
+
+        # Ensure we don't exceed max_results if the loop didn't break early enough
+        video_urls = video_urls[:max_results]
+        print(f"[TopicSearch] Found {len(video_urls)} video URLs for topic '{topic}': {video_urls}")
+    except Exception as e:
+        print(f"[TopicSearch] Error finding videos for topic '{topic}': {e}")
+        import traceback
+        traceback.print_exc()
+    return video_urls
 
 # Helper function (not a Modal function) to extract video URLs from search results
 def extract_video_urls_from_search(search_results: List[Dict[str, str]], max_urls: int = 3) -> List[str]:
@@ -588,25 +753,15 @@ def extract_video_urls_from_search(search_results: List[Dict[str, str]], max_url
                         break
         if len(video_urls) >= max_urls:
             break
-    
-    print(f"[URL Extraction] Extracted {len(video_urls)} video URLs: {video_urls}")
-    return video_urls
-
-
 # === 7. Topic-Based Video Analysis Orchestrator ===
 @app.function(
-    image=video_analysis_image,
+    image=video_analysis_image_v2,
     secrets=[HF_TOKEN_SECRET],
-    gpu="any", # Child functions use GPU
-    timeout=3600  # Allow up to 1 hour for multiple video analyses
+    gpu="any",
+    timeout=3600
 )
-async def _download_and_analyze_one_video(client: httpx.AsyncClient, video_url: str, topic: str) -> Dict[str, Any]:
-    """Helper to download and analyze a single video. Returns result or error dict."""
-    print(f"[TopicAnalysisWorker] Processing video URL for topic '{topic}': {video_url}")
-    try:
-        # 1. Download video
-        print(f"[TopicAnalysisWorker] Downloading video from: {video_url}")
-        response = await client.get(video_url)
+async def analyze_videos_by_topic(video_urls: list, topic: str) -> dict:
+    # Analyze videos concurrently
         response.raise_for_status()  # Raise HTTPError for bad responses (4XX or 5XX)
         video_bytes = await response.aread()
         print(f"[TopicAnalysisWorker] Downloaded {len(video_bytes)} bytes from {video_url}")