| import requests_proxy | |
| 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 | |
| MIN_WIDTH = 1920 | |
| MIN_HEIGHT = 1080 | |
| TARGET_ASPECT_RATIO = 16/9 | |
| output_folder = "outputs" | |
| temp_dir = "temp_files" | |
| os.makedirs(output_folder, exist_ok=True) | |
| os.makedirs(temp_dir, exist_ok=True) | |
| FOLDER_ID = "12S6adpanAXjf71pKKGRRPqpzbJa5XEh3" | |
| def load_proxies(proxy_file="proxys.txt"): | |
| try: | |
| with open(proxy_file, 'r') as f: | |
| proxies = [line.strip() for line in f if line.strip()] | |
| print(f"Loaded {len(proxies)} proxies from file") | |
| return [requests_proxy.ProxyInfo(proxy=f"http://{proxy}") for proxy in proxies] | |
| except Exception as e: | |
| print(f"Error loading proxies: {e}") | |
| return [] | |
| def search_google_images(query, num_images=1): | |
| try: | |
| api_key = os.getenv('GOOGLE_API_KEY') | |
| cse_id = os.getenv('GOOGLE_CSE_ID') | |
| proxies = load_proxies() | |
| print(f"Buscando imágenes para: {query}") | |
| if not proxies: | |
| print("No proxies available, trying without proxy") | |
| with requests.Session() as session: | |
| result = session.get( | |
| f"https://www.googleapis.com/customsearch/v1", | |
| params={ | |
| "q": query, | |
| "cx": cse_id, | |
| "searchType": "image", | |
| "num": num_images * 3, | |
| "safe": 'off', | |
| "imgSize": 'LARGE', | |
| "imgType": 'imgTypeUndefined', | |
| "rights": 'cc_publicdomain|cc_attribute|cc_sharealike', | |
| "key": api_key | |
| } | |
| ).json() | |
| else: | |
| with requests_proxy.ProxyManager(proxies) as proxy_manager: | |
| result = proxy_manager.get( | |
| f"https://www.googleapis.com/customsearch/v1", | |
| params={ | |
| "q": query, | |
| "cx": cse_id, | |
| "searchType": "image", | |
| "num": num_images * 3, | |
| "safe": 'off', | |
| "imgSize": 'LARGE', | |
| "imgType": 'imgTypeUndefined', | |
| "rights": 'cc_publicdomain|cc_attribute|cc_sharealike', | |
| "key": api_key | |
| } | |
| ).json() | |
| if 'items' in result: | |
| image_urls = [] | |
| for item in result.get('items', []): | |
| if 'image' in item: | |
| width = int(item['image'].get('width', 0)) | |
| height = int(item['image'].get('height', 0)) | |
| if width >= MIN_WIDTH and height >= MIN_HEIGHT: | |
| image_urls.append(item['link']) | |
| if len(image_urls) >= num_images: | |
| break | |
| print(f"Encontradas {len(image_urls)} imágenes de tamaño adecuado") | |
| return image_urls | |
| print("No se encontraron imágenes después de probar todos los proxies") | |
| return [] | |
| except Exception as e: | |
| print(f"Error general en la búsqueda de imágenes: {str(e)}") | |
| return [] | |
| def process_image(image): | |
| try: | |
| width, height = image.size | |
| current_ratio = width / height | |
| if current_ratio > TARGET_ASPECT_RATIO: | |
| new_width = max(MIN_WIDTH, width) | |
| new_height = int(new_width / TARGET_ASPECT_RATIO) | |
| else: | |
| new_height = max(MIN_HEIGHT, height) | |
| new_width = int(new_height * TARGET_ASPECT_RATIO) | |
| image = image.resize((new_width, new_height), Image.Resampling.LANCZOS) | |
| background = Image.new('RGB', (max(new_width, MIN_WIDTH), max(new_height, MIN_HEIGHT)), 'black') | |
| offset = ((background.width - image.width) // 2, | |
| (background.height - image.height) // 2) | |
| background.paste(image, offset) | |
| return background | |
| except Exception as e: | |
| print(f"Error processing image: {e}") | |
| return None | |
| def download_image(url): | |
| proxies = load_proxies() | |
| if not proxies: | |
| proxies = [None] | |
| for proxy in proxies: | |
| try: | |
| response = requests.get(url, proxies=proxy, timeout=10) | |
| image = Image.open(BytesIO(response.content)) | |
| if image.mode in ('RGBA', 'LA') or (image.mode == 'P' and 'transparency' in image.info): | |
| background = Image.new('RGB', image.size, (0, 0, 0)) | |
| background.paste(image, mask=image.split()[-1]) | |
| image = background | |
| processed_image = process_image(image) | |
| if processed_image: | |
| return processed_image | |
| except Exception as e: | |
| print(f"Error downloading image with proxy {proxy}: {e}") | |
| continue | |
| return None | |
| def create_animated_clip(image, duration=5, zoom_factor=1.1): | |
| img_array = np.array(image) | |
| img_clip = ImageClip(img_array).set_duration(duration) | |
| if img_clip.size[0] < MIN_WIDTH or img_clip.size[1] < MIN_HEIGHT: | |
| img_clip = img_clip.resize(width=MIN_WIDTH, height=MIN_HEIGHT) | |
| 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): | |
| keyword_list = [keyword.strip() for keyword in keywords.split(",") if keyword.strip()] | |
| if not keyword_list: | |
| keyword_list = ["nature"] | |
| 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) | |
| time.sleep(1) | |
| except Exception as e: | |
| print(f"Error processing keyword '{keyword}': {e}") | |
| continue | |
| if not video_clips: | |
| return ColorClip(size=(MIN_WIDTH, MIN_HEIGHT), 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): | |
| 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): | |
| try: | |
| audio_clip = AudioFileClip(audio_file) | |
| total_duration = audio_clip.duration + 2 | |
| video_clip = video_clip.loop(duration=total_duration) | |
| video_clip = video_clip.set_duration(total_duration).fadeout(2) | |
| final_clip = video_clip.set_audio(audio_clip) | |
| 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])) | |
| output_filename = f"final_video_{int(time.time())}.mp4" | |
| output_path = os.path.join(output_folder, output_filename) | |
| final_clip.write_videofile(output_path, codec="libx264", audio_codec="aac", fps=24) | |
| 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): | |
| 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): | |
| try: | |
| 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") | |
| audio_file = asyncio.run(text_to_speech(final_text, selected_voice, rate, pitch)) | |
| if not audio_file: | |
| raise ValueError("Failed to generate audio") | |
| video_clip = concatenate_google_images(keywords, clip_duration=5, num_images_per_keyword=1) | |
| if not video_clip: | |
| raise ValueError("Failed to generate video") | |
| music_clip = None | |
| if mp3_file is not None: | |
| music_clip = adjust_background_music(video_clip.duration, mp3_file.name) | |
| 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") | |
| 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() | |
| 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", | |
| 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("") | |
| 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 | |
| ) | |
| port = int(os.getenv("PORT", 7860)) | |
| demo.launch(server_name="0.0.0.0", server_port=port, share=True, show_error=True) |