update for streaming youtube url

app.py

@@ -5,6 +5,7 @@ from qdrant_client import QdrantClient
 import subprocess
 import os
 import uuid
+import yt_dlp
 
 # Setup CLIP
 device = "cuda" if torch.cuda.is_available() else "cpu"
@@ -18,37 +19,145 @@ client = QdrantClient(
 COLLECTION_NAME = "video_segments"
 
 # Paths
-VIDEO_BASE_DIR = "/project/phan/tqn/RAG-VideoReferencing/"
 CLIP_OUTPUT_DIR = "generated_clips"
 os.makedirs(CLIP_OUTPUT_DIR, exist_ok=True)
 
-DEFAULT_VIDEO_FILENAME = "temp_video_0.mp4"
-DEFAULT_VIDEO_PATH = os.path.join(VIDEO_BASE_DIR, DEFAULT_VIDEO_FILENAME)
-
-
-def extract_video_clip(input_path, start_time, end_time):
+# YouTube URLs mapping
+VIDEO_URLS = {
+    "temp_video_0.mp4": 'https://www.youtube.com/watch?v=9CGGh6ivg68',
+    "temp_video_1.mp4": 'https://www.youtube.com/watch?v=WXoOohWU28Y', 
+    "temp_video_2.mp4": 'https://www.youtube.com/watch?v=TV-DjM8242s', 
+    "temp_video_3.mp4": 'https://www.youtube.com/watch?v=rCVlIVKqqGE',
+    "temp_video_4.mp4": 'https://www.youtube.com/watch?v=lb_5AdUpfuA',
+    "temp_video_5.mp4": 'https://www.youtube.com/watch?v=FCQ-rih6cHY', 
+    "temp_video_6.mp4": 'https://www.youtube.com/watch?v=eQ6UE968Xe4', 
+    "temp_video_7.mp4": 'https://www.youtube.com/watch?v=eFgkZKhNUdM'
+}
+
+DEFAULT_VIDEO_URL = VIDEO_URLS["temp_video_0.mp4"]
+
+def extract_video_clip(video_url, start_time, end_time):
     """
-    Use ffmpeg to extract a clip from the video.
+    Use yt-dlp and ffmpeg to extract a clip directly from YouTube.
     """
     clip_name = f"clip_{uuid.uuid4().hex}.mp4"
     output_path = os.path.join(CLIP_OUTPUT_DIR, clip_name)
-
-    command = [
-        "ffmpeg",
-        "-ss", str(start_time),
-        "-i", input_path,
-        "-to", str(end_time - start_time),
-        "-c", "copy",
-        output_path,
-        "-y"  # Overwrite if exists
-    ]
-
+    duration = end_time - start_time
+    
+    print(f"[INFO] Attempting to extract clip from {video_url} ({start_time} - {end_time})")
+    
+    # Method 1: Use youtube-dl with ffmpeg
+    try:
+        # First, get the best direct URL from youtube-dl
+        ydl_opts = {
+            'format': 'bestvideo[ext=mp4]+bestaudio[ext=m4a]/best[ext=mp4]/best',
+            'quiet': True
+        }
+        
+        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+            info = ydl.extract_info(video_url, download=False)
+            formats = info.get('formats', [info])
+            
+            # Get the best format URL
+            best_url = None
+            for f in formats:
+                if f.get('ext') == 'mp4' and f.get('url'):
+                    best_url = f['url']
+                    break
+            
+            if not best_url and info.get('url'):
+                best_url = info['url']
+                
+            if not best_url:
+                print("[WARN] Could not find a suitable direct URL")
+                raise Exception("No suitable URL found")
+                
+            # Use ffmpeg to download the segment
+            command = [
+                "ffmpeg",
+                "-ss", str(start_time),
+                "-i", best_url,
+                "-t", str(duration),
+                "-c:v", "libx264",
+                "-c:a", "aac",
+                "-preset", "ultrafast",
+                output_path,
+                "-y"
+            ]
+            
+            print(f"[INFO] Running ffmpeg command: {' '.join(command)}")
+            result = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+            
+            if result.returncode != 0:
+                print(f"[WARN] ffmpeg command failed with code {result.returncode}")
+                print(f"[WARN] stderr: {result.stderr.decode('utf-8')}")
+                raise Exception(f"ffmpeg failed with code {result.returncode}")
+            
+            if os.path.exists(output_path) and os.path.getsize(output_path) > 0:
+                print(f"[INFO] Successfully extracted clip to {output_path}")
+                return output_path
+            else:
+                print(f"[WARN] Output file missing or empty: {output_path}")
+                raise Exception("Output file missing or empty")
+                
+    except Exception as e:
+        print(f"[ERROR] Method 1 failed: {str(e)}")
+    
+    # Method 2: Download whole video first
     try:
-        subprocess.run(command, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
-        return output_path
-    except subprocess.CalledProcessError as e:
-        print(f"[ERROR] FFmpeg failed: {e}")
-        return None
+        print("[INFO] Trying Method 2: Download full video first")
+        temp_video = os.path.join(CLIP_OUTPUT_DIR, f"temp_{uuid.uuid4().hex}.mp4")
+        
+        ydl_opts = {
+            'format': 'best[ext=mp4]/best',
+            'outtmpl': temp_video,
+            'quiet': True
+        }
+        
+        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+            ydl.download([video_url])
+        
+        if os.path.exists(temp_video) and os.path.getsize(temp_video) > 0:
+            # Extract clip from the downloaded video
+            command = [
+                "ffmpeg",
+                "-ss", str(start_time),
+                "-i", temp_video,
+                "-t", str(duration),
+                "-c:v", "copy",
+                "-c:a", "copy",
+                output_path,
+                "-y"
+            ]
+            
+            print(f"[INFO] Running ffmpeg command: {' '.join(command)}")
+            result = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+            
+            # Clean up temp file regardless of success
+            try:
+                os.remove(temp_video)
+                print(f"[INFO] Removed temporary file {temp_video}")
+            except Exception as cleanup_error:
+                print(f"[WARN] Failed to remove temp file: {cleanup_error}")
+            
+            if result.returncode != 0:
+                print(f"[WARN] ffmpeg command failed with code {result.returncode}")
+                print(f"[WARN] stderr: {result.stderr.decode('utf-8')}")
+                raise Exception(f"ffmpeg failed with code {result.returncode}")
+            
+            if os.path.exists(output_path) and os.path.getsize(output_path) > 0:
+                print(f"[INFO] Successfully extracted clip to {output_path}")
+                return output_path
+            else:
+                print(f"[WARN] Output file missing or empty: {output_path}")
+                raise Exception("Output file missing or empty")
+    
+    except Exception as e:
+        print(f"[ERROR] Method 2 failed: {str(e)}")
+    
+    # Both methods failed
+    print("[ERROR] All extraction methods failed")
+    return None
 
 def time_to_seconds(time_str):
     h, m, s = time_str.split(':')
@@ -62,8 +171,8 @@ def search_and_clip_video(text_query: str):
         text_tokens = clip.tokenize([text_query]).to(device)
         text_features = model.encode_text(text_tokens)
         text_features /= text_features.norm(dim=1, keepdim=True)
-
-    # Query Qdrant
+    
+    # Always use the original search method
     search_result = client.search(
         collection_name=COLLECTION_NAME,
         query_vector=text_features.cpu().numpy()[0].tolist(),
@@ -72,34 +181,40 @@ def search_and_clip_video(text_query: str):
 
     if not search_result:
         print("[WARN] No result found.")
-        return DEFAULT_VIDEO_PATH
+        return DEFAULT_VIDEO_URL
 
     hit = search_result[0]
     start = hit.payload.get("start", 0)
     end = hit.payload.get("end", 0)
-    start = time_to_seconds(start)
-    end = time_to_seconds(end)
-    video_filename = hit.payload.get("video_path", DEFAULT_VIDEO_FILENAME)
+    start = time_to_seconds(start) if isinstance(start, str) else float(start)
+    end = time_to_seconds(end) if isinstance(end, str) else float(end)
+    video_filename = hit.payload.get("video_path", "temp_video_0.mp4")
 
-    full_video_path = os.path.join(VIDEO_BASE_DIR, video_filename)
+    # Get YouTube URL from filename
+    video_url = VIDEO_URLS.get(video_filename, DEFAULT_VIDEO_URL)
 
     print(f"[INFO] Found: {video_filename} ({start} - {end})")
+    print(f"[INFO] Using YouTube URL: {video_url}")
+
+    # Handle very short clips or invalid timestamps
+    if end <= start or end - start < 1.0:
+        print("[WARN] Invalid clip duration, using default start/end times")
+        start = max(0, start - 5)  # Start 5 seconds before
+        end = start + 10           # 10 second clip
 
-    # Extract clip using ffmpeg
-    clip_path = extract_video_clip(full_video_path, float(start), float(end))
+    # Extract clip from YouTube
+    clip_path = extract_video_clip(video_url, float(start), float(end))
     if clip_path and os.path.exists(clip_path):
         print(f"[INFO] Returning clip: {clip_path}")
-        return clip_path
+        return clip_path 
     else:
         print("[WARN] Failed to extract clip, returning default video.")
-        return DEFAULT_VIDEO_PATH
+        return DEFAULT_VIDEO_URL
 
-
-# Fallback test interface
+# Function to get a test video
 def get_test_video():
-    print("[INFO] Returning test video path")
-    return DEFAULT_VIDEO_PATH
-
+    print("[INFO] Returning test YouTube URL")
+    return DEFAULT_VIDEO_URL
 
 # Gradio Interfaces
 search_demo = gr.Interface(
@@ -124,4 +239,4 @@ demo = gr.TabbedInterface(
 )
 
 if __name__ == "__main__":
-    demo.launch(share=True)
+    demo.launch(share=True)

requirements.txt

@@ -5,4 +5,6 @@ opencv-python
 qdrant-client
 uuid
 git+https://github.com/openai/CLIP.git
+ffmpeg-python
+yt-dlp
 ffmpeg-python