import gradio as gr from google import genai from google.genai import types from PIL import Image from io import BytesIO import base64 import os import json import random import urllib.parse import time # Check Gradio version required_version = "4.44.0" current_version = gr.__version__ if current_version < required_version: raise ValueError(f"Gradio version {current_version} is outdated. Please upgrade to {required_version} or later using 'pip install gradio=={required_version}'.") # Initialize the Google Generative AI client with the API key from environment variables try: api_key = os.environ['GEMINI_API_KEY'] except KeyError: raise ValueError("Please set the GEMINI_API_KEY environment variable.") client = genai.Client(api_key=api_key) # Define safety settings to disable all filters for content generation SAFETY_SETTINGS = [ types.SafetySetting( category=types.HarmCategory.HARM_CATEGORY_HARASSMENT, threshold=types.HarmBlockThreshold.BLOCK_NONE, ), types.SafetySetting( category=types.HarmCategory.HARM_CATEGORY_HATE_SPEECH, threshold=types.HarmBlockThreshold.BLOCK_NONE, ), types.SafetySetting( category=types.HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT, threshold=types.HarmBlockThreshold.BLOCK_NONE, ), types.SafetySetting( category=types.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT, threshold=types.HarmBlockThreshold.BLOCK_NONE, ), types.SafetySetting( category=types.HarmCategory.HARM_CATEGORY_CIVIC_INTEGRITY, threshold=types.HarmBlockThreshold.BLOCK_NONE, ), ] def clean_response_text(response_text): """ Clean the API response by removing Markdown code block markers. """ cleaned_text = response_text.strip() if cleaned_text.startswith("```json"): cleaned_text = cleaned_text[len("```json"):].strip() if cleaned_text.endswith("```"): cleaned_text = cleaned_text[:-len("```")].strip() return cleaned_text def generate_ideas(user_input): """ Generate a diverse set of ideas based on the user's input concept using the LLM. Yields progress updates for the loading UI. """ yield (10, f"Brainstorming epic ideas for {user_input}... 🌟") prompt = f""" The user has provided the concept: "{user_input}". You must generate 5 diverse and creative ideas for a TikTok video that are directly and explicitly related to "{user_input}". Each idea must clearly incorporate and focus on the core theme of "{user_input}" without deviating into unrelated topics. Each idea should be a short sentence describing a specific scene or concept. Return the response as a JSON object with a single key 'ideas' containing a list of 5 ideas. Ensure the response is strictly in JSON format. """ try: response = client.models.generate_content( model='gemini-2.0-flash-lite', contents=[prompt], config=types.GenerateContentConfig( temperature=1.2, safety_settings=SAFETY_SETTINGS ) ) if not response.text or response.text.isspace(): raise ValueError("Empty response from API") cleaned_text = clean_response_text(response.text) response_json = json.loads(cleaned_text) if 'ideas' not in response_json or not isinstance(response_json['ideas'], list) or len(response_json['ideas']) != 5: raise ValueError("Invalid JSON format: 'ideas' key missing, not a list, or incorrect length") ideas = response_json['ideas'] yield (20, f"Ideas locked in for {user_input}! 🚀") return ideas except Exception as e: print(f"Error generating ideas: {e}") yield (20, f"Oops, tweaking the plan for {user_input}... 🔧") return [ f"A dramatic {user_input} scene with cinematic lighting", f"A close-up of {user_input} in a futuristic setting", f"A high-energy {user_input} moment with vibrant colors", f"A serene {user_input} scene with soft focus", f"An action-packed {user_input} challenge with dynamic angles" ] def generate_item(user_input, ideas, generate_video=False, max_retries=3): """ Generate a single feed item (image and optionally one video) using one of the ideas. Yields progress updates for the loading UI. """ video_base64 = None max_total_attempts = 3 total_attempts = 0 while total_attempts < max_total_attempts: total_attempts += 1 yield (20 + total_attempts * 10, f"Attempt {total_attempts} to craft your {user_input} masterpiece... 🎨") generated_image = None text = None img_str = None image_prompt = None for image_attempt in range(max_retries): yield (20 + total_attempts * 10 + image_attempt * 5, f"Crafting a stunning image for {user_input}... 📸") selected_idea = random.choice(ideas) prompt = f""" The user has provided the concept: "{user_input}". Based on this concept and the specific idea "{selected_idea}", create content for a TikTok video. Return a JSON object with two keys: - 'caption': A short, viral TikTok-style caption with hashtags that reflects "{user_input}". - 'image_prompt': A detailed image prompt for generating a high-quality visual scene, ensuring the theme of "{user_input}" is central. The image prompt should describe the scene vividly, specify a perspective and style, and ensure no text or letters are included. Ensure the response is strictly in JSON format. """ try: response = client.models.generate_content( model='gemini-2.0-flash-lite', contents=[prompt], config=types.GenerateContentConfig( temperature=1.2, safety_settings=SAFETY_SETTINGS ) ) if not response.text or response.text.isspace(): raise ValueError("Empty response from API") cleaned_text = clean_response_text(response.text) response_json = json.loads(cleaned_text) if 'caption' not in response_json or 'image_prompt' not in response_json: raise ValueError("Invalid JSON format: 'caption' or 'image_prompt' key missing") text = response_json['caption'] image_prompt = response_json['image_prompt'] except Exception as e: print(f"Error generating item (image attempt {image_attempt + 1}): {e}") text = f"Amazing {user_input}! 🔥 #{user_input.replace(' ', '')}" image_prompt = f"A vivid scene of {selected_idea} related to {user_input}, in a vibrant pop art style, no text or letters" try: yield (40 + image_attempt * 5, f"Rendering your {user_input} vision... ✨") imagen = client.models.generate_images( model='imagen-3.0-generate-002', prompt=image_prompt, config=types.GenerateImagesConfig( aspect_ratio="9:16", number_of_images=1 ) ) if imagen.generated_images and len(imagen.generated_images) > 0: generated_image = imagen.generated_images[0] image = Image.open(BytesIO(generated_image.image.image_bytes)) target_width = 360 target_height = int(target_width / 9 * 16) image = image.resize((target_width, target_height), Image.LANCZOS) buffered = BytesIO() image.save(buffered, format="PNG") img_str = base64.b64encode(buffered.getvalue()).decode() yield (50, f"Image for {user_input} is ready! 🎉") break else: if image_attempt == max_retries - 1: yield (50, f"Tweaking the image for {user_input}... 🔄") if total_attempts == max_total_attempts: image = Image.new('RGB', (360, 640), color='gray') buffered = BytesIO() image.save(buffered, format="PNG") img_str = base64.b64encode(buffered.getvalue()).decode() yield (60, f"Using a placeholder for {user_input}... 🖼️") return { 'text': text, 'image_base64': img_str, 'video_base64': None, 'ideas': ideas } break except Exception as e: print(f"Error generating image (image attempt {image_attempt + 1}): {e}") if image_attempt == max_retries - 1: yield (50, f"Retrying image for {user_input}... 🔄") if total_attempts == max_total_attempts: image = Image.new('RGB', (360, 640), color='gray') buffered = BytesIO() image.save(buffered, format="PNG") img_str = base64.b64encode(buffered.getvalue()).decode() yield (60, f"Using a placeholder for {user_input}... 🖼️") return { 'text': text, 'image_base64': img_str, 'video_base64': None, 'ideas': ideas } break if generate_video and generated_image is not None: max_video_retries_per_image = 2 video_generated = False # Image-to-video generation try: yield (60, f"Filming a viral video for {user_input}... 🎥") video_prompt = f""" The user concept is "{user_input}". Based on this and the scene: {image_prompt}, create a video. Use a close-up shot with a slow dolly shot circling around the subject, using shallow focus on the main subject to emphasize details, in a realistic style with cinematic lighting. """ print(f"Attempting image-to-video generation: {video_prompt}") operation = client.models.generate_videos( model="veo-2.0-generate-001", prompt=video_prompt, image=generated_image.image, config=types.GenerateVideosConfig( aspect_ratio="9:16", number_of_videos=1, duration_seconds=8, negative_prompt="blurry, low quality, text, letters" ) ) while not operation.done: time.sleep(20) operation = client.operations.get(operation) print(f"Image-to-video operation: done={operation.done}, error={operation.error}, response={operation.response}") if operation.error: raise ValueError(f"Video generation failed: {operation.error.message}") if operation.response is None or not hasattr(operation.response, 'generated_videos') or operation.response.generated_videos is None: raise ValueError("Video generation failed: No generated_videos in response") if len(operation.response.generated_videos) > 0: video = operation.response.generated_videos[0] if video is None or not hasattr(video, 'video'): raise ValueError("Video is invalid or missing video data") video_data = client.files.download(file=video.video) if isinstance(video_data, bytes): video_bytes = video_data else: video_buffer = BytesIO() for chunk in video_data: video_buffer.write(chunk) video_bytes = video_buffer.getvalue() video_base64 = base64.b64encode(video_bytes).decode() video_generated = True yield (90, f"Video for {user_input} is a wrap! 🎬") return { 'text': text, 'image_base64': img_str, 'video_base64': video_base64, 'ideas': ideas } else: raise ValueError("No video was generated") except Exception as e: print(f"Error generating video (image-to-video): {e}") yield (70, f"Switching to a new video approach for {user_input}... 🎞️") # Text-to-video generation (fallback) if not video_generated: for video_attempt in range(max_video_retries_per_image): try: yield (75 + video_attempt * 5, f"Trying a fresh video take for {user_input}... 📹") video_prompt_base = f""" The user concept is "{user_input}". Based on this and the scene: {image_prompt}, create a video. Use a close-up shot with a slow dolly shot circling around the subject, using shallow focus on the main subject to emphasize details, in a realistic style with cinematic lighting. """ video_prompt = video_prompt_base if video_attempt == 0 else f""" The user concept is "{user_input}". Based on this and a simplified scene: {image_prompt}, create a video. Use a static close-up shot of the subject in a realistic style. """ print(f"Attempting text-to-video generation (attempt {video_attempt + 1}): {video_prompt}") operation = client.models.generate_videos( model="veo-2.0-generate-001", prompt=video_prompt, config=types.GenerateVideosConfig( aspect_ratio="9:16", number_of_videos=1, duration_seconds=8, negative_prompt="blurry, low quality, text, letters" ) ) while not operation.done: time.sleep(20) operation = client.operations.get(operation) print(f"Text-to-video operation (attempt {video_attempt + 1}): done={operation.done}, error={operation.error}, response={operation.response}") if operation.error: raise ValueError(f"Video generation failed: {operation.error.message}") if operation.response is None or not hasattr(operation.response, 'generated_videos') or operation.response.generated_videos is None: raise ValueError("Video generation failed: No generated_videos in response") if len(operation.response.generated_videos) > 0: video = operation.response.generated_videos[0] if video is None or not hasattr(video, 'video'): raise ValueError("Video is invalid or missing video data") video_data = client.files.download(file=video.video) if isinstance(video_data, bytes): video_bytes = video_data else: video_buffer = BytesIO() for chunk in video_data: video_buffer.write(chunk) video_bytes = video_buffer.getvalue() video_base64 = base64.b64encode(video_bytes).decode() video_generated = True yield (90, f"Video for {user_input} is a wrap! 🎬") return { 'text': text, 'image_base64': img_str, 'video_base64': video_base64, 'ideas': ideas } else: raise ValueError("No video was generated") except Exception as e: print(f"Error generating video (text-to-video attempt {video_attempt + 1}): {e}") if video_attempt == max_video_retries_per_image - 1: yield (85, f"Finalizing without video for {user_input}... 📌") if total_attempts == max_total_attempts: yield (95, f"Polishing your {user_input} masterpiece... ✨") return { 'text': text, 'image_base64': img_str, 'video_base64': video_base64, 'ideas': ideas } break if img_str is not None: yield (95, f"Polishing your {user_input} masterpiece... ✨") return { 'text': text, 'image_base64': img_str, 'video_base64': video_base64, 'ideas': ideas } print("Max total attempts reached without successful image generation. Using placeholder.") yield (95, f"Falling back to a placeholder for {user_input}... 🖼️") image = Image.new('RGB', (360, 640), color='gray') buffered = BytesIO() image.save(buffered, format="PNG") img_str = base64.b64encode(buffered.getvalue()).decode() yield (100, f"Ready to roll with {user_input}! 🚀") return { 'text': f"Amazing {user_input}! 🔥 #{user_input.replace(' ', '')}", 'image_base64': img_str, 'video_base64': None, 'ideas': ideas } def generate_progress_html(progress, message, user_input): """ Generate HTML for the progress bar and witty text. """ return f"""
{message}
{int(progress)}% Complete
""" def process_generator(generator): """ Process a generator to collect progress updates and the final result. Returns a tuple of (progress_updates, final_result). """ progress_updates = [] final_result = None for item in generator: if isinstance(item, tuple): progress_updates.append(item) else: final_result = item return progress_updates, final_result def start_feed(user_input, generate_video, current_index, feed_items): """ Start or update the feed based on the user input. Returns all outputs expected by Gradio. """ if not user_input.strip(): user_input = "trending" current_user_input = user_input is_loading = True progress_html = generate_progress_html(0, f"Getting started with {user_input}... 🚀", user_input) share_links = "" # Ensure feed_items is a list feed_items = feed_items if isinstance(feed_items, list) else [] print(f"start_feed: feed_items type={type(feed_items)}, value={feed_items}") try: # Generate ideas ideas_gen = generate_ideas(user_input) progress_updates, ideas = process_generator(ideas_gen) for progress, message in progress_updates: progress_html = generate_progress_html(progress, message, user_input) yield (progress_html, None, None, None, None, is_loading, progress_html) # Generate item item_gen = generate_item(user_input, ideas, generate_video=generate_video) progress_updates, item = process_generator(item_gen) for progress, message in progress_updates: progress_html = generate_progress_html(progress, message, user_input) yield (progress_html, None, None, None, None, is_loading, progress_html) feed_items = [item] current_index = 0 feed_html = generate_html(feed_items, False, current_index, user_input, is_loading=False) share_links = generate_share_links( item['image_base64'], item['video_base64'], item['text'] ) except Exception as e: print(f"Error in start_feed: {e}") feed_items = [] current_index = 0 feed_html = """

