video_generation.py

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")