import gradio as gr import logging import json import subprocess import sys import tempfile import os from urllib.parse import urlparse, parse_qs logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Try to import docker, but handle gracefully if not available try: import docker DOCKER_AVAILABLE = True docker_client = docker.from_env() docker_client.ping() logger.info("✅ Docker client connected successfully") except ImportError: DOCKER_AVAILABLE = False docker_client = None logger.warning("⚠️ Docker module not available, using subprocess fallback") except Exception as e: DOCKER_AVAILABLE = False docker_client = None logger.warning(f"⚠️ Docker connection failed: {e}") # Fallback: Use subprocess with yt-dlp and yt-search-python installed directly def search_youtube_subprocess(query, max_results=15): """Search YouTube using subprocess with yt-search-python""" try: logger.info(f"🔍 Subprocess search: {query}") # Create Python script for searching search_script = f''' import json from yt_search import YouTubeSearch import sys try: results = YouTubeSearch('{query}', max_results={max_results}).to_dict() videos = [] for video in results: videos.append({{ 'title': video.get('title', 'No Title'), 'url': 'https://youtube.com' + video.get('url_suffix', ''), 'thumbnail': video.get('thumbnails', [{{}}])[0].get('url', ''), 'channel': video.get('channel', 'Unknown'), 'duration': video.get('duration', 'N/A'), 'views': video.get('views', 'N/A') }}) print(json.dumps(videos)) except Exception as e: print(f"ERROR: {{e}}") sys.exit(1) ''' # Run the script result = subprocess.run( [sys.executable, "-c", search_script], capture_output=True, text=True, timeout=30 ) if result.returncode != 0: logger.error(f"Search error: {result.stderr}") return [], f"Search error: {result.stderr}" output = result.stdout.strip() if output.startswith("ERROR:"): return [], output videos = json.loads(output) logger.info(f"✅ Found {len(videos)} videos") return videos, None except subprocess.TimeoutExpired: return [], "Search timed out" except Exception as e: logger.error(f"❌ Search error: {e}") return [], str(e) def play_youtube_subprocess(url): """Play YouTube video using subprocess with yt-dlp""" if not url or not url.strip(): return "

⚠️ Please enter a YouTube URL

" try: logger.info(f"🎬 Subprocess playback: {url}") # Create Python script for playing play_script = f''' import json import yt_dlp import sys import os # Disable logging import logging logging.disable(logging.CRITICAL) try: ydl_opts = {{ 'quiet': True, 'no_warnings': True, 'extract_flat': False, 'format': 'best[ext=mp4]', 'no_cache': True, 'ignoreerrors': True, 'geo_bypass': True, }} with yt_dlp.YoutubeDL(ydl_opts) as ydl: info = ydl.extract_info('{url}', download=False) if info: result = {{ 'title': info.get('title', 'Video'), 'id': info.get('id', ''), 'url': info.get('url', ''), 'thumbnail': info.get('thumbnail', ''), 'duration': info.get('duration', 0), 'formats': [] }} # Get formats formats = info.get('formats', []) video_url = None # Try to get best quality mp4 for f in formats: if f.get('ext') == 'mp4': if not video_url or (f.get('height') and f.get('height') > 720): video_url = f.get('url') if f.get('height') and f.get('height') >= 720: break if not video_url and info.get('url'): video_url = info.get('url') result['video_url'] = video_url print(json.dumps(result)) else: print('ERROR: Could not extract video info') sys.exit(1) except Exception as e: print(f"ERROR: {{e}}") sys.exit(1) ''' # Run the script result = subprocess.run( [sys.executable, "-c", play_script], capture_output=True, text=True, timeout=60 ) if result.returncode != 0: logger.error(f"Playback error: {result.stderr}") return f"

❌ Playback error: {result.stderr}

" output = result.stdout.strip() if output.startswith("ERROR:"): return f"

❌ {output}

" video_data = json.loads(output) video_url = video_data.get('video_url') video_title = video_data.get('title', 'Video') video_id = video_data.get('id', '') if video_url: logger.info(f"✅ Playing: {video_title}") logger.info(f"📊 Video ID: {video_id}") logger.info(f"🔗 URL: {url}") return f"""

