Create media_processing.py

media_processing.py

@@ -0,0 +1,1167 @@
+import os
+import base64
+import cv2
+import numpy as np
+from PIL import Image
+import pytesseract
+import requests
+from urllib.parse import urlparse, urljoin
+from bs4 import BeautifulSoup
+import html2text
+import json
+import time
+import webbrowser
+import urllib.parse
+import copy
+import html
+import tempfile
+import uuid
+import datetime
+import threading
+import atexit
+from huggingface_hub import HfApi
+import gradio as gr
+import subprocess
+import re
+
+# ---------------------------------------------------------------------------
+# Video temp-file management (per-session tracking and cleanup)
+# ---------------------------------------------------------------------------
+VIDEO_TEMP_DIR = os.path.join(tempfile.gettempdir(), "anycoder_videos")
+VIDEO_FILE_TTL_SECONDS = 6 * 60 * 60  # 6 hours
+_SESSION_VIDEO_FILES: Dict[str, List[str]] = {}
+_VIDEO_FILES_LOCK = threading.Lock()
+
+def _ensure_video_dir_exists() -> None:
+    try:
+        os.makedirs(VIDEO_TEMP_DIR, exist_ok=True)
+    except Exception:
+        pass
+
+def _register_video_for_session(session_id: Optional[str], file_path: str) -> None:
+    if not session_id or not file_path:
+        return
+    with _VIDEO_FILES_LOCK:
+        if session_id not in _SESSION_VIDEO_FILES:
+            _SESSION_VIDEO_FILES[session_id] = []
+        _SESSION_VIDEO_FILES[session_id].append(file_path)
+
+def cleanup_session_videos(session_id: Optional[str]) -> None:
+    if not session_id:
+        return
+    with _VIDEO_FILES_LOCK:
+        file_list = _SESSION_VIDEO_FILES.pop(session_id, [])
+    for path in file_list:
+        try:
+            if path and os.path.exists(path):
+                os.unlink(path)
+        except Exception:
+            # Best-effort cleanup
+            pass
+
+def reap_old_videos(ttl_seconds: int = VIDEO_FILE_TTL_SECONDS) -> None:
+    """Delete old video files in the temp directory based on modification time."""
+    try:
+        _ensure_video_dir_exists()
+        now_ts = time.time()
+        for name in os.listdir(VIDEO_TEMP_DIR):
+            path = os.path.join(VIDEO_TEMP_DIR, name)
+            try:
+                if not os.path.isfile(path):
+                    continue
+                mtime = os.path.getmtime(path)
+                if now_ts - mtime > ttl_seconds:
+                    os.unlink(path)
+            except Exception:
+                pass
+    except Exception:
+        # Temp dir might not exist or be accessible; ignore
+        pass
+
+# ---------------------------------------------------------------------------
+# Audio temp-file management (per-session tracking and cleanup)
+# ---------------------------------------------------------------------------
+AUDIO_TEMP_DIR = os.path.join(tempfile.gettempdir(), "anycoder_audio")
+AUDIO_FILE_TTL_SECONDS = 6 * 60 * 60  # 6 hours
+_SESSION_AUDIO_FILES: Dict[str, List[str]] = {}
+_AUDIO_FILES_LOCK = threading.Lock()
+
+def _ensure_audio_dir_exists() -> None:
+    try:
+        os.makedirs(AUDIO_TEMP_DIR, exist_ok=True)
+    except Exception:
+        pass
+
+def _register_audio_for_session(session_id: Optional[str], file_path: str) -> None:
+    if not session_id or not file_path:
+        return
+    with _AUDIO_FILES_LOCK:
+        if session_id not in _SESSION_AUDIO_FILES:
+            _SESSION_AUDIO_FILES[session_id] = []
+        _SESSION_AUDIO_FILES[session_id].append(file_path)
+
+def cleanup_session_audio(session_id: Optional[str]) -> None:
+    if not session_id:
+        return
+    with _AUDIO_FILES_LOCK:
+        file_list = _SESSION_AUDIO_FILES.pop(session_id, [])
+    for path in file_list:
+        try:
+            if path and os.path.exists(path):
+                os.unlink(path)
+        except Exception:
+            pass
+
+def reap_old_audio(ttl_seconds: int = AUDIO_FILE_TTL_SECONDS) -> None:
+    try:
+        _ensure_audio_dir_exists()
+        now_ts = time.time()
+        for name in os.listdir(AUDIO_TEMP_DIR):
+            path = os.path.join(AUDIO_TEMP_DIR, name)
+            try:
+                if not os.path.isfile(path):
+                    continue
+                mtime = os.path.getmtime(path)
+                if now_ts - mtime > ttl_seconds:
+                    os.unlink(path)
+            except Exception:
+                pass
+    except Exception:
+        pass
+
+# ---------------------------------------------------------------------------
+# General temp media file management (per-session tracking and cleanup)
+# ---------------------------------------------------------------------------
+MEDIA_TEMP_DIR = os.path.join(tempfile.gettempdir(), "anycoder_media")
+MEDIA_FILE_TTL_SECONDS = 6 * 60 * 60  # 6 hours
+_SESSION_MEDIA_FILES: Dict[str, List[str]] = {}
+_MEDIA_FILES_LOCK = threading.Lock()
+
+# Global dictionary to store temporary media files for the session
+temp_media_files = {}
+
+def _ensure_media_dir_exists() -> None:
+    """Ensure the media temp directory exists."""
+    try:
+        os.makedirs(MEDIA_TEMP_DIR, exist_ok=True)
+    except Exception:
+        pass
+
+def track_session_media_file(session_id: Optional[str], file_path: str) -> None:
+    """Track a media file for session-based cleanup."""
+    if not session_id or not file_path:
+        return
+    with _MEDIA_FILES_LOCK:
+        if session_id not in _SESSION_MEDIA_FILES:
+            _SESSION_MEDIA_FILES[session_id] = []
+        _SESSION_MEDIA_FILES[session_id].append(file_path)
+
+def cleanup_session_media(session_id: Optional[str]) -> None:
+    """Clean up media files for a specific session."""
+    if not session_id:
+        return
+    with _MEDIA_FILES_LOCK:
+        files_to_clean = _SESSION_MEDIA_FILES.pop(session_id, [])
+    
+    for path in files_to_clean:
+        try:
+            if path and os.path.exists(path):
+                os.unlink(path)
+        except Exception:
+            # Best-effort cleanup
+            pass
+
+def reap_old_media(ttl_seconds: int = MEDIA_FILE_TTL_SECONDS) -> None:
+    """Delete old media files in the temp directory based on modification time."""
+    try:
+        _ensure_media_dir_exists()
+        now_ts = time.time()
+        for name in os.listdir(MEDIA_TEMP_DIR):
+            path = os.path.join(MEDIA_TEMP_DIR, name)
+            if os.path.isfile(path):
+                try:
+                    mtime = os.path.getmtime(path)
+                    if (now_ts - mtime) > ttl_seconds:
+                        os.unlink(path)
+                except Exception:
+                    pass
+    except Exception:
+        # Temp dir might not exist or be accessible; ignore
+        pass
+
+def cleanup_all_temp_media_on_startup() -> None:
+    """Clean up all temporary media files on app startup."""
+    try:
+        # Clean up temp_media_files registry
+        temp_media_files.clear()
+        
+        # Clean up actual files from disk (assume all are orphaned on startup)
+        _ensure_media_dir_exists()
+        for name in os.listdir(MEDIA_TEMP_DIR):
+            path = os.path.join(MEDIA_TEMP_DIR, name)
+            if os.path.isfile(path):
+                try:
+                    os.unlink(path)
+                except Exception:
+                    pass
+        
+        # Clear session tracking
+        with _MEDIA_FILES_LOCK:
+            _SESSION_MEDIA_FILES.clear()
+            
+        print("[StartupCleanup] Cleaned up orphaned temporary media files")
+    except Exception as e:
+        print(f"[StartupCleanup] Error during media cleanup: {str(e)}")
+
+def cleanup_all_temp_media_on_shutdown() -> None:
+    """Clean up all temporary media files on app shutdown."""
+    try:
+        print("[ShutdownCleanup] Cleaning up temporary media files...")
+        
+        # Clean up temp_media_files registry and remove files
+        for file_id, file_info in temp_media_files.items():
+            try:
+                if os.path.exists(file_info['path']):
+                    os.unlink(file_info['path'])
+            except Exception:
+                pass
+        temp_media_files.clear()
+        
+        # Clean up all session files
+        with _MEDIA_FILES_LOCK:
+            for session_id, file_paths in _SESSION_MEDIA_FILES.items():
+                for path in file_paths:
+                    try:
+                        if path and os.path.exists(path):
+                            os.unlink(path)
+                    except Exception:
+                        pass
+            _SESSION_MEDIA_FILES.clear()
+        
+        print("[ShutdownCleanup] Temporary media cleanup completed")
+    except Exception as e:
+        print(f"[ShutdownCleanup] Error during cleanup: {str(e)}")
+
+# Register shutdown cleanup handler
+atexit.register(cleanup_all_temp_media_on_shutdown)
+
+def create_temp_media_url(media_bytes: bytes, filename: str, media_type: str = "image", session_id: Optional[str] = None) -> str:
+    """Create a temporary file and return a local URL for preview."""
+    try:
+        # Create unique filename with timestamp and UUID
+        timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
+        unique_id = str(uuid.uuid4())[:8]
+        base_name, ext = os.path.splitext(filename)
+        unique_filename = f"{media_type}_{timestamp}_{unique_id}_{base_name}{ext}"
+        
+        # Create temporary file in the dedicated directory
+        _ensure_media_dir_exists()
+        temp_path = os.path.join(MEDIA_TEMP_DIR, unique_filename)
+        
+        # Write media bytes to temporary file
+        with open(temp_path, 'wb') as f:
+            f.write(media_bytes)
+        
+        # Track file for session-based cleanup
+        if session_id:
+            track_session_media_file(session_id, temp_path)
+        
+        # Store the file info for later upload
+        file_id = f"{media_type}_{unique_id}"
+        temp_media_files[file_id] = {
+            'path': temp_path,
+            'filename': filename,
+            'media_type': media_type,
+            'media_bytes': media_bytes
+        }
+        
+        # Return file:// URL for preview
+        file_url = f"file://{temp_path}"
+        print(f"[TempMedia] Created temporary {media_type} file: {file_url}")
+        return file_url
+        
+    except Exception as e:
+        print(f"[TempMedia] Failed to create temporary file: {str(e)}")
+        return f"Error creating temporary {media_type} file: {str(e)}"
+
+def upload_media_to_hf(media_bytes: bytes, filename: str, media_type: str = "image", token: gr.OAuthToken | None = None, use_temp: bool = True) -> str:
+    """Upload media file to user's Hugging Face account or create temporary file."""
+    try:
+        # If use_temp is True, create temporary file for preview
+        if use_temp:
+            return create_temp_media_url(media_bytes, filename, media_type)
+        
+        # Otherwise, upload to Hugging Face for permanent URL
+        # Try to get token from OAuth first, then fall back to environment variable
+        hf_token = None
+        if token and token.token:
+            hf_token = token.token
+        else:
+            hf_token = os.getenv('HF_TOKEN')
+        
+        if not hf_token:
+            return "Error: Please log in with your Hugging Face account to upload media, or set HF_TOKEN environment variable."
+        
+        # Initialize HF API
+        api = HfApi(token=hf_token)
+        
+        # Get current user info to determine username
+        try:
+            user_info = api.whoami()
+            username = user_info.get('name', 'unknown-user')
+        except Exception as e:
+            print(f"[HFUpload] Could not get user info: {e}")
+            username = 'anycoder-user'
+        
+        # Create repository name for media storage
+        repo_name = f"{username}/anycoder-media"
+        
+        # Try to create the repository if it doesn't exist
+        try:
+            api.create_repo(
+                repo_id=repo_name,
+                repo_type="dataset",
+                private=False,
+                exist_ok=True
+            )
+            print(f"[HFUpload] Repository {repo_name} ready")
+        except Exception as e:
+            print(f"[HFUpload] Repository creation/access issue: {e}")
+            # Continue anyway, repo might already exist
+        
+        # Create unique filename with timestamp and UUID
+        timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
+        unique_id = str(uuid.uuid4())[:8]
+        base_name, ext = os.path.splitext(filename)
+        unique_filename = f"{media_type}/{timestamp}_{unique_id}_{base_name}{ext}"
+        
+        # Create temporary file for upload
+        with tempfile.NamedTemporaryFile(delete=False, suffix=ext) as temp_file:
+            temp_file.write(media_bytes)
+            temp_path = temp_file.name
+        
+        try:
+            # Upload file to HF repository
+            api.upload_file(
+                path_or_fileobj=temp_path,
+                path_in_repo=unique_filename,
+                repo_id=repo_name,
+                repo_type="dataset",
+                commit_message=f"Upload {media_type} generated by AnyCoder"
+            )
+            
+            # Generate permanent URL
+            permanent_url = f"https://huggingface.co/datasets/{repo_name}/resolve/main/{unique_filename}"
+            print(f"[HFUpload] Successfully uploaded {media_type} to {permanent_url}")
+            return permanent_url
+            
+        finally:
+            # Clean up temporary file
+            try:
+                os.unlink(temp_path)
+            except Exception:
+                pass
+                
+    except Exception as e:
+        print(f"[HFUpload] Upload failed: {str(e)}")
+        return f"Error uploading {media_type} to Hugging Face: {str(e)}"
+
+def upload_temp_files_to_hf_and_replace_urls(html_content: str, token: gr.OAuthToken | None = None) -> str:
+    """Upload all temporary media files to HF and replace their URLs in HTML content."""
+    try:
+        if not temp_media_files:
+            print("[DeployUpload] No temporary media files to upload")
+            return html_content
+        
+        print(f"[DeployUpload] Uploading {len(temp_media_files)} temporary media files to HF")
+        updated_content = html_content
+        
+        for file_id, file_info in temp_media_files.items():
+            try:
+                # Upload to HF with permanent URL
+                permanent_url = upload_media_to_hf(
+                    file_info['media_bytes'],
+                    file_info['filename'], 
+                    file_info['media_type'],
+                    token,
+                    use_temp=False  # Force permanent upload
+                )
+                
+                if not permanent_url.startswith("Error"):
+                    # Replace the temporary file URL with permanent URL
+                    temp_url = f"file://{file_info['path']}"
+                    updated_content = updated_content.replace(temp_url, permanent_url)
+                    print(f"[DeployUpload] Replaced {temp_url} with {permanent_url}")
+                else:
+                    print(f"[DeployUpload] Failed to upload {file_id}: {permanent_url}")
+            
+            except Exception as e:
+                print(f"[DeployUpload] Error uploading {file_id}: {str(e)}")
+                continue
+        
+        # Clean up temporary files after upload
+        cleanup_temp_media_files()
+        
+        return updated_content
+        
+    except Exception as e:
+        print(f"[DeployUpload] Failed to upload temporary files: {str(e)}")
+        return html_content
+
+def cleanup_temp_media_files():
+    """Clean up temporary media files from disk and memory."""
+    try:
+        for file_id, file_info in temp_media_files.items():
+            try:
+                if os.path.exists(file_info['path']):
+                    os.remove(file_info['path'])
+                    print(f"[TempCleanup] Removed {file_info['path']}")
+            except Exception as e:
+                print(f"[TempCleanup] Failed to remove {file_info['path']}: {str(e)}")
+        
+        # Clear the global dictionary
+        temp_media_files.clear()
+        print("[TempCleanup] Cleared temporary media files registry")
+        
+    except Exception as e:
+        print(f"[TempCleanup] Error during cleanup: {str(e)}")
+
+def generate_image_with_qwen(prompt: str, image_index: int = 0, token: gr.OAuthToken | None = None) -> str:
+    """Generate image using Qwen image model via Hugging Face InferenceClient and upload to HF for permanent URL"""
+    try:
+        # Check if HF_TOKEN is available
+        if not os.getenv('HF_TOKEN'):
+            return "Error: HF_TOKEN environment variable is not set. Please set it to your Hugging Face API token."
+        
+        # Create InferenceClient for Qwen image generation
+        client = InferenceClient(
+            provider="auto",
+            api_key=os.getenv('HF_TOKEN'),
+            bill_to="huggingface",
+        )
+        
+        # Generate image using Qwen/Qwen-Image model
+        image = client.text_to_image(
+            prompt,
+            model="Qwen/Qwen-Image",
+        )
+        
+        # Resize image to reduce size while maintaining quality
+        max_size = 1024  # Increased size since we're not using data URIs
+        if image.width > max_size or image.height > max_size:
+            image.thumbnail((max_size, max_size), Image.Resampling.LANCZOS)
+        
+        # Convert PIL Image to bytes for upload
+        import io
+        buffer = io.BytesIO()
+        # Save as JPEG with good quality since we're not embedding
+        image.convert('RGB').save(buffer, format='JPEG', quality=90, optimize=True)
+        image_bytes = buffer.getvalue()
+        
+        # Create temporary URL for preview (will be uploaded to HF during deploy)
+        filename = f"generated_image_{image_index}.jpg"
+        temp_url = upload_media_to_hf(image_bytes, filename, "image", token, use_temp=True)
+        
+        # Check if creation was successful
+        if temp_url.startswith("Error"):
+            return temp_url
+        
+        # Return HTML img tag with temporary URL
+        return f'<img src="{temp_url}" alt="{prompt}" style="max-width: 100%; height: auto; border-radius: 8px; margin: 10px 0;" loading="lazy" />'
+        
+    except Exception as e:
+        print(f"Image generation error: {str(e)}")
+        return f"Error generating image: {str(e)}"
+
+def generate_image_to_image(input_image_data, prompt: str, token: gr.OAuthToken | None = None) -> str:
+    """Generate an image using image-to-image with Qwen-Image-Edit via Hugging Face InferenceClient."""
+    try:
+        # Check token
+        if not os.getenv('HF_TOKEN'):
+            return "Error: HF_TOKEN environment variable is not set. Please set it to your Hugging Face API token."
+
+        # Prepare client
+        client = InferenceClient(
+            provider="auto",
+            api_key=os.getenv('HF_TOKEN'),
+            bill_to="huggingface",
+        )
+
+        # Normalize input image to bytes
+        import io
+        from PIL import Image
+        try:
+            import numpy as np
+        except Exception:
+            np = None
+
+        if hasattr(input_image_data, 'read'):
+            # File-like object
+            raw = input_image_data.read()
+            pil_image = Image.open(io.BytesIO(raw))
+        elif hasattr(input_image_data, 'mode') and hasattr(input_image_data, 'size'):
+            # PIL Image
+            pil_image = input_image_data
+        elif np is not None and isinstance(input_image_data, np.ndarray):
+            pil_image = Image.fromarray(input_image_data)
+        elif isinstance(input_image_data, (bytes, bytearray)):
+            pil_image = Image.open(io.BytesIO(input_image_data))
+        else:
+            # Fallback: try to convert via bytes
+            pil_image = Image.open(io.BytesIO(bytes(input_image_data)))
+
+        # Ensure RGB
+        if pil_image.mode != 'RGB':
+            pil_image = pil_image.convert('RGB')
+
+        # Resize input image to avoid request body size limits
+        max_input_size = 1024
+        if pil_image.width > max_input_size or pil_image.height > max_input_size:
+            pil_image.thumbnail((max_input_size, max_input_size), Image.Resampling.LANCZOS)
+
+        buf = io.BytesIO()
+        pil_image.save(buf, format='JPEG', quality=85, optimize=True)
+        input_bytes = buf.getvalue()
+
+        # Call image-to-image
+        image = client.image_to_image(
+            input_bytes,
+            prompt=prompt,
+            model="Qwen/Qwen-Image-Edit",
+        )
+
+        # Resize/optimize (larger since not using data URIs)
+        max_size = 1024
+        if image.width > max_size or image.height > max_size:
+            image.thumbnail((max_size, max_size), Image.Resampling.LANCZOS)
+
+        out_buf = io.BytesIO()
+        image.convert('RGB').save(out_buf, format='JPEG', quality=90, optimize=True)
+        image_bytes = out_buf.getvalue()
+
+        # Create temporary URL for preview (will be uploaded to HF during deploy)
+        filename = "image_to_image_result.jpg"
+        temp_url = upload_media_to_hf(image_bytes, filename, "image", token, use_temp=True)
+        
+        # Check if creation was successful
+        if temp_url.startswith("Error"):
+            return temp_url
+
+        return f"<img src=\"{temp_url}\" alt=\"{prompt}\" style=\"max-width: 100%; height: auto; border-radius: 8px; margin: 10px 0;\" loading=\"lazy\" />"
+    except Exception as e:
+        print(f"Image-to-image generation error: {str(e)}")
+        return f"Error generating image (image-to-image): {str(e)}"
+
+def generate_video_from_image(input_image_data, prompt: str, session_id: Optional[str] = None, token: gr.OAuthToken | None = None) -> str:
+    """Generate a video from an input image and prompt using Hugging Face InferenceClient."""
+    try:
+        print("[Image2Video] Starting video generation")
+        if not os.getenv('HF_TOKEN'):
+            print("[Image2Video] Missing HF_TOKEN")
+            return "Error: HF_TOKEN environment variable is not set. Please set it to your Hugging Face API token."
+
+        # Prepare client
+        client = InferenceClient(
+            provider="auto",
+            api_key=os.getenv('HF_TOKEN'),
+            bill_to="huggingface",
+        )
+        print(f"[Image2Video] InferenceClient initialized (provider=auto)")
+
+        # Normalize input image to bytes, with downscale/compress to cap request size
+        import io
+        from PIL import Image
+        try:
+            import numpy as np
+        except Exception:
+            np = None
+
+        def _load_pil(img_like) -> Image.Image:
+            if hasattr(img_like, 'read'):
+                return Image.open(io.BytesIO(img_like.read()))
+            if hasattr(img_like, 'mode') and hasattr(img_like, 'size'):
+                return img_like
+            if np is not None and isinstance(img_like, np.ndarray):
+                return Image.fromarray(img_like)
+            if isinstance(img_like, (bytes, bytearray)):
+                return Image.open(io.BytesIO(img_like))
+            return Image.open(io.BytesIO(bytes(img_like)))
+
+        pil_image = _load_pil(input_image_data)
+        if pil_image.mode != 'RGB':
+            pil_image = pil_image.convert('RGB')
+        try:
+            print(f"[Image2Video] Input PIL image size={pil_image.size} mode={pil_image.mode}")
+        except Exception:
+            pass
+
+        # Progressive encode to keep payload under ~3.9MB (below 4MB limit)
+        MAX_BYTES = 3_900_000
+        max_dim = 1024  # initial cap on longest edge
+        quality = 90
+
+        def encode_current(pil: Image.Image, q: int) -> bytes:
+            tmp = io.BytesIO()
+            pil.save(tmp, format='JPEG', quality=q, optimize=True)
+            return tmp.getvalue()
+
+        # Downscale while the longest edge exceeds max_dim
+        while max(pil_image.size) > max_dim:
+            ratio = max_dim / float(max(pil_image.size))
+            new_size = (max(1, int(pil_image.size[0] * ratio)), max(1, int(pil_image.size[1] * ratio)))
+            pil_image = pil_image.resize(new_size, Image.Resampling.LANCZOS)
+
+        encoded = encode_current(pil_image, quality)
+        # If still too big, iteratively reduce quality, then dimensions
+        while len(encoded) > MAX_BYTES and (quality > 40 or max(pil_image.size) > 640):
+            if quality > 40:
+                quality -= 10
+            else:
+                # reduce dims by 15% if already at low quality
+                new_w = max(1, int(pil_image.size[0] * 0.85))
+                new_h = max(1, int(pil_image.size[1] * 0.85))
+                pil_image = pil_image.resize((new_w, new_h), Image.Resampling.LANCZOS)
+            encoded = encode_current(pil_image, quality)
+
+        input_bytes = encoded
+
+        # Call image-to-video; require method support
+        model_id = "Lightricks/LTX-Video-0.9.8-13B-distilled"
+        image_to_video_method = getattr(client, "image_to_video", None)
+        if not callable(image_to_video_method):
+            print("[Image2Video] InferenceClient.image_to_video not available in this huggingface_hub version")
+            return (
+                "Error generating video (image-to-video): Your installed huggingface_hub version "
+                "does not expose InferenceClient.image_to_video. Please upgrade with "
+                "`pip install -U huggingface_hub` and try again."
+            )
+        print(f"[Image2Video] Calling image_to_video with model={model_id}, prompt length={len(prompt or '')}")
+        video_bytes = image_to_video_method(
+            input_bytes,
+            prompt=prompt,
+            model=model_id,
+        )
+        print(f"[Image2Video] Received video bytes: {len(video_bytes) if hasattr(video_bytes, '__len__') else 'unknown length'}")
+
+        # Create temporary URL for preview (will be uploaded to HF during deploy)
+        filename = "image_to_video_result.mp4"
+        temp_url = upload_media_to_hf(video_bytes, filename, "video", token, use_temp=True)
+        
+        # Check if creation was successful
+        if temp_url.startswith("Error"):
+            return temp_url
+        
+        video_html = (
+            f'<video controls autoplay muted loop playsinline '
+            f'style="max-width: 100%; height: auto; border-radius: 8px; margin: 10px 0; display: block;" '
+            f'onloadstart="this.style.backgroundColor=\'#f0f0f0\'" '
+            f'onerror="this.style.display=\'none\'; console.error(\'Video failed to load\')">'
+            f'<source src="{temp_url}" type="video/mp4" />'
+            f'<p style="text-align: center; color: #666;">Your browser does not support the video tag.</p>'
+            f'</video>'
+        )
+        
+        print(f"[Image2Video] Successfully generated video HTML tag with temporary URL: {temp_url}")
+        
+        # Validate the generated video HTML
+        if not validate_video_html(video_html):
+            print("[Image2Video] Generated video HTML failed validation")
+            return "Error: Generated video HTML is malformed"
+        
+        return video_html
+    except Exception as e:
+        import traceback
+        print("[Image2Video] Exception during generation:")
+        traceback.print_exc()
+        print(f"Image-to-video generation error: {str(e)}")
+        return f"Error generating video (image-to-video): {str(e)}"
+
+def generate_video_from_text(prompt: str, session_id: Optional[str] = None, token: gr.OAuthToken | None = None) -> str:
+    """Generate a video from a text prompt using Hugging Face InferenceClient."""
+    try:
+        print("[Text2Video] Starting video generation from text")
+        if not os.getenv('HF_TOKEN'):
+            print("[Text2Video] Missing HF_TOKEN")
+            return "Error: HF_TOKEN environment variable is not set. Please set it to your Hugging Face API token."
+
+        client = InferenceClient(
+            provider="auto",
+            api_key=os.getenv('HF_TOKEN'),
+            bill_to="huggingface",
+        )
+        print("[Text2Video] InferenceClient initialized (provider=auto)")
+
+        # Ensure the client has text_to_video (newer huggingface_hub)
+        text_to_video_method = getattr(client, "text_to_video", None)
+        if not callable(text_to_video_method):
+            print("[Text2Video] InferenceClient.text_to_video not available in this huggingface_hub version")
+            return (
+                "Error generating video (text-to-video): Your installed huggingface_hub version "
+                "does not expose InferenceClient.text_to_video. Please upgrade with "
+                "`pip install -U huggingface_hub` and try again."
+            )
+
+        model_id = "Wan-AI/Wan2.2-T2V-A14B"
+        prompt_str = (prompt or "").strip()
+        print(f"[Text2Video] Calling text_to_video with model={model_id}, prompt length={len(prompt_str)}")
+        video_bytes = text_to_video_method(
+            prompt_str,
+            model=model_id,
+        )
+        print(f"[Text2Video] Received video bytes: {len(video_bytes) if hasattr(video_bytes, '__len__') else 'unknown length'}")
+
+        # Create temporary URL for preview (will be uploaded to HF during deploy)
+        filename = "text_to_video_result.mp4"
+        temp_url = upload_media_to_hf(video_bytes, filename, "video", token, use_temp=True)
+        
+        # Check if creation was successful
+        if temp_url.startswith("Error"):
+            return temp_url
+        
+        video_html = (
+            f'<video controls autoplay muted loop playsinline '
+            f'style="max-width: 100%; height: auto; border-radius: 8px; margin: 10px 0; display: block;" '
+            f'onloadstart="this.style.backgroundColor=\'#f0f0f0\'" '
+            f'onerror="this.style.display=\'none\'; console.error(\'Video failed to load\')">'
+            f'<source src="{temp_url}" type="video/mp4" />'
+            f'<p style="text-align: center; color: #666;">Your browser does not support the video tag.</p>'
+            f'</video>'
+        )
+        
+        print(f"[Text2Video] Successfully generated video HTML tag with temporary URL: {temp_url}")
+        
+        # Validate the generated video HTML
+        if not validate_video_html(video_html):
+            print("[Text2Video] Generated video HTML failed validation")
+            return "Error: Generated video HTML is malformed"
+        
+        return video_html
+    except Exception as e:
+        import traceback
+        print("[Text2Video] Exception during generation:")
+        traceback.print_exc()
+        print(f"Text-to-video generation error: {str(e)}")
+        return f"Error generating video (text-to-video): {str(e)}"
+
+def generate_music_from_text(prompt: str, music_length_ms: int = 30000, session_id: Optional[str] = None, token: gr.OAuthToken | None = None) -> str:
+    """Generate music from a text prompt using ElevenLabs Music API and return an HTML <audio> tag."""
+    try:
+        api_key = os.getenv('ELEVENLABS_API_KEY')
+        if not api_key:
+            return "Error: ELEVENLABS_API_KEY environment variable is not set."
+
+        headers = {
+            'Content-Type': 'application/json',
+            'xi-api-key': api_key,
+        }
+        payload = {
+            'prompt': (prompt or 'Epic orchestral theme with soaring strings and powerful brass'),
+            'music_length_ms': int(music_length_ms) if music_length_ms else 30000,
+        }
+
+        resp = requests.post('https://api.elevenlabs.io/v1/music/compose', headers=headers, json=payload)
+        try:
+            resp.raise_for_status()
+        except Exception as e:
+            return f"Error generating music: {getattr(e, 'response', resp).text if hasattr(e, 'response') else resp.text}"
+
+        # Create temporary URL for preview (will be uploaded to HF during deploy)
+        filename = "generated_music.mp3"
+        temp_url = upload_media_to_hf(resp.content, filename, "audio", token, use_temp=True)
+        
+        # Check if creation was successful
+        if temp_url.startswith("Error"):
+            return temp_url
+        
+        audio_html = (
+            "<div class=\"anycoder-music\" style=\"max-width:420px;margin:16px auto;padding:12px 16px;border:1px solid #e5e7eb;border-radius:12px;background:linear-gradient(180deg,#fafafa,#f3f4f6);box-shadow:0 2px 8px rgba(0,0,0,0.06)\">"
+            "  <div style=\"font-size:13px;color:#374151;margin-bottom:8px;display:flex:align-items:center;gap:6px\">"
+            "    <span>🎵 Generated music</span>"
+            "  </div>"
+            f"  <audio controls autoplay loop style=\"width:100%;outline:none;\">"
+            f"    <source src=\"{temp_url}\" type=\"audio/mpeg\" />"
+            "    Your browser does not support the audio element."
+            "  </audio>"
+            "</div>"
+        )
+        
+        print(f"[Music] Successfully generated music HTML tag with temporary URL: {temp_url}")
+        return audio_html
+    except Exception as e:
+        return f"Error generating music: {str(e)}"
+
+def extract_image_prompts_from_text(text: str, num_images_needed: int = 1) -> list:
+    """Extract image generation prompts from the full text based on number of images needed"""
+    # Use the entire text as the base prompt for image generation
+    # Clean up the text and create variations for the required number of images
+    
+    # Clean the text
+    cleaned_text = text.strip()
+    if not cleaned_text:
+        return []
+    
+    # Create variations of the prompt for the required number of images
+    prompts = []
+    
+    # Generate exactly the number of images needed
+    for i in range(num_images_needed):
+        if i == 0:
+            # First image: Use the full prompt as-is
+            prompts.append(cleaned_text)
+        elif i == 1:
+            # Second image: Add "visual representation" to make it more image-focused
+            prompts.append(f"Visual representation of {cleaned_text}")
+        elif i == 2:
+            # Third image: Add "illustration" to create a different style
+            prompts.append(f"Illustration of {cleaned_text}")
+        else:
+            # For additional images, use different variations
+            variations = [
+                f"Digital art of {cleaned_text}",
+                f"Modern design of {cleaned_text}",
+                f"Professional illustration of {cleaned_text}",
+                f"Clean design of {cleaned_text}",
+                f"Beautiful visualization of {cleaned_text}",
+                f"Stylish representation of {cleaned_text}",
+                f"Contemporary design of {cleaned_text}",
+                f"Elegant illustration of {cleaned_text}"
+            ]
+            variation_index = (i - 3) % len(variations)
+            prompts.append(variations[variation_index])
+    
+    return prompts
+
+def create_image_replacement_blocks(html_content: str, user_prompt: str) -> str:
+    """Create search/replace blocks to replace placeholder images with generated Qwen images"""
+    if not user_prompt:
+        return ""
+    
+    # Find existing image placeholders in the HTML first
+    import re
+    
+    # Common patterns for placeholder images
+    placeholder_patterns = [
+        r'<img[^>]*src=["\'](?:placeholder|dummy|sample|example)[^"\']*["\'][^>]*>',
+        r'<img[^>]*src=["\']https?://via\.placeholder\.com[^"\']*["\'][^>]*>',
+        r'<img[^>]*src=["\']https?://picsum\.photos[^"\']*["\'][^>]*>',
+        r'<img[^>]*src=["\']https?://dummyimage\.com[^"\']*["\'][^>]*>',
+        r'<img[^>]*alt=["\'][^"\']*placeholder[^"\']*["\'][^>]*>',
+        r'<img[^>]*class=["\'][^"\']*placeholder[^"\']*["\'][^>]*>',
+        r'<img[^>]*id=["\'][^"\']*placeholder[^"\']*["\'][^>]*>',
+        r'<img[^>]*src=["\']data:image[^"\']*["\'][^>]*>',  # Base64 images
+        r'<img[^>]*src=["\']#["\'][^>]*>',  # Empty src
+        r'<img[^>]*src=["\']about:blank["\'][^>]*>',  # About blank
+    ]
+    
+    # Find all placeholder images
+    placeholder_images = []
+    for pattern in placeholder_patterns:
+        matches = re.findall(pattern, html_content, re.IGNORECASE)
+        placeholder_images.extend(matches)
+    
+    # Filter out HF URLs from placeholders (they are real generated content)
+    placeholder_images = [img for img in placeholder_images if 'huggingface.co/datasets/' not in img]
+    
+    # If no placeholder images found, look for any img tags
+    if not placeholder_images:
+        img_pattern = r'<img[^>]*>'
+        # Case-insensitive to catch <IMG> or mixed-case tags
+        placeholder_images = re.findall(img_pattern, html_content, re.IGNORECASE)
+    
+    # Also look for div elements that might be image placeholders
+    div_placeholder_patterns = [
+        r'<div[^>]*class=["\'][^"\']*(?:image|img|photo|picture)[^"\']*["\'][^>]*>.*?</div>',
+        r'<div[^>]*id=["\'][^"\']*(?:image|img|photo|picture)[^"\']*["\'][^>]*>.*?</div>',
+    ]
+    
+    for pattern in div_placeholder_patterns:
+        matches = re.findall(pattern, html_content, re.IGNORECASE | re.DOTALL)
+        placeholder_images.extend(matches)
+    
+    # Count how many images we need to generate
+    num_images_needed = len(placeholder_images)
+    
+    if num_images_needed == 0:
+        return ""
+    
+    # Generate image prompts based on the number of images found
+    image_prompts = extract_image_prompts_from_text(user_prompt, num_images_needed)
+    
+    # Generate images for each prompt
+    generated_images = []
+    for i, prompt in enumerate(image_prompts):
+        image_html = generate_image_with_qwen(prompt, i, token=None)  # TODO: Pass token from parent context
+        if not image_html.startswith("Error"):
+            generated_images.append((i, image_html))
+    
+    if not generated_images:
+        return ""
+    
+    # Create search/replace blocks
+    replacement_blocks = []
+    
+    for i, (prompt_index, generated_image) in enumerate(generated_images):
+        if i < len(placeholder_images):
+            # Replace existing placeholder
+            placeholder = placeholder_images[i]
+            # Clean up the placeholder for better matching
+            placeholder_clean = re.sub(r'\s+', ' ', placeholder.strip())
+            
+            # Try multiple variations of the placeholder for better matching
+            placeholder_variations = [
+                placeholder_clean,
+                placeholder_clean.replace('"', "'"),
+                placeholder_clean.replace("'", '"'),
+                re.sub(r'\s+', ' ', placeholder_clean),
+                placeholder_clean.replace('  ', ' '),
+            ]
+            
+            # Create a replacement block for each variation
+            for variation in placeholder_variations:
+                replacement_blocks.append(f"""{SEARCH_START}
+{variation}
+{DIVIDER}
+{generated_image}
+{REPLACE_END}""")
+        else:
+            # Add new image if we have more generated images than placeholders
+            # Find a good insertion point (after body tag or main content)
+            if '<body' in html_content:
+                body_end = html_content.find('>', html_content.find('<body')) + 1
+                insertion_point = html_content[:body_end] + '\n    '
+                replacement_blocks.append(f"""{SEARCH_START}
+{insertion_point}
+{DIVIDER}
+{insertion_point}
+    {generated_image}
+{REPLACE_END}""")
+    
+    return '\n\n'.join(replacement_blocks)
+
+def create_image_replacement_blocks_text_to_image_single(html_content: str, prompt: str) -> str:
+    """Create search/replace blocks that generate and insert ONLY ONE text-to-image result."""
+    if not prompt or not prompt.strip():
+        return ""
+
+    import re
+
+    # Detect placeholders similarly to the multi-image version
+    placeholder_patterns = [
+        r'<img[^>]*src=["\'](?:placeholder|dummy|sample|example)[^"\']*["\'][^>]*>',
+        r'<img[^>]*src=["\']https?://via\.placeholder\.com[^"\']*["\'][^>]*>',
+        r'<img[^>]*src=["\']https?://picsum\.photos[^"\']*["\'][^>]*>',
+        r'<img[^>]*src=["\']https?://dummyimage\.com[^"\']*["\'][^>]*>',
+        r'<img[^>]*alt=["\'][^"\']*placeholder[^"\']*["\'][^>]*>',
+        r'<img[^>]*class=["\'][^"\']*placeholder[^"\']*["\'][^>]*>',
+        r'<img[^>]*id=["\'][^"\']*placeholder[^"\']*["\'][^>]*>',
+        r'<img[^>]*src=["\']data:image[^"\']*["\'][^>]*>',
+        r'<img[^>]*src=["\']#["\'][^>]*>',
+        r'<img[^>]*src=["\']about:blank["\'][^>]*>',
+    ]
+
+    placeholder_images = []
+    for pattern in placeholder_patterns:
+        matches = re.findall(pattern, html_content, re.IGNORECASE)
+        if matches:
+            placeholder_images.extend(matches)
+    
+    # Filter out HF URLs from placeholders (they are real generated content)
+    placeholder_images = [img for img in placeholder_images if 'huggingface.co/datasets/' not in img]
+    
+    # Filter out HF URLs from placeholders (they are real generated content)
+    placeholder_images = [img for img in placeholder_images if 'huggingface.co/datasets/' not in img]
+
+    # Fallback to any <img> if no placeholders
+    if not placeholder_images:
+        img_pattern = r'<img[^>]*>'
+        placeholder_images = re.findall(img_pattern, html_content)
+
+    # Generate a single image
+    image_html = generate_image_with_qwen(prompt, 0, token=None)  # TODO: Pass token from parent context
+    if image_html.startswith("Error"):
+        return ""
+
+    # Replace first placeholder if present
+    if placeholder_images:
+        placeholder = placeholder_images[0]
+        placeholder_clean = re.sub(r'\s+', ' ', placeholder.strip())
+        placeholder_variations = [
+            placeholder_clean,
+            placeholder_clean.replace('"', "'"),
+            placeholder_clean.replace("'", '"'),
+            re.sub(r'\s+', ' ', placeholder_clean),
+            placeholder_clean.replace('  ', ' '),
+        ]
+        blocks = []
+        for variation in placeholder_variations:
+            blocks.append(f"""{SEARCH_START}
+{variation}
+{DIVIDER}
+{image_html}
+{REPLACE_END}""")
+        return '\n\n'.join(blocks)
+
+    # Otherwise insert after <body>
+    if '<body' in html_content:
+        body_end = html_content.find('>', html_content.find('<body')) + 1
+        insertion_point = html_content[:body_end] + '\n    '
+        return f"""{SEARCH_START}
+{insertion_point}
+{DIVIDER}
+{insertion_point}
+    {image_html}
+{REPLACE_END}"""
+
+    # If no <body>, just append
+    return f"{SEARCH_START}\n\n{DIVIDER}\n{image_html}\n{REPLACE_END}"
+
+def create_video_replacement_blocks_text_to_video(html_content: str, prompt: str, session_id: Optional[str] = None) -> str:
+    """Create search/replace blocks that generate and insert ONLY ONE text-to-video result."""
+    if not prompt or not prompt.strip():
+        return ""
+
+    import re
+
+    # Detect the same placeholders as image counterparts, to replace the first image slot with a video
+    placeholder_patterns = [
+        r'<img[^>]*src=["\'](?:placeholder|dummy|sample|example)[^"\']*["\'][^>]*>',
+        r'<img[^>]*src=["\']https?://via\.placeholder\.com[^"\']*["\'][^>]*>',
+        r'<img[^>]*src=["\']https?://picsum\.photos[^"\']*["\'][^>]*>',
+        r'<img[^>]*src=["\']https?://dummyimage\.com[^"\']*["\'][^>]*>',
+        r'<img[^>]*alt=["\'][^"\']*placeholder[^"\']*["\'][^>]*>',
+        r'<img[^>]*class=["\'][^"\']*placeholder[^"\']*["\'][^>]*>',
+        r'<img[^>]*id=["\'][^"\']*placeholder[^"\']*["\'][^>]*>',
+        r'<img[^>]*src=["\']data:image[^"\']*["\'][^>]*>',
+        r'<img[^>]*src=["\']#["\'][^>]*>',
+        r'<img[^>]*src=["\']about:blank["\'][^>]*>',
+    ]
+
+    placeholder_images = []
+    for pattern in placeholder_patterns:
+        matches = re.findall(pattern, html_content, re.IGNORECASE)
+        if matches:
+            placeholder_images.extend(matches)
+    
+    # Filter out HF URLs from placeholders (they are real generated content)
+    placeholder_images = [img for img in placeholder_images if 'huggingface.co/datasets/' not in img]
+
+    if not placeholder_images:
+        img_pattern = r'<img[^>]*>'
+        placeholder_images = re.findall(img_pattern, html_content)
+
+    video_html = generate_video_from_text(prompt, session_id=session_id, token=None)  # TODO: Pass token from parent context
+    if video_html.startswith("Error"):
+        return ""
+
+    # Replace first placeholder if present
+    if placeholder_images:
+        placeholder = placeholder_images[0]
+        placeholder_clean = re.sub(r'\s+', ' ', placeholder.strip())
+        placeholder_variations = [
+            placeholder,
+            placeholder_clean,
+            placeholder_clean.replace('"', "'"),
+            placeholder_clean.replace("'", '"'),
+            re.sub(r'\s+', ' ', placeholder_clean),
+            placeholder_clean.replace('  ', ' '),
+        ]
+        blocks = []
+        for variation in placeholder_variations:
+            blocks.append(f"""{SEARCH_START}
+{variation}
+{DIVIDER}
+{video_html}
+{REPLACE_END}""")
+        return '\n\n'.join(blocks)
+
+    # Otherwise insert after <body> with proper container
+    if '<body' in html_content:
+        body_start = html_content.find('<body')
+        body_end = html_content.find('>', body_start) + 1
+        opening_body_tag = html_content[body_start:body_end]
+        
+        # Look for existing container elements to insert into
+        body_content_start = body_end
+        
+        # Try to find a good insertion point within existing content structure
+        patterns_to_try = [
+            r'<main[^>]*>',
+            r'<section[^>]*class="[^"]*hero[^"]*"[^>]*>',
+            r'<div[^>]*class="[^"]*container[^"]*"[^>]*>',
+            r'<header[^>]*>',
+        ]
+        
+        insertion_point = None
+        for pattern in patterns_to_try:
+            import re
+            match = re.search(pattern, html_content[body_content_start:], re.IGNORECASE)
+            if match:
+                match_end = body_content_start + match.end()
+                # Find the end of this tag
+                tag_content = html_content[body_content_start + match.start():match_end]
+                insertion_point = html_content[:match_end] + '\n        '
+                break
+        
+        if not insertion_point:
+            # Fallback to right after body tag with container div
+            insertion_point = html_content[:body_end] + '\n    '
+            video_with_container = f'<div class="video-container" style="margin: 20px 0; text-align: center;">\n        {video_html}\n    </div>'
+            return f"""{SEARCH_START}
+{insertion_point}
+{DIVIDER}
+{insertion_point}
+    {video_with_container}
+{REPLACE_END}"""
+        else:
+            return f"""{SEARCH_START}
+{insertion_point}
+{DIVIDER}
+{insertion_point}
+        {video_html}
+{REPLACE_END}"""
+
+    # If no <body>, just append
+    return f"{SEARCH_START}\n\n{DIVIDER}\n{video_html}\n{REPLACE_END}"
+
+def create_music_replacement_blocks_text_to_music(html_content: str, prompt: str, session_id: Optional[str] = None) -> str:
+    """Create search/replace blocks that insert ONE generated <audio> near the top of <body>."""
+    if not prompt or not prompt.strip():
+        return ""
+
+    audio_html = generate_music_from_text(prompt, session_id=session_id, token=None)  # TODO: Pass token from parent context
+    if audio_html.startswith("Error"):
+        return ""
+
+    # Prefer inserting after the first <section>...</section> if present; else after <body>
+    import re
+    section_match = re.search(r"<section\b[\s\S]*?</section>", html_content, flags=re.IGNORECASE)
+    if section_match:
+        section_html = section_match.group(0)
+        section_clean = re.sub(r"\s+", " ", section_html.strip())
+        variations = [
+            section_html,
+            section_clean,
+            section_clean.replace('"', "'"),
+            section_clean.replace("'", '"'),
+            re.sub(r"\s+", " ", section_clean),
+        ]
+        blocks = []
+        for v in variations:
+            blocks.append(f"""{SEARCH_START}
+{v}
+{DIVIDER}
+{v}\n    {audio_html}
+{REPLACE_END}""")
+        return "\n\n".join(blocks)
+    if '<body' in html_content:
+        body_end = html_content.find('>', html_content.find('<body')) + 1
+        insertion_point = html_content[:body_end] + '\n    '
+        return f"""{SEARCH_START}
+{insertion_point}
+{DIVIDER}
+{insertion_point}
+    {audio_html}
+{REPLACE_END}"""
+
+    # If no <body>, just append
+    return f"{SEARCH_START}\n\n{DIVIDER}\n{audio_html}\n{REPLACE_END}"