Initial Commit

.gitignore

@@ -0,0 +1,11 @@
+venv
+videos
+__pycache__
+*.pyc
+*.pyo
+.vscode
+.idea
+.env
+
+Dockerfile
+.vscode

.gradio/certificate.pem

@@ -0,0 +1,31 @@
+-----BEGIN CERTIFICATE-----
+MIIFazCCA1OgAwIBAgIRAIIQz7DSQONZRGPgu2OCiwAwDQYJKoZIhvcNAQELBQAw
+TzELMAkGA1UEBhMCVVMxKTAnBgNVBAoTIEludGVybmV0IFNlY3VyaXR5IFJlc2Vh
+cmNoIEdyb3VwMRUwEwYDVQQDEwxJU1JHIFJvb3QgWDEwHhcNMTUwNjA0MTEwNDM4
+WhcNMzUwNjA0MTEwNDM4WjBPMQswCQYDVQQGEwJVUzEpMCcGA1UEChMgSW50ZXJu
+ZXQgU2VjdXJpdHkgUmVzZWFyY2ggR3JvdXAxFTATBgNVBAMTDElTUkcgUm9vdCBY
+MTCCAiIwDQYJKoZIhvcNAQEBBQADggIPADCCAgoCggIBAK3oJHP0FDfzm54rVygc
+h77ct984kIxuPOZXoHj3dcKi/vVqbvYATyjb3miGbESTtrFj/RQSa78f0uoxmyF+
+0TM8ukj13Xnfs7j/EvEhmkvBioZxaUpmZmyPfjxwv60pIgbz5MDmgK7iS4+3mX6U
+A5/TR5d8mUgjU+g4rk8Kb4Mu0UlXjIB0ttov0DiNewNwIRt18jA8+o+u3dpjq+sW
+T8KOEUt+zwvo/7V3LvSye0rgTBIlDHCNAymg4VMk7BPZ7hm/ELNKjD+Jo2FR3qyH
+B5T0Y3HsLuJvW5iB4YlcNHlsdu87kGJ55tukmi8mxdAQ4Q7e2RCOFvu396j3x+UC
+B5iPNgiV5+I3lg02dZ77DnKxHZu8A/lJBdiB3QW0KtZB6awBdpUKD9jf1b0SHzUv
+KBds0pjBqAlkd25HN7rOrFleaJ1/ctaJxQZBKT5ZPt0m9STJEadao0xAH0ahmbWn
+OlFuhjuefXKnEgV4We0+UXgVCwOPjdAvBbI+e0ocS3MFEvzG6uBQE3xDk3SzynTn
+jh8BCNAw1FtxNrQHusEwMFxIt4I7mKZ9YIqioymCzLq9gwQbooMDQaHWBfEbwrbw
+qHyGO0aoSCqI3Haadr8faqU9GY/rOPNk3sgrDQoo//fb4hVC1CLQJ13hef4Y53CI
+rU7m2Ys6xt0nUW7/vGT1M0NPAgMBAAGjQjBAMA4GA1UdDwEB/wQEAwIBBjAPBgNV
+HRMBAf8EBTADAQH/MB0GA1UdDgQWBBR5tFnme7bl5AFzgAiIyBpY9umbbjANBgkq
+hkiG9w0BAQsFAAOCAgEAVR9YqbyyqFDQDLHYGmkgJykIrGF1XIpu+ILlaS/V9lZL
+ubhzEFnTIZd+50xx+7LSYK05qAvqFyFWhfFQDlnrzuBZ6brJFe+GnY+EgPbk6ZGQ
+3BebYhtF8GaV0nxvwuo77x/Py9auJ/GpsMiu/X1+mvoiBOv/2X/qkSsisRcOj/KK
+NFtY2PwByVS5uCbMiogziUwthDyC3+6WVwW6LLv3xLfHTjuCvjHIInNzktHCgKQ5
+ORAzI4JMPJ+GslWYHb4phowim57iaztXOoJwTdwJx4nLCgdNbOhdjsnvzqvHu7Ur
+TkXWStAmzOVyyghqpZXjFaH3pO3JLF+l+/+sKAIuvtd7u+Nxe5AW0wdeRlN8NwdC
+jNPElpzVmbUq4JUagEiuTDkHzsxHpFKVK7q4+63SM1N95R1NbdWhscdCb+ZAJzVc
+oyi3B43njTOQ5yOf+1CceWxG1bQVs5ZufpsMljq4Ui0/1lvh+wjChP4kqKOJ2qxq
+4RgqsahDYVvTH9w7jXbyLeiNdd8XM2w9U/t7y0Ff/9yi0GE44Za4rF2LN9d11TPA
+mRGunUHBcnWEvgJBQl9nJEiU0Zsnvgc/ubhPgXRR4Xq37Z0j4r7g1SgEEzwxA57d
+emyPxgcYxn/eR44/KJ4EBs+lVDR3veyJm+kXQ99b21/+jh5Xos1AnX5iItreGCc=
+-----END CERTIFICATE-----

README.md

@@ -0,0 +1,13 @@
+---
+title: Video Editing Genie
+emoji: 🎥
+colorFrom: blue
+colorTo: purple
+sdk: gradio
+app_file: app.py
+pinned: false
+tags:
+  - mcp-server-track
+---
+
+Video Editing Genie is an AI-powered video editing assistant that helps users create, edit, and enhance videos using advanced AI models like Gemini.

app.py