Error generating content. Please try again!

""" progress_html = generate_progress_html(100, "Oops, something went wrong! 😅", user_input) is_loading = False return current_user_input, current_index, feed_items, feed_html, share_links, is_loading, progress_html is_loading = False progress_html = "" return current_user_input, current_index, feed_items, feed_html, share_links, is_loading, progress_html def load_next(user_input, generate_video, current_index, feed_items): """ Load the next item in the feed. Returns all outputs expected by Gradio. """ current_user_input = user_input if user_input.strip() else "trending" user_input = current_user_input is_loading = True progress_html = generate_progress_html(0, f"Loading next {user_input} vibe... 🚀", user_input) share_links = "" # Ensure feed_items is a list feed_items = feed_items if isinstance(feed_items, list) else [] print(f"load_next: feed_items type={type(feed_items)}, value={feed_items}, current_index={current_index}") try: if current_index + 1 < len(feed_items): current_index += 1 progress_html = generate_progress_html(50, f"Switching to the next {user_input} moment... 🔄", user_input) yield (progress_html, None, None, None, None, is_loading, progress_html) else: ideas = feed_items[-1]['ideas'] if feed_items else None if not ideas: ideas_gen = generate_ideas(user_input) progress_updates, ideas = process_generator(ideas_gen) for progress, message in progress_updates: progress_html = generate_progress_html(progress, message, user_input) yield (progress_html, None, None, None, None, is_loading, progress_html) new_item_gen = generate_item(user_input, ideas, generate_video=generate_video) progress_updates, new_item = process_generator(new_item_gen) for progress, message in progress_updates: progress_html = generate_progress_html(progress, message, user_input) yield (progress_html, None, None, None, None, is_loading, progress_html) feed_items.append(new_item) current_index = len(feed_items) - 1 feed_html = generate_html(feed_items, False, current_index, user_input, is_loading=False) share_links = generate_share_links( feed_items[current_index]['image_base64'], feed_items[current_index]['video_base64'], feed_items[current_index]['text'] ) except Exception as e: print(f"Error in load_next: {e}") feed_html = """

