TTV / app.py
Nafiul Huda
Update app.py
a4c0061 verified
Raw
History Blame Contribute Delete
24.5 kB
# Import necessary libraries
import os
import re
import cv2
import math
import time
import random
import shutil
import io
import tempfile
import requests
import numpy as np
import soundfile as sf
import torch
from kokoro import KPipeline
from pydub import AudioSegment
from PIL import Image, ImageDraw, ImageFont
from gtts import gTTS
# MoviePy integration
import moviepy.config as mpy_config
from moviepy.editor import (
VideoFileClip, concatenate_videoclips, AudioFileClip, ImageClip,
CompositeVideoClip, CompositeAudioClip, concatenate_audioclips
)
import moviepy.video.fx.all as vfx
import gradio as gr
# Initialize Kokoro TTS pipeline (American English configuration)
try:
pipeline = KPipeline(lang_code='a')
except Exception as e:
print(f"Warning initializing Kokoro Pipeline: {e}. Fallback systems active.")
# ---------------- Global Configuration ---------------- #
PEXELS_API_KEY = os.environ.get('PEXELS_API_KEY', '')
GROQ_API_KEY = os.environ.get('GROQ_API_KEY', '')
if not PEXELS_API_KEY:
PEXELS_API_KEY = 'YOUR_PEXELS_KEY_HERE'
if not GROQ_API_KEY:
GROQ_API_KEY = 'YOUR_GROQ_KEY_HERE'
GROQ_MODEL = "llama-3.3-70b-versatile"
OUTPUT_VIDEO_FILENAME = "final_video.mp4"
USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
# Execution state primitives
selected_voice = 'am_michael'
voice_speed = 1.0
font_size = 45
video_clip_probability = 0.65
bg_music_volume = 0.08
fps = 24
preset = "veryfast"
TARGET_RESOLUTION = (1920, 1080)
CAPTION_COLOR = "white"
TEMP_FOLDER = None
VOICE_CHOICES = {
'Emma (Female)': 'af_heart', 'Bella (Female)': 'af_bella', 'Nicole (Female)': 'af_nicole',
'Aoede (Female)': 'af_aoede', 'Kore (Female)': 'af_kore', 'Sarah (Female)': 'af_sarah',
'Nova (Female)': 'af_nova', 'Sky (Female)': 'af_sky', 'Alloy (Female)': 'af_alloy',
'Jessica (Female)': 'af_jessica', 'River (Female)': 'af_river', 'Michael (Male)': 'am_michael',
'Fenrir (Male)': 'am_fenrir', 'Puck (Male)': 'am_puck', 'Echo (Male)': 'am_echo',
'Eric (Male)': 'am_eric', 'Liam (Male)': 'am_liam', 'Onyx (Male)': 'am_onyx',
'Santa (Male)': 'am_santa', 'Adam (Male)': 'am_adam', 'Emma 🇬🇧 (Female)': 'bf_emma',
'Isabella 🇬🇧 (Female)': 'bf_isabella', 'Alice 🇬🇧 (Female)': 'bf_alice', 'Lily 🇬🇧 (Female)': 'bf_lily',
'George 🇬🇧 (Male)': 'bm_george', 'Fable 🇬🇧 (Male)': 'bm_fable', 'Lewis 🇬🇧 (Male)': 'bm_lewis',
'Daniel 🇬🇧 (Male)': 'bm_daniel'
}
# ---------------- Core Support Functions ---------------- #
def check_api_keys():
if not PEXELS_API_KEY or PEXELS_API_KEY == 'YOUR_PEXELS_KEY_HERE':
return False, "PEXELS_API_KEY is not configured in environment variables."
if not GROQ_API_KEY or GROQ_API_KEY == 'YOUR_GROQ_KEY_HERE':
return False, "GROQ_API_KEY is not configured in environment variables."
return True, "API keys configured successfully."
def generate_script(user_input, tone_style):
"""Generate high-quality, humanized documentary scripts using style presets via Groq."""
headers = {
'Authorization': f'Bearer {GROQ_API_KEY}',
'Content-Type': 'application/json'
}
# Tone definition matrix map
tone_directives = {
"Funny / Humorous": "Write with sharp comedic timing, witty punchlines, and playful sarcasm. Make light of the topic like a stand-up comedian presenting a documentary.",
"Serious / Dramatic": "Write with an authoritative, compelling, and intense cinematic tone. Focus on raw emotion, gravity, and high stakes without being dry.",
"Informal / Casual": "Write like a close friend talking over coffee. Use relaxed phrasing, casual observations, and a warm, approachable delivery.",
"Educational / Insightful": "Write with curiosity and infectious enthusiasm for facts. Keep it deeply engaging, clear, intellectual, yet completely accessible."
}
selected_tone_prompt = tone_directives.get(tone_style, tone_directives["Informal / Casual"])
prompt = f"""You are an authentic human video narrator. Your task is to write an engaging video script based on the topic provided.
TONE SETTING:
{selected_tone_prompt}
CRITICAL INSTRUCTIONS FOR NATURAL HUMAN TONE:
- Avoid ALL standard robotic AI tropes, rigid filler terms, and artificial structural transitions.
- Strictly DO NOT use words like: "delve", "tapestry", "testament", "furthermore", "moreover", "in conclusion", "look no further", "nestled", "beacon", or "revolutionize".
- Write with organic variety in sentence structure. Mix short, punchy phrases with authentic commentary.
Format Requirements:
- Break the script into distinct scenes using structural brackets: [Tag].
- The Tag must be a simple 1-2 word search query suitable for finding background stock video/images (e.g., [Cat Sleeping], [Stormy Sky]).
- Directly beneath each tag, write exactly one engaging sentence (maximum 15 words) continuing the narrative stream.
- Conclude the piece with a [Subscribe] tag containing a clever, customized parting remark matching your assigned tone.
Topic: {user_input}
"""
data = {
'model': GROQ_MODEL,
'messages': [{'role': 'user', 'content': prompt}],
'temperature': 0.72,
'max_tokens': 1500
}
try:
response = requests.post(
'https://api.groq.com/openai/v1/chat/completions',
headers=headers,
json=data,
timeout=25
)
if response.status_code == 200:
return response.json()['choices'][0]['message']['content'].strip()
else:
print(f"Groq API Error {response.status_code}: {response.text}")
return None
except Exception as e:
print(f"Network processing exception during script generation: {str(e)}")
return None
def parse_script(script_text):
if not script_text:
return []
pattern = r'\[(.*?)\]\s*([^\[]+)'
matches = re.findall(pattern, script_text)
elements = []
for tag, narration in matches:
clean_tag = tag.strip()
clean_narration = re.sub(r'\s+', ' ', narration.strip())
if not clean_tag or not clean_narration:
continue
elements.append({"type": "media", "prompt": clean_tag})
words = clean_narration.split()
calculated_duration = max(3.5, len(words) * 0.45)
elements.append({
"type": "tts",
"text": clean_narration,
"duration": calculated_duration
})
return elements
def search_pexels_videos(query, pexels_api_key):
headers = {'Authorization': pexels_api_key}
url = "https://api.pexels.com/videos/search"
params = {"query": query, "per_page": 8, "orientation": "landscape"}
try:
response = requests.get(url, headers=headers, params=params, timeout=12)
if response.status_code == 200:
videos = response.json().get("videos", [])
if videos:
selected_video = random.choice(videos)
video_files = selected_video.get("video_files", [])
for file in video_files:
if file.get("quality") == "hd" and file.get("width", 0) >= 1280:
return file.get("link")
if video_files:
return video_files[0].get("link")
except Exception as e:
print(f"Pexels video fetch exception: {e}")
return None
def search_pexels_images(query, pexels_api_key):
headers = {'Authorization': pexels_api_key}
url = "https://api.pexels.com/v1/search"
params = {"query": query, "per_page": 8, "orientation": "landscape"}
try:
response = requests.get(url, headers=headers, params=params, timeout=12)
if response.status_code == 200:
photos = response.json().get("photos", [])
if photos:
return random.choice(photos).get("src", {}).get("large2x")
except Exception as e:
print(f"Pexels image search exception: {e}")
return None
def download_asset_file(url, local_path):
try:
headers = {"User-Agent": USER_AGENT}
with requests.get(url, headers=headers, stream=True, timeout=20) as r:
r.raise_for_status()
with open(local_path, 'wb') as f:
for chunk in r.iter_content(chunk_size=16384):
if chunk: f.write(chunk)
return local_path
except Exception as e:
print(f"Error handling asset download: {e}")
if os.path.exists(local_path): os.remove(local_path)
return None
def generate_solid_fallback(prompt, target_res):
w, h = target_res
random.seed(prompt)
base_color = (random.randint(15, 35), random.randint(20, 40), random.randint(30, 55))
img = Image.new('RGB', (w, h), base_color)
fallback_path = os.path.join(TEMP_FOLDER, f"fallback_{int(time.time())}_{random.randint(0,99)}.jpg")
img.save(fallback_path, quality=90)
return fallback_path
def generate_media_asset(prompt):
safe_name = re.sub(r'[^\w\s-]', '', prompt).strip().replace(' ', '_')
if random.random() < video_clip_probability:
video_url = search_pexels_videos(prompt, PEXELS_API_KEY)
if video_url:
local_video_path = os.path.join(TEMP_FOLDER, f"vid_{safe_name}_{int(time.time())}.mp4")
if download_asset_file(video_url, local_video_path):
return {"path": local_video_path, "type": "video"}
img_url = search_pexels_images(prompt, PEXELS_API_KEY)
if img_url:
local_img_path = os.path.join(TEMP_FOLDER, f"img_{safe_name}_{int(time.time())}.jpg")
if download_asset_file(img_url, local_img_path):
return {"path": local_img_path, "type": "image"}
return {"path": generate_solid_fallback(prompt, TARGET_RESOLUTION), "type": "image"}
def generate_tts_audio(text):
safe_name = re.sub(r'[^\w\s-]', '', text[:10]).strip().replace(' ', '_')
output_audio_path = os.path.join(TEMP_FOLDER, f"tts_{safe_name}_{int(time.time())}.wav")
try:
generator = pipeline(text, voice=selected_voice, speed=voice_speed, split_pattern=r'\n+')
audio_blocks = [audio for _, _, audio in generator]
if audio_blocks:
merged_audio = np.concatenate(audio_blocks) if len(audio_blocks) > 1 else audio_blocks[0]
sf.write(output_audio_path, merged_audio, 24000)
return output_audio_path
except Exception as e:
print(f"Kokoro engine bypass, trying gTTS fallback: {e}")
try:
tts = gTTS(text=text, lang='en', slow=False)
temp_mp3 = os.path.join(TEMP_FOLDER, f"gtts_{int(time.time())}.mp3")
tts.save(temp_mp3)
AudioSegment.from_mp3(temp_mp3).export(output_audio_path, format="wav")
if os.path.exists(temp_mp3): os.remove(temp_mp3)
return output_audio_path
except Exception as fail_err:
print(f"Critical synthesis failure. Using silence audio pad: {fail_err}")
duration_sec = max(3, len(text.split()) * 0.5)
sf.write(output_audio_path, np.zeros(int(duration_sec * 24000), dtype=np.float32), 24000)
return output_audio_path
def build_wrapped_subtitle_layer(text, canvas_resolution, font_size_target):
width, height = canvas_resolution
max_text_boundary_width = int(width * 0.85)
font_engine = None
font_options = ["/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf", "C:\\Windows\\Fonts\\arialbd.ttf", "/usr/share/fonts/Arial.ttf"]
for path in font_options:
if os.path.exists(path):
try:
font_engine = ImageFont.truetype(path, font_size_target)
break
except: pass
if font_engine is None: font_engine = ImageFont.load_default()
words = text.split()
compiled_lines, current_line_build = [], []
measurement_canvas = Image.new('RGBA', (1, 1))
draw_inspector = ImageDraw.Draw(measurement_canvas)
for word in words:
test_string = " ".join(current_line_build + [word])
try:
box = draw_inspector.textbbox((0, 0), test_string, font=font_engine)
calculated_w = box[2] - box[0]
except: calculated_w = len(test_string) * (font_size_target * 0.55)
if calculated_w <= max_text_boundary_width:
current_line_build.append(word)
else:
if current_line_build: compiled_lines.append(" ".join(current_line_build))
current_line_build = [word]
if current_line_build: compiled_lines.append(" ".join(current_line_build))
line_stride = font_size_target + 12
computed_box_height = (len(compiled_lines) * line_stride) + 40
subtitle_strip_canvas = Image.new('RGBA', (width, computed_box_height), (0, 0, 0, 0))
context_drawer = ImageDraw.Draw(subtitle_strip_canvas)
for idx, line in enumerate(compiled_lines):
try:
box = context_drawer.textbbox((0, 0), line, font=font_engine)
w = box[2] - box[0]
except: w = len(line) * (font_size_target * 0.55)
target_x = (width - w) // 2
target_y = 20 + (idx * line_stride)
for dx, dy in [(-2, -2), (-2, 2), (2, -2), (2, 2), (0, -2), (0, 2), (-2, 0), (2, 0)]:
context_drawer.text((target_x + dx, target_y + dy), line, font=font_engine, fill=(0, 0, 0, 225))
context_drawer.text((target_x, target_y), line, font=font_engine, fill=(255, 255, 255, 255))
local_subtitle_png_path = os.path.join(TEMP_FOLDER, f"sub_{int(time.time())}_{random.randint(100,999)}.png")
subtitle_strip_canvas.save(local_subtitle_png_path)
return local_subtitle_png_path
def resize_and_crop_to_fill(clip, target_resolution):
tw, th = target_resolution
clip_w, clip_h = clip.w, clip.h
aspect_target = tw / th
aspect_clip = clip_w / clip_h
if aspect_clip > aspect_target:
scale_factor = th / clip_h
resized_clip = clip.resize(newsize=(int(clip_w * scale_factor), th))
return resized_clip.crop(x1=(resized_clip.w - tw) / 2, x2=((resized_clip.w - tw) / 2) + tw, y1=0, y2=th)
else:
scale_factor = tw / clip_w
resized_clip = clip.resize(newsize=(tw, int(clip_h * scale_factor)))
return resized_clip.crop(x1=0, x2=tw, y1=(resized_clip.h - th) / 2, y2=((resized_clip.h - th) / 2) + th)
def apply_cinematic_motion_effect(clip, target_resolution):
tw, th = target_resolution
base_clip = clip.resize(newsize=(int(tw * 1.2), int(th * 1.2)))
max_delta_x = base_clip.w - tw
max_delta_y = base_clip.h - th
motion_style = random.choice(["zoom-in", "zoom-out", "pan-right", "pan-left"])
clip_duration = clip.duration
def frame_transformation_matrix(get_frame, t):
frame = get_frame(t)
progress = (t / clip_duration) if clip_duration > 0 else 0
smooth_step = 0.5 * (1.0 - math.cos(math.pi * progress))
if motion_style == "zoom-in":
sc = 1.0 + (0.12 * smooth_step)
rf = cv2.resize(frame, (int(tw * sc), int(th * sc)), interpolation=cv2.INTER_LINEAR)
return rf[((rf.shape[0]-th)//2):((rf.shape[0]-th)//2)+th, ((rf.shape[1]-tw)//2):((rf.shape[1]-tw)//2)+tw]
elif motion_style == "zoom-out":
sc = 1.15 - (0.12 * smooth_step)
rf = cv2.resize(frame, (int(tw * sc), int(th * sc)), interpolation=cv2.INTER_LINEAR)
return rf[((rf.shape[0]-th)//2):((rf.shape[0]-th)//2)+th, ((rf.shape[1]-tw)//2):((rf.shape[1]-tw)//2)+tw]
elif motion_style == "pan-right":
x_offset = int(max_delta_x * smooth_step)
return frame[max_delta_y//2:(max_delta_y//2)+th, x_offset:x_offset+tw]
else:
x_offset = int(max_delta_x * (1.0 - smooth_step))
return frame[max_delta_y//2:(max_delta_y//2)+th, x_offset:x_offset+tw]
return base_clip.fl(frame_transformation_matrix)
def compile_individual_segment(media_path, asset_type, audio_path, script_text, segment_id):
video_sequence_clip, voice_track_clip, subtitle_overlay_clip = None, None, None
try:
if not os.path.exists(media_path) or not os.path.exists(audio_path): return None
voice_track_clip = AudioFileClip(audio_path).fx(vfx.audio_fadeout, 0.15)
allocated_duration = voice_track_clip.duration + 0.35
if asset_type == "video":
raw_video = VideoFileClip(media_path, audio=False)
normalized_video = resize_and_crop_to_fill(raw_video, TARGET_RESOLUTION)
video_sequence_clip = normalized_video.fx(vfx.loop, duration=allocated_duration) if normalized_video.duration < allocated_duration else normalized_video.subclip(0, allocated_duration)
raw_video.close()
else:
base_image = ImageClip(media_path).set_duration(allocated_duration)
video_sequence_clip = apply_cinematic_motion_effect(base_image, TARGET_RESOLUTION).fx(vfx.fadein, 0.25).fx(vfx.fadeout, 0.25)
base_image.close()
video_sequence_clip = video_sequence_clip.set_audio(voice_track_clip)
if CAPTION_COLOR == "white" and script_text:
sub_file = build_wrapped_subtitle_layer(script_text, TARGET_RESOLUTION, font_size)
subtitle_overlay_clip = ImageClip(sub_file).set_duration(allocated_duration).set_position(('center', int(TARGET_RESOLUTION[1] * 0.78)))
return CompositeVideoClip([video_sequence_clip, subtitle_overlay_clip], size=TARGET_RESOLUTION)
return video_sequence_clip
except Exception as err:
print(f"Segment compilation error #{segment_id}: {err}")
try:
for c in [video_sequence_clip, voice_track_clip, subtitle_overlay_clip]:
if c: c.close()
except: pass
return None
# ---------------- Pipeline Processing Entry ---------------- #
def generate_video(user_input, tone_style, resolution_mode, enable_captions):
global TARGET_RESOLUTION, CAPTION_COLOR, TEMP_FOLDER
api_ok, check_msg = check_api_keys()
if not api_ok: return None, f"Configuration Error: {check_msg}"
TARGET_RESOLUTION = (1920, 1080) if resolution_mode == "Full (16:9)" else (1080, 1920)
CAPTION_COLOR = "white" if enable_captions == "Yes" else "transparent"
TEMP_FOLDER = tempfile.mkdtemp()
master_clip_list = []
final_output_composition = None
try:
print(f"Requesting '{tone_style}' narrative track from Groq AI engines...")
generated_script_text = generate_script(user_input, tone_style)
if not generated_script_text: return None, "Failed to retrieve script from Groq endpoints."
script_execution_steps = parse_script(generated_script_text)
if not script_execution_steps: return None, "Parser layout error. Check script tags."
paired_segments = [(script_execution_steps[i], script_execution_steps[i+1]) for i in range(0, len(script_execution_steps), 2) if i + 1 < len(script_execution_steps)]
for idx, (media_cue, audio_cue) in enumerate(paired_segments):
media_asset = generate_media_asset(media_cue['prompt'])
audio_track_file = generate_tts_audio(audio_cue['text'])
constructed_segment = compile_individual_segment(media_asset['path'], media_asset['type'], audio_track_file, audio_cue['text'], idx)
if constructed_segment: master_clip_list.append(constructed_segment)
if not master_clip_list: return None, "Pipeline Error: Could not render individual timeline tracks."
final_output_composition = concatenate_videoclips(master_clip_list, method="compose")
bg_music_source = "music.mp3"
if os.path.exists(bg_music_source):
try:
ambient_music_clip = AudioFileClip(bg_music_source)
if ambient_music_clip.duration < final_output_composition.duration:
ambient_music_clip = concatenate_audioclips([ambient_music_clip] * math.ceil(final_output_composition.duration / ambient_music_clip.duration))
ambient_music_clip = ambient_music_clip.subclip(0, final_output_composition.duration).fx(vfx.volumex, bg_music_volume)
final_output_composition = final_output_composition.set_audio(CompositeAudioClip([final_output_composition.audio, ambient_music_clip]))
except Exception as music_err: print(f"Background audio mixing bypassed: {music_err}")
final_output_composition.write_videofile(OUTPUT_VIDEO_FILENAME, codec='libx264', audio_codec='aac', fps=fps, preset=preset, threads=4, logger=None)
return OUTPUT_VIDEO_FILENAME, "Cinematic video generation completed successfully!"
except Exception as pipeline_fault:
return None, f"Execution Failure: {str(pipeline_fault)}"
finally:
try:
if final_output_composition: final_output_composition.close()
for clip in master_clip_list:
if clip: clip.close()
except: pass
time.sleep(1.0)
if TEMP_FOLDER and os.path.exists(TEMP_FOLDER):
try: shutil.rmtree(TEMP_FOLDER)
except: pass
# ---------------- Gradio Interface Binding Maps ---------------- #
def UI_interaction_bridge(prompt, tone, res_mode, caption_flag, music_upload, voice, v_prob, music_vol, frames_per_sec, speed_preset, tts_speed, size_font):
global selected_voice, voice_speed, font_size, video_clip_probability, bg_music_volume, fps, preset
selected_voice = VOICE_CHOICES.get(voice, 'am_michael')
voice_speed, font_size, video_clip_probability, bg_music_volume, fps, preset = tts_speed, size_font, v_prob / 100.0, music_vol, frames_per_sec, speed_preset
if music_upload is not None:
try: shutil.copy(music_upload.name, "music.mp3")
except Exception as e: print(f"Error mounting audio file: {e}")
return generate_video(prompt, tone, res_mode, caption_flag)
with gr.Blocks(title="AI Cinematic Documentary Generator") as app_interface:
gr.Markdown("# 🎬 AI Cinematic Documentary Video Generator")
with gr.Row():
with gr.Column(scale=1):
user_prompt_input = gr.Textbox(label="Documentary Video Concept / Prompt", placeholder="Ex: The secret life of honeybees...", lines=3)
ui_tone = gr.Dropdown(choices=["Funny / Humorous", "Serious / Dramatic", "Informal / Casual", "Educational / Insightful"], value="Informal / Casual", label="Narrator Presentation Tone Preset")
with gr.Row():
ui_resolution = gr.Radio(["Full (16:9)", "Shorts (9:16)"], label="Video Dimensions", value="Full (16:9)")
ui_captions = gr.Radio(["Yes", "No"], label="Render Captions", value="Yes")
ui_audio_upload = gr.File(label="Optional Background Music (MP3)", file_types=[".mp3"])
with gr.Accordion("Advanced Cinema Tuning Configuration", open=False):
ui_voice_selection = gr.Dropdown(choices=list(VOICE_CHOICES.keys()), label="Voice Model", value="Michael (Male)")
ui_video_probability = gr.Slider(0, 100, value=70, step=5, label="Video vs Photo Ratio (%)")
ui_music_level = gr.Slider(0.00, 0.40, value=0.06, step=0.01, label="Music Mix Volume")
ui_fps_target = gr.Slider(15, 60, value=24, step=1, label="Target Output FPS")
ui_encoding_speed = gr.Dropdown(choices=["ultrafast", "superfast", "veryfast", "medium", "slow"], value="veryfast", label="FFmpeg Speed Preset")
ui_voice_tempo = gr.Slider(0.6, 1.4, value=1.05, step=0.05, label="Voice Tempo Scale")
ui_caption_font_size = gr.Slider(20, 90, value=48, step=2, label="Caption Size")
trigger_generation_button = gr.Button("🎬 Render Cinematic Video Sequence", variant="primary")
with gr.Column(scale=1):
video_output_display = gr.Video(label="Final Render Preview")
pipeline_status_log = gr.Textbox(label="System Operational Pipeline Logs", interactive=False)
trigger_generation_button.click(
fn=UI_interaction_bridge,
inputs=[user_prompt_input, ui_tone, ui_resolution, ui_captions, ui_audio_upload, ui_voice_selection, ui_video_probability, ui_music_level, ui_fps_target, ui_encoding_speed, ui_voice_tempo, ui_caption_font_size],
outputs=[video_output_display, pipeline_status_log]
)
if __name__ == "__main__":
app_interface.launch()