@@ -0,0 +1,249 @@
+import gradio as gr
+import os
+from dotenv import load_dotenv
+load_dotenv()
+from tools.video_tools import (
+    change_aspect_ratio,
+    add_text_overlay,
+    change_speed,
+    extract_audio,
+    add_image_overlay,
+    apply_color_filter,
+    merge_videos,
+    edit_video_segment,
+)
+from tools.audio_tools import (
+    add_audio_to_video,
+    search_audio_from_envato,
+)
+from typing import List
+from tools.gemini_tool import analyze_video_context
+from tools.subtitle_tools import generate_word_level_subtitles_from_url
+from pathlib import Path
+import ast
+videos_dir = Path.cwd().absolute() / "videos"
+if not videos_dir.exists():
+    videos_dir.mkdir(parents=True, exist_ok=True)
+
+
+
+
+demo = gr.TabbedInterface(
+    [
+        gr.Interface(
+            fn=change_aspect_ratio,
+            inputs=[
+                gr.Textbox(label="Video URL"),
+                gr.Dropdown(["9:16", "1:1", "4:3", "16:9"], label="Target Ratio", value="9:16"),
+                gr.Dropdown(["blur_bg", "pad", "crop", "stretch"], label="Method", value="blur_bg", 
+                    info="blur_bg: Recommended for YouTube style videos - adds blurred background with centered video")
+            ],
+            outputs=gr.Textbox(label="Result URL"),
+            description="Convert video aspect ratio with YouTube-style blurred background",
+            api_name="change_aspect_ratio"
+        ),
+        gr.Interface(
+            fn=add_text_overlay,
+            inputs=[
+                gr.Textbox(label="Video URL"),
+                gr.JSON(label="Text Elements", value=[{
+                    "text": "Sample Text",
+                    "start_time": 0,
+                    "end_time": 5,
+                    "font_size": 24,
+                    "font_color": "white",
+                    "x_pos": "center",
+                    "y_pos": "bottom",
+                    "box": True,
+                    "box_color": "[email protected]"
+                }])
+            ],
+            outputs=gr.Textbox(label="Result URL"),
+            api_name="add_text_overlay"
+        ),
+        gr.Interface(
+            fn=change_speed,
+            inputs=[
+                gr.Textbox(label="Video URL"),
+                gr.Slider(0.25, 4, value=1.0, label="Speed Factor")
+            ],
+            outputs=gr.Textbox(label="Result URL"),
+            api_name="change_speed"
+        ),
+        gr.Interface(
+            fn=extract_audio,
+            inputs=[
+                gr.Textbox(label="Video URL"),
+                gr.Dropdown(["mp3", "aac", "wav", "ogg", "flac"], label="Audio Format", value="mp3")
+            ],
+            outputs=gr.Textbox(label="Result URL"),
+            api_name="extract_audio"
+        ),
+        
+        gr.Interface(
+            fn=add_image_overlay,
+            inputs=[
+                gr.Textbox(label="Video URL"),
+                gr.Textbox(label="Image Path"),
+                gr.Dropdown(["top_left", "top_right", "bottom_left", "bottom_right", "center"], 
+                           label="Position", value="top_right"),
+                gr.Slider(0, 1, value=1.0, label="Opacity"),
+                gr.Number(label="Start Time (seconds, optional)", value=None),
+                gr.Number(label="End Time (seconds, optional)", value=None),
+                gr.Textbox(label="Width (optional, e.g. '100', 'iw*0.1')", value=None),
+                gr.Textbox(label="Height (optional, e.g. '50', 'ih*0.1')", value=None)
+            ],
+            outputs=gr.Textbox(label="Result URL"),
+            api_name="add_image_overlay"
+        ),
+        gr.Interface(
+            fn=apply_color_filter,
+            inputs=[
+                gr.Textbox(label="Video URL"),
+                gr.Dropdown(["sepia", "grayscale", "warm", "cool", "vintage"], label="Filter Type", value="sepia"),
+                gr.Slider(0, 1, value=1.0, label="Intensity")
+            ],
+            outputs=gr.Textbox(label="Result URL"),
+            api_name="apply_color_filter"
+        ),
+        gr.Interface(
+            fn=merge_videos,
+            inputs=[
+                gr.Textbox(
+                    label="Video URLs (Python list format)", 
+                    placeholder='["https://url1.com", "https://url2.com"]',
+                    lines=4,
+                    info="Enter a list of video URLs to merge with transitions"
+                ),
+                # gr.Dropdown(
+                #     ["fade", "none"], 
+                #     label="Transition Type", 
+                #     value="fade",
+                #     info="Select the type of transition to apply between videos (only fade is supported)"
+                # ),
+                # gr.Number(
+                #     0.5, 3.0, 
+                #     value=1.0, 
+                #     label="Transition Duration (seconds)",
+                #     info="Duration of each transition between clips"
+                # )
+            ],
+            outputs=gr.Textbox(label="Result URL"),
+            title="Merge Videos",
+            description="Combine multiple videos",
+            api_name="merge_videos"
+        ),
+        gr.Interface(
+            fn=add_audio_to_video,
+            inputs=[
+                gr.Textbox(label="Video URL"),
+                gr.Textbox(label="Audio URL"),
+                gr.Number(label="Start Time (seconds)", value=0),
+                gr.Slider(0, 2, value=1.0, label="Volume"),
+                gr.Radio(["true", "false"], label="Keep Original Audio", value="true")
+            ],
+            outputs=gr.Textbox(label="Result URL"),
+            api_name="add_audio"
+        ),
+
+        gr.Interface(
+            fn=analyze_video_context,
+            inputs=[
+                gr.Textbox(label="Video URL"),
+                gr.Textbox(label="Analysis Task", value="Describe what's happening in this video")
+            ],
+            outputs=gr.Textbox(label="Analysis Result"),
+            api_name="analyze_video"
+        ),
+        gr.Interface(
+            fn=edit_video_segment,
+            inputs=[
+                gr.Textbox(label="Video URL"),
+                gr.Number(label="Start Time (seconds)", value=0),
+                gr.Number(label="End Time (seconds)", value=10),
+                gr.Radio(["cut", "keep"], label="Action", value="keep")
+            ],
+            outputs=gr.Textbox(label="Result URL"),
+            api_name="edit_video_segment"
+        ),
+        # Add this to your Gradio app
+        gr.Interface(
+        fn=generate_word_level_subtitles_from_url,
+        inputs=[
+            gr.Textbox(
+                label="Video URL", 
+                placeholder="Enter the video URL to add word-level subtitles",
+                lines=1,
+                info="Required: Direct URL to the video file"
+            ),
+            gr.Textbox(
+                label="Custom API URL (Optional)", 
+                placeholder="Leave blank to use default transcription API",
+                lines=1,
+                info="Optional: Custom transcription API endpoint"
+            )
+        ],
+        outputs=gr.Textbox(
+            label="Result", 
+            placeholder="Processing status and download URL will appear here...",
+            lines=3,
+            info="Check here for the final video URL or any error messages"
+        ),
+        title="Word-Level Subtitle Generator",
+        description="""
+        **MCP Server Tool: Add word-by-word subtitles to videos**
+        
+        This tool processes videos to add precise word-level subtitles where each word appears exactly when spoken.
+        Perfect for creating engaging content with synchronized text overlay.
+        
+        Simply paste your video URL and click Submit. Processing typically takes 1-3 minutes.
+        """,
+        examples=[
+            ["https://example.com/sample-video.mp4", ""],
+            ["https://example.com/another-video.mp4", "https://custom-api.com/transcribe"]
+        ],
+        api_name="word_level_subtitles",
+       
+    ),
+    gr.Interface(
+        fn=search_audio_from_envato,
+        inputs=[
+            gr.Textbox(label="Search Query"),
+            gr.Slider(1, 50, value=10, step=1, label="Limit"),
+            gr.Dropdown(
+                ["music", "effects"], 
+                label="Category", 
+                value=None, 
+                multiselect=False
+            ),
+            gr.Number(label="Min Length (seconds)", value=None),
+            gr.Number(label="Max Length (seconds)", value=None),
+            gr.Textbox(label="Tags (comma-separated)", value=None),
+            gr.Checkbox(label="Free Only", value=False)
+        ],
+        outputs=gr.JSON(label="Sound Effects"),
+        title="Sound Effect Search",
+        description="Search for royalty-free sound effects using Envato API",
+        api_name="search_audio_from_envato"
+    )
+    ],
+
+    [
+        "Change Aspect Ratio",
+        "Add Text Overlay",
+        "Change Speed",
+        "Extract Audio",
+        "Add Image Overlay",
+        "Apply Color Filter",
+        "Merge Videos",
+        "Add Audio",
+        "Analyze Video",
+        "Edit Video Segment",
+        "Add Subtitles",
+        "Search Audio from Envato"
+    ]
+)
+
+if __name__ == "__main__":
+
+    demo.launch(mcp_server=True, share=True, allowed_paths=[videos_dir])

deploy_whisper_on_modal.py

@@ -0,0 +1,105 @@
+import modal
+from fastapi import Form, HTTPException
+
+
+cuda_version = "12.4.0"  # should be no greater than host CUDA version
+flavor = "devel"  #  includes full CUDA toolkit
+operating_sys = "ubuntu22.04"
+tag = f"{cuda_version}-{flavor}-{operating_sys}"
+
+image = (
+    modal.Image.from_registry(f"nvidia/cuda:{tag}", add_python="3.11")
+    .apt_install(
+        "git",
+        "ffmpeg",
+        "libcudnn8",
+        "libcudnn8-dev",
+    )
+    .pip_install(
+        "fastapi[standard]",
+        "httpx",
+        "torch==2.0.0",
+        "torchaudio==2.0.0",
+        "numpy<2.0",
+        extra_index_url="https://download.pytorch.org/whl/cu118",
+    )
+    .pip_install(
+        "git+https://github.com/m-bain/[email protected]",
+        "ffmpeg-python",
+        "ctranslate2==4.4.0",
+    )
+)
+app = modal.App("whisperx-api", image=image)
+
+GPU_CONFIG = "L4"
+
+CACHE_DIR = "/cache"
+cache_vol = modal.Volume.from_name("whisper-cache", create_if_missing=True)
+
+@app.cls(
+    gpu=GPU_CONFIG,
+    volumes={CACHE_DIR: cache_vol},
+    scaledown_window=60 * 10,
+    timeout=60 * 60,
+)
+@modal.concurrent(max_inputs=15)
+class Model:
+    @modal.enter()
+    def setup(self):
+        import whisperx
+
+        device = "cuda"
+        compute_type = (
+            "float16"  # change to "int8" if low on GPU mem (may reduce accuracy)
+        )
+
+        # 1. Transcribe with original whisper (batched)
+        self.model = whisperx.load_model("large-v2", device, compute_type=compute_type, download_root=CACHE_DIR)
+
+    @modal.method()
+    def transcribe(self, audio_url: str):
+        import requests
+        import whisperx
+
+        batch_size = 16
+
+        response = requests.get(audio_url)
+        audio_path = "downloaded_audio.wav"
+        with open(audio_path, "wb") as audio_file:
+            audio_file.write(response.content)
+
+        audio = whisperx.load_audio(audio_path)
+
+        result = self.model.transcribe(audio, batch_size=batch_size)
+
+        model_a, metadata = whisperx.load_align_model(language_code=result["language"], device="cuda")
+        aligned_result = whisperx.align(result["segments"], model_a, metadata, audio_path, device="cuda")
+
+        results = {
+            "language": result["language"],
+            "language_probability": result.get("language_probability", 1.0),
+            "words": []
+        }
+
+        for word in aligned_result["word_segments"]:
+            results["words"].append({
+                "start": word["start"],
+                "end": word["end"],
+                "word": word["word"]
+            })
+
+        return results
+
+@app.function()
+@modal.fastapi_endpoint(docs=True, method="POST")
+async def transcribe_endpoint(url: str = Form(...)):
+    if not url.startswith(("http://", "https://")):
+        raise HTTPException(status_code=400, detail="URL must start with http:// or https://")
+    return  Model().transcribe.remote(audio_url=url)
+
+# ## Run the model
+@app.local_entrypoint()
+def main():
+    url = "https://pub-ebe9e51393584bf5b5bea84a67b343c2.r2.dev/examples_english_english.wav"
+    print(Model().transcribe.remote(url))
+

helpers/video_tool_helper.py