Error generating content. Please try again!

""" progress_html = generate_progress_html(100, "Oops, something went wrong! 😅", user_input) is_loading = False return current_user_input, current_index, feed_items, feed_html, share_links, is_loading, progress_html is_loading = False progress_html = "" return current_user_input, current_index, feed_items, feed_html, share_links, is_loading, progress_html def load_previous(user_input, generate_video, current_index, feed_items): """ Load the previous item in the feed. """ current_user_input = user_input if user_input.strip() else "trending" # Ensure feed_items is a list feed_items = feed_items if isinstance(feed_items, list) else [] print(f"load_previous: feed_items type={type(feed_items)}, value={feed_items}, current_index={current_index}") if current_index > 0: current_index -= 1 feed_html = generate_html(feed_items, False, current_index, user_input, is_loading=False) share_links = generate_share_links( feed_items[current_index]['image_base64'] if feed_items else "", feed_items[current_index]['video_base64'] if feed_items else None, feed_items[current_index]['text'] if feed_items else "" ) return current_user_input, current_index, feed_items, feed_html, share_links, False, "" def generate_share_links(image_base64, video_base64, caption): """ Generate share links for social media platforms with download links for image and video. """ image_data_url = f"data:image/png;base64,{image_base64}" if image_base64 else "#" encoded_caption = urllib.parse.quote(caption if caption else "") download_links = f"""

