import os import streamlit as st import subprocess import replicate import openai import requests from PIL import Image import tempfile import base64 from dotenv import load_dotenv from io import BytesIO from openai import OpenAI import re load_dotenv() OpenAI.api_key = os.getenv("OPENAI_API_KEY") if not OpenAI.api_key: raise ValueError("The OpenAI API key must be set in the OPENAI_API_KEY environment variable.") client = OpenAI() def execute_ffmpeg_command(command): try: result = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE) if result.returncode == 0: print("FFmpeg command executed successfully.") return result.stdout, result.stderr else: print("Error executing FFmpeg command:") return None, result.stderr except Exception as e: print("An error occurred during FFmpeg execution:") return None, str(e) def execute_fmpeg_command(command): try: result = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True) return result.stdout # Return just the stdout part, not a tuple except subprocess.CalledProcessError as e: print(f"FFmpeg command failed with error: {e.stderr.decode()}") return None def search_keyword(keyword, frame_texts): return [index for index, text in st.session_state.frame_texts.items() if keyword.lower() in text.lower()] frame_numbers = [] # Function to generate description for video frames def generate_description(base64_frames): try: prompt_messages = [ { "role": "user", "content": [ " Find the most interesting / impactful portions of a video. The output will be targeted towards social media (like TikTok or Reels) or to news broadcasts. For the provided frames return the most interesting / impactful frames that will hold the interest of an audience and also describe why you chose it. I am trying to fill these frames for a TikTok video. Hence while selecting the frames keep that in mind. You do not have to give me the script of the Tiktok vfideo. Just return the most interesting frames in a sequence that will come for a 10 second tiktok video. List all frame numbers separated by commas at the end like this for eg, Frames : 1,2,4,7,9", *map(lambda x: {"image": x, "resize": 428}, base64_frames), ], }, ] response = client.chat.completions.create( model="gpt-4-vision-preview", messages=prompt_messages, max_tokens=3000, ) description = response.choices[0].message.content # Use regular expression to find frame numbers frame_numbers = re.findall(r'Frames\s*:\s*(\d+(?:,\s*\d+)*)', response.choices[0].message.content) # Convert the string of numbers into a list of integers if frame_numbers: frame_numbers = [int(num) for num in frame_numbers[0].split(',')] else: frame_numbers = [] print("Frame numbers to extract:", frame_numbers) return description, frame_numbers except Exception as e: print(f"Error in generate_description: {e}") return None, [] def generate_video(prompt, num_frames): # Run the text-to-video generation job output = replicate.run( "cjwbw/damo-text-to-video:1e205ea73084bd17a0a3b43396e49ba0d6bc2e754e9283b2df49fad2dcf95755", input={"prompt": prompt, "num_frames": num_frames} ) # Print the output URL st.success(f"Video generation successful! Output URL: {output}") return output def overlay_text(video_path, dynamic_text, text_effect): # Define the text filter based on the selected effect if text_effect == "Vertical scroll": text_filter = f"drawtext=text='{dynamic_text}':fontsize=24:fontfile=Opensans.ttf:fontcolor=white:box=1:boxcolor=black@0.5:boxborderw=5:x=(w-text_w)/2:y='if(lt(mod(t,10),5*(n-1)/10),(h-text_h)/2+((h+text_h)/10)*mod(t,5), (h-text_h)/2+((h+text_h)/10)*(1-(mod(t,10)/10)))':enable='between(t,0,10*(n-1)/10)" elif text_effect == "typing": text_filter = f"drawtext=text='{dynamic_text}':subtitles=typewriter.ass:force_style='FontName=Ubuntu Mono,FontSize=100,PrimaryColour=&H00FFFFFF&'" elif text_effect == "Horizontal scroll": text_filter = f"drawtext=text='{dynamic_text}':fontsize=24:fontfile=OpenSans-Regular.ttf:fontcolor=white:box=1:boxcolor=black@0.5:boxborderw=5:x=(w-text_w)/2:y=(h-text_h)/2:enable='between(t,0,10*(n-1)/10)':x='if(lt(t,10*(n-1)/10),(w-text_w)/2-((w+text_w)/10)*mod(t,10),NAN)'" else: # Handle the case where none of the conditions are met print("Invalid text effect selected.") return None # Print the FFmpeg command ffmpeg_command = [ "ffmpeg", "-i", "uploaded_video.mp4", "-vf", text_filter, "-c:a", "copy", "-y", "text_overlay_video.mp4" ] print("FFmpeg Command:", " ".join(ffmpeg_command)) # Run FFmpeg command to overlay text onto the video with the selected effect result = subprocess.run(ffmpeg_command, capture_output=True, text=True) # Check if the process was successful if result.returncode == 0: # Print the standard output of the command print("FFmpeg output:", result.stdout) else: # Print error message if the process failed print("Error running FFmpeg command:", result.stderr) return None # Example usage overlay_text("input_video.mp4", "Dynamic Text", "Vertical scroll") def text_to_video_section(): # Set your Replicate API token apikey=os.environ["REPLICATE_API_TOKEN"] st.title("Text-to-Video Generation") # Input prompt from user prompt = st.text_input("Enter your prompt:", "batman riding a horse") # Number of frames input from user num_frames = st.slider("Select number of frames:", min_value=1, max_value=100, value=50) # Button to trigger text-to-video generation if st.button("Generate Video"): video_path = generate_video(prompt, num_frames) st.video(video_path) # Input field for dynamic text dynamic_text = st.text_input("Enter dynamic text:", "Your dynamic text here") # Dropdown for selecting text effects text_effects = ["None", "Vertical scroll", "Typing", "Horizontal scroll"] selected_effect = st.selectbox("Select text effect:", text_effects) # Button to overlay text onto the video if st.button("Add Text"): if not os.path.exists("uploaded_video.mp4"): st.error("Please upload a video first.") else: if selected_effect != "None": # Convert selected effect to lowercase selected_effect_lower = selected_effect.lower() result_video = overlay_text("uploaded_video.mp4", dynamic_text, selected_effect_lower) if result_video: st.video(result_video) else: st.error("Please select a text effect.") def extract_keywords(article): prompt = f"Read the article provided below ,pick out the 5 important keywords ,add .jpg at the end of the keywords. Do not provide any additional text or explanations. Article: {article}" completions = client.chat.completions.create( model="gpt-4-1106-preview", messages=[ {"role": "user", "content": prompt} ], max_tokens=200, n=1, stop=None, temperature=0.0, ) # Extract keywords from the OpenAI response keywords =completions.choices[0].message.content return keywords def fetch_images(query): api_key = os.getenv("serp_key") # Replace "YOUR_API_KEY" with your actual API key endpoint = f"https://serpapi.com/search?engine=google_images&q={query}&api_key={api_key}" try: response = requests.get(endpoint) data = response.json() print("API Response:", data) image_urls = [result['original'] for result in data['images_results']] return image_urls[:10] # Return only the first 5 images except Exception as e: st.error(f"Error fetching images: {e}") return [] def resize_images(image_files): resized_image_files = [] for image_file in image_files: try: with Image.open(image_file) as img: # Convert image to RGB mode if it has an alpha channel if img.mode == 'RGBA': img = img.convert('RGB') # Ensure width and height are divisible by 2 width = img.width - (img.width % 2) height = img.height - (img.height % 2) resized_img = img.resize((width, height)) resized_image_file = f"{image_file.split('.')[0]}_resized.jpg" resized_img.save(resized_image_file) resized_image_files.append(resized_image_file) except (OSError) as e: st.warning(f"Skipping image {image_file} as it cannot be identified.") continue return resized_image_files def create_video_slideshow(image_urls): # Create temporary directory to store image files temp_dir = "temp_images" os.makedirs(temp_dir, exist_ok=True) # Download and save images image_files = [] for i, image_url in enumerate(image_urls): image_path = os.path.join(temp_dir, f"image_{i}.jpg") with open(image_path, 'wb') as f: response = requests.get(image_url) f.write(response.content) image_files.append(image_path) # Resize images resized_image_files = resize_images(image_files) # Run FFmpeg command to create video slideshow output_video_path = "slideshow_video.mp4" subprocess.run([ "ffmpeg", "-y", "-framerate", "1", "-i", os.path.join(temp_dir, "image_%d.jpg"), '-c:v', 'libx264','-r', '30', output_video_path ]) # Cleanup temporary directory for image_file in image_files: os.remove(image_file) for resized_image_file in resized_image_files: os.remove(resized_image_file) os.rmdir(temp_dir) return output_video_path def add_text_to_video(input_video_path, output_video_path, text_input, text_animation): # Define text animation filter based on dropdown selection if text_animation == "fade_in_out": text_animation_filter = f"drawtext=text='{text_input}':fontsize=24:fontcolor=darkslategray:fontfile=Opensans.ttf:box=1:boxcolor=white@0.8:boxborderw=5:x=w+tw-55*t:y=h-line_h-20:enable='between(t,0,10*(n-1)/10)':x='if(lt(t,10*(n-1)/10),(w-text_w)/2+(w/10)*mod(t,10),NAN)', drawtext=textfile=latest.txt:fontsize=24:fontcolor=white:fontfile=Opensans.ttf:y=h-line_h-20:x=13:box=1:boxcolor=darkorange:boxborderw=8'" elif text_animation == "slide_from_left": text_animation_filter = "split[text][tmp];[tmp]crop=w='min(iw\\,iw*max(1,(iw/2-2*t)/iw)):h='min(ih\\,ih*max(1,(ih/2-2*t)/ih)):x=-100+t*300:y=0[tleft];[text]crop=w='min(iw\\,iw*max(1,(iw/2-2*t)/iw)):h='min(ih\\,ih*max(1,(ih/2-2*t)/ih)):x=100-t*300:y=0[tright];[tmp][tleft]overlay=x='min(0,-100+t*300)':y=0[tmp];[tmp][tright]overlay=x='min(0,100-t*300)':y=0" else: text_animation_filter = "" # No animation # Check if text_input is empty or text_animation is "None" if not text_input or text_animation == "None": # Return the input video path without any modifications return input_video_path print("text filter:",text_animation_filter) # Run ffmpeg command to overlay text onto the video with animation cmd = [ "ffmpeg","-y", "-i", input_video_path, "-vf", text_animation_filter, "-c:a", "copy", output_video_path ] print(" ".join(cmd)) subprocess.run(cmd) return output_video_path def image_to_video_section(): st.title("Image-to-Video Generation") # Multi-input box for entering the article article = st.text_area("Enter the article:", "Your article here") # Define video filter options video_filter_options=["None","Vintage warm","Grayscale","Invert","Sepia"] # Add text overlay options outside of the button block text_input = st.text_input("Enter text to overlay") text_animation_options = ["None", "fade_in_out", "Horizontal scroll"] text_animation = st.selectbox("Select text animation", text_animation_options) print("text applied:",text_input) # Select video filter video_filter = st.selectbox("Select video filter", video_filter_options) # Button to trigger keyword extraction and video generation if st.button("Generate Video"): if article.strip() != "": # Extract keywords using OpenAI API keywords = extract_keywords(article) st.write(keywords) # Fetch images from Google Images based on keywords image_urls = fetch_images(" ".join(keywords.split()[:10])) # Fetch images based on the first 5 keywords if image_urls: st.success("Images fetched successfully!") # Display the first 5 images for image_url in image_urls: st.image(image_url, caption='Image from Google', use_column_width=True) # Create video slideshow from fetched images video_path = create_video_slideshow(image_urls) # Display generated video st.video(video_path) #video filter if video_filter=="Vintage warm": video_filt="eq=brightness=0.05:saturation=1.5" elif video_filter=="Grayscale": video_filt="hue=s=0" elif video_filter=="Invert": video_filt="lutrgb='r=negval:g=negval:b=negval'" elif video_filter=="Sepia": video_filt="colorchannelmixer=.393:.769:.189:0:.349:.686:.168:0:.272:.534:.131" else: video_filt="" #no filter if not video_filter=="None": cmdvid=["ffmpeg","-y","-i", video_path,"-vf",video_filt,"-c:a", "copy","videofilter.mp4"] print(" ".join(cmdvid)) subprocess.run(cmdvid) st.video("videofilter.mp4") #text filter if text_animation == "fade_in_out": text_animation_filter = f"drawtext=text='{text_input}':fontsize=24:fontcolor=darkslategray:fontfile=Opensans.ttf:box=1:boxcolor=white@0.8:boxborderw=5:x=w+tw-55*t:y=h-line_h-20:enable='between(t,0,10*(n-1)/10)':x='if(lt(t,10*(n-1)/10),(w-text_w)/2+(w/10)*mod(t,10),NAN)', drawtext=textfile=latest.txt:fontsize=24:fontcolor=white:fontfile=Opensans.ttf:y=h-line_h-20:x=13:box=1:boxcolor=darkorange:boxborderw=8'" elif text_animation == "Horizontal scroll": text_animation_filter =f"drawtext=text='{text_input}':fontsize=24:fontcolor=darkslategray:fontfile=Opensans.ttf:box=1:boxcolor=white@0.8:boxborderw=5:x=w+tw-55*t:y=h-line_h-20:enable='between(t,0,10*(n-1)/10)':x='if(lt(t,10*(n-1)/10),(w-text_w)/2+(w/10)*mod(t,10),NAN)', drawtext=textfile=latest.txt:fontsize=24:fontcolor=white:fontfile=Opensans.ttf:y=h-line_h-20:x=13:box=1:boxcolor=darkorange:boxborderw=8'" else: text_animation_filter = "" # No animation # Check if text_input is empty or text_animation is "None" if not text_input or text_animation == "None": # Return the input video path without any modifications return video_path print("text filter:",text_animation_filter) # Run ffmpeg command to overlay text onto the video with animation cmd = [ "ffmpeg","-y", "-i", video_path, "-vf", text_animation_filter, "-c:a", "copy", "output_video.mp4" ] print(" ".join(cmd)) subprocess.run(cmd) #return output_video_path # Add text overlay to the generated video st.video("output_video.mp4") else: st.error("No images found for the given keywords.") else: st.warning("Please enter an article before generating the video.") def frame_extraction(): st.title("Insightly Video") # stream_url = st.text_input("Enter the live stream URL (YouTube, Twitch, etc.):") # keyword = st.text_input("Enter a keyword to filter the frames (optional):") uploaded_video = st.file_uploader("Or upload a video file (MP4):", type=["mp4"]) # Slider to select the number of seconds for extraction seconds = st.slider("Select the number of seconds for extraction:", min_value=1, max_value=60, value=10) extract_frames_button = st.button("Extract Frames") if uploaded_video is not None and extract_frames_button: with tempfile.NamedTemporaryFile(delete=False, suffix=".mp4") as tmpfile: tmpfile.write(uploaded_video.getvalue()) video_file_path = tmpfile.name ffmpeg_command = [ 'ffmpeg', # Input stream URL '-i', video_file_path, '-t', str(seconds), # Duration to process the input (selected seconds) '-vf', 'fps=1', # Extract one frame per second '-f', 'image2pipe', # Output format as image2pipe '-c:v', 'mjpeg', # Codec for output video '-an', # No audio '-' ] ffmpeg_output = execute_fmpeg_command(ffmpeg_command) if ffmpeg_output: st.write("Frames Extracted:") frame_bytes_list = ffmpeg_output.split(b'\xff\xd8')[1:] # Correct splitting for JPEG frames n_frames = len(frame_bytes_list) base64_frames = [base64.b64encode(b'\xff\xd8' + frame_bytes).decode('utf-8') for frame_bytes in frame_bytes_list] frame_dict = {} for idx, frame_base64 in enumerate(base64_frames): col1, col2 = st.columns([3, 2]) with col1: frame_bytes = base64.b64decode(frame_base64) frame_dict[idx + 1] = frame_bytes st.image(Image.open(BytesIO(frame_bytes)), caption=f'Frame {idx + 1}', use_column_width=True) with col2: pass # Here, you might want to process combined_analysis_results to summarize or just display them # Extract audio audio_command = [ 'ffmpeg', '-i', video_file_path, '-t', str(seconds), '-vf', 'fps=1', # Input stream URL '-vn', # Ignore the video for the audio output '-acodec', 'libmp3lame', # Set the audio codec to MP3 # Duration for the audio extraction (selected seconds) '-f', 'mp3', # Output format as MP3 '-' ] audio_output, _ = execute_ffmpeg_command(audio_command) st.write("Extracted Audio:") audio_tempfile = tempfile.NamedTemporaryFile(delete=False, suffix=".mp3") audio_tempfile.write(audio_output) audio_tempfile.close() st.audio(audio_output, format='audio/mpeg', start_time=0) # Get consolidated description for all frames if ffmpeg_output: description, frame_numbers = generate_description(base64_frames) if description: st.header("Frame Description:") st.write(description) else: st.write("Failed to generate description.") if frame_numbers: print("Frame numbers to extract:", frame_numbers) # Check frame numbers # Create a mapping from original frame numbers to sequential numbers frame_mapping = {} new_frame_numbers = [] for idx, frame_number in enumerate(sorted(frame_numbers)): frame_mapping[frame_number] = idx + 1 new_frame_numbers.append(idx + 1) print("New frame numbers:", new_frame_numbers) print("Frame mapping:", frame_mapping) # Create a temporary directory to store images with tempfile.TemporaryDirectory() as temp_dir: image_paths = [] for frame_number in frame_numbers: if frame_number in frame_dict: frame_path = os.path.join(temp_dir, f'frame_{frame_mapping[frame_number]:03}.jpg') # Updated file naming image_paths.append(frame_path) with open(frame_path, 'wb') as f: f.write(frame_dict[frame_number]) # Once all selected frames are saved as images, create a video from them using FFmpeg video_output_path = os.path.join(temp_dir, 'output.mp4') framerate = 1 # Adjust framerate based on the number of frames ffmpeg_command = [ 'ffmpeg', '-framerate', str(framerate), # Set framerate based on the number of frames '-i', os.path.join(temp_dir, 'frame_%03d.jpg'), # Input pattern for all frame files '-c:v', 'libx264', '-pix_fmt', 'yuv420p', video_output_path ] print("FFmpeg command:", ' '.join(ffmpeg_command)) # Debug FFmpeg command subprocess.run(ffmpeg_command, stdout=subprocess.PIPE, stderr=subprocess.PIPE) # Display or provide a download link for the created video st.header("Final Video") st.video(video_output_path) else: st.write(" ") def main(): # st.title("Video Uploader and Player") # uploaded_file = st.file_uploader("Upload a video", type=["mp4", "mov"]) # if uploaded_file is not None: # Save the uploaded video to disk # with open("uploaded_video.mp4", "wb") as f: # f.write(uploaded_file.getbuffer()) # st.success("Video uploaded successfully!") # Display the uploaded video # st.video("uploaded_video.mp4") # Add accordion menu for text to video and image to video sections menu_selection = st.sidebar.selectbox("Select:", ["Text to video", "Image to video","Frame Extraction"]) if menu_selection == "Text to video": text_to_video_section() elif menu_selection == "Image to video": image_to_video_section() elif menu_selection == "Frame Extraction": frame_extraction() if __name__ == "__main__": main()