app.py

import gradio as gr
import tempfile
import os
import yt_dlp
import re
from pathlib import Path
import google.generativeai as genai
from google.generativeai import upload_file, get_file
import time

# Configure Google AI
GOOGLE_API_KEY = os.getenv("GOOGLE_API_KEY")
if GOOGLE_API_KEY:
    genai.configure(api_key=GOOGLE_API_KEY)
else:
    raise ValueError("GOOGLE_API_KEY environment variable not set")

def is_valid_url(url):
    """Check if URL is from supported platforms"""
    patterns = [
        r'(https?://)?(www\.)?(youtube\.com|youtu\.be)',
        r'(https?://)?(www\.)?(instagram\.com|instagr\.am)',
        r'(https?://)?(www\.)?(tiktok\.com)',
        r'(https?://)?(www\.)?(twitter\.com|x\.com)',
    ]
    
    for pattern in patterns:
        if re.search(pattern, url, re.IGNORECASE):
            return True
    return False

def get_video_info(url):
    """Get video information without downloading"""
    try:
        # Extract video ID for YouTube
        if 'youtube.com' in url or 'youtu.be' in url:
            video_id = None
            if 'youtu.be/' in url:
                video_id = url.split('youtu.be/')[-1].split('?')[0]
            elif 'watch?v=' in url:
                video_id = url.split('watch?v=')[-1].split('&')[0]
            
            if video_id:
                # Use YouTube oEmbed API (no key required, often works better)
                import requests
                oembed_url = f"https://www.youtube.com/oembed?url=https://www.youtube.com/watch?v={video_id}&format=json"
                response = requests.get(oembed_url, timeout=10)
                if response.status_code == 200:
                    data = response.json()
                    return {
                        'title': data.get('title', 'Unknown Video'),
                        'author': data.get('author_name', 'Unknown Author'),
                        'thumbnail': data.get('thumbnail_url', ''),
                    }
    except Exception as e:
        print(f"Could not get video info: {e}")
    
    return None

def analyze_video_from_url(url, query):
    """Analyze video based on URL and metadata instead of downloading"""
    # Get video information
    video_info = get_video_info(url)
    
    # Create context from URL and available info
    context = f"Video URL: {url}\n"
    if video_info:
        context += f"Title: {video_info['title']}\n"
        context += f"Author: {video_info['author']}\n"
    
    # Generate AI response based on URL and context
    try:
        model = genai.GenerativeModel('gemini-2.0-flash-exp')
        
        prompt = f"""
        I have a video with the following information:
        {context}
        
        User question: {query}
        
        Based on the video URL and available metadata, please provide helpful analysis and insights. 
        Since I cannot directly access the video content, please:
        
        1. Analyze what type of content this might be based on the URL and title
        2. Provide general guidance about analyzing this type of video
        3. Suggest what insights could typically be extracted
        4. Give relevant advice based on the platform (YouTube, Instagram, etc.)
        
        Be helpful and informative while acknowledging the limitations of not having direct video access.
        If this appears to be a specific type of content (tutorial, entertainment, news, etc.), 
        provide relevant analysis frameworks and questions that would be useful.
        
        Be conversational and engaging.
        """
        
        response = model.generate_content(prompt)
        return response.text
        
    except Exception as e:
        raise gr.Error(f"AI analysis failed: {str(e)}")

