Add MoviePy video overlay - complete pipeline

app.py

@@ -2,9 +2,12 @@ import gradio as gr
 import os
 import requests
 import random
+import tempfile
 from openai import OpenAI
 from smolagents import CodeAgent, MCPClient, tool
 from huggingface_hub import InferenceClient
+from moviepy.editor import VideoFileClip, TextClip, CompositeVideoClip
+from moviepy.config import change_settings
 
 # Initialize clients
 openai_client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
@@ -184,6 +187,107 @@ def search_pexels_video_tool(style: str, niche: str) -> dict:
             "error": str(e)
         }
 
+@tool
+def create_quote_video_tool(video_url: str, quote_text: str, output_path: str) -> dict:
+    """
+    Create a final quote video by overlaying text on the background video.
+    
+    Args:
+        video_url: URL of the background video from Pexels
+        quote_text: The quote text to overlay
+        output_path: Path where to save the final video
+    
+    Returns:
+        Dictionary with success status and output path
+    """
+    
+    try:
+        # Step 1: Download the video
+        response = requests.get(video_url, stream=True, timeout=30)
+        response.raise_for_status()
+        
+        # Create temporary file for downloaded video
+        temp_video = tempfile.NamedTemporaryFile(delete=False, suffix='.mp4')
+        
+        with open(temp_video.name, 'wb') as f:
+            for chunk in response.iter_content(chunk_size=8192):
+                f.write(chunk)
+        
+        # Step 2: Load video with MoviePy
+        video = VideoFileClip(temp_video.name)
+        
+        # Get video dimensions
+        w, h = video.size
+        
+        # Step 3: Create text overlay
+        # Calculate font size based on video height (responsive)
+        font_size = int(h * 0.06)  # 6% of video height
+        
+        # Split quote into multiple lines for better readability
+        words = quote_text.split()
+        lines = []
+        current_line = []
+        max_words_per_line = 4
+        
+        for word in words:
+            current_line.append(word)
+            if len(current_line) >= max_words_per_line:
+                lines.append(' '.join(current_line))
+                current_line = []
+        
+        if current_line:
+            lines.append(' '.join(current_line))
+        
+        # Join lines with newline
+        formatted_text = '\n'.join(lines)
+        
+        # Create text clip with styling
+        txt_clip = TextClip(
+            formatted_text,
+            fontsize=font_size,
+            color='white',
+            font='Arial-Bold',
+            stroke_color='black',
+            stroke_width=2,
+            method='caption',
+            size=(int(w * 0.9), None),  # 90% of video width
+            align='center'
+        )
+        
+        # Position text in center
+        txt_clip = txt_clip.set_position('center').set_duration(video.duration)
+        
+        # Step 4: Composite video with text
+        final_video = CompositeVideoClip([video, txt_clip])
+        
+        # Step 5: Export final video
+        final_video.write_videofile(
+            output_path,
+            codec='libx264',
+            audio_codec='aac',
+            temp_audiofile='temp-audio.m4a',
+            remove_temp=True,
+            fps=24
+        )
+        
+        # Cleanup
+        video.close()
+        final_video.close()
+        os.unlink(temp_video.name)
+        
+        return {
+            "success": True,
+            "output_path": output_path,
+            "message": "Video created successfully!"
+        }
+    
+    except Exception as e:
+        return {
+            "success": False,
+            "output_path": None,
+            "message": f"Error creating video: {str(e)}"
+        }
+
 # Initialize the MCP-powered agent
 def initialize_agent():
     """Initialize the CodeAgent with MCP capabilities"""
@@ -193,10 +297,10 @@ def initialize_agent():
         
         # Create agent with custom tools
         agent = CodeAgent(
-            tools=[generate_quote_tool, search_pexels_video_tool],
+            tools=[generate_quote_tool, search_pexels_video_tool, create_quote_video_tool],
             model=model,
-            additional_authorized_imports=["requests", "openai", "random"],
-            max_steps=5
+            additional_authorized_imports=["requests", "openai", "random", "tempfile", "os"],
+            max_steps=10
         )
         
         # Add MCP client if available
