Create app.py

app.py

@@ -0,0 +1,244 @@
+import gradio as gr
+from tts_module import get_voices, text_to_speech
+from moviepy.editor import (
+    AudioFileClip, ImageClip, CompositeAudioClip,
+    concatenate_audioclips, concatenate_videoclips, vfx, CompositeVideoClip,
+    ColorClip
+)
+import asyncio
+import os
+import json
+import time
+import requests
+import random
+from googleapiclient.discovery import build
+from google.oauth2 import service_account
+from googleapiclient.http import MediaFileUpload
+from io import BytesIO
+from PIL import Image
+import numpy as np
+
+# Define output and temp directories
+output_folder = "outputs"
+temp_dir = "temp_files"
+os.makedirs(output_folder, exist_ok=True)
+os.makedirs(temp_dir, exist_ok=True)
+
+# Google Drive folder ID
+FOLDER_ID = "12S6adpanAXjf71pKKGRRPqpzbJa5XEh3"  # Replace with your folder ID
+
+def cleanup_temp_files():
+    """Delete all temporary files from the temp_files folder."""
+    for filename in os.listdir(temp_dir):
+        file_path = os.path.join(temp_dir, filename)
+        try:
+            if os.path.isfile(file_path):
+                os.remove(file_path)
+        except Exception as e:
+            print(f"Error deleting {file_path}: {e}")
+
+def search_google_images(query, num_images=1):
+    """Search for images using Google Custom Search API."""
+    try:
+        service = build("customsearch", "v1", developerKey=os.getenv('GOOGLE_API_KEY'))
+        result = service.cse().list(q=query, cx=os.getenv('GOOGLE_CSE_ID'), searchType="image", num=num_images).execute()
+        return [item['link'] for item in result.get('items', [])]
+    except Exception as e:
+        print(f"Error searching Google Images: {e}")
+        return []
+
+def download_image(url):
+    """Download an image from a URL and return a PIL Image object."""
+    try:
+        response = requests.get(url)
+        return Image.open(BytesIO(response.content))
+    except Exception as e:
+        print(f"Error downloading image: {e}")
+        return None
+
+def create_animated_clip(image, duration=5, zoom_factor=1.1):
+    """Create an animated clip from a still image with a zoom effect."""
+    img_array = np.array(image)
+    img_clip = ImageClip(img_array).set_duration(duration)
+    return img_clip.resize(lambda t: 1 + (zoom_factor - 1) * t / duration)
+
+def concatenate_google_images(keywords, clip_duration=5, num_images_per_keyword=1):
+    """Concatenate Google Images based on keywords."""
+    keyword_list = [keyword.strip() for keyword in keywords.split(",") if keyword.strip()]
+    if not keyword_list:
+        keyword_list = ["nature"]  # Default keyword
+    video_clips = []
+    for keyword in keyword_list:
+        try:
+            print(f"Searching images for keyword '{keyword}'...")
+            image_urls = search_google_images(keyword, num_images=num_images_per_keyword)
+            for url in image_urls:
+                image = download_image(url)
+                if image:
+                    clip = create_animated_clip(image, duration=clip_duration)
+                    video_clips.append(clip)
+        except Exception as e:
+            print(f"Error processing keyword '{keyword}': {e}")
+            continue
+    if not video_clips:
+        # If no images, create a black clip
+        return ColorClip(size=(1920, 1080), color=[0, 0, 0], duration=5)
+    random.shuffle(video_clips)
+    return concatenate_videoclips(video_clips, method="compose")
+
+def adjust_background_music(video_duration, music_file):
+    """Adjust background music to match video duration."""
+    try:
+        music = AudioFileClip(music_file)
+        if music.duration < video_duration:
+            repetitions = int(video_duration / music.duration) + 1
+            music_clips = [music] * repetitions
+            music = concatenate_audioclips(music_clips)
+        music = music.subclip(0, video_duration)
+        return music.volumex(0.2)
+    except Exception as e:
+        print(f"Error adjusting music: {e}")
+        return None
+
+def combine_audio_video(audio_file, video_clip, music_clip=None):
+    """Combine audio and video into a final file."""
+    try:
+        audio_clip = AudioFileClip(audio_file)
+        total_duration = audio_clip.duration + 2  # Add 2 extra seconds
+        
+        # Ensure video has correct duration
+        video_clip = video_clip.loop(duration=total_duration)
+        video_clip = video_clip.set_duration(total_duration).fadeout(2)
+        
+        # Combine main audio
+        final_clip = video_clip.set_audio(audio_clip)
+        # Add background music if it exists
+        if music_clip:
+            music_clip = music_clip.set_duration(total_duration).audio_fadeout(2)
+            final_clip = final_clip.set_audio(CompositeAudioClip([audio_clip, music_clip]))
+        
+        # Generate filename and path
+        output_filename = f"final_video_{int(time.time())}.mp4"
+        output_path = os.path.join(output_folder, output_filename)
+        
+        # Save video
+        final_clip.write_videofile(output_path, codec="libx264", audio_codec="aac", fps=24)
+        
+        # Clean up clips
+        final_clip.close()
+        video_clip.close()
+        audio_clip.close()
+        if music_clip:
+            music_clip.close()
+            
+        return output_path
+    except Exception as e:
+        print(f"Error combining audio and video: {e}")
+        if 'final_clip' in locals():
+            final_clip.close()
+        return None
+
+def upload_to_google_drive(file_path, folder_id):
+    """Upload a file to Google Drive and return the public link."""
+    try:
+        creds = service_account.Credentials.from_service_account_info(
+            json.loads(os.getenv('GOOGLE_SERVICE_ACCOUNT')),
+            scopes=['https://www.googleapis.com/auth/drive']
+        )
+        service = build('drive', 'v3', credentials=creds)
+
+        file_metadata = {
+            'name': os.path.basename(file_path),
+            'parents': [folder_id]
+        }
+        media = MediaFileUpload(file_path, resumable=True)
+        file = service.files().create(body=file_metadata, media_body=media, fields='id').execute()
+
+        permission = {
+            'type': 'anyone',
+            'role': 'reader'
+        }
+        service.permissions().create(fileId=file['id'], body=permission).execute()
+
+        file_id = file['id']
+        download_link = f"https://drive.google.com/uc?export=download&id={file_id}"
+        return download_link
+    except Exception as e:
+        print(f"Error uploading to Google Drive: {e}")
+        return None
+
+def process_input(text, txt_file, mp3_file, selected_voice, rate, pitch, keywords):
+    """Process user input and generate the final video."""
+    try:
+        # Determine the text to use
+        if text.strip():
+            final_text = text
+        elif txt_file is not None:
+            final_text = txt_file.decode("utf-8")
+        else:
+            raise ValueError("No text input provided")
+        
+        # Generate audio
+        audio_file = asyncio.run(text_to_speech(final_text, selected_voice, rate, pitch))
+        if not audio_file:
+            raise ValueError("Failed to generate audio")
+        
+        # Generate video from Google Images
+        video_clip = concatenate_google_images(keywords, clip_duration=5, num_images_per_keyword=1)
+        if not video_clip:
+            raise ValueError("Failed to generate video")
+        
+        # Process background music if it exists
+        music_clip = None
+        if mp3_file is not None:
+            music_clip = adjust_background_music(video_clip.duration, mp3_file.name)
+        
+        # Combine everything
+        final_video_path = combine_audio_video(audio_file, video_clip, music_clip)
+        if not final_video_path:
+            raise ValueError("Failed to combine audio and video")
+        
+        # Upload to Google Drive and get the link
+        download_link = upload_to_google_drive(final_video_path, folder_id=FOLDER_ID)
+        if download_link:
+            print(f"Video uploaded to Google Drive. Download link: {download_link}")
+            return f"[Download video]({download_link})"
+        else:
+            raise ValueError("Error uploading video to Google Drive")
+    except Exception as e:
+        print(f"Error during processing: {e}")
+        return None
+    finally:
+        cleanup_temp_files()  # Clean up temporary files when finished
+
+# Gradio interface
+with gr.Blocks() as demo:
+    gr.Markdown("# Text-to-Video Generator")
+    with gr.Row():
+        with gr.Column():
+            text_input = gr.Textbox(label="Write your text here", lines=5)
+            txt_file_input = gr.File(label="Or upload a .txt file", file_types=[".txt"])
+            mp3_file_input = gr.File(label="Upload background music (.mp3)", file_types=[".mp3"])
+            keyword_input = gr.Textbox(
+                label="Enter keywords separated by commas (e.g., universe, galaxy, forest, cat)",
+                value="fear, religion, god, demons, aliens, possession, galaxy, mysterious, dystopian, astral, warfare, space, space, galaxy, moon, fear, astral, god, evil, mystery, cosmos, stars, paranormal, inexplicable, hidden, enigma, unknown, unusual, intriguing, curious, strange, supernatural, esoteric, arcane, occultism, supernatural, mystery, phenomenon, rare, unusual, enigmatic, sinister, gloomy, dark, shadowy, macabre, eerie, chilling, cursed, fantastic, unreal, unknown, mysterious, enigmatic, inexplicable, unusual, strange, unusual, arcane, esoteric, hidden, shadowy, dark, gloomy, sinister, macabre, eerie, chilling, cursed, fantastic, unreal, paranormal, supernatural, occultism, phenomenon, rare, intriguing, curious"
+            )
+            voices = asyncio.run(get_voices())
+            voice_dropdown = gr.Dropdown(choices=list(voices.keys()), label="Select Voice")
+            rate_slider = gr.Slider(minimum=-50, maximum=50, value=0, label="Speech Rate Adjustment (%)", step=1)
+            pitch_slider = gr.Slider(minimum=-20, maximum=20, value=0, label="Pitch Adjustment (Hz)", step=1)
+        with gr.Column():
+           output_link = gr.Markdown("")  # Display download link
+
+    btn = gr.Button("Generate Video")
+    btn.click(
+        process_input,
+        inputs=[text_input, txt_file_input, mp3_file_input, voice_dropdown, rate_slider, pitch_slider, keyword_input],
+        outputs=output_link
+    )
+
+# Read the port assigned by Hugging Face
+port = int(os.getenv("PORT", 7860))
+
+# Launch the application
+demo.launch(server_name="0.0.0.0", server_port=port, share=True, show_error=True)