Update streamlit_app.py

streamlit_app.py

@@ -1,224 +1,224 @@
-import streamlit as st
-import os
-import re
-import base64
-import tempfile
-import google.generativeai as genai
-import time
-
-# Local imports from your other project files
-from agents import get_content_enhancer_agent
-from tools.image_tool import ImageGenerationTool
-from local_knowledge_base import load_knowledge_base
-
-# --- Page and API Configuration ---
-st.set_page_config(page_title="Video Analyzer", layout="wide")
-st.title("🎥 Video Analysis & Content Generation")
-st.markdown("Provide a video file or a YouTube link to generate an enriched summary with AI-generated images.")
-
-# --- Authentication and Initialization ---
-try:
-    genai.configure(api_key=st.secrets["GOOGLE_API_KEY"])
-except KeyError as e:
-    st.error(f"Secret key '{e.args[0]}' not found. Please add it to your .streamlit/secrets.toml file.")
-    st.stop()
-
-@st.cache_resource(show_spinner="Loading Local Knowledge Base...")
-def get_local_kb():
-    return load_knowledge_base()
-
-
-local_kb = get_local_kb()
-local_kb.load(recreate=False)  # Load the knowledge base into memory
-# --- Core Logic Functions ---
-
-def enrich_and_generate_images(video_summary: str) -> str:
-    """
-    Takes a raw summary, enriches it using an AI agent, and generates images for placeholders.
-    This is a helper function used by both local file and YouTube workflows.
-    """
-    # --- Step 2: Enhance Content with AI Agent ---
-    st.write("### Step 2: Enhancing Content with AI Agent...")
-    with st.spinner("The content agent is adding details and finding opportunities for images..."):
-        content_agent = get_content_enhancer_agent(knowledge_base=local_kb)
-        agent_response_generator = content_agent.run(video_summary)
-        enriched_text_with_placeholders = "".join(list(agent_response_generator))
-    
-    st.success("✅ Content enhancement complete.")
-    with st.expander("See Enriched Text with Image Placeholders"):
-        st.text(enriched_text_with_placeholders)
-
-    # --- Step 3: Generate Images and Assemble Final Document ---
-    st.write("### Step 3: Generating Images and Assembling Final Document...")
-    final_output = enriched_text_with_placeholders
-    
-    try:
-        # Get Clipdrop API key from secrets
-        image_tool = ImageGenerationTool(api_key=st.secrets["CLIPDROP_API_KEY"])
-    except KeyError:
-        st.error("CLIPDROP_API_KEY not found in Streamlit Secrets. Please add it to continue.")
-        st.stop()
-
-    placeholders = re.findall(r'\[IMAGE: caption="(.*?)"\]', final_output)
-    if not placeholders:
-        st.warning("The agent did not add any image placeholders. Returning the enhanced text.")
-        return final_output
-
-    total_images = len(placeholders)
-    progress_bar = st.progress(0, text=f"Preparing to generate {total_images} images...")
-
-    for i, caption_prompt in enumerate(placeholders):
-        progress_text = f"Generating image {i+1} of {total_images}: '{caption_prompt[:40]}...'"
-        progress_bar.progress((i) / total_images, text=progress_text)
-        
-        image_path = image_tool.generate_image(prompt=caption_prompt)
-        
-        placeholder_to_replace = f'[IMAGE: caption="{caption_prompt}"]'
-        if "Error:" in image_path or not os.path.exists(image_path):
-            markdown_tag = f'\n> *Image generation failed for prompt: "{caption_prompt}". Reason: {image_path}*\n'
-        else:
-            with open(image_path, "rb") as f:
-                image_bytes = f.read()
-            base64_image = base64.b64encode(image_bytes).decode("utf-8")
-            markdown_tag = f'\n<div align="center" style="margin: 1rem 0;"><img src="data:image/png;base64,{base64_image}" alt="{caption_prompt}" style="max-width: 100%; max-height: 500px; border-radius: 8px; box-shadow: 0 4px 8px rgba(0,0,0,0.1);"><br><em style="font-size: 0.9em; color: #555;">{caption_prompt}</em></div>\n'
-        
-        final_output = final_output.replace(placeholder_to_replace, markdown_tag, 1)
-
-        if i < total_images - 1:
-            time.sleep(5)  # Respect API rate limits
-
-    progress_bar.progress(1.0, text="All images generated!")
-    st.success("✅ All images generated.")
-    return final_output
-
-def run_youtube_analysis_workflow(yt_url: str):
-    """
-    Analyzes a YouTube video directly from its URL without downloading.
-    """
-    st.write("### Step 1: Analyzing YouTube Video...")
-    st.info("Sending URL to Gemini for analysis...")
-
-    try:
-        # The key is to use a tool that tells the model how to handle the YouTube URL.
-        from google.ai.generativelanguage import Part, FileData
-        
-        video_analysis_model = genai.GenerativeModel('models/gemini-1.5-flash-latest')
-        
-        prompt = "Analyze this video then compile the top 2-3 core topics and then write a detailed summary on it. Maintain the original order of events."
-        
-        # Create the FileData part using the YouTube URI
-        video_part = Part(
-            file_data=FileData(
-                mime_type="video/youtube", 
-                file_uri=yt_url
-            )
-        )
-        
-        with st.spinner("Gemini is processing the video... This may take a few minutes for longer videos."):
-            response = video_analysis_model.generate_content([prompt, video_part], request_options={"timeout": 600})
-
-        video_summary = response.text
-        st.success("✅ YouTube video analysis complete.")
-        with st.expander("See Raw Video Summary"):
-            st.markdown(video_summary)
-
-        # Now, pass the summary to the shared enrichment/image generation stage
-        return enrich_and_generate_images(video_summary)
-
-    except Exception as e:
-        st.error(f"An error occurred during YouTube analysis: {e}")
-        st.info("This can sometimes happen with private, age-restricted, or very new videos. Please check the URL and try again.")
-        return None
-
-def run_local_file_analysis_workflow(video_path: str):
-    """
-    Analyzes a video from a local file path by uploading it to Google.
-    """
-    video_file_obj = None
-    try:
-        st.write("### Step 1: Uploading & Analyzing Local Video File...")
-        st.info("Uploading video file to Google. This may take a moment...")
-        video_file_obj = genai.upload_file(path=video_path)
-        st.info(f"Video uploaded: {video_file_obj.name}. Waiting for processing...")
-
-        while video_file_obj.state.name == "PROCESSING":
-            time.sleep(5)
-            video_file_obj = genai.get_file(video_file_obj.name)
-
-        if video_file_obj.state.name == "FAILED":
-            st.error(f"Video processing failed: {video_file_obj.state}")
-            return None
-
-        st.success("✅ Video is processed and ready for analysis.")
-        
-        video_analysis_model = genai.GenerativeModel('models/gemini-1.5-flash-latest')
-        prompt = ("Analyze this video in detail. First, identify the 2-3 core topics discussed. "
-                  "Then, write a comprehensive summary of the video's content, maintaining the original order of events.")
-        
-        response = video_analysis_model.generate_content([prompt, video_file_obj], request_options={"timeout": 600})
-        video_summary = response.text
-        st.success("✅ Local video analysis complete.")
-        with st.expander("See Raw Video Summary"):
-            st.markdown(video_summary)
-
-        # Pass the summary to the shared enrichment/image generation stage
-        return enrich_and_generate_images(video_summary)
-
-    except Exception as e:
-        st.error(f"An error occurred during local file analysis: {e}")
-        return None
-    finally:
-        if video_file_obj:
-            genai.delete_file(video_file_obj.name)
-            st.info(f"Cleaned up cloud file: {video_file_obj.name}")
-
-
-# --- Streamlit UI ---
-
-st.divider()
-st.subheader("Select Video Source")
-
-# Use tabs for a clean UI. We recommend YouTube for speed and reliability.
-tab1, tab2 = st.tabs(["🔗 **Use a YouTube Link (Recommended)**", "📤 Upload a Video File"])
-video_path = None
-final_response = None
-
-with tab1:
-    yt_url = st.text_input(
-        "Enter YouTube Video URL",
-        placeholder="https://www.youtube.com/watch?v=..."
-    )
-
-with tab2:
-    video_file = st.file_uploader(
-        "Choose a video file",
-        type=["mp4", "mov", "avi", "mpeg", "webm"]
-    )
-
-if st.button("🚀 Analyze and Generate Content", use_container_width=True):
-    # Prioritize YouTube URL if both are provided
-    if yt_url:
-        final_response = run_youtube_analysis_workflow(yt_url)
-    elif video_file:
-        with st.spinner("Saving uploaded file locally..."):
-            with tempfile.NamedTemporaryFile(delete=False, suffix=".mp4") as tmp_file:
-                tmp_file.write(video_file.getbuffer())
-                video_path = tmp_file.name
-        
-        try:
-            final_response = run_local_file_analysis_workflow(video_path)
-        finally:
-            if video_path and os.path.exists(video_path):
-                os.remove(video_path)
-                st.info(f"Cleaned up local temporary file: {video_path}")
-    else:
-        st.warning("Please provide a YouTube URL or upload a video file.")
-        st.stop()
-
-    # Display the final generated content if the workflow was successful
-    if final_response:
-        st.divider()
-        st.markdown("## ✨ Final Generated Content")
-        st.markdown(final_response, unsafe_allow_html=True)
-    else:
+import streamlit as st
+import os
+import re
+import base64
+import tempfile
+import google.generativeai as genai
+import time
+
+# Local imports from your other project files
+from agents import get_content_enhancer_agent
+from tools.image_tool import ImageGenerationTool
+from local_knowledge_base import load_knowledge_base
+
+# --- Page and API Configuration ---
+st.set_page_config(page_title="Video Analyzer", layout="wide")
+st.title("🎥 Video Analysis & Content Generation")
+st.markdown("Provide a video file or a YouTube link to generate an enriched summary with AI-generated images.")
+
+# --- Authentication and Initialization ---
+try:
+    genai.configure(api_key=os.getenv("GOOGLE_API_KEY"))
+except KeyError as e:
+    st.error(f"Secret key '{e.args[0]}' not found. Please add it to your .streamlit/secrets.toml file.")
+    st.stop()
+
+@st.cache_resource(show_spinner="Loading Local Knowledge Base...")
+def get_local_kb():
+    return load_knowledge_base()
+
+
+local_kb = get_local_kb()
+local_kb.load(recreate=False)  # Load the knowledge base into memory
+# --- Core Logic Functions ---
+
+def enrich_and_generate_images(video_summary: str) -> str:
+    """
+    Takes a raw summary, enriches it using an AI agent, and generates images for placeholders.
+    This is a helper function used by both local file and YouTube workflows.
+    """
+    # --- Step 2: Enhance Content with AI Agent ---
+    st.write("### Step 2: Enhancing Content with AI Agent...")
+    with st.spinner("The content agent is adding details and finding opportunities for images..."):
+        content_agent = get_content_enhancer_agent(knowledge_base=local_kb)
+        agent_response_generator = content_agent.run(video_summary)
+        enriched_text_with_placeholders = "".join(list(agent_response_generator))
+    
+    st.success("✅ Content enhancement complete.")
+    with st.expander("See Enriched Text with Image Placeholders"):
+        st.text(enriched_text_with_placeholders)
+
+    # --- Step 3: Generate Images and Assemble Final Document ---
+    st.write("### Step 3: Generating Images and Assembling Final Document...")
+    final_output = enriched_text_with_placeholders
+    
+    try:
+        # Get Clipdrop API key from secrets
+        image_tool = ImageGenerationTool(api_key=os.getenv("CLIPDROP_API_KEY"))
+    except KeyError:
+        st.error("CLIPDROP_API_KEY not found in Streamlit Secrets. Please add it to continue.")
+        st.stop()
+
+    placeholders = re.findall(r'\[IMAGE: caption="(.*?)"\]', final_output)
+    if not placeholders:
+        st.warning("The agent did not add any image placeholders. Returning the enhanced text.")
+        return final_output
+
+    total_images = len(placeholders)
+    progress_bar = st.progress(0, text=f"Preparing to generate {total_images} images...")
+
+    for i, caption_prompt in enumerate(placeholders):
+        progress_text = f"Generating image {i+1} of {total_images}: '{caption_prompt[:40]}...'"
+        progress_bar.progress((i) / total_images, text=progress_text)
+        
+        image_path = image_tool.generate_image(prompt=caption_prompt)
+        
+        placeholder_to_replace = f'[IMAGE: caption="{caption_prompt}"]'
+        if "Error:" in image_path or not os.path.exists(image_path):
+            markdown_tag = f'\n> *Image generation failed for prompt: "{caption_prompt}". Reason: {image_path}*\n'
+        else:
+            with open(image_path, "rb") as f:
+                image_bytes = f.read()
+            base64_image = base64.b64encode(image_bytes).decode("utf-8")
+            markdown_tag = f'\n<div align="center" style="margin: 1rem 0;"><img src="data:image/png;base64,{base64_image}" alt="{caption_prompt}" style="max-width: 100%; max-height: 500px; border-radius: 8px; box-shadow: 0 4px 8px rgba(0,0,0,0.1);"><br><em style="font-size: 0.9em; color: #555;">{caption_prompt}</em></div>\n'
+        
+        final_output = final_output.replace(placeholder_to_replace, markdown_tag, 1)
+
+        if i < total_images - 1:
+            time.sleep(5)  # Respect API rate limits
+
+    progress_bar.progress(1.0, text="All images generated!")
+    st.success("✅ All images generated.")
+    return final_output
+
+def run_youtube_analysis_workflow(yt_url: str):
+    """
+    Analyzes a YouTube video directly from its URL without downloading.
+    """
+    st.write("### Step 1: Analyzing YouTube Video...")
+    st.info("Sending URL to Gemini for analysis...")
+
+    try:
+        # The key is to use a tool that tells the model how to handle the YouTube URL.
+        from google.ai.generativelanguage import Part, FileData
+        
+        video_analysis_model = genai.GenerativeModel('models/gemini-1.5-flash-latest')
+        
+        prompt = "Analyze this video then compile the top 2-3 core topics and then write a detailed summary on it. Maintain the original order of events."
+        
+        # Create the FileData part using the YouTube URI
+        video_part = Part(
+            file_data=FileData(
+                mime_type="video/youtube", 
+                file_uri=yt_url
+            )
+        )
+        
+        with st.spinner("Gemini is processing the video... This may take a few minutes for longer videos."):
+            response = video_analysis_model.generate_content([prompt, video_part], request_options={"timeout": 600})
+
+        video_summary = response.text
+        st.success("✅ YouTube video analysis complete.")
+        with st.expander("See Raw Video Summary"):
+            st.markdown(video_summary)
+
+        # Now, pass the summary to the shared enrichment/image generation stage
+        return enrich_and_generate_images(video_summary)
+
+    except Exception as e:
+        st.error(f"An error occurred during YouTube analysis: {e}")
+        st.info("This can sometimes happen with private, age-restricted, or very new videos. Please check the URL and try again.")
+        return None
+
+def run_local_file_analysis_workflow(video_path: str):
+    """
+    Analyzes a video from a local file path by uploading it to Google.
+    """
+    video_file_obj = None
+    try:
+        st.write("### Step 1: Uploading & Analyzing Local Video File...")
+        st.info("Uploading video file to Google. This may take a moment...")
+        video_file_obj = genai.upload_file(path=video_path)
+        st.info(f"Video uploaded: {video_file_obj.name}. Waiting for processing...")
+
+        while video_file_obj.state.name == "PROCESSING":
+            time.sleep(5)
+            video_file_obj = genai.get_file(video_file_obj.name)
+
+        if video_file_obj.state.name == "FAILED":
+            st.error(f"Video processing failed: {video_file_obj.state}")
+            return None
+
+        st.success("✅ Video is processed and ready for analysis.")
+        
+        video_analysis_model = genai.GenerativeModel('models/gemini-1.5-flash-latest')
+        prompt = ("Analyze this video in detail. First, identify the 2-3 core topics discussed. "
+                  "Then, write a comprehensive summary of the video's content, maintaining the original order of events.")
+        
+        response = video_analysis_model.generate_content([prompt, video_file_obj], request_options={"timeout": 600})
+        video_summary = response.text
+        st.success("✅ Local video analysis complete.")
+        with st.expander("See Raw Video Summary"):
+            st.markdown(video_summary)
+
+        # Pass the summary to the shared enrichment/image generation stage
+        return enrich_and_generate_images(video_summary)
+
+    except Exception as e:
+        st.error(f"An error occurred during local file analysis: {e}")
+        return None
+    finally:
+        if video_file_obj:
+            genai.delete_file(video_file_obj.name)
+            st.info(f"Cleaned up cloud file: {video_file_obj.name}")
+
+
+# --- Streamlit UI ---
+
+st.divider()
+st.subheader("Select Video Source")
+
+# Use tabs for a clean UI. We recommend YouTube for speed and reliability.
+tab1, tab2 = st.tabs(["🔗 **Use a YouTube Link (Recommended)**", "📤 Upload a Video File"])
+video_path = None
+final_response = None
+
+with tab1:
+    yt_url = st.text_input(
+        "Enter YouTube Video URL",
+        placeholder="https://www.youtube.com/watch?v=..."
+    )
+
+with tab2:
+    video_file = st.file_uploader(
+        "Choose a video file",
+        type=["mp4", "mov", "avi", "mpeg", "webm"]
+    )
+
+if st.button("🚀 Analyze and Generate Content", use_container_width=True):
+    # Prioritize YouTube URL if both are provided
+    if yt_url:
+        final_response = run_youtube_analysis_workflow(yt_url)
+    elif video_file:
+        with st.spinner("Saving uploaded file locally..."):
+            with tempfile.NamedTemporaryFile(delete=False, suffix=".mp4") as tmp_file:
+                tmp_file.write(video_file.getbuffer())
+                video_path = tmp_file.name
+        
+        try:
+            final_response = run_local_file_analysis_workflow(video_path)
+        finally:
+            if video_path and os.path.exists(video_path):
+                os.remove(video_path)
+                st.info(f"Cleaned up local temporary file: {video_path}")
+    else:
+        st.warning("Please provide a YouTube URL or upload a video file.")
+        st.stop()
+
+    # Display the final generated content if the workflow was successful
+    if final_response:
+        st.divider()
+        st.markdown("## ✨ Final Generated Content")
+        st.markdown(final_response, unsafe_allow_html=True)
+    else:
         st.error("The analysis workflow did not produce a final output. Please check the errors above.")