Update app.py

app.py

@@ -1,84 +1,132 @@
-import requests
-import json
+import subprocess
+import sys
+
+# Ensure compatible versions of httpx and httpcore are installed
+subprocess.check_call([sys.executable, "-m", "pip", "install", "httpx==0.18.2", "httpcore==0.13.6"])
+
 import gradio as gr
+import requests
+import re
+import yt_dlp
 import logging
-from bs4 import BeautifulSoup
 
 # Configure logging for debugging purposes
 logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')
 
-# Function to search YouTube without using the API
-def youtube_search(query, max_results=10):
-    # Create the YouTube search URL
-    search_url = f"https://www.youtube.com/results?search_query={query.replace(' ', '+')}"
-    headers = {
-        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/85.0.4183.83 Safari/537.36"
+# Function to search YouTube videos using yt-dlp for better reliability
+def youtube_search(query, max_results=50):
+    ydl_opts = {
+        'quiet': False,  # Set to False to get more detailed output from yt-dlp
+        'extract_flat': 'in_playlist',
+        'logger': logging.getLogger(),  # Use the logging module to capture yt-dlp logs
+        'simulate': True,
+        'noplaylist': True,  # To avoid playlist entries
     }
+    search_url = f"ytsearch{max_results}:{query}"
 
     logging.debug(f"Starting YouTube search for query: {query}")
 
     try:
-        response = requests.get(search_url, headers=headers)
-        response.raise_for_status()  # Raise an error for bad status codes
-
-        # Parse the HTML response using BeautifulSoup
-        soup = BeautifulSoup(response.text, "html.parser")
-
-        # Look for the initial JSON data block within the script tags
-        scripts = soup.find_all("script")
-        for script in scripts:
-            if 'var ytInitialData = ' in script.text:
-                json_text = script.text.split('var ytInitialData = ')[1].split("};")[0] + "}"
-                data = json.loads(json_text)
-
-                # Traverse through the JSON to find video entries
-                video_items = []
-                contents = data['contents']['twoColumnSearchResultsRenderer']['primaryContents']['sectionListRenderer']['contents']
-                for content in contents:
-                    video_entries = content.get('itemSectionRenderer', {}).get('contents', [])
-                    for entry in video_entries:
-                        video_renderer = entry.get('videoRenderer')
-                        if video_renderer:
-                            video_id = video_renderer.get('videoId', 'N/A')
-                            video_title = video_renderer.get('title', {}).get('runs', [{}])[0].get('text', 'N/A')
-                            video_url = f"https://www.youtube.com/watch?v={video_id}"
-                            thumbnail_url = video_renderer.get('thumbnail', {}).get('thumbnails', [{}])[0].get('url', 'N/A')
-
-                            video_items.append((thumbnail_url, video_title, video_url))
-
-                            if len(video_items) >= max_results:
-                                break
-
-                if video_items:
-                    return video_items, ""  # Return the list of video items and no error message
-                else:
-                    logging.warning("No video entries found.")
-                    return [], "No video entries found."
-        logging.warning("JSON data block not found in the page.")
-        return [], "Unable to find video data."
-
-    except requests.exceptions.RequestException as e:
-        error_message = f"Request error occurred: {e}"
-        logging.error(error_message)
-        return [], error_message
-    except json.JSONDecodeError as e:
-        error_message = f"JSON decoding error occurred: {e}"
-        logging.error(error_message)
-        return [], error_message
+        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+            result = ydl.extract_info(search_url, download=False)
+            gallery_items = []
+
+            if 'entries' in result:
+                logging.debug(f"Number of entries found: {len(result['entries'])}")
+                for entry in result['entries']:
+                    video_id = entry.get('id')
+                    thumbnail_url = entry.get('thumbnail') if entry.get('thumbnail') else "https://via.placeholder.com/150"
+                    video_title = entry.get('title', "Unknown Title")
+
+                    if video_id:
+                        gallery_items.append((thumbnail_url, video_id, video_title))
+                        logging.debug(f"Added video: ID={video_id}, Thumbnail={thumbnail_url}, Title={video_title}")
+                    else:
+                        logging.debug(f"Missing video ID for entry: {entry}")
+            else:
+                logging.warning("No entries found in search result.")
+            return gallery_items, ""
     except Exception as e:
-        error_message = f"An unexpected error occurred: {e}"
+        error_message = f"Error during YouTube yt-dlp request: {e}"
         logging.error(error_message)
         return [], error_message
 
-# Function to create a gallery for Gradio
-def display_videos(query):
-    gallery_items, error_message = youtube_search(query)
-    if error_message:
-        return gr.update(value=[], label="Error: Unable to fetch videos"), error_message
+# Function to display the video using the video URL
+def show_video(video_url):
+    # Regular expression to extract the YouTube video ID from the URL
+    video_id = None
+    patterns = [
+        r"youtube\.com/watch\?v=([^&?\/]+)",
+        r"youtube\.com/embed/([^&?\/]+)",
+        r"youtube\.com/v/([^&?\/]+)",
+        r"youtu\.be/([^&?\/]+)"
+    ]
+
+    for pattern in patterns:
+        match = re.search(pattern, video_url)
+        if match:
+            video_id = match.group(1)
+            logging.debug(f"Extracted video ID: {video_id}")
+            break
+
+    # If no video ID is found, return an error message
+    if not video_id:
+        logging.error("Invalid YouTube URL. Please enter a valid YouTube video link.")
+        return "Invalid YouTube URL. Please enter a valid YouTube video link."
+
+    # Create the embed URL
+    embed_url = f"https://www.youtube.com/embed/{video_id}"
+    logging.debug(f"Embed URL generated: {embed_url}")
+
+    # Return an iframe with the video
+    html_code = f'''
+    <iframe width="560" height="315" src="{embed_url}" 
+    frameborder="0" allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture" 
+    allowfullscreen></iframe>
+    '''
+    return html_code
+
+# Create the Gradio interface
+with gr.Blocks() as demo:
+    gr.Markdown("## YouTube Video Search, Selection, and Playback")
+
+    with gr.Row():
+        with gr.Column(scale=3):
+            search_query_input = gr.Textbox(label="Search YouTube", placeholder="Enter your search query here")
+            search_button = gr.Button("Search")
+            search_output = gr.Gallery(label="Search Results", columns=5, height="1500px")
+            error_output = gr.Textbox(label="Error Message", interactive=False, visible=False)
+        
+        with gr.Column(scale=2):
+            selected_video_link = gr.Textbox(label="Selected Video Link", interactive=False)
+            play_video_button = gr.Button("Play Video")
+            video_output = gr.HTML(label="Video Player")
+
+    # Define search button behavior
+    def update_search_results(query):
+        gallery_items, error_message = youtube_search(query)
+        if error_message:
+            return [], error_message, gr.update(visible=True)
+        # Display videos even if the title or thumbnail is missing by using placeholders
+        gallery_items_display = [(item[0], item[1]) for item in gallery_items]  # Show thumbnail and video ID only
+        return gallery_items_display, "", gr.update(visible=False)
+
+    # Update the selected video link field when a video is clicked in the gallery
+    def on_video_select(evt: gr.SelectData):
+        # Extract the video ID from the event value, which is a dictionary containing details of the selected item
+        selected_video_id = evt.value["caption"]
+        video_url = f"https://www.youtube.com/watch?v={selected_video_id}"
+        logging.debug(f"Video selected: {video_url}")
+        return video_url
+
+    # Play the video when the Play Video button is clicked
+    def play_video(video_url):
+        logging.debug(f"Playing video with URL: {video_url}")
+        return show_video(video_url)
 
-    # Display thumbnails and titles as gallery items
-    formatted_gallery = [(f"<img src='{item[0]}' width='250'/>", f"<a href='{item[2]}' target='_blank'>{item[1]}</a>") for item in gallery_items]
-    return formatted_gallery, ""
+    search_button.click(update_search_results, inputs=search_query_input, outputs=[search_output, error_output, error_output])
+    search_output.select(on_video_select, inputs=None, outputs=selected_video_link)
+    play_video_button.click(play_video, inputs=selected_video_link, outputs=video_output)
 
-# Gradio interface
-with gr
+# Launch the Gradio interface
+demo.launch()