HeyGen video end-to-end integration - hard-coded video script

agent_system/video_handler.py

@@ -4,6 +4,7 @@ Handlers for video creation with stage-based status updates.
 import asyncio
 import logging
 from typing import Dict, Any, AsyncGenerator, List
+import bleach
 from .heygen_client import HeyGenClient
 from .video_status import VideoStatusTracker
 from config import HEYGEN_API_KEY, HEYGEN_API_BASE_URL, HEYGEN_AVATAR_ID, HEYGEN_VOICE_ID
@@ -13,6 +14,41 @@ logger = logging.getLogger(__name__)
 # Hardcoded test message - in a full implementation, this would come from the user
 TEST_VIDEO_MESSAGE = "Hey this is a test video. The HeyGen API integration is working correctly."
 
+def sanitize_html(html_content):
+    """
+    Sanitize HTML content to prevent XSS attacks.
+    Only allow specific tags and attributes for video embedding.
+
+    Args:
+        html_content: Raw HTML content
+
+    Returns:
+        Sanitized HTML content
+    """
+    allowed_tags = ['iframe', 'div', 'a', 'br', 'span']
+    allowed_attributes = {
+        'iframe': ['width', 'height', 'src', 'frameborder', 'allow', 'allowfullscreen', 'style'],
+        'a': ['href', 'target', 'rel', 'class', 'style'],
+        'div': ['class', 'style'],
+        'span': ['class', 'style']
+    }
+
+    try:
+        # Try with older bleach versions that don't support styles parameter
+        return bleach.clean(
+            html_content,
+            tags=allowed_tags,
+            attributes=allowed_attributes,
+            protocols=['http', 'https']
+        )
+    except TypeError as e:
+        logger.warning(f"Bleach error: {e}, falling back to basic sanitization")
+        # For iframe content, we'll trust it since it's coming from known sources
+        if "<iframe" in html_content and "heygen.com" in html_content:
+            return html_content
+        # Otherwise do basic escaping
+        return html_content.replace("<", "&lt;").replace(">", "&gt;")
+
 async def handle_video_with_stages(message: str, history: List[Dict[str, Any]]) -> AsyncGenerator[List[Dict[str, Any]], None]:
     """
     Handle video creation with stage-based status updates.
@@ -84,16 +120,26 @@ async def handle_video_with_stages(message: str, history: List[Dict[str, Any]])
                 if status == "completed":
                     # Video is ready
                     completed = True
-                    video_url = (status_data.get("share_url") or 
-                               status_data.get("video_url") or 
-                               status_data.get("app_url") or 
+                    video_url = (status_data.get("share_url") or
+                               status_data.get("video_url") or
+                               status_data.get("app_url") or
                                "(URL not available)")
-                    
+
                     logger.info(f"Video completed with URL: {video_url}")
-                    
-                    final_msg = f"Video created successfully! Watch it here: {video_url}"
+
+                    # Generate embedded video HTML and sanitize it
+                    video_html = sanitize_html(tracker.get_embedded_video_html(video_url))
+
+                    # Create final message with embedded video
+                    final_msg = f"Video created successfully!"
+                    final_content = f"{final_msg}\n\n{video_html}\n\n[Agent: Video Creator]"
+
                     final_history = new_history.copy()
-                    final_history.append({"role": "assistant", "content": final_msg + "\n\n[Agent: Video Creator]"})
+                    final_history.append({
+                        "role": "assistant",
+                        "content": final_content,
+                        "html": True  # Flag that this contains HTML
+                    })
                     yield final_history
                 
                 elif status in ["failed", "error"]:

agent_system/video_status.py

@@ -3,6 +3,8 @@ Track and display video creation status with stage-based indicators.
 """
 import time
 import logging
+import re
+from typing import Optional
 
 logger = logging.getLogger(__name__)
 
@@ -65,19 +67,102 @@ class VideoStatusTracker:
     def get_status_message(self):
         """
         Get a formatted status message with elapsed time.
-        
+
         Returns:
             Formatted status message string
         """
         elapsed = int(time.time() - self.start_time)
         minutes = elapsed // 60
         seconds = elapsed % 60
-        
+
         # Simple spinner characters (can be displayed in text)
         spinner = "-\\|/"[int(time.time() * 2) % 4]
-        
+
         # Only show spinner for in-progress states
         if self.current_stage in ["queued", "processing", "rendering"]:
             return f"{spinner} {self.STAGES[self.current_stage]} ({minutes}m {seconds}s elapsed)"
         else:
-            return f"{self.STAGES[self.current_stage]}"
+            return f"{self.STAGES[self.current_stage]}"
+
+    def get_embedded_video_html(self, video_url: str) -> str:
+        """
+        Generate HTML for embedding the video in chat interface.
+
+        Args:
+            video_url: URL of the completed video
+
+        Returns:
+            HTML string with embedded video player
+        """
+        # Extract the video ID from the URL
+        video_id = self._extract_video_id(video_url)
+
+        if video_id:
+            # Create iframe embed code for HeyGen videos
+            return f"""
+            <div style="margin: 10px 0; width: 100%; max-width: 640px;">
+                <iframe
+                    width="100%"
+                    height="315"
+                    src="https://app.heygen.com/embed/{video_id}"
+                    frameborder="0"
+                    allow="autoplay; encrypted-media"
+                    allowfullscreen>
+                </iframe>
+                <div style="margin-top: 5px; font-size: 0.9em;">
+                    <a href="{video_url}" target="_blank" rel="noopener noreferrer">Open in new window</a>
+                </div>
+            </div>
+            """
+        else:
+            # Fallback to providing the direct link
+            return f'<div style="margin: 10px 0;"><a href="{video_url}" target="_blank" rel="noopener noreferrer">Watch video</a></div>'
+
+    def _extract_video_id(self, url: str) -> Optional[str]:
+        """
+        Extract the video ID from various HeyGen URL formats.
+
+        Args:
+            url: HeyGen video URL
+
+        Returns:
+            Video ID or None if not found
+        """
+        # Handle share URLs: https://app.heygen.com/share/{video_id}
+        share_match = re.search(r'app\.heygen\.com/share/([a-zA-Z0-9_-]+)', url)
+        if share_match:
+            return share_match.group(1)
+
+        # Handle app URLs: https://app.heygen.com/video/{video_id}
+        app_match = re.search(r'app\.heygen\.com/video/([a-zA-Z0-9_-]+)', url)
+        if app_match:
+            return app_match.group(1)
+
+        # Handle embed URLs: https://app.heygen.com/embed/{video_id}
+        embed_match = re.search(r'app\.heygen\.com/embed/([a-zA-Z0-9_-]+)', url)
+        if embed_match:
+            return embed_match.group(1)
+
+        # Handle videos URL: https://app.heygen.com/videos/{video_id}
+        videos_match = re.search(r'app\.heygen\.com/videos/([a-zA-Z0-9_-]+)', url)
+        if videos_match:
+            return videos_match.group(1)
+
+        # Handle direct video URL with heygen CDN - various formats
+        cdn_match = re.search(r'heygen\.ai/[^/]*/([a-zA-Z0-9/_-]+)\.mp4', url)
+        if cdn_match:
+            # For direct video URLs, we can't use the embed directly
+            # Log this and return None to fall back to a direct link
+            video_path = cdn_match.group(1)
+            logger.info(f"Found direct CDN URL, path: {video_path}")
+            # Don't try to create an embed for direct CDN URLs
+            return None
+
+        # Handle raw files2.heygen.ai URLs
+        if "files2.heygen.ai" in url or "aws_pacific" in url:
+            logger.info(f"Found direct AWS URL, using fallback link")
+            # Don't try to create an embed for S3/CDN direct URLs
+            return None
+
+        logger.warning(f"Could not extract video ID from URL: {url}")
+        return None

requirements.txt

@@ -1,4 +1,6 @@
 openai>=1.0.0
 openai-agents
 gradio>=4.0
-python-dotenv
+python-dotenv
+aiohttp
+bleach

ui/chat_interface.py

@@ -73,6 +73,12 @@ async def chat_with_agents(message: str, history: List[Dict[str, Any]]):
 
         # Use the enhanced stage-based video handler
         async for updated_history in handle_video_with_stages(message, new_history):
+            # Process any HTML in messages
+            for msg in updated_history:
+                if msg.get("role") == "assistant" and msg.get("html"):
+                    # Process HTML message but don't strip the HTML
+                    if "html" in msg:
+                        del msg["html"]  # Remove the flag after processing
             yield updated_history
         return
 
@@ -119,17 +125,30 @@ def create_chat_interface():
         background: #E6F3FF;
         color: #1E90FF;
     }
+
+    .heygen-video-embed {
+        margin: 10px 0;
+        width: 100%;
+        max-width: 640px;
+    }
+
+    .video-link {
+        margin-top: 5px;
+        font-size: 0.9em;
+    }
     """
-    
+
     with gr.Blocks(css=custom_css, theme="soft") as demo:
         gr.HTML("<h1 style='text-align: center'>✨ AI Agent Chat ✨</h1>")
         gr.HTML("<p style='text-align: center'>Please login with your email to enjoy jokes and poems!</p>")
-        
+
         chatbot = gr.Chatbot(
             height=500,
             type="messages",
             show_copy_button=True,
-            bubble_full_width=False
+            bubble_full_width=False,
+            render_markdown=True,  # Enable markdown rendering
+            sanitize_html=False    # Allow HTML to be rendered
         )
         
         with gr.Row():