@@ -229,14 +333,14 @@ def mcp_agent_pipeline(niche, style):
         status_log.append("📋 **TASK RECEIVED:**")
         status_log.append(f"   → Generate {niche} quote with {style} aesthetic")
         status_log.append(f"   → Find matching video")
-        status_log.append(f"   → Prepare for video creation\n")
+        status_log.append(f"   → Create final quote video\n")
         
         # STEP 2: Agent executes quote generation
         status_log.append("🧠 **MCP TOOL: generate_quote_tool**")
         quote = generate_quote_tool(niche, style)
         
         if "Error" in quote:
-            return "\n".join(status_log) + f"\n❌ Failed: {quote}", None
+            return "\n".join(status_log) + f"\n❌ Failed: {quote}", None, None
         
         status_log.append(f"   ✅ Generated: \"{quote}\"\n")
         
@@ -246,12 +350,40 @@ def mcp_agent_pipeline(niche, style):
         
         if not video_result["success"]:
             error_msg = video_result.get("error", "Unknown error")
-            return "\n".join(status_log) + f"\n❌ Video search failed: {error_msg}", None
+            return "\n".join(status_log) + f"\n❌ Video search failed: {error_msg}", None, None
         
         status_log.append(f"   ✅ Found video: {video_result['search_query']}")
         status_log.append(f"   📥 Video URL: {video_result['video_url']}\n")
         
-        # STEP 4: MCP Server integration status
+        # STEP 4: Agent creates final video with text overlay
+        status_log.append("🎬 **MCP TOOL: create_quote_video_tool**")
+        status_log.append("   ⏳ Creating video with text overlay...")
+        
+        # Create output directory if it doesn't exist
+        output_dir = "/tmp/quote_videos"
+        os.makedirs(output_dir, exist_ok=True)
+        
+        # Generate unique filename
+        import time
+        output_filename = f"quote_video_{int(time.time())}.mp4"
+        output_path = os.path.join(output_dir, output_filename)
+        
+        # Create the video
+        creation_result = create_quote_video_tool(
+            video_result["video_url"],
+            quote,
+            output_path
+        )
+        
+        if not creation_result["success"]:
+            status_log.append(f"   ❌ Video creation failed: {creation_result['message']}")
+            status_log.append("\n📺 **PREVIEW MODE:**")
+            status_log.append("   Showing background video preview instead")
+            return "\n".join(status_log), video_result["video_url"], None
+        
+        status_log.append(f"   ✅ Video created successfully!\n")
+        
+        # STEP 5: MCP Server integration status
         status_log.append("🔗 **MCP SERVER STATUS:**")
         if mcp_enabled:
             status_log.append("   ✅ Connected to: abidlabs-mcp-tools.hf.space")
@@ -260,21 +392,17 @@ def mcp_agent_pipeline(niche, style):
             status_log.append("   ⚠️ MCP server connection pending")
         status_log.append("")
         
-        # STEP 5: Next steps
-        status_log.append("🚧 **NEXT STEPS (Coming Soon):**")
-        status_log.append("   → Text overlay with MoviePy")
-        status_log.append("   → Voice narration with ElevenLabs")
-        status_log.append("   → Final video export\n")
-        
-        status_log.append("🎬 **PIPELINE COMPLETE!**")
-        status_log.append(f"📱 View on Pexels: {video_result['pexels_url']}")
+        # STEP 6: Success!
+        status_log.append("✨ **PIPELINE COMPLETE!**")
+        status_log.append(f"   📱 Original video: {video_result['pexels_url']}")
+        status_log.append(f"   🎬 Final video ready for download!")
         
         final_status = "\n".join(status_log)
-        return final_status, video_result["video_url"]
+        return final_status, video_result["video_url"], creation_result["output_path"]
     
     except Exception as e:
         status_log.append(f"\n❌ Pipeline error: {str(e)}")
-        return "\n".join(status_log), None
+        return "\n".join(status_log), None, None
 
 def fallback_pipeline(niche, style):
     """Fallback pipeline if MCP agent fails"""
@@ -286,7 +414,7 @@ def fallback_pipeline(niche, style):
     quote = generate_quote_tool(niche, style)
     
     if "Error" in quote:
