Create video_generation.py

video_generation.py

@@ -0,0 +1,632 @@
+import os
+import io
+import base64
+import time
+import requests
+import json
+import logging
+from PIL import Image
+import numpy as np
+from typing import List, Optional, Dict, Any
+import cv2
+import tempfile
+import random
+
+logger = logging.getLogger(__name__)
+
+class FreeVideoGenerator:
+    """
+    Free video generation using open-source models on Hugging Face
+    """
+    
+    def __init__(self, hf_token: Optional[str] = None):
+        self.hf_token = hf_token or os.getenv('HF_TOKEN', '')
+        self.base_url = "https://api-inference.huggingface.co/models"
+        
+        # Available free models for different tasks
+        self.models = {
+            # Text-to-Video models (FREE)
+            "text_to_video": {
+                "zeroscope_v2": "cerspense/zeroscope_v2_576w",
+                "modelscope": "damo-vilab/text-to-video-ms-1.7b",
+                "stable_video": "stabilityai/stable-video-diffusion-img2vid-xt",
+                "video_crafter": "VideoCrafter/VideoCrafter2",
+                "animatediff": "guoyww/animatediff"
+            },
+            
+            # Image-to-Video models (FREE)
+            "image_to_video": {
+                "stable_video": "stabilityai/stable-video-diffusion-img2vid-xt",
+                "img2vid_xt": "stabilityai/stable-video-diffusion-img2vid-xt-1-1",
+                "zeroscope_img2vid": "cerspense/zeroscope_v2_XL"
+            },
+            
+            # Animation models (FREE)
+            "animation": {
+                "animate_diff": "guoyww/animatediff",
+                "magic_animate": "zcxu-eric/MagicAnimate",
+                "text2video_zero": "PAIR/Text2Video-Zero"
+            }
+        }
+        
+        # Free API endpoints that work without token
+        self.free_endpoints = {
+            "text_to_video": "https://api-inference.huggingface.co/models/cerspense/zeroscope_v2_576w",
+            "image_to_video": "https://api-inference.huggingface.co/models/stabilityai/stable-video-diffusion-img2vid-xt",
+            "animation": "https://api-inference.huggingface.co/models/PAIR/Text2Video-Zero"
+        }
+        
+        # Performance settings
+        self.timeout = 120  # Longer timeout for videos
+        self.max_retries = 3
+        self.wait_between_retries = [10, 20, 30]  # Progressive waiting
+        
+        # Video settings
+        self.default_fps = 8
+        self.default_frames = 24
+        self.default_width = 576
+        self.default_height = 320
+        
+        # Cache for generated videos
+        self.video_cache = {}
+        self.cache_size = 50
+        
+    def detect_video_request(self, text: str) -> bool:
+        """Detect if user wants to generate a video"""
+        video_triggers = [
+            'generate video', 'create video', 'make a video', 'video of',
+            'animate', 'animation', 'moving picture', 'motion picture',
+            'video generation', 'create animation', 'make animation',
+            'video clip', 'short video', 'motion graphics', 'cinematic',
+            'film', 'movie', 'moving image', 'dynamic image', 'animated video'
+        ]
+        text_lower = text.lower()
+        return any(trigger in text_lower for trigger in video_triggers)
+    
+    def extract_video_prompt(self, text: str) -> str:
+        """Extract video description from user message"""
+        prompt = text.lower()
+        
+        # Remove common video request phrases
+        remove_phrases = [
+            'generate video of', 'create video of', 'make a video of',
+            'create animation of', 'make animation of', 'animate',
+            'generate animation of', 'video of', 'animation of',
+            'make a film about', 'create a film about', 'produce video of',
+            'can you make a video', 'i want a video', 'show me a video',
+            'video showing', 'animate this', 'create moving image of'
+        ]
+        
+        for phrase in remove_phrases:
+            prompt = prompt.replace(phrase, '')
+        
+        # Remove question words
+        question_words = ['how to', 'what is', 'can you', 'could you', 'would you']
+        for word in question_words:
+            if prompt.startswith(word):
+                prompt = prompt[len(word):].strip()
+        
+        return prompt.strip().capitalize()
+    
+    def enhance_prompt_with_context(self, prompt: str, context_type: str = "general") -> str:
+        """Enhance video prompts with cinematic and cultural context"""
+        
+        # Basic cinematic enhancements
+        cinematic_enhancements = [
+            "cinematic, 8k, ultra detailed, high quality, masterpiece",
+            "epic, dramatic lighting, film grain, cinematic shot, professional",
+            "beautiful, stunning, visually striking, vivid colors, trending",
+            "high resolution, detailed, sharp focus, studio quality, professional",
+            "film still, movie scene, cinematic photography, 35mm film"
+        ]
+        
+        # Cultural/Kiswahili enhancements
+        cultural_enhancements = {
+            "safari": "African safari, wildlife documentary style, national geographic, savanna",
+            "cultural": "traditional African culture, vibrant colors, community celebration, authentic",
+            "coastal": "Swahili coast, Indian Ocean, dhows sailing, traditional architecture, beach",
+            "urban": "modern African city, bustling streets, contemporary life, urban landscape",
+            "historical": "historical Africa, ancient kingdoms, traditional ceremonies, heritage",
+            "wildlife": "African wildlife, natural habitat, animal behavior, nature documentary",
+            "village": "traditional African village, community life, rural setting, authentic"
+        }
+        
+        # Motion and animation enhancements
+        motion_enhancements = [
+            "smooth motion, fluid animation, dynamic movement, cinematic motion",
+            "slow motion, dramatic pacing, epic timing, filmic movement",
+            "fast paced, energetic movement, dynamic action, lively animation"
+        ]
+        
+        enhanced_prompt = prompt
+        
+        # Add cinematic quality
+        enhanced_prompt += f", {random.choice(cinematic_enhancements)}"
+        
+        # Add motion enhancement
+        enhanced_prompt += f", {random.choice(motion_enhancements)}"
+        
+        # Add context-specific enhancements
+        context_keywords = {
+            "safari": ["safari", "wildlife", "animal", "lion", "elephant", "giraffe"],
+            "cultural": ["culture", "traditional", "dance", "ceremony", "ritual"],
+            "coastal": ["coast", "beach", "ocean", "sea", "dhow", "swahili"],
+            "urban": ["city", "urban", "street", "building", "modern", "skyline"],
+            "historical": ["history", "ancient", "kingdom", "heritage", "traditional"],
+            "wildlife": ["animal", "bird", "nature", "wild", "savanna", "forest"],
+            "village": ["village", "rural", "community", "hut", "traditional"]
+        }
+        
+        prompt_lower = enhanced_prompt.lower()
+        for theme, keywords in context_keywords.items():
+            if any(keyword in prompt_lower for keyword in keywords):
+                enhanced_prompt += f", {cultural_enhancements.get(theme, '')}"
+                break
+        
+        # Add technical specifications for better results
+        technical_specs = [
+            f"{self.default_width}x{self.default_height} resolution",
+            f"{self.default_fps} fps",
+            "high bitrate",
+            "stable diffusion",
+            "consistent quality"
+        ]
+        
+        enhanced_prompt += f", {', '.join(random.sample(technical_specs, 2))}"
+        
+        return enhanced_prompt
+    
+    def get_cached_video(self, prompt: str) -> Optional[str]:
+        """Get cached video if available"""
+        cache_key = prompt.lower().strip()[:100]
+        return self.video_cache.get(cache_key)
+    
+    def cache_video(self, prompt: str, video_data: str):
+        """Cache generated video"""
+        cache_key = prompt.lower().strip()[:100]
+        
+        # Limit cache size
+        if len(self.video_cache) >= self.cache_size:
+            # Remove oldest entry
+            self.video_cache.pop(next(iter(self.video_cache)))
+        
+        self.video_cache[cache_key] = video_data
+    
+    def generate_text_to_video(self, prompt: str, model: str = "zeroscope_v2") -> Optional[str]:
+        """
+        Generate video from text prompt using free models
+        
+        Args:
+            prompt: Text description of the video
+            model: Model to use ('zeroscope_v2', 'modelscope', etc.)
+            
+        Returns:
+            Base64 encoded video or None
+        """
+        
+        # Check cache first
+        cached_video = self.get_cached_video(prompt)
+        if cached_video:
+            logger.info("🎬 Using cached video")
+            return cached_video
+        
+        model_id = self.models["text_to_video"].get(model, "cerspense/zeroscope_v2_576w")
+        api_url = f"{self.base_url}/{model_id}"
+        
+        headers = {}
+        if self.hf_token:
+            headers["Authorization"] = f"Bearer {self.hf_token}"
+        
+        # Optimized parameters for faster generation
+        payload = {
+            "inputs": prompt,
+            "parameters": {
+                "num_frames": self.default_frames,
+                "num_inference_steps": 25,  # Reduced for speed
+                "guidance_scale": 7.5,
+                "fps": self.default_fps,
+                "height": self.default_height,
+                "width": self.default_width,
+                "negative_prompt": "blurry, low quality, distorted, bad anatomy, watermark, text"
+            }
+        }
+        
+        for attempt in range(self.max_retries):
+            try:
+                logger.info(f"🎬 Generating video (attempt {attempt + 1}): {prompt[:50]}...")
+                
+                response = requests.post(
+                    api_url,
+                    headers=headers,
+                    json=payload,
+                    timeout=self.timeout
+                )
+                
+                if response.status_code == 200:
+                    # Convert to base64
+                    video_bytes = response.content
+                    video_b64 = base64.b64encode(video_bytes).decode('utf-8')
+                    
+                    # Determine format
+                    content_type = response.headers.get('content-type', 'video/mp4')
+                    if 'webm' in content_type:
+                        format_str = "webm"
+                    else:
+                        format_str = "mp4"
+                    
+                    video_data = f"data:video/{format_str};base64,{video_b64}"
+                    
+                    # Cache the result
+                    self.cache_video(prompt, video_data)
+                    
+                    return video_data
+                    
+                elif response.status_code == 503:
+                    # Model is loading
+                    wait_time = self.wait_between_retries[min(attempt, len(self.wait_between_retries)-1)]
+                    logger.info(f"⏳ Video model loading, waiting {wait_time}s...")
+                    time.sleep(wait_time)
+                    continue
+                    
+                else:
+                    logger.error(f"Video API error {response.status_code}: {response.text[:200]}")
+                    
+            except requests.exceptions.Timeout:
+                logger.warning(f"⏰ Video generation timeout, attempt {attempt + 1}")
+                time.sleep(self.wait_between_retries[min(attempt, len(self.wait_between_retries)-1)])
+                continue
+            except Exception as e:
+                logger.error(f"Video generation error: {e}")
+                if attempt < self.max_retries - 1:
+                    time.sleep(self.wait_between_retries[min(attempt, len(self.wait_between_retries)-1)])
+                    continue
+                break
+        
+        # Fallback to simpler animation if video generation fails
+        logger.info("🔄 Falling back to text animation")
+        return self.generate_animation_from_text(prompt)
+    
+    def generate_image_to_video(self, image_data: str, prompt: str = "") -> Optional[str]:
+        """
+        Generate video from an image using free models
+        
+        Args:
+            image_data: Base64 encoded image or image URL
+            prompt: Optional text prompt for guidance
+            
+        Returns:
+            Base64 encoded video or None
+        """
+        try:
+            # Prepare image
+            if image_data.startswith('data:image'):
+                # Extract base64 from data URL
+                image_b64 = image_data.split(',')[1]
+                image_bytes = base64.b64decode(image_b64)
+                image = Image.open(io.BytesIO(image_bytes))
+            else:
+                # Assume it's a file path or URL
+                if image_data.startswith('http'):
+                    response = requests.get(image_data, timeout=30)
+                    image = Image.open(io.BytesIO(response.content))
+                else:
+                    image = Image.open(image_data)
+            
+            # Resize image for faster processing
+            image = image.resize((self.default_width, self.default_height), Image.Resampling.LANCZOS)
+            
+            # Convert to bytes
+            img_byte_arr = io.BytesIO()
+            image.save(img_byte_arr, format='PNG')
+            img_byte_arr = img_byte_arr.getvalue()
+            
+            # Use free model (Stable Video Diffusion)
+            model_id = "stabilityai/stable-video-diffusion-img2vid-xt"
+            api_url = f"{self.base_url}/{model_id}"
+            
+            headers = {
+                "Authorization": f"Bearer {self.hf_token}" if self.hf_token else ""
+            }
+            
+            # If prompt is provided, use it as guidance
+            params = {}
+            if prompt:
+                params = {
+                    "parameters": {
+                        "motion_bucket_id": 127,
+                        "noise_aug_strength": 0.02
+                    }
+                }
+            
+            response = requests.post(
+                api_url,
+                headers=headers,
+                data=img_byte_arr,
+                json=params if params else None,
+                timeout=150  # Longer timeout for image-to-video
+            )
+            
+            if response.status_code == 200:
+                video_b64 = base64.b64encode(response.content).decode('utf-8')
+                return f"data:video/mp4;base64,{video_b64}"
+            else:
+                logger.error(f"Image-to-video API error: {response.status_code}")
+                return None
+            
+        except Exception as e:
+            logger.error(f"Image to video error: {e}")
+            return None
+    
+    def create_slideshow_video(self, images: List[str], duration_per_image: float = 2.0) -> Optional[str]:
+        """
+        Create a simple slideshow video from multiple images
+        
+        Args:
+            images: List of base64 encoded images
+            duration_per_image: Duration for each image in seconds
+            
+        Returns:
+            Base64 encoded video
+        """
+        try:
+            # Create temporary directory
+            with tempfile.TemporaryDirectory() as tmpdir:
+                image_paths = []
+                
+                # Save all images
+                for i, img_data in enumerate(images):
+                    if img_data.startswith('data:image'):
+                        img_b64 = img_data.split(',')[1]
+                        img_bytes = base64.b64decode(img_b64)
+                    else:
+                        img_bytes = base64.b64decode(img_data)
+                    
+                    img_path = os.path.join(tmpdir, f'frame_{i:03d}.png')
+                    with open(img_path, 'wb') as f:
+                        f.write(img_bytes)
+                    image_paths.append(img_path)
+                
+                # Read first image to get dimensions
+                first_img = cv2.imread(image_paths[0])
+                if first_img is None:
+                    logger.error("Failed to read first image")
+                    return None
+                    
+                height, width = first_img.shape[:2]
+                
+                # Create video writer
+                fps = 10
+                output_path = os.path.join(tmpdir, 'output.mp4')
+                fourcc = cv2.VideoWriter_fourcc(*'mp4v')
+                out = cv2.VideoWriter(output_path, fourcc, fps, (width, height))
+                
+                # Write frames with smooth transitions
+                frames_per_image = int(fps * duration_per_image)
+                transition_frames = int(fps * 0.5)  # Half second transition
+                
+                for i in range(len(image_paths)):
+                    current_img = cv2.imread(image_paths[i])
+                    if current_img is None:
+                        continue
+                    
+                    # Resize to match dimensions
+                    current_img = cv2.resize(current_img, (width, height))
+                    
+                    # Write main frames
+                    main_frames = frames_per_image - transition_frames
+                    for _ in range(main_frames):
+                        out.write(current_img)
+                    
+                    # Add transition to next image if exists
+                    if i < len(image_paths) - 1:
+                        next_img = cv2.imread(image_paths[i + 1])
+                        if next_img is not None:
+                            next_img = cv2.resize(next_img, (width, height))
+                            
+                            # Create crossfade transition
+                            for t in range(transition_frames):
+                                alpha = t / transition_frames
+                                beta = 1.0 - alpha
+                                blended = cv2.addWeighted(current_img, beta, next_img, alpha, 0)
+                                out.write(blended)
+                
+                out.release()
+                
+                # Read and encode video
+                with open(output_path, 'rb') as f:
+                    video_bytes = f.read()
+                
+                video_b64 = base64.b64encode(video_bytes).decode('utf-8')
+                return f"data:video/mp4;base64,{video_b64}"
+                
+        except Exception as e:
+            logger.error(f"Slideshow error: {e}")
+            return None
+    
+    def generate_animation_from_text(self, text: str) -> Optional[str]:
+        """
+        Create simple text animation
+        
+        Args:
+            text: Text to animate
+            
+        Returns:
+            Base64 encoded video
+        """
+        try:
+            # Create temporary directory
+            with tempfile.TemporaryDirectory() as tmpdir:
+                # Create frames with text
+                fps = 10
+                duration = 4  # seconds
+                total_frames = fps * duration
+                height, width = self.default_height, self.default_width
+                
+                output_path = os.path.join(tmpdir, 'animation.mp4')
+                fourcc = cv2.VideoWriter_fourcc(*'mp4v')
+                out = cv2.VideoWriter(output_path, fourcc, fps, (width, height))
+                
+                # Create gradient background colors
+                colors = [
+                    (41, 128, 185),   # Blue
+                    (39, 174, 96),    # Green
+                    (142, 68, 173),   # Purple
+                    (230, 126, 34),   # Orange
+                    (231, 76, 60)     # Red
+                ]
+                
+                for frame_num in range(total_frames):
+                    # Create gradient background
+                    frame = np.zeros((height, width, 3), dtype=np.uint8)
+                    
+                    # Select color based on frame
+                    color_idx = (frame_num // (total_frames // len(colors))) % len(colors)
+                    bg_color = colors[color_idx]
+                    
+                    # Apply gradient
+                    for i in range(height):
+                        # Gradient from top to bottom
+                        factor = i / height
+                        r = int(bg_color[2] * (1 - factor) + 10 * factor)
+                        g = int(bg_color[1] * (1 - factor) + 10 * factor)
+                        b = int(bg_color[0] * (1 - factor) + 10 * factor)
+                        
+                        frame[i, :, 0] = b  # OpenCV uses BGR
+                        frame[i, :, 1] = g
+                        frame[i, :, 2] = r
+                    
+                    # Add text with animation
+                    font = cv2.FONT_HERSHEY_SIMPLEX
+                    
+                    # Calculate text position (center)
+                    text_lines = text.split(' ')
+                    y_start = height // 2 - (len(text_lines) * 40) // 2
+                    
+                    for i, line in enumerate(text_lines):
+                        # Calculate font size with pulse effect
+                        pulse = 0.7 + 0.3 * np.sin(2 * np.pi * (frame_num / fps) + i * 0.5)
+                        font_scale = 1.2 * pulse
+                        thickness = int(2 * pulse)
+                        
+                        # Calculate text size and position
+                        text_size = cv2.getTextSize(line, font, font_scale, thickness)[0]
+                        text_x = (width - text_size[0]) // 2
+                        text_y = y_start + i * 40
+                        
+                        # Add text shadow
+                        shadow_color = (0, 0, 0)
+                        cv2.putText(frame, line, (text_x + 2, text_y + 2), font, 
+                                   font_scale, shadow_color, thickness + 1)
+                        
+                        # Add main text
+                        text_color = (255, 255, 255)  # White
+                        cv2.putText(frame, line, (text_x, text_y), font, 
+                                   font_scale, text_color, thickness)
+                    
+                    # Add decorative elements
+                    if frame_num % 10 < 5:
+                        # Add twinkling stars
+                        for _ in range(3):
+                            star_x = random.randint(0, width)
+                            star_y = random.randint(0, height)
+                            cv2.circle(frame, (star_x, star_y), 2, (255, 255, 255), -1)
+                    
+                    out.write(frame)
+                
+                out.release()
+                
+                # Read and encode video
+                with open(output_path, 'rb') as f:
+                    video_bytes = f.read()
+                
+                video_b64 = base64.b64encode(video_bytes).decode('utf-8')
+                return f"data:video/mp4;base64,{video_b64}"
+                
+        except Exception as e:
+            logger.error(f"Text animation error: {e}")
+            return None
+    
+    def create_cultural_video(self, theme: str, style: str = "animated") -> Optional[str]:
+        """
+        Create videos with Kiswahili cultural themes
+        
+        Args:
+            theme: Cultural theme (safari, ceremony, dance, etc.)
+            style: Animation style
+            
+        Returns:
+            Base64 encoded video
+        """
+        # Cultural themes and prompts
+        cultural_themes = {
+            "safari": "African safari sunset with elephants and giraffes walking, majestic savanna landscape",
+            "dance": "Traditional Maasai warriors dancing, vibrant colors, cultural celebration, energetic movement",
+            "market": "Busy African market scene, vibrant colors, people trading goods, lively atmosphere",
+            "coastal": "Swahili coast with traditional dhows sailing, Indian Ocean waves, beach scenery",
+            "wildlife": "African wildlife documentary style, lions hunting on savanna, dramatic nature scene",
+            "village": "Traditional African village life, community activities, sunset over huts",
+            "ceremony": "African wedding ceremony, traditional attire, dancing, celebration, cultural rituals",
+            "sunset": "African sunset over savanna, acacia trees silhouette, warm colors, peaceful scene",
+            "city": "Modern African city at night, Nairobi skyline, lights, urban life, contemporary"
+        }
+        
+        # Get prompt for theme
+        base_prompt = cultural_themes.get(theme, f"African {theme}, cultural, vibrant, dynamic")
+        
+        # Add style-specific enhancements
+        style_enhancements = {
+            "animated": "animated, cartoon style, smooth motion, vibrant colors, lively",
+            "realistic": "realistic, documentary style, cinematic, natural lighting, photorealistic",
+            "painting": "painting style, brush strokes, artistic, masterpiece, textured",
+            "watercolor": "watercolor painting, soft edges, dreamy, artistic, blended colors",
+            "cinematic": "cinematic, film grain, dramatic lighting, movie scene, professional"
+        }
+        
+        style_enhancement = style_enhancements.get(style, "animated, vibrant, smooth motion")
+        
+        full_prompt = f"{base_prompt}, {style_enhancement}, {self.default_width}x{self.default_height}, {self.default_fps} fps"
+        
+        return self.generate_text_to_video(full_prompt)
+    
+    def get_video_info(self) -> Dict[str, Any]:
+        """Get information about available video generation options"""
+        return {
+            "available_models": {
+                "text_to_video": list(self.models["text_to_video"].keys()),
+                "image_to_video": list(self.models["image_to_video"].keys()),
+                "animation": list(self.models["animation"].keys())
+            },
+            "free_models": ["zeroscope_v2", "stable_video", "text2video_zero"],
+            "max_duration": "4 seconds",
+            "max_frames": self.default_frames,
+            "resolution": f"{self.default_width}x{self.default_height}",
+            "fps": self.default_fps,
+            "formats": ["MP4", "WebM"],
+            "features": [
+                "Text-to-Video",
+                "Image-to-Video",
+                "Slideshow Creation",
+                "Text Animation",
+                "Cultural Themes",
+                "Crossfade Transitions",
+                "Animated Text Effects"
+            ],
+            "cultural_themes": [
+                "safari", "dance", "market", "coastal", 
+                "wildlife", "village", "ceremony", "sunset", "city"
+            ],
+            "styles": ["animated", "realistic", "painting", "watercolor", "cinematic"],
+            "cache_enabled": True,
+            "cache_size": self.cache_size,
+            "timeout_seconds": self.timeout,
+            "max_retries": self.max_retries
+        }
+    
+    def cleanup_cache(self):
+        """Cleanup old cache entries"""
+        if len(self.video_cache) > self.cache_size:
+            # Remove oldest entries
+            keys_to_remove = list(self.video_cache.keys())[:len(self.video_cache) - self.cache_size]
+            for key in keys_to_remove:
+                del self.video_cache[key]
+            logger.info(f"🧹 Cleaned up {len(keys_to_remove)} cache entries")