Spaces:
Running
Running
| """ | |
| Full Code: Orbit Video Engine with Advanced Gradio UI | |
| This script combines the video-generation code (using Kokoro for TTS, | |
| MoviePy for video operations, Pexels/Google image/video search, etc.) with a | |
| Gradio Blocks UI that allows: | |
| 1. Content input and script generation. | |
| 2. Dynamic clip editing (change prompt, narration, or upload custom media). | |
| 3. Video settings (resolution, render speed, background music, subtitle settings). | |
| 4. Final video generation and preview/download. | |
| """ | |
| # --------- IMPORTS --------- | |
| from kokoro import KPipeline | |
| import soundfile as sf | |
| import torch | |
| import os, time, random, math, json, tempfile, shutil, re, requests | |
| from moviepy.editor import VideoFileClip, AudioFileClip, ImageClip, concatenate_videoclips, CompositeVideoClip, TextClip, CompositeAudioClip | |
| import moviepy.video.fx.all as vfx | |
| import moviepy.config as mpy_config | |
| from pydub import AudioSegment | |
| from pydub.generators import Sine | |
| from PIL import Image, ImageDraw, ImageFont | |
| import numpy as np | |
| from bs4 import BeautifulSoup | |
| from urllib.parse import quote | |
| import pysrt | |
| from gtts import gTTS | |
| import cv2 | |
| import gradio as gr | |
| # --------- GLOBAL CONFIGURATION --------- | |
| pipeline = KPipeline(lang_code='a') # Use American English voice. (Uses 'af_heart' for American English) | |
| mpy_config.change_settings({"IMAGEMAGICK_BINARY": "/usr/bin/convert"}) | |
| PEXELS_API_KEY = 'BhJqbcdm9Vi90KqzXKAhnEHGsuFNv4irXuOjWtT761U49lRzo03qBGna' | |
| OPENROUTER_API_KEY = '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' | |
| OPENROUTER_MODEL = "mistralai/mistral-small-3.1-24b-instruct:free" | |
| OUTPUT_VIDEO_FILENAME = "final_video.mp4" | |
| USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36" | |
| # These globals are later set per run | |
| TARGET_RESOLUTION = None | |
| CAPTION_COLOR = None | |
| TEMP_FOLDER = None | |
| # --------- HELPER FUNCTIONS --------- | |
| def generate_script(user_input): | |
| """Generate documentary script using OpenRouter API.""" | |
| headers = { | |
| 'Authorization': f'Bearer {OPENROUTER_API_KEY}', | |
| 'HTTP-Referer': 'https://your-domain.com', | |
| 'X-Title': 'AI Documentary Maker' | |
| } | |
| prompt = f"""Short Documentary Script GeneratorInstructions: | |
| If I say "use this," just output the script exactly as I gave it. | |
| If I only give topics, generate a script based on them. | |
| If I provide a full script, rewrite it without any changes. Make everything short simple and humarous funny and act as serious but humarous. And don't say anything off topic. Also alway say a funny statement to subscribe based on the video topic at the end. Use normal conversational text like a normal person talking and avoid AI phase make the statements humanize and normal conversational | |
| And always talk negative views don't really wants any positive statement and the full script will be one video about one topic so keep in mind | |
| Formatting Rules: | |
| Title in Square Brackets: | |
| Each section starts with a one-word title inside [ ] (max two words if necessary). | |
| This title will be used as a search term for Pexels footage. | |
| Casual & Funny Narration: | |
| Each section has 5-10 words of narration. | |
| Keep it natural, funny, and unpredictable (not robotic, poetic, or rhythmic). | |
| No Special Formatting: | |
| No bold, italics, or special characters. You are a assistant AI your task is to create script. You aren't a chatbot. So, don't write extra text | |
| Generalized Search Terms: | |
| If a term is too specific, make it more general for Pexels search. | |
| Scene-Specific Writing: | |
| Each section describes only what should be shown in the video. | |
| Output Only the Script, and also make it funny and humarous and helirous and also add to subscribe with a funny statement like subscribe now or ..... | |
| No extra text, just the script. | |
| Example Output: | |
| [North Korea] | |
| Top 5 unknown facts about North Korea. | |
| [Invisibility] | |
| North Korea’s internet speed is so fast… it doesn’t exist. | |
| [Leadership] | |
| Kim Jong-un once won an election with 100% votes… against himself. | |
| [Magic] | |
| North Korea discovered time travel. That’s why their news is always from the past. | |
| [Warning] | |
| Subscribe now, or Kim Jong-un will send you a free one-way ticket… to North Korea. | |
| [Freedom] | |
| North Korean citizens can do anything… as long as it's government-approved. | |
| Now here is the Topic/scrip: {user_input} | |
| """ | |
| data = { | |
| 'model': OPENROUTER_MODEL, | |
| 'messages': [{'role': 'user', 'content': prompt}], | |
| 'temperature': 0.4, | |
| 'max_tokens': 5000 | |
| } | |
| try: | |
| response = requests.post( | |
| 'https://openrouter.ai/api/v1/chat/completions', | |
| headers=headers, | |
| json=data, | |
| timeout=30 | |
| ) | |
| if response.status_code == 200: | |
| response_data = response.json() | |
| if 'choices' in response_data and len(response_data['choices']) > 0: | |
| return response_data['choices'][0]['message']['content'] | |
| else: | |
| print("Unexpected response format:", response_data) | |
| return None | |
| else: | |
| print(f"API Error {response.status_code}: {response.text}") | |
| return None | |
| except Exception as e: | |
| print(f"Request failed: {str(e)}") | |
| return None | |
| def parse_script(script_text): | |
| """ | |
| Parse the generated script into a list of clip elements. | |
| Each clip consists of a media element (with a 'prompt') and a TTS element with narration. | |
| """ | |
| sections = {} | |
| current_title = None | |
| current_text = "" | |
| try: | |
| for line in script_text.splitlines(): | |
| line = line.strip() | |
| if line.startswith("[") and "]" in line: | |
| bracket_start = line.find("[") | |
| bracket_end = line.find("]", bracket_start) | |
| if bracket_start != -1 and bracket_end != -1: | |
| if current_title is not None: | |
| sections[current_title] = current_text.strip() | |
| current_title = line[bracket_start+1:bracket_end] | |
| current_text = line[bracket_end+1:].strip() | |
| elif current_title: | |
| current_text += line + " " | |
| if current_title: | |
| sections[current_title] = current_text.strip() | |
| elements = [] | |
| for title, narration in sections.items(): | |
| if not title or not narration: | |
| continue | |
| media_element = {"type": "media", "prompt": title, "effects": "fade-in"} | |
| words = narration.split() | |
| duration = max(3, len(words) * 0.5) | |
| tts_element = {"type": "tts", "text": narration, "voice": "en", "duration": duration} | |
| elements.append(media_element) | |
| elements.append(tts_element) | |
| return elements | |
| except Exception as e: | |
| print(f"Error parsing script: {e}") | |
| return [] | |
| def search_pexels_videos(query, pexels_api_key): | |
| """Search for a video on Pexels by query and return a random HD video link.""" | |
| headers = {'Authorization': pexels_api_key} | |
| base_url = "https://api.pexels.com/videos/search" | |
| num_pages = 3 | |
| videos_per_page = 15 | |
| max_retries = 3 | |
| retry_delay = 1 | |
| search_query = query | |
| all_videos = [] | |
| for page in range(1, num_pages + 1): | |
| for attempt in range(max_retries): | |
| try: | |
| params = {"query": search_query, "per_page": videos_per_page, "page": page} | |
| response = requests.get(base_url, headers=headers, params=params, timeout=10) | |
| if response.status_code == 200: | |
| data = response.json() | |
| videos = data.get("videos", []) | |
| if not videos: | |
| print(f"No videos found on page {page}.") | |
| break | |
| for video in videos: | |
| video_files = video.get("video_files", []) | |
| for file in video_files: | |
| if file.get("quality") == "hd": | |
| all_videos.append(file.get("link")) | |
| break | |
| break | |
| elif response.status_code == 429: | |
| print(f"Rate limit hit (attempt {attempt+1}/{max_retries}). Retrying...") | |
| time.sleep(retry_delay) | |
| retry_delay *= 2 | |
| else: | |
| print(f"Error fetching videos: {response.status_code} {response.text}") | |
| if attempt < max_retries - 1: | |
| time.sleep(retry_delay) | |
| retry_delay *= 2 | |
| else: | |
| break | |
| except requests.exceptions.RequestException as e: | |
| print(f"Request exception: {e}") | |
| if attempt < max_retries - 1: | |
| time.sleep(retry_delay) | |
| retry_delay *= 2 | |
| else: | |
| break | |
| if all_videos: | |
| random_video = random.choice(all_videos) | |
| print(f"Selected random video from {len(all_videos)} HD videos") | |
| return random_video | |
| else: | |
| print("No suitable videos found.") | |
| return None | |
| def search_pexels_images(query, pexels_api_key): | |
| """Search for an image on Pexels by query.""" | |
| headers = {'Authorization': pexels_api_key} | |
| url = "https://api.pexels.com/v1/search" | |
| params = {"query": query, "per_page": 5, "orientation": "landscape"} | |
| max_retries = 3 | |
| retry_delay = 1 | |
| for attempt in range(max_retries): | |
| try: | |
| response = requests.get(url, headers=headers, params=params, timeout=10) | |
| if response.status_code == 200: | |
| data = response.json() | |
| photos = data.get("photos", []) | |
| if photos: | |
| photo = random.choice(photos[:min(5, len(photos))]) | |
| img_url = photo.get("src", {}).get("original") | |
| return img_url | |
| else: | |
| print(f"No images found for query: {query}") | |
| return None | |
| elif response.status_code == 429: | |
| print(f"Rate limit hit (attempt {attempt+1}/{max_retries}). Retrying...") | |
| time.sleep(retry_delay) | |
| retry_delay *= 2 | |
| else: | |
| print(f"Error fetching images: {response.status_code} {response.text}") | |
| if attempt < max_retries - 1: | |
| time.sleep(retry_delay) | |
| retry_delay *= 2 | |
| except requests.exceptions.RequestException as e: | |
| print(f"Request exception: {e}") | |
| if attempt < max_retries - 1: | |
| time.sleep(retry_delay) | |
| retry_delay *= 2 | |
| print(f"No Pexels images found for query: {query}") | |
| return None | |
| def search_google_images(query): | |
| """Search for images on Google Images.""" | |
| try: | |
| search_url = f"https://www.google.com/search?q={quote(query)}&tbm=isch" | |
| headers = {"User-Agent": USER_AGENT} | |
| response = requests.get(search_url, headers=headers, timeout=10) | |
| soup = BeautifulSoup(response.text, "html.parser") | |
| img_tags = soup.find_all("img") | |
| image_urls = [] | |
| for img in img_tags: | |
| src = img.get("src", "") | |
| if src.startswith("http") and "gstatic" not in src: | |
| image_urls.append(src) | |
| if image_urls: | |
| return random.choice(image_urls[:5]) if len(image_urls) >= 5 else image_urls[0] | |
| else: | |
| print(f"No Google Images found for query: {query}") | |
| return None | |
| except Exception as e: | |
| print(f"Error in Google Images search: {e}") | |
| return None | |
| def download_image(image_url, filename): | |
| """Download an image from a URL to a local file with verification.""" | |
| try: | |
| headers = {"User-Agent": USER_AGENT} | |
| print(f"Downloading image from: {image_url} to {filename}") | |
| response = requests.get(image_url, headers=headers, stream=True, timeout=15) | |
| response.raise_for_status() | |
| with open(filename, 'wb') as f: | |
| for chunk in response.iter_content(chunk_size=8192): | |
| f.write(chunk) | |
| print(f"Image downloaded to: {filename}") | |
| try: | |
| img = Image.open(filename) | |
| img.verify() | |
| img = Image.open(filename) | |
| if img.mode != 'RGB': | |
| img = img.convert('RGB') | |
| img.save(filename) | |
| print(f"Image validated and processed: {filename}") | |
| return filename | |
| except Exception as e_validate: | |
| print(f"Downloaded file is not a valid image: {e_validate}") | |
| if os.path.exists(filename): | |
| os.remove(filename) | |
| return None | |
| except Exception as e: | |
| print(f"Image download error: {e}") | |
| if os.path.exists(filename): | |
| os.remove(filename) | |
| return None | |
| def download_video(video_url, filename): | |
| """Download a video from a URL to a local file.""" | |
| try: | |
| response = requests.get(video_url, stream=True, timeout=30) | |
| response.raise_for_status() | |
| with open(filename, 'wb') as f: | |
| for chunk in response.iter_content(chunk_size=8192): | |
| f.write(chunk) | |
| print(f"Video downloaded to: {filename}") | |
| return filename | |
| except Exception as e: | |
| print(f"Video download error: {e}") | |
| if os.path.exists(filename): | |
| os.remove(filename) | |
| return None | |
| def generate_media(prompt, user_image=None, current_index=0, total_segments=1): | |
| """ | |
| Generate a visual asset for the clip. | |
| If user_image is provided (from custom media upload) use it; otherwise, | |
| use Pexels (and Google Images for news related queries) with fallbacks. | |
| """ | |
| safe_prompt = re.sub(r'[^\w\s-]', '', prompt).strip().replace(' ', '_') | |
| if user_image is not None: | |
| # Assume user_image is a local path or file-like object. | |
| print(f"Using custom media provided for prompt: {prompt}") | |
| return {"path": user_image, "asset_type": "image"} | |
| if "news" in prompt.lower(): | |
| print(f"News query detected: {prompt}. Using Google Images...") | |
| image_file = os.path.join(TEMP_FOLDER, f"{safe_prompt}_news.jpg") | |
| image_url = search_google_images(prompt) | |
| if image_url: | |
| downloaded_image = download_image(image_url, image_file) | |
| if downloaded_image: | |
| return {"path": downloaded_image, "asset_type": "image"} | |
| if random.random() < 0.25: | |
| video_file = os.path.join(TEMP_FOLDER, f"{safe_prompt}_video.mp4") | |
| video_url = search_pexels_videos(prompt, PEXELS_API_KEY) | |
| if video_url: | |
| downloaded_video = download_video(video_url, video_file) | |
| if downloaded_video: | |
| return {"path": downloaded_video, "asset_type": "video"} | |
| image_file = os.path.join(TEMP_FOLDER, f"{safe_prompt}.jpg") | |
| image_url = search_pexels_images(prompt, PEXELS_API_KEY) | |
| if image_url: | |
| downloaded_image = download_image(image_url, image_file) | |
| if downloaded_image: | |
| return {"path": downloaded_image, "asset_type": "image"} | |
| fallback_terms = ["nature", "people", "landscape", "technology", "business"] | |
| for term in fallback_terms: | |
| fallback_file = os.path.join(TEMP_FOLDER, f"fallback_{term}.jpg") | |
| fallback_url = search_pexels_images(term, PEXELS_API_KEY) | |
| if fallback_url: | |
| downloaded_fallback = download_image(fallback_url, fallback_file) | |
| if downloaded_fallback: | |
| return {"path": downloaded_fallback, "asset_type": "image"} | |
| print(f"Failed to generate asset for: {prompt}") | |
| return None | |
| def generate_silent_audio(duration, sample_rate=24000): | |
| """Generate a silent WAV file for TTS fallback.""" | |
| num_samples = int(duration * sample_rate) | |
| silence = np.zeros(num_samples, dtype=np.float32) | |
| silent_path = os.path.join(TEMP_FOLDER, f"silent_{int(time.time())}.wav") | |
| sf.write(silent_path, silence, sample_rate) | |
| return silent_path | |
| def generate_tts(text, voice): | |
| """Generate TTS audio using Kokoro, falling back to gTTS if needed.""" | |
| safe_text = re.sub(r'[^\w\s-]', '', text[:10]).strip().replace(' ', '_') | |
| file_path = os.path.join(TEMP_FOLDER, f"tts_{safe_text}.wav") | |
| if os.path.exists(file_path): | |
| return file_path | |
| try: | |
| kokoro_voice = 'af_heart' if voice == 'en' else voice | |
| generator = pipeline(text, voice=kokoro_voice, speed=0.9, split_pattern=r'\n+') | |
| audio_segments = [] | |
| for i, (gs, ps, audio) in enumerate(generator): | |
| audio_segments.append(audio) | |
| full_audio = np.concatenate(audio_segments) if len(audio_segments) > 1 else audio_segments[0] | |
| sf.write(file_path, full_audio, 24000) | |
| return file_path | |
| except Exception as e: | |
| try: | |
| tts = gTTS(text=text, lang='en') | |
| mp3_path = os.path.join(TEMP_FOLDER, f"tts_{safe_text}.mp3") | |
| tts.save(mp3_path) | |
| audio = AudioSegment.from_mp3(mp3_path) | |
| audio.export(file_path, format="wav") | |
| os.remove(mp3_path) | |
| return file_path | |
| except Exception as fallback_error: | |
| return generate_silent_audio(duration=max(3, len(text.split()) * 0.5)) | |
| def apply_kenburns_effect(clip, target_resolution, effect_type=None): | |
| """Apply a smooth Ken Burns effect to images.""" | |
| target_w, target_h = target_resolution | |
| clip_aspect = clip.w / clip.h | |
| target_aspect = target_w / target_h | |
| if clip_aspect > target_aspect: | |
| new_height = target_h | |
| new_width = int(new_height * clip_aspect) | |
| else: | |
| new_width = target_w | |
| new_height = int(new_width / clip_aspect) | |
| clip = clip.resize(newsize=(new_width, new_height)) | |
| base_scale = 1.15 | |
| new_width = int(new_width * base_scale) | |
| new_height = int(new_height * base_scale) | |
| clip = clip.resize(newsize=(new_width, new_height)) | |
| max_offset_x = new_width - target_w | |
| max_offset_y = new_height - target_h | |
| available_effects = ["zoom-in", "zoom-out", "pan-left", "pan-right", "up-left"] | |
| if effect_type is None or effect_type == "random": | |
| effect_type = random.choice(available_effects) | |
| if effect_type == "zoom-in": | |
| start_zoom = 0.9 | |
| end_zoom = 1.1 | |
| start_center = (new_width / 2, new_height / 2) | |
| end_center = start_center | |
| elif effect_type == "zoom-out": | |
| start_zoom = 1.1 | |
| end_zoom = 0.9 | |
| start_center = (new_width / 2, new_height / 2) | |
| end_center = start_center | |
| elif effect_type == "pan-left": | |
| start_zoom = 1.0 | |
| end_zoom = 1.0 | |
| start_center = (max_offset_x + target_w / 2, (max_offset_y // 2) + target_h / 2) | |
| end_center = (target_w / 2, (max_offset_y // 2) + target_h / 2) | |
| elif effect_type == "pan-right": | |
| start_zoom = 1.0 | |
| end_zoom = 1.0 | |
| start_center = (target_w / 2, (max_offset_y // 2) + target_h / 2) | |
| end_center = (max_offset_x + target_w / 2, (max_offset_y // 2) + target_h / 2) | |
| elif effect_type == "up-left": | |
| start_zoom = 1.0 | |
| end_zoom = 1.0 | |
| start_center = (max_offset_x + target_w / 2, max_offset_y + target_h / 2) | |
| end_center = (target_w / 2, target_h / 2) | |
| else: | |
| raise ValueError(f"Unsupported effect_type: {effect_type}") | |
| def transform_frame(get_frame, t): | |
| frame = get_frame(t) | |
| ratio = t / clip.duration if clip.duration > 0 else 0 | |
| ratio = 0.5 - 0.5 * math.cos(math.pi * ratio) | |
| current_zoom = start_zoom + (end_zoom - start_zoom) * ratio | |
| crop_w = int(target_w / current_zoom) | |
| crop_h = int(target_h / current_zoom) | |
| current_center_x = start_center[0] + (end_center[0] - start_center[0]) * ratio | |
| current_center_y = start_center[1] + (end_center[1] - start_center[1]) * ratio | |
| min_center_x = crop_w / 2 | |
| max_center_x = new_width - crop_w / 2 | |
| min_center_y = crop_h / 2 | |
| max_center_y = new_height - crop_h / 2 | |
| current_center_x = max(min_center_x, min(current_center_x, max_center_x)) | |
| current_center_y = max(min_center_y, min(current_center_y, max_center_y)) | |
| cropped_frame = cv2.getRectSubPix(frame, (crop_w, crop_h), (current_center_x, current_center_y)) | |
| resized_frame = cv2.resize(cropped_frame, (target_w, target_h), interpolation=cv2.INTER_LANCZOS4) | |
| return resized_frame | |
| return clip.fl(transform_frame) | |
| def resize_to_fill(clip, target_resolution): | |
| """Resize and crop a clip to fill the target resolution.""" | |
| target_w, target_h = target_resolution | |
| clip_aspect = clip.w / clip.h | |
| target_aspect = target_w / target_h | |
| if clip_aspect > target_aspect: | |
| clip = clip.resize(height=target_h) | |
| crop_amount = (clip.w - target_w) / 2 | |
| clip = clip.crop(x1=crop_amount, x2=clip.w - crop_amount, y1=0, y2=clip.h) | |
| else: | |
| clip = clip.resize(width=target_w) | |
| crop_amount = (clip.h - target_h) / 2 | |
| clip = clip.crop(x1=0, x2=clip.w, y1=crop_amount, y2=clip.h - crop_amount) | |
| return clip | |
| def find_mp3_files(): | |
| """Find an MP3 file for background music.""" | |
| mp3_files = [] | |
| for root, dirs, files in os.walk('.'): | |
| for file in files: | |
| if file.endswith('.mp3'): | |
| mp3_path = os.path.join(root, file) | |
| mp3_files.append(mp3_path) | |
| return mp3_files[0] if mp3_files else None | |
| def add_background_music(final_video, bg_music_volume=0.08): | |
| """Add background music to the final video.""" | |
| try: | |
| bg_music_path = find_mp3_files() | |
| if bg_music_path and os.path.exists(bg_music_path): | |
| bg_music = AudioFileClip(bg_music_path) | |
| if bg_music.duration < final_video.duration: | |
| loops_needed = math.ceil(final_video.duration / bg_music.duration) | |
| bg_segments = [bg_music] * loops_needed | |
| bg_music = concatenate_videoclips(bg_segments) | |
| bg_music = bg_music.subclip(0, final_video.duration) | |
| bg_music = bg_music.volumex(bg_music_volume) | |
| video_audio = final_video.audio | |
| mixed_audio = CompositeAudioClip([video_audio, bg_music]) | |
| final_video = final_video.set_audio(mixed_audio) | |
| return final_video | |
| except Exception as e: | |
| print(f"Error adding background music: {e}") | |
| return final_video | |
| def create_clip(media_path, asset_type, tts_path, duration=None, effects=None, narration_text=None, segment_index=0): | |
| """Create a video clip from a media asset and its corresponding TTS audio, optionally with subtitles.""" | |
| try: | |
| if not os.path.exists(media_path) or not os.path.exists(tts_path): | |
| return None | |
| audio_clip = AudioFileClip(tts_path).audio_fadeout(0.2) | |
| audio_duration = audio_clip.duration | |
| target_duration = audio_duration + 0.2 | |
| if asset_type == "video": | |
| clip = VideoFileClip(media_path) | |
| clip = resize_to_fill(clip, TARGET_RESOLUTION) | |
| if clip.duration < target_duration: | |
| clip = clip.loop(duration=target_duration) | |
| else: | |
| clip = clip.subclip(0, target_duration) | |
| elif asset_type == "image": | |
| img = Image.open(media_path) | |
| if img.mode != 'RGB': | |
| with tempfile.NamedTemporaryFile(suffix='.jpg', delete=False) as temp: | |
| img.convert('RGB').save(temp.name) | |
| media_path = temp.name | |
| img.close() | |
| clip = ImageClip(media_path).set_duration(target_duration) | |
| clip = apply_kenburns_effect(clip, TARGET_RESOLUTION) | |
| clip = clip.fadein(0.3).fadeout(0.3) | |
| else: | |
| return None | |
| if narration_text and CAPTION_COLOR != "transparent": | |
| try: | |
| words = narration_text.split() | |
| chunks = [] | |
| current_chunk = [] | |
| for word in words: | |
| current_chunk.append(word) | |
| if len(current_chunk) >= 5: | |
| chunks.append(' '.join(current_chunk)) | |
| current_chunk = [] | |
| if current_chunk: | |
| chunks.append(' '.join(current_chunk)) | |
| chunk_duration = audio_duration / len(chunks) | |
| subtitle_clips = [] | |
| subtitle_y_position = int(TARGET_RESOLUTION[1] * 0.70) | |
| for i, chunk_text in enumerate(chunks): | |
| start_time = i * chunk_duration | |
| end_time = (i + 1) * chunk_duration | |
| txt_clip = TextClip( | |
| chunk_text, | |
| fontsize=45, | |
| font='Arial-Bold', | |
| color=CAPTION_COLOR, | |
| bg_color='rgba(0, 0, 0, 0.25)', | |
| method='caption', | |
| align='center', | |
| stroke_width=2, | |
| stroke_color=CAPTION_COLOR, | |
| size=(TARGET_RESOLUTION[0] * 0.8, None) | |
| ).set_start(start_time).set_end(end_time) | |
| txt_clip = txt_clip.set_position(('center', subtitle_y_position)) | |
| subtitle_clips.append(txt_clip) | |
| clip = CompositeVideoClip([clip] + subtitle_clips) | |
| except Exception as sub_error: | |
| txt_clip = TextClip( | |
| narration_text, | |
| fontsize=28, | |
| color=CAPTION_COLOR, | |
| align='center', | |
| size=(TARGET_RESOLUTION[0] * 0.7, None) | |
| ).set_position(('center', int(TARGET_RESOLUTION[1] / 3))).set_duration(clip.duration) | |
| clip = CompositeVideoClip([clip, txt_clip]) | |
| clip = clip.set_audio(audio_clip) | |
| return clip | |
| except Exception as e: | |
| print(f"Error in create_clip: {str(e)}") | |
| return None | |
| def fix_imagemagick_policy(): | |
| """Attempt to fix ImageMagick security policies.""" | |
| try: | |
| policy_paths = [ | |
| "/etc/ImageMagick-6/policy.xml", | |
| "/etc/ImageMagick-7/policy.xml", | |
| "/etc/ImageMagick/policy.xml", | |
| "/usr/local/etc/ImageMagick-7/policy.xml" | |
| ] | |
| found_policy = next((path for path in policy_paths if os.path.exists(path)), None) | |
| if not found_policy: | |
| return False | |
| os.system(f"sudo cp {found_policy} {found_policy}.bak") | |
| os.system(f"sudo sed -i 's/rights=\"none\"/rights=\"read|write\"/g' {found_policy}") | |
| os.system(f"sudo sed -i 's/<policy domain=\"path\" pattern=\"@\\*\"[^>]*>/<policy domain=\"path\" pattern=\"@*\" rights=\"read|write\"/g' {found_policy}") | |
| os.system(f"sudo sed -i 's/<policy domain=\"coder\" rights=\"none\" pattern=\"PDF\"[^>]*>/<!-- <policy domain=\"coder\" rights=\"none\" pattern=\"PDF\"> -->/g' {found_policy}") | |
| return True | |
| except Exception as e: | |
| return False | |
| def generate_video(user_input, resolution, caption_option): | |
| """ | |
| Original video generation function. | |
| This version takes only the basic inputs and uses the global clip data generated by the script. | |
| You may need to extend this to use updated clip data from the dynamic clip editor. | |
| """ | |
| global TARGET_RESOLUTION, CAPTION_COLOR, TEMP_FOLDER | |
| if resolution == "Full HD (1920x1080)": | |
| TARGET_RESOLUTION = (1920, 1080) | |
| elif resolution == "Short (1080x1920)": | |
| TARGET_RESOLUTION = (1080, 1920) | |
| else: | |
| TARGET_RESOLUTION = (1920, 1080) | |
| CAPTION_COLOR = "white" if caption_option == "Yes" else "transparent" | |
| TEMP_FOLDER = tempfile.mkdtemp() | |
| fix_success = fix_imagemagick_policy() | |
| if not fix_success: | |
| print("Unable to fix ImageMagick policies; proceeding with alternative methods.") | |
| print("Generating script from API...") | |
| script = generate_script(user_input) | |
| if not script: | |
| shutil.rmtree(TEMP_FOLDER) | |
| return None | |
| print("Generated Script:\n", script) | |
| elements = parse_script(script) | |
| if not elements: | |
| shutil.rmtree(TEMP_FOLDER) | |
| return None | |
| paired_elements = [] | |
| for i in range(0, len(elements), 2): | |
| if i + 1 < len(elements): | |
| paired_elements.append((elements[i], elements[i + 1])) | |
| if not paired_elements: | |
| shutil.rmtree(TEMP_FOLDER) | |
| return None | |
| clips = [] | |
| for idx, (media_elem, tts_elem) in enumerate(paired_elements): | |
| print(f"Processing segment {idx+1} with prompt: '{media_elem['prompt']}'") | |
| media_asset = generate_media(media_elem['prompt']) | |
| if not media_asset: | |
| continue | |
| tts_path = generate_tts(tts_elem['text'], tts_elem['voice']) | |
| if not tts_path: | |
| continue | |
| clip = create_clip( | |
| media_path=media_asset['path'], | |
| asset_type=media_asset['asset_type'], | |
| tts_path=tts_path, | |
| duration=tts_elem['duration'], | |
| effects=media_elem.get('effects', 'fade-in'), | |
| narration_text=tts_elem['text'], | |
| segment_index=idx | |
| ) | |
| if clip: | |
| clips.append(clip) | |
| if not clips: | |
| shutil.rmtree(TEMP_FOLDER) | |
| return None | |
| final_video = concatenate_videoclips(clips, method="compose") | |
| final_video = add_background_music(final_video, bg_music_volume=0.08) | |
| final_video.write_videofile(OUTPUT_VIDEO_FILENAME, codec='libx264', fps=24, preset='veryfast') | |
| shutil.rmtree(TEMP_FOLDER) | |
| return OUTPUT_VIDEO_FILENAME | |
| # --------- NEW CALLBACK FUNCTIONS FOR THE NEW UI --------- | |
| # Global variable to store clip data from script-generation. | |
| generated_clip_data = [] | |
| def generate_script_clips(topic_input, full_script): | |
| """ | |
| Callback when the user clicks "📝 Generate Script & Load Clips". | |
| Uses full_script if provided, else generates a script based on topic_input. | |
| Returns generated raw script and a JSON string representing clip data. | |
| """ | |
| input_text = full_script if full_script.strip() != "" else topic_input | |
| script = generate_script(input_text) | |
| if not script: | |
| return "Error: failed to generate script", "{}" | |
| elements = parse_script(script) | |
| clip_list = [] | |
| for i in range(0, len(elements), 2): | |
| if i + 1 < len(elements): | |
| media_elem = elements[i] | |
| tts_elem = elements[i+1] | |
| clip_info = { | |
| "prompt": media_elem.get("prompt", ""), | |
| "narration": tts_elem.get("text", ""), | |
| "custom_media": None | |
| } | |
| clip_list.append(clip_info) | |
| global generated_clip_data | |
| generated_clip_data = clip_list | |
| return script, json.dumps(clip_list) | |
| def update_clip_editor(clip_json): | |
| """ | |
| Generate a dynamic interface (using Gradio Blocks) for editing clip details. | |
| Returns a list of Accordion components. | |
| """ | |
| clip_list = json.loads(clip_json) | |
| editors = [] | |
| for idx, clip in enumerate(clip_list, start=1): | |
| accordion_title = f"Clip {idx}: {clip['prompt']}" | |
| # Create an accordion for each clip. | |
| with gr.Accordion(label=accordion_title, open=(idx<=2)) as accordion: | |
| prompt_box = gr.Textbox(label="Visual Prompt", value=clip["prompt"]) | |
| narration_box = gr.Textbox(label="Narration Text", value=clip["narration"], lines=3) | |
| custom_media_box = gr.File(label="Custom Media Upload (overrides prompt)", file_types=[".jpg", ".png", ".mp4"]) | |
| # Pack into a dictionary structure. | |
| editors.append({ | |
| "prompt": prompt_box, | |
| "narration": narration_box, | |
| "custom_media": custom_media_box | |
| }) | |
| editors.append(accordion) | |
| return editors | |
| def generate_final_video(topic_input, full_script, clip_data_json, resolution, render_speed, | |
| video_clip_percent, zoom_pan, bg_music_file, bg_music_volume, | |
| subtitle_enabled, font_dropdown, font_size, outline_width, | |
| font_color, outline_color, subtitle_position): | |
| """ | |
| Callback for "🎬 Generate Video" button. | |
| Here we would combine the user-edited clip data with settings and call video generation. | |
| Note: For demonstration, we will use the original generate_video function. | |
| """ | |
| # In a real integration you would process clip_data_json to update each clip with custom overrides. | |
| print("Final settings:") | |
| print(f"Resolution: {resolution}, Render Speed: {render_speed}, Video Clip %: {video_clip_percent}, Zoom/Pan: {zoom_pan}") | |
| if bg_music_file is not None: | |
| print("Custom background music provided.") | |
| # For now, we simply call generate_video using topic_input and full_script. | |
| video_file = generate_video(topic_input, resolution, "Yes") | |
| return video_file | |
| # --------- GRADIO BLOCKS UI --------- | |
| with gr.Blocks(title="🚀 Orbit Video Engine") as demo: | |
| with gr.Row(): | |
| # Column 1: Content Input & Script Generation | |
| with gr.Column(scale=1): | |
| gr.Markdown("## 1. Content Input") | |
| topic_input = gr.Textbox(label="Topic Input", placeholder="Enter your video topic here...") | |
| full_script = gr.Textbox(label="Or Paste Full Script", placeholder="Paste full script here...", lines=5) | |
| generate_script_btn = gr.Button("📝 Generate Script & Load Clips") | |
| generated_script_disp = gr.Textbox(label="Generated Script", interactive=False, visible=False) | |
| clip_data_storage = gr.Textbox(visible=False) | |
| # Column 2: Clip Editor (Dynamic) | |
| with gr.Column(scale=1): | |
| gr.Markdown("## 2. Edit Clips") | |
| gr.Markdown("Modify each clip's visual prompt, narration or upload custom media.") | |
| clip_editor_container = gr.Column(visible=False) | |
| # Column 3: Settings & Output | |
| with gr.Column(scale=1): | |
| gr.Markdown("## 3. Video Settings") | |
| resolution = gr.Radio(choices=["Short (1080x1920)", "Full HD (1920x1080)"], | |
| label="Resolution", value="Full HD (1920x1080)") | |
| render_speed = gr.Dropdown(choices=["ultrafast", "veryfast", "fast", "medium", "slow", "veryslow"], | |
| label="Render Speed", value="fast") | |
| video_clip_percent = gr.Slider(0, 100, step=5, label="Video Clip Percentage", value=25) | |
| zoom_pan = gr.Checkbox(label="Add Zoom/Pan Effect (Images)", value=True) | |
| with gr.Accordion("Background Music", open=False): | |
| bg_music_file = gr.File(label="Upload Background Music (MP3)", file_types=[".mp3"]) | |
| bg_music_volume = gr.Slider(0.0, 1.0, step=0.05, label="BGM Volume", value=0.15) | |
| with gr.Accordion("Subtitle Settings", open=True): | |
| subtitle_enabled = gr.Checkbox(label="Enable Subtitles", value=True) | |
| font_dropdown = gr.Dropdown(choices=["Impact", "Arial", "Helvetica", "Times New Roman"], | |
| label="Font", value="Arial") | |
| font_size = gr.Number(label="Font Size", value=45) | |
| outline_width = gr.Number(label="Outline Width", value=2) | |
| font_color = gr.ColorPicker(label="Font Color", value="#FFFFFF") | |
| outline_color = gr.ColorPicker(label="Outline Color", value="#000000") | |
| subtitle_position = gr.Radio(choices=["center", "bottom", "top"], label="Subtitle Position", value="center") | |
| gr.Markdown("## 4. Output") | |
| generate_video_btn = gr.Button("🎬 Generate Video") | |
| video_preview = gr.Video(label="Generated Video") | |
| download_video_file = gr.File(label="Download Video", interactive=False) | |
| # Interactions | |
| generate_script_btn.click( | |
| fn=generate_script_clips, | |
| inputs=[topic_input, full_script], | |
| outputs=[generated_script_disp, clip_data_storage] | |
| ).then( | |
| fn=lambda script: gr.update(visible=True), | |
| inputs=[generated_script_disp], | |
| outputs=[generated_script_disp] | |
| ).then( | |
| fn=update_clip_editor, | |
| inputs=[clip_data_storage], | |
| outputs=[clip_editor_container] | |
| ).then( | |
| fn=lambda _: gr.update(visible=True), | |
| inputs=[clip_editor_container], | |
| outputs=[clip_editor_container] | |
| ) | |
| generate_video_btn.click( | |
| fn=generate_final_video, | |
| inputs=[topic_input, full_script, clip_data_storage, resolution, render_speed, | |
| video_clip_percent, zoom_pan, bg_music_file, bg_music_volume, | |
| subtitle_enabled, font_dropdown, font_size, outline_width, font_color, outline_color, subtitle_position], | |
| outputs=[video_preview] | |
| ) | |
| demo.launch(share=True) | |