-        return "\n".join(status_log) + f"\n❌ {quote}", None
+        return "\n".join(status_log) + f"\n❌ {quote}", None, None
     
     status_log.append(f"   ✅ Quote: \"{quote}\"\n")
     
@@ -295,12 +423,33 @@ def fallback_pipeline(niche, style):
     video_result = search_pexels_video_tool(style, niche)
     
     if not video_result["success"]:
-        return "\n".join(status_log) + f"\n❌ {video_result.get('error', 'Failed')}", None
+        return "\n".join(status_log) + f"\n❌ {video_result.get('error', 'Failed')}", None, None
     
     status_log.append(f"   ✅ Found: {video_result['search_query']}\n")
+    
+    # Create video
+    status_log.append("🎬 Creating final video...")
+    output_dir = "/tmp/quote_videos"
+    os.makedirs(output_dir, exist_ok=True)
+    
+    import time
+    output_filename = f"quote_video_{int(time.time())}.mp4"
+    output_path = os.path.join(output_dir, output_filename)
+    
+    creation_result = create_quote_video_tool(
+        video_result["video_url"],
+        quote,
+        output_path
+    )
+    
+    if not creation_result["success"]:
+        status_log.append(f"   ❌ {creation_result['message']}")
+        return "\n".join(status_log), video_result["video_url"], None
+    
+    status_log.append("   ✅ Video created!\n")
     status_log.append("🎬 **COMPLETE!**")
     
-    return "\n".join(status_log), video_result["video_url"]
+    return "\n".join(status_log), video_result["video_url"], creation_result["output_path"]
 
 # Gradio Interface
 with gr.Blocks(title="AIQuoteClipGenerator - MCP Edition", theme=gr.themes.Soft()) as demo:
@@ -310,7 +459,7 @@ with gr.Blocks(title="AIQuoteClipGenerator - MCP Edition", theme=gr.themes.Soft(
     
     **MCP Integration Features:**
     - 🔗 **MCP Server:** Connected to smolagents framework
-    - 🛠️ **Custom Tools:** Quote generation + Video search
+    - 🛠️ **Custom Tools:** Quote generation + Video search + Video creation
     - 🤖 **Agent Reasoning:** Autonomous task execution
     - ⚡ **Tool Orchestration:** Intelligent pipeline management
     """)
@@ -351,29 +500,35 @@ with gr.Blocks(title="AIQuoteClipGenerator - MCP Edition", theme=gr.themes.Soft(
             output = gr.Textbox(label="Agent Status", lines=20, show_label=False)
     
     with gr.Row():
-        video_output = gr.Video(label="🎥 Video Preview")
+        with gr.Column():
+            gr.Markdown("### 🎥 Background Video Preview")
+            preview_video = gr.Video(label="Original Pexels Video")
+        
+        with gr.Column():
+            gr.Markdown("### ✨ Final Quote Video")
+            final_video = gr.Video(label="Download Your Video")
     
     gr.Markdown("""
     ---
     ### ✨ MCP Implementation
     - ✅ **smolagents Framework** - Proper MCP integration
-    - ✅ **Custom MCP Tools** - Quote generation & video search
+    - ✅ **Custom MCP Tools** - Quote generation, video search & video creation
     - ✅ **CodeAgent** - Autonomous reasoning and execution
     - ✅ **MCP Client** - Connected to external MCP servers
-    - 🚧 **MoviePy Processing** - Text overlay (next)
-    - 🚧 **ElevenLabs Integration** - Voice narration (next)
+    - ✅ **MoviePy Processing** - Text overlay with professional styling
+    - 🚧 **ElevenLabs Integration** - Voice narration (future)
     
     ### 🏆 Hackathon: MCP 1st Birthday
     **Track:** Track 2 - MCP in Action  
     **Category:** Productivity Tools  
-    **Built with:** Gradio + smolagents + OpenAI + Pexels + MCP
+    **Built with:** Gradio + smolagents + OpenAI + Pexels + MoviePy + MCP
     """)
     
     generate_btn.click(
         mcp_agent_pipeline, 
         inputs=[niche, style], 
-        outputs=[output, video_output]
+        outputs=[output, preview_video, final_video]
     )
 
 if __name__ == "__main__":
-    demo.launch()
+    demo.launch()