▶️ {video_title}

📺 Playing: {url}
⏱️ Duration: {video_data.get('duration', 0)} seconds
""" else: return f"""

❌ Could not get video URL

💡 Try opening directly: {url}

""" except subprocess.TimeoutExpired: return "

⏱️ Playback timed out. Try a different video.

" except Exception as e: logger.error(f"❌ Playback error: {e}") return f"""

❌ Error: {str(e)}

💡 Try this URL directly: {url}

""" # Choose method based on availability if DOCKER_AVAILABLE: def search_youtube(query, max_results=15): """Search using Docker if available""" try: videos, error = search_youtube_docker(query, max_results) if error: logger.warning(f"Docker search failed, falling back to subprocess: {error}") return search_youtube_subprocess(query, max_results) return videos, None except: return search_youtube_subprocess(query, max_results) def play_youtube(url): """Play using Docker if available""" try: result = play_youtube_docker(url) # Check if result contains error if "❌" in result and "Docker" in result: logger.warning("Docker playback failed, falling back to subprocess") return play_youtube_subprocess(url) return result except: return play_youtube_subprocess(url) else: search_youtube = search_youtube_subprocess play_youtube = play_youtube_subprocess def create_video_grid(videos): """Create HTML grid from video results""" if not videos: return "

❌ No videos found. Please try a different search term.

" html = """
""" for video in videos: video_id = video['url'].split('watch?v=')[-1] if 'watch?v=' in video['url'] else video['url'] html += f"""
{video['title']}
{video['title'][:50]}{'...' if len(video['title']) > 50 else ''}
📺 {video['channel'][:25]}
⏱️ {video['duration']} | 👁️ {video['views']}
""" html += """
""" return html def handle_search(query): """Handle search request""" if not query or not query.strip(): return "

⚠️ Please enter a search term.

" videos, error = search_youtube(query, max_results=15) if error: return f"

❌ Search error: {error}

" return create_video_grid(videos) # Gradio Interface with gr.Blocks(title="YouTube Browser", theme=gr.themes.Soft()) as demo: gr.Markdown(""" # 🎬 YouTube Browser Search YouTube, browse thumbnails, and play videos directly in this Space! """) # Status indicator if DOCKER_AVAILABLE: gr.Markdown("✅ **Docker mode**: Using Docker for isolation") else: gr.Markdown("⚠️ **Subprocess mode**: Running directly (Docker not available)") with gr.Row(): with gr.Column(scale=4): search_input = gr.Textbox( label="Search YouTube", placeholder="Enter search term...", value="", container=False ) with gr.Column(scale=1): search_btn = gr.Button("🔍 Search", variant="primary", size="lg") # Search results results_display = gr.HTML( label="Search Results", value="

🔍 Search for videos to see results here!

" ) gr.Markdown("---") # Player section gr.Markdown("### 🎥 Video Player") with gr.Row(): with gr.Column(scale=4): url_input = gr.Textbox( label="YouTube URL", placeholder="https://www.youtube.com/watch?v=...", container=False, elem_id="url_input" ) with gr.Column(scale=1): play_btn = gr.Button("▶️ Play Video", variant="primary", size="lg", elem_id="play_btn") player_display = gr.HTML( label="Player", value="

💡 Search for a video or paste a URL above to play.

" ) # Event handlers search_btn.click( fn=handle_search, inputs=search_input, outputs=results_display ) search_input.submit( fn=handle_search, inputs=search_input, outputs=results_display ) play_btn.click( fn=play_youtube, inputs=url_input, outputs=player_display ) url_input.submit( fn=play_youtube, inputs=url_input, outputs=player_display ) gr.Markdown(""" ### 💡 How to use: 1. **Search**: Enter a search term and click "Search" 2. **Browse**: Scroll through thumbnails 3. **Watch**: Click "Play" on any video 4. **Direct**: Or paste any YouTube URL directly ### 📊 Features: - Search YouTube with thumbnails - Play videos directly in the Space - All URLs logged to console - Click thumbnails to load videos """) demo.launch()