def download_video(url, progress=gr.Progress()):
    """Download video from URL using yt-dlp with better error handling"""
    if not is_valid_url(url):
        raise gr.Error("Please enter a valid YouTube, Instagram, TikTok, or X video URL")
    
    progress(0.1, desc="Starting download...")
    
    # Create temp file
    temp_video = tempfile.NamedTemporaryFile(delete=False, suffix='.mp4')
    temp_video.close()

    # yt-dlp options optimized for HF Spaces with better network handling
    ydl_opts = {
        'outtmpl': temp_video.name,
        'format': 'best[filesize<50M]/worst[filesize<50M]/best',  # Smaller files for HF
        'quiet': True,
        'no_warnings': True,
        'nooverwrites': False,
        'user_agent': 'Mozilla/5.0 (compatible; GradioApp/1.0)',
        'retries': 1,  # Reduced for HF Spaces
        'fragment_retries': 1,
        'extractor_retries': 1,
        'socket_timeout': 15,  # Shorter timeout
        'nocheckcertificate': True,
        'prefer_insecure': True,  # Sometimes helps with network issues
    }

    progress(0.3, desc="Downloading video...")
    
    try:
        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
            ydl.download([url])

        # Verify download
        if os.path.exists(temp_video.name) and os.path.getsize(temp_video.name) > 0:
            file_size_mb = os.path.getsize(temp_video.name) / (1024 * 1024)
            
            if file_size_mb > 50:  # Smaller limit for HF
                os.unlink(temp_video.name)
                raise gr.Error(f"Video too large ({file_size_mb:.1f}MB). Please use a smaller video.")
            
            progress(0.7, desc="Processing video...")
            return temp_video.name
        else:
            raise gr.Error("Downloaded file is empty or doesn't exist")

    except Exception as e:
        # Clean up on error
        if os.path.exists(temp_video.name):
            try:
                os.unlink(temp_video.name)
            except:
                pass
        
        error_msg = str(e)
        
        # Handle specific network errors common in HF Spaces
        if "Failed to resolve" in error_msg or "No address associated with hostname" in error_msg:
            raise gr.Error("🌐 Network connectivity issue. Hugging Face Spaces may have restricted access to this platform. Try a different video or platform.")
        elif "403" in error_msg or "Forbidden" in error_msg:
            raise gr.Error("🚫 Video download was blocked by the platform. Try a different video.")
        elif "404" in error_msg or "not found" in error_msg.lower():
            raise gr.Error("📹 Video not found. Please check the URL.")
        elif "timeout" in error_msg.lower():
            raise gr.Error("⏱️ Download timed out. Try a shorter video.")
        else:
            # For network errors, fall back to URL-based analysis
            print(f"Download failed, falling back to URL analysis: {error_msg}")
            raise Exception("download_failed")  # Special exception for fallback

def analyze_video_with_ai(video_path, query, progress=gr.Progress()):
    """Analyze video using Google Gemini AI"""
    if not video_path or not os.path.exists(video_path):
        raise gr.Error("No video file found")
    
    progress(0.1, desc="Uploading video to AI...")
    
    try:
        # Upload video to Google AI
        processed_video = upload_file(video_path)
        
        progress(0.5, desc="Processing video...")
        
        # Wait for processing
        while processed_video.state.name == "PROCESSING":
            time.sleep(2)
            processed_video = get_file(processed_video.name)
        
        if processed_video.state.name == "FAILED":
            raise gr.Error("Video processing failed")
        
        progress(0.8, desc="Generating response...")
        
        # Generate AI response
        model = genai.GenerativeModel('gemini-2.0-flash-exp')
        
        prompt = f"""
        Analyze this video and respond to the user's question: {query}
        
        Provide a comprehensive, insightful response that includes:
        1. Direct analysis of the video content
        2. Key insights and observations
        3. Specific details from what you can see/hear
        4. Actionable takeaways if relevant
        
        Be conversational and engaging while being thorough and accurate.
        """
        
        response = model.generate_content([processed_video, prompt])
        
        progress(1.0, desc="Complete!")
        
        return response.text
        
    except Exception as e:
        raise gr.Error(f"AI analysis failed: {str(e)}")
    
    finally:
        # Clean up video file
        try:
            if video_path and os.path.exists(video_path):
                os.unlink(video_path)
        except:
            pass