@@ -0,0 +1,220 @@
+import os
+import subprocess
+import tempfile
+import shutil
+from videos_management import save_video_to_static_folder, download_video_from_url
+import uuid
+
+def ratio_to_resolution(ratio: str, base_height=720):
+    """Convert ratio string W:H to pixel resolution keeping height fixed."""
+    try:
+        w, h = map(int, ratio.split(':'))
+        if w <= 0 or h <= 0:
+            raise ValueError()
+    except Exception:
+        raise ValueError(f"Invalid ratio format: {ratio}. Expected 'W:H' with positive ints.")
+    width = int(base_height * w / h)
+    height = base_height
+    # Ensure even numbers for ffmpeg
+    width = width + (width % 2)
+    height = height + (height % 2)
+    return width, height
+
+def probe_video_dimensions(video_path):
+    """Use ffprobe to get width and height of video."""
+    cmd = [
+        "ffprobe", "-v", "error",
+        "-select_streams", "v:0",
+        "-show_entries", "stream=width,height",
+        "-of", "csv=p=0",
+        video_path
+    ]
+    output = subprocess.check_output(cmd).decode().strip()
+    width, height = map(int, output.split(','))
+    return width, height
+
+
+
+def get_duration(video_path: str) -> float:
+    """Get duration of a video using ffprobe."""
+    cmd = [
+        'ffprobe', '-v', 'error', '-show_entries', 'format=duration',
+        '-of', 'default=noprint_wrappers=1:nokey=1', video_path
+    ]
+    result = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+    return float(result.stdout.strip())
+
+def add_transition(
+    first_video_url: str,
+    second_video_url: str,
+    transition_type: str = "fade",
+    transition_duration: float = 1.0
+) -> str:
+    """
+    Add a simple transition between two videos using fade in/out
+    
+    Args:
+        first_video_url: URL to the first video
+        second_video_url: URL to the second video
+        transition_type: Type of transition (only "fade" is supported)
+        transition_duration: Duration of the transition in seconds
+        
+    Returns:
+        URL to the merged video with transition
+    """
+    # Download videos
+    first_video_path = download_video_from_url(first_video_url)
+    second_video_path = download_video_from_url(second_video_url)
+    
+    if not first_video_path or not second_video_path:
+        print("Failed to download one or both videos")
+        return None
+
+    # Create temp directory and files
+    temp_dir = tempfile.mkdtemp()
+    first_fade_out = os.path.join(temp_dir, "fade_out.mp4")
+    second_fade_in = os.path.join(temp_dir, "fade_in.mp4")
+    concat_file = os.path.join(temp_dir, "concat.txt")
+    output_path = os.path.join(temp_dir, "output.mp4")
+    
+    try:
+        # Get durations
+        dur1 = get_duration(first_video_path)
+        dur2 = get_duration(second_video_path)
+
+        # Validate transition duration
+        if transition_duration <= 0 or transition_duration > min(dur1, dur2):
+            print(f"Invalid transition duration (max: {min(dur1, dur2):.2f}s)")
+            return None
+
+        fade_out_start = dur1 - transition_duration
+        subprocess.run([
+            "ffmpeg", "-y", "-i", first_video_path,
+            "-vf", f"fade=t=out:st={fade_out_start}:d={transition_duration}",
+            "-c:a", "copy", first_fade_out
+        ], check=True, capture_output=True)
+        
+        subprocess.run([
+            "ffmpeg", "-y", "-i", second_video_path,
+            "-vf", f"fade=t=in:st=0:d={transition_duration}",
+            "-c:a", "copy", second_fade_in
+        ], check=True, capture_output=True)
+        
+        with open(concat_file, 'w') as f:
+            f.write(f"file '{first_fade_out}'\n")
+            f.write(f"file '{second_fade_in}'\n")
+        
+        subprocess.run([
+            "ffmpeg", "-y", "-f", "concat", "-safe", "0",
+            "-i", concat_file, "-c", "copy", output_path
+        ], check=True, capture_output=True)
+        
+        result_url = save_video_to_static_folder(output_path)
+        
+        shutil.rmtree(temp_dir)
+        
+        return result_url
+        
+    except Exception as e:
+        print(f"Error creating transition: {str(e)}")
+        if os.path.exists(temp_dir):
+            shutil.rmtree(temp_dir)
+    return None
+
+
+
+def crop_video(
+    video_url: str,
+    x: int,
+    y: int,
+    width: int,
+    height: int
+) -> str:
+    """
+    Crop a video to the specified dimensions.
+    
+    Args:
+        video_url (str): URL to the input video file
+        x (int): X coordinate of the top-left corner
+        y (int): Y coordinate of the top-left corner
+        width (int): Width of the cropped area
+        height (int): Height of the cropped area
+        
+    Returns:
+        str: URL to the output video file
+    """
+    # Download the video from URL
+    video_path = download_video_from_url(video_url)
+    if not video_path:
+        return None
+    
+    output_path = generate_temp_path()
+    
+    try:
+        subprocess.run([
+            "ffmpeg", "-i", video_path,
+            "-vf", f"crop={width}:{height}:{x}:{y}",
+            "-c:a", "copy", output_path
+        ], check=True, capture_output=True)
+        
+        # Save to static folder and get URL
+        result_url = save_video_to_static_folder(output_path)
+        
+        # Clean up temp file (only the output_path, not the original video)
+        os.remove(output_path)
+            
+        return result_url
+    except Exception as e:
+        print(f"Error cropping video: {e}")
+        return None
+
+
+
+
+def resize_video(
+    video_url: str, 
+    width: int, 
+    height: int
+) -> str:
+    """
+    Resize a video to specified dimensions
+    
+    Args:
+        video_url: URL to input video
+        width: Width in pixels
+        height: Height in pixels
+        
+    Returns:
+        URL to output video
+    """
+    # Download the video from URL
+    video_path = download_video_from_url(video_url)
+    if not video_path:
+        return None
+    
+    output_path = generate_temp_path()
+    
+    try:
+        subprocess.run([
+            "ffmpeg", "-i", video_path,
+            "-vf", f"scale={width}:{height}",
+            "-c:a", "copy", output_path
+        ], check=True, capture_output=True)
+        
+        # Save to static folder and get URL
+        result_url = save_video_to_static_folder(output_path)
+        
+        # Clean up temp file (only the output_path, not the original video)
+        os.remove(output_path)
+            
+        return result_url
+    
+    except Exception as e:
+        print(f"Error resizing video: {e}")
+        return None
+
+
+def generate_temp_path(extension="mp4"):
+    """Generate a temporary file path"""
+    temp_dir = tempfile.gettempdir()
+    return os.path.join(temp_dir, f"output_{uuid.uuid4()}.{extension}")

test.py

@@ -0,0 +1,6 @@
+def test():
+    """
+    This function is a placeholder for testing purposes.
+    It currently does nothing but can be expanded with test cases.
+    """
+    pass

tools/audio_tools.py

