app.py

import subprocess
import sys
import random
import time

# 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
import os
from fake_useragent import UserAgent

# Configure logging for debugging purposes
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')

# Set up User-Agent rotation
ua = UserAgent()

# Define a list of proxies (example placeholders, replace with your working proxies)
PROXIES = [
    "http://proxy1.example.com:8080",
    "http://proxy2.example.com:8080",
    "http://proxy3.example.com:8080",
]

# Function to search YouTube videos using yt-dlp for better reliability
def youtube_search(query, max_results=10):
    cookies_file = "cookies.txt"  # You need to provide this file with cookies exported from YouTube
    proxies_attempted = 0
    max_proxy_attempts = len(PROXIES)
    success = False
    gallery_items = []
    error_message = ""

    while not success and proxies_attempted <= max_proxy_attempts:
        if proxies_attempted < max_proxy_attempts:
            # Use a proxy
            proxy = PROXIES[proxies_attempted]
            logging.debug(f"Trying proxy: {proxy}")
            ydl_opts = {
                'quiet': False,  # Set to False to get more detailed output from yt-dlp
                'logger': logging.getLogger(),  # Use the logging module to capture yt-dlp logs
                'simulate': True,
                'noplaylist': True,  # Avoid extracting playlists
                'format': 'best',
                'proxy': proxy,
                'http_headers': {
                    'User-Agent': ua.random  # Rotate user agents to make requests appear less like a bot
                },
            }
        else:
            # Fallback to direct access after all proxies fail
            logging.debug("Falling back to direct connection without a proxy.")
            ydl_opts = {
                'quiet': False,
                'logger': logging.getLogger(),
                'simulate': True,
                'noplaylist': True,
                'format': 'best',
                'http_headers': {
                    'User-Agent': ua.random
                },
            }

        if os.path.exists(cookies_file):
            ydl_opts['cookiefile'] = cookies_file
            logging.debug("Using cookies for YouTube authentication.")

        search_url = f"ytsearch{max_results}:{query}"
        logging.debug(f"Starting YouTube search for query: {query}")

        try:
            # Introduce a random delay to avoid rate-limiting issues
            time.sleep(random.uniform(2, 5))

            with yt_dlp.YoutubeDL(ydl_opts) as ydl:
                result = ydl.extract_info(search_url, download=False)

                if 'entries' in result:
                    logging.debug(f"Number of entries found: {len(result['entries'])}")
                    for entry in result['entries']:
                        video_id = entry.get('id')
                        # Fallback to YouTube static thumbnails if missing
                        thumbnail_url = entry.get('thumbnail') if entry.get('thumbnail') else f"https://img.youtube.com/vi/{video_id}/hqdefault.jpg"
                        video_title = entry.get('title', "Unknown Title")
                        video_description = entry.get('description', "No description available.")

                        if video_id:
                            gallery_items.append({
                                "thumbnail": thumbnail_url,
                                "video_id": video_id,
                                "title": video_title,
                                "description": video_description
                            })
                            logging.debug(f"Added video: ID={video_id}, Thumbnail={thumbnail_url}, Title={video_title}")
                        else:
                            logging.debug(f"Missing video ID for entry: {entry}")

                    success = True
                else:
                    logging.warning("No entries found in search result.")

        except Exception as e:
            error_message = f"Error during YouTube yt-dlp request: {e}"
            logging.error(error_message)
            proxies_attempted += 1  # Increment the proxy attempt counter

    if not success:
        return [], error_message
    return gallery_items, ""

# 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=2, height="1000px", elem_id="gallery")
            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)
        
        # Prepare gallery items
        gallery_items_display = [(item["thumbnail"], f"{item['title']}\n{item['description']}", item["video_id"]) for item in gallery_items]
        
        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)

    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)

# Launch the Gradio interface
demo.launch()