Download the media to share:

Download Image """ if video_base64: video_data_url = f"data:video/mp4;base64,{video_base64}" download_links += f""" Download Video """ download_links += "
" instruction = """

Click a share button below to start a post with the caption, then manually upload the downloaded image or video.

""" share_links = """
Share on TikTok Share on Instagram Share on Facebook Share on X Share on Pinterest
""" youtube_share = "" if video_base64: youtube_share = f"""
Share to YouTube as a Short
""" return f"""
{download_links} {instruction} {share_links} {youtube_share}
""".format(caption=encoded_caption) def generate_html(feed_items, scroll_to_latest=False, current_index=0, user_input="", is_loading=False, progress=0, message=""): """ Generate an HTML string to display the current feed item or loading state. """ # Ensure feed_items is a list feed_items = feed_items if isinstance(feed_items, list) else [] print(f"generate_html: feed_items type={type(feed_items)}, value={feed_items}, current_index={current_index}") if is_loading: return """
""" if not feed_items or current_index >= len(feed_items): return """

Enter a concept or idea to start your feed!

""" item = feed_items[current_index] if item['video_base64']: media_element = f""" """ else: media_element = f""" """ html_str = f"""
{media_element}
{item['text']}
""" return html_str # Define the Gradio interface with gr.Blocks( css=""" body { background-color: #000; color: #fff; font-family: Arial, sans-serif; } .gradio-container { max-width: 400px; margin: 0 auto; padding: 10px; } input, select, button, .gr-checkbox { border-radius: 5px; background-color: #222; color: #fff; border: 1px solid #444; } button { background-color: #ff2d55; border: none; } button:hover { background-color: #e0264b; } .gr-button { width: 100%; margin-top: 10px; } .gr-form { background-color: #111; padding: 15px; border-radius: 10px; } """, title="Create Your Feed" ) as demo: current_user_input = gr.State(value="") current_index = gr.State(value=0) feed_items = gr.State(value=[]) # Ensure initial value is an empty list is_loading = gr.State(value=False) share_links = gr.State(value="") with gr.Column(elem_classes="gr-form"): gr.Markdown("### Create Your Feed") user_input = gr.Textbox( label="Enter Concept or Ideas", value="", placeholder="e.g., sushi adventure, neon tech", submit_btn=False ) generate_video_checkbox = gr.Checkbox( label="Generate Video (may take longer)", value=False ) magic_button = gr.Button("✨ Generate Next Item", elem_classes="gr-button") progress_html = gr.HTML(label="Progress", visible=True) feed_html = gr.HTML() share_html = gr.HTML(label="Share this item:") user_input.submit( fn=start_feed, inputs=[user_input, generate_video_checkbox, current_index, feed_items], outputs=[current_user_input, current_index, feed_items, feed_html, share_html, is_loading, progress_html] ) magic_button.click( fn=load_next, inputs=[user_input, generate_video_checkbox, current_index, feed_items], outputs=[current_user_input, current_index, feed_items, feed_html, share_html, is_loading, progress_html] ) previous_button = gr.Button("Previous", elem_id="previous-button", visible=False) previous_button.click( fn=load_previous, inputs=[user_input, generate_video_checkbox, current_index, feed_items], outputs=[current_user_input, current_index, feed_items, feed_html, share_html, is_loading, progress_html] ) # Launch the app demo.launch()