@@ -0,0 +1,190 @@
+import os
+import tempfile
+import subprocess
+import uuid
+import requests
+from typing import List
+from videos_management import save_video_to_static_folder, download_video_from_url
+
+
+
+def generate_temp_path(extension="mp4"):
+    """Generate a temporary file path"""
+    temp_dir = tempfile.gettempdir()
+    return os.path.join(temp_dir, f"output_{uuid.uuid4()}.{extension}")
+
+
+
+def search_audio_from_envato(
+    query: str,
+    limit: int = 10,
+    category: str = None,
+    length_min: int = None,
+    length_max: int = None,
+    tags: str = None,
+    free_only: bool = False
+) -> List[dict]:
+    """
+    This function is used to search for royalty-free sound effects using Envato API.
+    It returns a list of sound effects with metadata and download URLs or preview URLs.
+    
+    Args:
+        query (str): Search query
+        limit (int): Maximum number of results to return
+        category (str, optional): Category to filter by (e.g., "music", "effects")
+        length_min (int, optional): Minimum audio length in seconds
+        length_max (int, optional): Maximum audio length in seconds
+        tags (str, optional): Comma-separated list of tags to match
+        free_only (bool, optional): If True, only return free items
+        
+    Returns:
+        List[dict]: List of sound effects with metadata and download URLs
+    """
+    api_key = os.getenv("ENVATO_API_KEY")
+    if not api_key:
+        print("ENVATO_API_KEY not found in environment variables")
+        return []
+    
+    try:
+        # Envato API endpoint for AudioJungle
+        url = "https://api.envato.com/v1/discovery/search/search/item"
+        
+        # Prepare parameters
+        params = {
+            "site": "audiojungle.net",
+            "term": query,
+            "page_size": limit
+        }
+        
+        if category:
+            params["category"] = category
+        
+        if length_min is not None:
+            params["length_min"] = length_min
+        
+        if length_max is not None:
+            params["length_max"] = length_max
+        
+        if tags:
+            params["tags"] = tags
+        
+        headers = {"Authorization": f"Bearer {api_key}"}
+        response = requests.get(url, headers=headers, params=params)
+        
+        if response.status_code != 200:
+            print(f"API request failed with status code {response.status_code}: {response.text}")
+            return []
+        
+        # Parse response
+        data = response.json()
+        results = []
+        
+        for item in data.get("matches", []):
+            if free_only and item.get("price_cents", 0) > 0:
+                continue
+                
+            preview_url = None
+            if "previews" in item and "icon_with_audio_preview" in item["previews"]:
+                preview_url = item["previews"]["icon_with_audio_preview"].get("mp3_url")
+            
+            if not preview_url:
+                continue
+                
+            length_info = item.get("previews", {}).get("icon_with_audio_preview", {}).get("length", {})
+            length_seconds = (length_info.get("hours", 0) * 3600 + 
+                              length_info.get("minutes", 0) * 60 + 
+                              length_info.get("seconds", 0))
+            
+            
+            results.append({
+                "id": item.get("id"),
+                "name": item.get("name"),
+                "description": item.get("description"),
+                "preview_url": preview_url,
+                "thumbnail_url": item.get("previews", {}).get("icon_with_audio_preview", {}).get("icon_url"),
+                "length_seconds": length_seconds,
+           
+            })
+        print(results)
+        return results
+    except Exception as e:
+        print(f"Error searching sound effects: {str(e)}")
+        return []
+
+
+def add_audio_to_video(
+    video_url: str,
+    audio_path: str,
+    start_time: float = 0,
+    volume: float = 1.0,
+    keep_original_audio: bool = True
+) -> str:
+    """
+    Add audio to a video
+    
+    Args:
+        video_url: URL to input video
+        audio_path: Path or URL to audio file
+        start_time: Time to start adding audio (seconds)
+        volume: Volume adjustment (1.0 is normal)
+        keep_original_audio: Whether to keep original audio
+        
+    Returns:
+        URL to output video
+    """
+    # Download the video from URL
+    video_path = download_video_from_url(video_url)
+    if not video_path:
+        return None
+    
+    # If audio_path is a URL, download it first
+    temp_audio_file = None
+    try:
+        if audio_path.startswith('http://') or audio_path.startswith('https://'):
+            # Download the audio from URL
+            response = requests.get(audio_path, stream=True)
+            if response.status_code != 200:
+                print(f"Failed to download audio: HTTP {response.status_code}")
+                return None
+                
+            temp_audio_file = generate_temp_path(extension="mp3")
+            with open(temp_audio_file, 'wb') as f:
+                for chunk in response.iter_content(chunk_size=8192):
+                    f.write(chunk)
+            
+            audio_path = temp_audio_file
+        
+        output_path = generate_temp_path()
+        
+        if keep_original_audio:
+            # Mix original audio with new audio
+            subprocess.run([
+                "ffmpeg", "-y", "-i", video_path, "-i", audio_path,
+                "-filter_complex", f"[1:a]adelay={int(start_time*1000)}|{int(start_time*1000)},volume={volume}[added];[0:a][added]amix=duration=first[a]",
+                "-map", "0:v", "-map", "[a]", "-c:v", "copy", output_path
+            ], check=True, capture_output=True)
+        else:
+            # Replace original audio with new audio
+            subprocess.run([
+                "ffmpeg", "-y", "-i", video_path, "-i", audio_path,
+                "-filter_complex", f"[1:a]adelay={int(start_time*1000)}|{int(start_time*1000)},volume={volume}[a]",
+                "-map", "0:v", "-map", "[a]", "-c:v", "copy", output_path
+            ], check=True, capture_output=True)
+        
+        # Save to static folder and get URL
+        result_url = save_video_to_static_folder(output_path)
+        
+        # Clean up temp files
+        if os.path.exists(output_path):
+            os.remove(output_path)
+        if temp_audio_file and os.path.exists(temp_audio_file):
+            os.remove(temp_audio_file)
+            
+        return result_url
+    
+    except Exception as e:
+        print(f"Error adding audio to video: {e}")
+        # Clean up temp files
+        if temp_audio_file and os.path.exists(temp_audio_file):
+            os.remove(temp_audio_file)
+        return None

tools/gemini_tool.py

@@ -0,0 +1,104 @@
+# To run this code you need to install the following dependencies:
+# pip install google-genai
+
+import base64
+import os
+import tempfile
+import mimetypes
+from google import genai
+from google.genai import types
+from videos_management import download_video_from_url
+import time
+
+def analyze_video_context(video_url: str, task: str):
+    """
+    This is an AI modal that can analyze the entire video along with video. You can use this to analyze the context of a video.
+    Args:
+        video_url (str): The URL to the video file.
+        task (str): The task to analyze the video for.
+    Returns:
+        str: The response from the GEMINI AI MODEL.
+    """
+    
+    video_path = download_video_from_url(video_url)
+    if not video_path:
+        return "Error: Could not download video from URL"
+
+    try:
+       
+        client = genai.Client(
+            api_key=os.environ.get("GEMINI_API_KEY"),
+        )
+
+        instructions = """You are a video editing assistant that processes long videos and helps identify key moments for creating shorts. Analyze the video and provide clear, concise answers to timestamp-related questions. For example, if asked 'At what time does the hero die?' or 'What happens at [timestamp]?', give the exact time and a brief description of the scene. Use your analysis to pinpoint important segments to assist with video editing."""
+        
+        mime_type, _ = mimetypes.guess_type(video_path)
+        if not mime_type or not mime_type.startswith('video/'):
+            mime_type = "video/mp4"  # Default to mp4 if we can't determine or it's not a video
+        
+        with open(video_path, "rb") as f:
+            video_bytes = f.read()
+        
+        file_size = os.path.getsize(video_path)
+        if file_size > 20 * 1024 * 1024:  # if larger than 20MB
+            with open(video_path, "rb") as f:
+                # Explicitly specify MIME type when uploading
+                video_file = client.files.upload(
+                    file=f, 
+                    config=types.UploadFileConfigDict(mime_type=mime_type)
+                )
+                
+                # Wait for processing to complete
+                max_retries = 12  # 1 minute max wait (5 sec * 12)
+                retry_count = 0
+                while video_file.state == "PROCESSING" and retry_count < max_retries:
+                    print(f'Waiting for video to be processed. Attempt {retry_count+1}/{max_retries}')
+                    time.sleep(5)
+                    video_file = client.files.get(name=video_file.name)
+                    retry_count += 1
+                
+                if video_file.state != "SUCCEEDED":
+                    return f"Error: Video processing failed or timed out. Current state: {video_file.state}"
+            
+            response = client.models.generate_content(
+                model="gemini-2.5-pro-preview-05-06",
+                contents=[
+                    video_file,
+                    types.Part.from_text(text=instructions),
+                    types.VideoMetadata(
+                fps=5,
+         
+               )
+                ],
+                config=types.GenerateContentConfig(
+                    system_instruction=instructions,
+                    response_mime_type="text/plain",
+                
+                )
+            )
+        else:
+            response = client.models.generate_content(
+                model="gemini-2.0-flash",
+                contents=types.Content(
+                    parts=[
+                    types.Part(
+                        inline_data=types.Blob(
+                            data=video_bytes,
+                            mime_type=mime_type
+                        ),
+                        video_metadata=types.VideoMetadata(fps=5)
+                    ),
+                    types.Part(text=instructions)   
+                ]
+                ),
+                config=types.GenerateContentConfig(
+                    system_instruction=task,
+                    response_mime_type="text/plain",
+                )
+            )
+            
+        return response.text
+    except Exception as e:
+        return f"Error analyzing video: {str(e)}"
+
+

tools/subtitle_tools.py