def process_video_and_chat(url, query):
    """Main function to download video and get AI response with fallback"""
    if not url.strip():
        raise gr.Error("Please enter a video URL")
    
    if not query.strip():
        raise gr.Error("Please enter a question about the video")
    
    progress = gr.Progress()
    progress(0.0, desc="Starting...")
    
    try:
        # Try to download video first
        video_path = download_video(url, progress)
        
        # Analyze with AI using actual video
        response = analyze_video_with_ai(video_path, query, progress)
        
        return response
        
    except Exception as e:
        # If download fails due to network issues, fall back to URL-based analysis
        if str(e) == "download_failed" or "Network connectivity" in str(e) or "Failed to resolve" in str(e):
            progress(0.5, desc="Switching to URL-based analysis...")
            
            # Add a notice about the fallback
            fallback_notice = "🔄 **Note**: Direct video download failed due to network restrictions, so I'm providing analysis based on the video URL and metadata.\n\n"
            
            try:
                response = analyze_video_from_url(url, query)
                return fallback_notice + response
            except Exception as fallback_error:
                raise gr.Error(f"Both video download and URL analysis failed: {str(fallback_error)}")
        else:
            # Re-raise the original error if it's not a network issue
            raise e

# Create Gradio interface
def create_interface():
    with gr.Blocks(
        title="The Plug - AI Video Analyzer",
        theme=gr.themes.Soft(),
        css="""
        .gradio-container {
            max-width: 800px !important;
            margin: auto !important;
        }
        .header {
            text-align: center;
            margin-bottom: 2rem;
        }
        .footer {
            text-align: center;
            margin-top: 2rem;
            color: #666;
        }
        """
    ) as demo:
        
        # Header
        gr.HTML("""
        <div class="header">
            <h1>🎥 The Plug - AI Video Analyzer</h1>
            <p>Analyze videos from YouTube, Instagram, TikTok, and X with AI</p>
            <div style="background: #f0f0f0; padding: 10px; border-radius: 8px; margin: 10px 0;">
                <small><strong>🔄 Smart Fallback:</strong> If video download fails due to network restrictions, 
                the app will analyze based on video metadata and URL instead.</small>
            </div>
        </div>
        """)
        
        with gr.Row():
            with gr.Column():
                # Video URL input
                video_url = gr.Textbox(
                    label="Video URL",
                    placeholder="https://youtube.com/watch?v=... or Instagram/TikTok/X link",
                    lines=1,
                    info="Paste a video URL from YouTube, Instagram, TikTok, or X"
                )
                
                # Query input
                query = gr.Textbox(
                    label="Your Question",
                    placeholder="What is the main topic? Summarize the key points...",
                    lines=3,
                    info="Ask anything about the video content"
                )
                
                # Submit button
                submit_btn = gr.Button("🚀 Analyze Video", variant="primary", size="lg")
        
        with gr.Row():
            with gr.Column():
                # Response output
                response = gr.Textbox(
                    label="AI Analysis",
                    lines=15,
                    max_lines=25,
                    interactive=False,
                    show_copy_button=True
                )
        
        # Example queries
        gr.Examples(
            examples=[
                ["https://www.youtube.com/watch?v=dQw4w9WgXcQ", "What is this video about?"],
                ["", "Summarize the key points mentioned"],
                ["", "What are the main takeaways?"],
                ["", "Who is the target audience?"],
            ],
            inputs=[video_url, query],
            label="Example Questions"
        )
        
        # Footer
        gr.HTML("""
        <div class="footer">
            <p>Built with ❤️ using Gradio and Google Gemini AI</p>
            <p>Supports YouTube, Instagram, TikTok, and X video URLs</p>
        </div>
        """)
        
        # Connect the function
        submit_btn.click(
            fn=process_video_and_chat,
            inputs=[video_url, query],
            outputs=response,
            show_progress=True
        )
        
        # Also allow Enter key to submit
        query.submit(
            fn=process_video_and_chat,
            inputs=[video_url, query],
            outputs=response,
            show_progress=True
        )
    
    return demo

# Launch the app
if __name__ == "__main__":
    demo = create_interface()
    demo.launch(
        share=True,
        server_name="0.0.0.0",
        server_port=7860,
        show_error=True
    )