@@ -0,0 +1,304 @@
+import httpx
+import os
+import tempfile
+import subprocess
+import json
+from typing import List, Dict, Any, Optional
+from videos_management import download_video_from_url, save_video_to_static_folder
+import uuid
+import shutil
+import requests
+from functools import lru_cache
+from dotenv import load_dotenv
+import gradio as gr
+import ffmpeg   
+import shlex
+
+# Load environment variables
+load_dotenv()
+
+@lru_cache(maxsize=100)
+def fetch_word_level_transcription(video_url: str, api_url: str) -> Dict[str, Any]:
+    """
+    Fetch word-level transcription from the API.
+    
+    Args:
+        video_url (str): URL of the video to transcribe
+        api_url (str, optional): URL of the transcription API. If None, uses environment variable
+        
+    Returns:
+        Dict[str, Any]: Dictionary containing transcription data with 'words' key, or None if failed
+        
+    Raises:
+        Exception: If API request fails or returns invalid response
+    """
+    try:
+        if not api_url:
+            api_url = os.getenv("TRANSCRIPTION_API_URL", "https://malikibrarbhutta--whisperx-api-transcribe-endpoint.modal.run/")
+        
+        print(f"Sending request to: {api_url}")
+        
+        # Use httpx for better timeout handling
+        with httpx.Client(timeout=120.0) as client:  # 2 minute timeout
+            response = client.post(
+                api_url,
+                data={"url": video_url},
+                headers={
+                    "accept": "application/json",
+                    "Content-Type": "application/x-www-form-urlencoded"
+                }
+            )
+                        
+        print(response.text)
+        if response.status_code != 200:
+            print(f"API error: {response.status_code} - {response.text}")
+            return None
+            
+        return response.json()
+    except Exception as e:
+        print(f"Error fetching transcription: {e}")
+        return None
+
+def create_simple_srt_file(transcription: dict) -> str:
+    """
+    Create a simple SRT subtitle file for word-level subtitles.
+    
+    Args:
+        transcription (dict): Dictionary containing transcription data with 'words' key
+        
+    Returns:
+        str: Path to the created SRT file, or None if creation failed
+        
+    Raises:
+        Exception: If file creation fails or transcription data is invalid
+    """
+    words = transcription.get("words", [])
+    if not words:
+        return None
+
+    # Create temporary SRT file
+    temp_dir = tempfile.gettempdir()
+    srt_file = os.path.join(temp_dir, f"subtitles_{uuid.uuid4()}.srt")
+    
+    def format_time(seconds):
+        """Convert seconds to SRT time format (HH:MM:SS,mmm)"""
+        hours = int(seconds // 3600)
+        minutes = int((seconds % 3600) // 60)
+        secs = int(seconds % 60)
+        millis = int((seconds % 1) * 1000)
+        return f"{hours:02d}:{minutes:02d}:{secs:02d},{millis:03d}"
+    
+    with open(srt_file, "w", encoding="utf-8") as f:
+        for i, word in enumerate(words, 1):
+            word_text = word.get("word", "").strip()
+            if not word_text:
+                continue
+                
+            start_time = format_time(word["start"])
+            end_time = format_time(word["end"])
+            
+            f.write(f"{i}\n")
+            f.write(f"{start_time} --> {end_time}\n")
+            f.write(f"{word_text}\n\n")
+    
+    return srt_file
+
+def get_position_filter(position: str) -> str:
+    """
+    Convert position string to FFmpeg drawtext filter parameters with better spacing.
+    
+    Args:
+        position (str): Position identifier ('bottom-center', 'top-center', 'middle', 'bottom-left', 'bottom-right')
+        
+    Returns:
+        str: FFmpeg position parameters string
+    """
+    # Adjusted positions with better spacing from edges
+    positions = {
+        "bottom-center": "x=(w-text_w)/2:y=h-text_h-80",  # More space from bottom
+        "top-center": "x=(w-text_w)/2:y=80",              # More space from top
+        "middle": "x=(w-text_w)/2:y=(h-text_h)/2",
+        "bottom-left": "x=80:y=h-text_h-80",              # More space from edges
+        "bottom-right": "x=w-text_w-80:y=h-text_h-80"     # More space from edges
+    }
+    return positions.get(position, positions["bottom-center"])
+
+
+def add_simple_subtitles_to_video(
+    video_path: str,
+    srt_file: str,
+    output_path: str,
+    font_size: int = 36,
+    font_color: str = "white",
+    position: str = "bottom-center"
+) -> bool:
+    try:
+        # Map friendly names to ASS format values
+        alignment_map = {
+            "bottom-center": 2,
+            "top-center": 8,
+            "center": 5
+        }
+        alignment = alignment_map.get(position, 2)
+
+        color_map = {
+            "white": "&HFFFFFF&",
+            "yellow": "&H00FFFF&",
+            "red": "&H0000FF&",
+            "green": "&H00FF00&",
+            "blue": "&HFF0000&",
+            "black": "&H000000&"
+        }
+        color = color_map.get(font_color.lower(), "&HFFFFFF&")
+
+        # Safely quote file paths
+        quoted_srt = shlex.quote(srt_file)
+
+        # Subtitle filter with style
+        subtitle_filter = (
+            f"subtitles={quoted_srt}:"
+            f"force_style='Fontsize={font_size},PrimaryColour={color},"
+            f"OutlineColour=&H000000&,Alignment={alignment},MarginV=20'"
+        )
+
+        # Run ffmpeg command
+        (
+            ffmpeg
+            .input(video_path)
+            .output(output_path, vf=subtitle_filter, vcodec='libx264', acodec='copy')
+            .overwrite_output()
+            .run()
+        )
+
+        print("✅ Subtitles added successfully.")
+        return True
+
+    except ffmpeg.Error as e:
+        print("❌ FFmpeg error:")
+        print(e.stderr.decode() if e.stderr else str(e))
+        return False
+    
+
+def get_color_hex(color: str) -> str:
+    """
+    Convert color name or hex to FFmpeg hex format.
+    
+    Args:
+        color (str): Color name (e.g., 'white', 'red') or hex color (e.g., '#FFFFFF')
+        
+    Returns:
+        str: Hex color code in uppercase format without '#' prefix
+    """
+    color_map = {
+        "white": "FFFFFF",
+        "black": "000000",
+        "red": "FF0000",
+        "green": "00FF00",
+        "blue": "0000FF",
+        "yellow": "FFFF00",
+        "cyan": "00FFFF",
+        "magenta": "FF00FF"
+    }
+    
+    if color.startswith("#"):
+        return color[1:].upper()
+    
+    return color_map.get(color.lower(), "FFFFFF")
+
+def generate_word_level_subtitles_from_url(
+    video_url: str,
+    transcription_api_url: str = ""
+) -> str:
+    """
+    Generate word-level subtitles for a video from URL (MCP Server Function).
+    
+    This function downloads a video, fetches word-level transcription, creates subtitles
+    where each word appears individually at the correct timing, and returns the final video URL.
+    
+    Args:
+        video_url (str): URL of the video to process. Must be a valid, accessible video URL
+        transcription_api_url (str, optional): URL of the transcription API. If empty, uses default from environment
+        
+    Returns:
+        str: URL to the output video with word-level subtitles, or error message if processing failed
+        
+    Raises:
+        Exception: Various exceptions may occur during video download, transcription, or processing
+    """
+    try:
+        
+        if not video_url or not video_url.strip():
+            return "Error: Video URL is required"
+        
+        video_path = download_video_from_url(video_url)
+        if not video_path:
+            return "Error: Failed to download video. Please check the URL and try again."
+        
+        
+        # Fetch transcription
+        print("Fetching transcription...")
+        api_url = transcription_api_url.strip() if transcription_api_url.strip() else None
+        transcription = fetch_word_level_transcription(video_url, api_url)
+        
+        if not transcription:
+            return "Error: Failed to get transcription from API. Please check your API URL or try again later."
+        
+        words_count = len(transcription.get('words', []))
+        print(f"Transcription received with {words_count} words")
+        
+        if words_count == 0:
+            return "Error: No words found in transcription. The video might be too short or contain no speech."
+        
+        # Create SRT file
+        print("Creating subtitle file...")
+        srt_file = create_simple_srt_file(transcription)
+        
+        if not srt_file:
+            return "Error: Failed to create subtitle file from transcription data."
+        
+        print(f"Subtitle file created: {srt_file}")
+        
+        # Create output path
+        temp_dir = tempfile.mkdtemp()
+        output_path = os.path.join(temp_dir, "subtitled_video.mp4")
+        
+        # Add subtitles to video with improved positioning
+        print("Adding subtitles to video...")
+        success = add_simple_subtitles_to_video(
+            video_path=video_path,
+            srt_file=srt_file,
+            output_path=output_path,
+            font_size=38,  # Slightly larger for better readability
+            font_color="white",
+            position="bottom-center"
+        )
+        
+        if not success:
+            return "Error: Failed to add subtitles to video. FFmpeg processing failed."
+        
+        print("Subtitles added successfully")
+        
+        # Save to static folder
+        print("Saving final video...")
+        result_url = save_video_to_static_folder(output_path)
+        
+        if result_url:
+            print(f"Video saved successfully: {result_url}")
+            return f"Success! Video with word-level subtitles: {result_url}"
+        else:
+            return "Error: Failed to save final video to static folder."
+            
+    except Exception as e:
+        error_msg = f"Error: {str(e)}"
+        print(error_msg)
+        return error_msg
+    
+    finally:
+        # Clean up temporary files
+        try:
+            if 'srt_file' in locals() and os.path.exists(srt_file):
+                os.remove(srt_file)
+            if 'temp_dir' in locals():
+                shutil.rmtree(temp_dir, ignore_errors=True)
+        except:
+            pass

tools/video_tools.py

@@ -0,0 +1,783 @@
+import os
+import tempfile
+import subprocess
+import uuid
+import requests
+import json
+import dotenv
+import shutil
+from typing import Optional, Tuple, Union, List
+from pathlib import Path
+from videos_management import save_video_to_static_folder, download_video_from_url
+from helpers.video_tool_helper import generate_temp_path, get_duration
+dotenv.load_dotenv()
+
+
+def edit_video_segment(
+    video_url: str, 
+    start_time: float, 
+    end_time: float,
+    action: str = "cut"
+) -> str:
+    """
+    Edit a video segment - can cut out a segment or keep only that segment
+    
+    Args:
+        video_url: URL to input video
+        start_time: Start time in seconds
+        end_time: End time in seconds
+        action: Either "cut" (remove segment) or "keep" (keep only segment)
+        
+    Returns:
+        URL to output video
+    """
+    # Download the video from URL
+    video_path = download_video_from_url(video_url)
+    if not video_path:
+        return None
+    
+    output_path = generate_temp_path()
+    
+    try:
+        if action == "keep":
+            # Keep only the specified segment
+            subprocess.run([
+                "ffmpeg", "-i", video_path,
+                "-ss", str(start_time), "-to", str(end_time),
+                "-c", "copy", output_path
+            ], check=True, capture_output=True)
+            
+            # Save to static folder and get URL
+            result_url = save_video_to_static_folder(output_path)
+            
+            # Clean up temp files (only the output_path, not the original video)
+            os.remove(output_path)
+                
+            return result_url
+            
+        elif action == "cut":
+            # Get video duration
+            duration_cmd = [
+                "ffprobe", "-v", "error", "-show_entries", "format=duration",
+                "-of", "default=noprint_wrappers=1:nokey=1", video_path
+            ]
+            duration = float(subprocess.check_output(duration_cmd).decode().strip())
+            
+            # Create temp directory for segments
+            temp_dir = tempfile.mkdtemp()
+            segments = []
+            
+            # First segment (before cut)
+            if start_time > 0:
+                segment1 = os.path.join(temp_dir, "segment1.mp4")
+                subprocess.run([
+                    "ffmpeg", "-i", video_path,
+                    "-ss", "0", "-to", str(start_time),
+                    "-c", "copy", segment1
+                ], check=True, capture_output=True)
+                segments.append(segment1)
+            
+            # Second segment (after cut)
+            if end_time < duration:
+                segment2 = os.path.join(temp_dir, "segment2.mp4")
+                subprocess.run([
+                    "ffmpeg", "-i", video_path,
+                    "-ss", str(end_time), "-to", str(duration),
+                    "-c", "copy", segment2
+                ], check=True, capture_output=True)
+                segments.append(segment2)
+            
+            # If no segments to keep
+            if not segments:
+                # Clean up
+                shutil.rmtree(temp_dir)
+                return None
+                
+            # If only one segment
+            if len(segments) == 1:
+                shutil.copy(segments[0], output_path)
+                
+                # Save to static folder and get URL
+                result_url = save_video_to_static_folder(output_path)
+                
+                # Clean up
+                shutil.rmtree(temp_dir)
+                os.remove(output_path)
+                
+                return result_url
+            
+            # Concatenate segments
+            concat_file = os.path.join(temp_dir, "concat.txt")
+            with open(concat_file, 'w') as f:
+                for segment in segments:
+                    f.write(f"file '{segment}'\n")
+            
+            subprocess.run([
+                "ffmpeg", "-f", "concat", "-safe", "0",
+                "-i", concat_file, "-c", "copy", output_path
+            ], check=True, capture_output=True)
+            
+            # Save to static folder and get URL
+            result_url = save_video_to_static_folder(output_path)
+            
+            # Clean up
+            shutil.rmtree(temp_dir)
+            os.remove(output_path)
+            
+            return result_url
+        else:
+            print(f"Unknown action: {action}")
+            return None
+    
+    except Exception as e:
+        print(f"Error editing video segment: {e}")
+        return None
+
+def add_text_overlay(
+    video_url: str, 
+    text_elements: list[dict]
+) -> str:
+    """
+    Add one or more text overlays to a video at specified times and positions.
+    
+    Args:
+        video_url: URL to the input video file.
+        text_elements: A list of dictionaries, where each dictionary defines a text overlay.
+            Required keys for each text_element dict:
+            - 'text': str - The text to display.
+            - 'start_time': str or float - Start time (seconds).
+            - 'end_time': str or float - End time (seconds).
+            Optional keys for each text_element dict:
+            - 'font_size': int (default: 24)
+            - 'font_color': str (default: 'white')
+            - 'x_pos': str or int (default: 'center')
+            - 'y_pos': str or int (default: 'h-th-10')
+            - 'box': bool (default: False)
+            - 'box_color': str (default: '[email protected]')
+            - 'box_border_width': int (default: 0)
+            
+    Returns:
+        URL to output video
+    """
+    # Download the video from URL
+    video_path = download_video_from_url(video_url)
+    if not video_path:
+        return None
+    
+    if not text_elements:
+        return video_url  # Return original URL if no text elements
+    
+    output_path = generate_temp_path()
+    
+    try:
+        # Build filter string for each text element
+        filter_parts = []
+        
+        for element in text_elements:
+            text = element.get('text')
+            start_time = element.get('start_time')
+            end_time = element.get('end_time')
+            
+            if text is None or start_time is None or end_time is None:
+                print("Text element is missing required keys (text, start_time, end_time)")
+                continue
+            
+            # Escape special characters in text
+            safe_text = text.replace('\\', '\\\\').replace("'", "\\'").replace(':', '\\:').replace(',', '\\,')
+            
+            # Build filter parameters
+            params = [
+                f"text='{safe_text}'",
+                f"fontsize={element.get('font_size', 24)}",
+                f"fontcolor={element.get('font_color', 'white')}"
+            ]
+            
+            # Handle position
+            x_pos = element.get('x_pos', 'center')
+            if x_pos == 'center':
+                x_pos = '(w-text_w)/2'
+            params.append(f"x={x_pos}")
+            
+            y_pos = element.get('y_pos', 'h-th-10')
+            if y_pos == 'bottom':
+                y_pos = 'h-text_h-10'
+            elif y_pos == 'top':
+                y_pos = '10'
+            params.append(f"y={y_pos}")
+            
+            # Enable text during specified time range
+            params.append(f"enable=between(t\\,{start_time}\\,{end_time})")
+            
+            # Add box if specified
+            if element.get('box', False):
+                params.append("box=1")
+                params.append(f"boxcolor={element.get('box_color', '[email protected]')}")
+                if 'box_border_width' in element:
+                    params.append(f"boxborderw={element['box_border_width']}")
+            
+            # Add font file if specified
+            if 'font_file' in element:
+                font_path = element['font_file'].replace('\\', '\\\\').replace("'", "\\'").replace(':', '\\:')
+                params.append(f"fontfile='{font_path}'")
+            
+            # Join parameters
+            filter_parts.append(f"drawtext={':'.join(params)}")
+        
+        # Join all filter parts
+        filter_string = ','.join(filter_parts)
+        
+        # Apply filters
+        subprocess.run([
+            "ffmpeg", "-i", video_path,
+            "-vf", filter_string,
+            "-c:a", "copy",
+            output_path
+        ], check=True, capture_output=True)
+        
+        # Save to static folder and get URL
+        result_url = save_video_to_static_folder(output_path)
+        
+        # Clean up temp file
+        os.remove(output_path)
+            
+        return result_url
+    
+    except Exception as e:
+        print(f"Error adding text overlay: {e}")
+        return None
+
+def change_speed(
+    video_url: str, 
+    speed_factor: float
+) -> str:
+    """
+    Change video playback speed
+    
+    Args:
+        video_url: URL to input video
+        speed_factor: Speed factor (1.0 is normal, <1 is slower, >1 is faster)
+        
+    Returns:
+        URL to output video
+    """
+    video_path = download_video_from_url(video_url)
+    if not video_path:
+        return None
+    
+    output_path = generate_temp_path()
+    
+    try:
+        # For video: setpts=1/speed_factor*PTS
+        # For audio: atempo=speed_factor (limited to 0.5-2.0 range)
+        
+        # Handle audio speed limits
+        audio_filter = ""
+        remaining_factor = speed_factor
+        
+        while remaining_factor > 2.0:
+            audio_filter += "atempo=2.0,"
+            remaining_factor /= 2.0
+        
+        while remaining_factor < 0.5:
+            audio_filter += "atempo=0.5,"
+            remaining_factor *= 2.0
+        
+        audio_filter += f"atempo={remaining_factor}"
+        
+        subprocess.run([
+            "ffmpeg", "-i", video_path,
+            "-filter_complex", f"[0:v]setpts={1/speed_factor}*PTS[v];[0:a]{audio_filter}[a]",
+            "-map", "[v]", "-map", "[a]", output_path
+        ], check=True, capture_output=True)
+        
+        # Save to static folder and get URL
+        result_url = save_video_to_static_folder(output_path)
+        
+        # Clean up temp file (only the output_path, not the original video)
+        os.remove(output_path)
+            
+        return result_url
+    
+    except Exception as e:
+        print(f"Error changing video speed: {e}")
+        return None
+
+def extract_audio(
+    video_url: str, 
+    audio_format: str = "mp3"
+) -> str:
+    """
+    Extract audio from a video
+    
+    Args:
+        video_url: URL to input video
+        audio_format: Output audio format (mp3, aac, wav, etc.)
+        
+    Returns:
+        URL to output audio
+    """
+    print("extension", audio_format)
+    video_path = download_video_from_url(video_url)
+    if not video_path:
+        return None
+    
+    output_path = generate_temp_path(extension=audio_format)
+    
+    try:
+        # Map audio codecs to formats
+        codec_map = {
+            "mp3": "libmp3lame",
+            "aac": "aac",
+            "wav": "pcm_s16le",
+            "ogg": "libvorbis",
+            "flac": "flac"
+        }
+        
+        codec = codec_map.get(audio_format, audio_format)
+        
+        subprocess.run([
+            "ffmpeg", "-i", video_path,
+            "-vn", "-acodec", codec, output_path
+        ], check=True, capture_output=True)
+        
+        result_url = save_video_to_static_folder(video_path=output_path, extension=audio_format)
+        
+        os.remove(output_path)
+            
+        return result_url
+    
+    except Exception as e:
+        print(f"Error extracting audio: {e}")
+        return None
+
+
+def add_image_overlay(
+    video_url: str,
+    image_path: str,
+    position: str = 'top_right',
+    opacity: float = 1.0,
+    start_time: float = None,
+    end_time: float = None,
+    width: str = None,
+    height: str = None
+) -> str:
+    """
+    Add an image overlay (watermark/logo) to a video.
+    
+    Args:
+        video_url: URL to the input video file.
+        image_path: Path to the image file for the overlay.
+        position: Position of the overlay. 
+            Options: 'top_left', 'top_right', 'bottom_left', 'bottom_right', 'center'.
+            Or specify custom coordinates like 'x=10:y=10'.
+        opacity: Opacity of the overlay (0.0 to 1.0).
+        start_time: Start time for the overlay in seconds. If None, starts from beginning.
+        end_time: End time for the overlay in seconds. If None, lasts till end.
+        width: Width for the overlay image (e.g., '100', 'iw*0.1'). Original if None.
+        height: Height for the overlay image (e.g., '50', 'ih*0.1'). Original if None.
+        
+    Returns:
+        URL to output video
+    """
+    # Download the video from URL
+    video_path = download_video_from_url(video_url)
+    if not video_path:
+        return None
+    
+    if not os.path.exists(image_path):
+        print(f"Image file not found: {image_path}")
+        return None
+    
+    output_path = generate_temp_path()
+    
+    try:
+        # Process the image (scale, set opacity)
+        image_filters = []
+        
+        # Apply scaling if requested
+        if width or height:
+            scale_params = []
+            if width:
+                scale_params.append(f"width={width}")
+            else:
+                scale_params.append("width=-1")  # Auto-width maintaining aspect
+                
+            if height:
+                scale_params.append(f"height={height}")
+            else:
+                scale_params.append("height=-1")  # Auto-height maintaining aspect
+                
+            image_filters.append(f"scale={':'.join(scale_params)}")
+        
+        # Apply opacity if requested
+        if opacity is not None and opacity < 1.0:
+            image_filters.append("format=rgba")  # Ensure alpha channel exists
+            image_filters.append(f"colorchannelmixer=aa={opacity}")
+        
+        # Determine overlay position coordinates
+        overlay_x = '0'
+        overlay_y = '0'
+        
+        if position == 'top_left':
+            overlay_x, overlay_y = '10', '10'
+        elif position == 'top_right':
+            overlay_x, overlay_y = 'main_w-overlay_w-10', '10'
+        elif position == 'bottom_left':
+            overlay_x, overlay_y = '10', 'main_h-overlay_h-10'
+        elif position == 'bottom_right':
+            overlay_x, overlay_y = 'main_w-overlay_w-10', 'main_h-overlay_h-10'
+        elif position == 'center':
+            overlay_x, overlay_y = '(main_w-overlay_w)/2', '(main_h-overlay_h)/2'
+        elif ':' in position:
+            pos_parts = position.split(':')
+            for part in pos_parts:
+                if part.startswith('x='):
+                    overlay_x = part.split('=')[1]
+                if part.startswith('y='):
+                    overlay_y = part.split('=')[1]
+        
+        # Create a temporary file for the processed image
+        temp_image = generate_temp_path(extension="png")
+        
+        # Process the image if needed
+        if image_filters:
+            image_filter_string = ','.join(image_filters)
+            subprocess.run([
+                "ffmpeg", "-i", image_path,
+                "-vf", image_filter_string,
+                temp_image
+            ], check=True, capture_output=True)
+        else:
+            # Just copy the image
+            shutil.copy(image_path, temp_image)
+        
+        # Prepare overlay filter parameters
+        overlay_params = [f"x={overlay_x}", f"y={overlay_y}"]
+        
+        # Add time-based enabling condition if specified
+        if start_time is not None or end_time is not None:
+            enable_expr = []
+            if start_time is not None:
+                enable_expr.append(f"gte(t,{start_time})")
+            if end_time is not None:
+                enable_expr.append(f"lte(t,{end_time})")
+            
+            if enable_expr:
+                overlay_params.append(f"enable={'*'.join(enable_expr)}")
+        
+        # Build the filter complex
+        filter_complex = f"[0:v][1:v]overlay={':'.join(overlay_params)}[out]"
+        
+        # Apply the overlay
+        subprocess.run([
+            "ffmpeg", "-i", video_path, "-i", temp_image,
+            "-filter_complex", filter_complex,
+            "-map", "[out]", "-map", "0:a?",
+            "-c:a", "copy",
+            output_path
+        ], check=True, capture_output=True)
+        
+        # Clean up temp image
+        os.remove(temp_image)
+        
+        # Save to static folder and get URL
+        result_url = save_video_to_static_folder(output_path)
+        
+        # Clean up temp file
+        os.remove(output_path)
+            
+        return result_url
+    
+    except Exception as e:
+        print(f"Error adding image overlay: {e}")
+        return None
+
+def apply_color_filter(
+    video_url: str,
+    filter_type: str,
+    intensity: float = 1.0
+) -> str:
+    """
+    Apply a color filter/effect to a video.
+    
+    Args:
+        video_url (str): URL to the input video file
+        filter_type (str): Type of filter ('sepia', 'grayscale', 'warm', 'cool', 'vintage')
+        intensity (float): Intensity of the effect (0-1)
+        
+    Returns:
+        str: URL to the output video file
+    """
+    # Download the video from URL
+    video_path = download_video_from_url(video_url)
+    if not video_path:
+        return None
+    
+    output_path = generate_temp_path()
+    
+    try:
+        # Define filter expressions based on filter type
+        if filter_type == "sepia":
+            filter_expr = (
+                f"colorchannelmixer="
+                f"rr={1-0.393*intensity}:rg={0.769*intensity}:rb={0.189*intensity}:"
+                f"gr={0.349*intensity}:gg={1-0.686*intensity}:gb={0.168*intensity}:"
+                f"br={0.272*intensity}:bg={0.534*intensity}:bb={1-0.131*intensity}"
+            )
+        elif filter_type == "grayscale":
+            filter_expr = f"colorchannelmixer=.3:.4:.3:0:.3:.4:.3:0:.3:.4:.3"
+        elif filter_type == "warm":
+            filter_expr = f"colortemperature=temperature={6500+1500*intensity}"
+        elif filter_type == "cool":
+            filter_expr = f"colortemperature=temperature={6500-1500*intensity}"
+        elif filter_type == "vintage":
+            # Vintage look: slight sepia tone + vignette + grain
+            filter_expr = (
+                f"colorchannelmixer="
+                f"rr=.93:rg=.01:rb=.07:"
+                f"gr=.22:gg=.85:gb=.05:"
+                f"br=.05:bg=.20:bb=.90,"
+                f"vignette=angle=PI/4:x0=(W/2):y0=(H/2)"
+            )
+        else:
+            print(f"Unsupported filter type: {filter_type}")
+            return None
+        
+        subprocess.run([
+            "ffmpeg", "-i", video_path,
+            "-vf", filter_expr,
+            "-c:a", "copy", output_path
+        ], check=True, capture_output=True)
+        
+        # Save to static folder and get URL
+        result_url = save_video_to_static_folder(output_path)
+        
+        # Clean up temp file (only the output_path, not the original video)
+        os.remove(output_path)
+            
+        return result_url
+    except Exception as e:
+        print(f"Error applying color filter: {e}")
+        return None
+
+def merge_videos(
+    video_urls: List[str]
+) -> str:
+    """
+    Merge multiple videos without transitions
+    
+    Args:
+        video_urls: List of video URLs to merge
+        
+    Returns:
+        URL to the merged video
+    """
+    # Validate input
+    if not video_urls:
+        print("No video URLs provided")
+        return None
+    
+    if not isinstance(video_urls, list):
+        print(f"Expected list of video URLs, got {type(video_urls)}")
+        # Try to handle single video case
+        if isinstance(video_urls, str):
+            video_urls = [video_urls]
+        else:
+            return None
+    
+    # Download all videos
+    video_paths = []
+    for url in video_urls:
+        if url is None:
+            continue
+            
+        print(f"Processing video: {url}")
+        path = download_video_from_url(url)
+        if path:
+            video_paths.append(path)
+        else:
+            print(f"Failed to download video: {url}")
+            # Continue with other videos instead of failing completely
+    
+    # Check if we have any valid videos
+    if not video_paths:
+        print("No valid videos to merge")
+        return None
+    
+    if len(video_paths) == 1:
+        # Just return a copy of the single video
+        result_url = save_video_to_static_folder(video_paths[0])
+        return result_url
+    
+    # Create temp files
+    temp_dir = tempfile.mkdtemp()
+    output_path = os.path.join(temp_dir, "output.mp4")
+    concat_file = os.path.join(temp_dir, "concat.txt")
+    
+    try:
+        # Simple concatenation without transitions
+        with open(concat_file, 'w') as f:
+            for path in video_paths:
+                f.write(f"file '{path}'\n")
+        
+        subprocess.run([
+            "ffmpeg", "-y", "-f", "concat", "-safe", "0",
+            "-i", concat_file, "-c", "copy", output_path
+        ], check=True, capture_output=True)
+        
+        # Save to static folder and get URL
+        result_url = save_video_to_static_folder(output_path)
+        
+        # Clean up temp files
+        shutil.rmtree(temp_dir)
+        
+        return result_url
+    except Exception as e:
+        print(f"Error merging videos: {str(e)}")
+        
+        # Clean up temp files
+        if os.path.exists(temp_dir):
+            shutil.rmtree(temp_dir)
+        
+        return None
+
+
+def change_aspect_ratio(video_url: str, target_ratio: str = "9:16", method: str = "smart") -> str:
+    """    Change the aspect ratio of a video to a target ratio using specified method.
+    Args:
+        video_url (str): URL of the video to process.
+        target_ratio (str): Target aspect ratio in format "W:H" (default: "9:16").
+        method (str): Method to use for changing aspect ratio ("pad", "crop", "stretch", "smart").
+    Returns:
+        str: URL to the processed video with changed aspect ratio.
+    """
+    if not video_url:
+        print("No video URL provided")
+        return None
+
+    valid_methods = ["pad", "crop", "stretch", "smart"]
+    if method not in valid_methods:
+        print(f"Invalid method: {method}. Must be one of: {', '.join(valid_methods)}")
+        return None
+
+    try:
+        target_w, target_h = map(int, target_ratio.split(':'))
+        if target_w <= 0 or target_h <= 0:
+            raise ValueError("Ratio values must be positive integers")
+    except Exception:
+        print(f"Invalid target ratio format: {target_ratio}. Must be in format 'W:H' with positive integers")
+        return None
+
+    video_path = download_video_from_url(video_url)
+    if not video_path:
+        print("Failed to download video from URL")
+        return None
+
+    output_path = generate_temp_path()
+    temp_dir = tempfile.mkdtemp()
+    blurred_bg = None
+
+    try:
+        # Get original video dimensions
+        probe_cmd = [
+            "ffprobe", "-v", "error", "-select_streams", "v:0",
+            "-show_entries", "stream=width,height",
+            "-of", "csv=p=0", video_path
+        ]
+        width, height = map(int, subprocess.check_output(probe_cmd).decode().strip().split(','))
+        target_aspect = target_w / target_h
+        current_aspect = width / height
+
+        if method == "pad":
+            filter_expr = f"scale=iw*min({target_aspect}*ih/iw,1):ih*min({1/target_aspect}*iw/ih,1),pad={target_w}:{target_h}:(ow-iw)/2:(oh-ih)/2:black"
+
+        elif method == "crop":
+            if current_aspect > target_aspect:
+                new_width = int(height * target_aspect)
+                x_offset = int((width - new_width) / 2)
+                filter_expr = f"crop={new_width}:{height}:{x_offset}:0,scale={target_w}:{target_h}"
+            else:
+                new_height = int(width / target_aspect)
+                y_offset = int((height - new_height) / 2)
+                filter_expr = f"crop={width}:{new_height}:0:{y_offset},scale={target_w}:{target_h}"
+
+        elif method == "stretch":
+            filter_expr = f"scale={target_w}:{target_h},setdar={target_ratio}"
+
+        # Smart → internally uses blur_bg when converting landscape to portrait
+        if method == "blur_bg" or (method == "smart" and target_aspect < current_aspect and target_ratio == "9:16"):
+            # Use "blur_bg" logic
+            target_w, target_h = 1080, 1920
+            scale_factor_w = target_w / width
+            scale_factor_h = target_h / height
+            main_scale = min(scale_factor_w, scale_factor_h)
+            main_width = int(width * main_scale)
+            main_height = int(height * main_scale)
+            main_y = int((target_h - main_height) / 2)
+            bg_scale = max(target_w / width, target_h / height)
+
+            if width * bg_scale > 3840:
+                bg_scale = 3840 / width
+
+            bg_width = (int(width * bg_scale) + 1) & ~1
+            bg_height = (int(height * bg_scale) + 1) & ~1
+            main_width = (main_width + 1) & ~1
+            main_height = (main_height + 1) & ~1
+
+            blurred_bg = os.path.join(temp_dir, "blurred_bg.mp4")
+
+            # Try to create blurred background
+            try:
+                filter_string = f"scale={bg_width}:{bg_height},boxblur=8:1"
+                subprocess.run([
+                    "ffmpeg", "-i", video_path,
+                    "-vf", filter_string,
+                    "-c:v", "libx264", "-preset", "fast", "-crf", "23",
+                    "-c:a", "copy", blurred_bg
+                ], check=True, capture_output=True)
+
+                subprocess.run([
+                    "ffmpeg",
+                    "-i", blurred_bg,
+                    "-i", video_path,
+                    "-filter_complex",
+                    f"[0:v]scale={target_w}:{target_h},setsar=1[bg];" +
+                    f"[1:v]scale={main_width}:{main_height}[main];" +
+                    f"[bg][main]overlay=(W-w)/2:{main_y}:format=auto",
+                    "-map", "[v]", "-map", "1:a?",
+                    "-c:v", "libx264", "-preset", "fast", "-crf", "23",
+                    "-c:a", "copy", output_path
+                ], check=True, capture_output=True)
+
+                return save_video_to_static_folder(output_path)
+
+            except subprocess.CalledProcessError as e:
+                print("FFmpeg error:", e.stderr.decode())
+                return None
+
+        elif method == "smart":
+            # Fallback if not a 9:16 conversion, use crop logic
+            return change_aspect_ratio(video_url, target_ratio, method="crop")
+
+        # For simple methods (pad/crop/stretch)
+        subprocess.run([
+            "ffmpeg", "-i", video_path,
+            "-vf", filter_expr,
+            "-c:v", "libx264", "-preset", "fast", "-crf", "23",
+            "-c:a", "copy", output_path
+        ], check=True, capture_output=True)
+
+        return save_video_to_static_folder(output_path)
+
+    except Exception as e:
+        print("Unexpected error:", str(e))
+        return None
+
+    finally:
+        for path in [output_path, blurred_bg]:
+            if path and os.path.exists(path):
+                try: os.remove(path)
+                except: pass
+        if os.path.exists(temp_dir):
+            try: shutil.rmtree(temp_dir)
+            except: pass

videos_management.py

@@ -0,0 +1,106 @@
+from pathlib import Path
+import uuid
+import os
+import shutil
+
+APP_BASE_URL = os.getenv("APP_BASE_URL", "http://127.0.0.1:7860/gradio_api")
+
+def generate_unique_id():
+    """Generate a unique ID for a video file"""
+    return str(uuid.uuid4())
+
+def save_video_to_static_folder(video_path, unique_id=None, extension="mp4"):
+    """
+    Save a video to the static videos folder
+    
+    Args:
+        video_path (str): Path to the video file
+        unique_id (str, optional): Unique ID for the video. If None, one will be generated
+        
+    Returns:
+        str: URL to access the video
+    """
+    if unique_id is None:
+        unique_id = generate_unique_id()
+        
+    output_dir = Path.cwd().absolute() / "videos"
+    if not output_dir.exists():
+        output_dir.mkdir(parents=True, exist_ok=True)
+    
+    output_path = output_dir / f"{unique_id}.{extension}"
+    
+    # Copy the file to the output path
+    shutil.copy(video_path, output_path)
+    
+    # Generate URL
+    video_url = f"{APP_BASE_URL}/file=videos/{unique_id}.{extension}"
+    
+    return video_url
+
+def download_video_from_url(video_url):
+    """
+    Download a video from a URL to a temporary file or handle local file paths
+    
+    Args:
+        video_url (str): URL of the video or path to local file
+        
+    Returns:
+        str: Path to the downloaded temporary file or local file
+    """
+    import tempfile
+    import requests
+    from pathlib import Path
+    import os
+    
+    print("Video url", video_url)
+    
+    if video_url is None:
+        print("Video URL is None")
+        return None
+    
+    video_url = str(video_url)
+    
+    # Check if it's a local file path
+    if os.path.exists(video_url):
+        print(f"Using existing local file: {video_url}")
+        return video_url
+    
+    # Check if it's a Gradio API URL
+    if video_url.startswith(f"{APP_BASE_URL}/file=videos/"):
+        # Extract the unique ID from the URL
+        unique_id = video_url.split('/')[-1].split('.')[0]
+        video_path = Path.cwd().absolute() / "videos" / f"{unique_id}.mp4"
+        if video_path.exists():
+            print(f"Found local file for URL: {str(video_path)}")
+            return str(video_path)
+    
+    # Handle file paths that may have been incorrectly passed as URLs
+    # For Windows paths that may be misinterpreted as URLs
+    if ":\\" in video_url or (os.path.sep == "\\" and "\\" in video_url):
+        potential_path = video_url
+        if os.path.exists(potential_path):
+            print(f"Using existing Windows path: {potential_path}")
+            return potential_path
+    
+    # If not a local file or path, download it from URL
+    try:
+        print(f"Downloading from URL: {video_url}")
+        response = requests.get(video_url, stream=True)
+        if response.status_code != 200:
+            print(f"Failed to download video: HTTP {response.status_code}")
+            return None
+            
+        # Create a temporary file
+        temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=".mp4")
+        
+        # Write the content to the file
+        for chunk in response.iter_content(chunk_size=8192):
+            temp_file.write(chunk)
+            
+        temp_file.close()
+        print(f"Downloaded to: {temp_file.name}")
+        return temp_file.name
+        
+    except Exception as e:
+        print(f"Error downloading video: {str(e)}")
+        return None