Update streamlit_app.py

streamlit_app.py

@@ -1,215 +1,191 @@
 import os
-import uuid
-import yt_dlp
 import string
-import ffmpeg
-from glob import glob
-import streamlit as st
-from phi.agent import Agent
-from phi.model.google import Gemini
-from phi.tools.duckduckgo import DuckDuckGo
-from google.generativeai import upload_file, get_file
 import time
-import google.generativeai as genai
+from glob import glob
 from pathlib import Path
-import tempfile
 
+import yt_dlp
+import ffmpeg
+import streamlit as st
 from dotenv import load_dotenv
+
 load_dotenv()
 
-# Page Configuration
-st.set_page_config(
-    page_title="Generate the story of videos:",
-    layout="wide"
-)
+# Optional imports for Phi/Gemini and Google genai
+try:
+    from phi.agent import Agent
+    from phi.model.google import Gemini
+    from phi.tools.duckduckgo import DuckDuckGo
+    PHI_AVAILABLE = True
+except Exception:
+    PHI_AVAILABLE = False
+
+try:
+    import google.generativeai as genai
+    from google.generativeai import upload_file, get_file
+    GENAI_AVAILABLE = True
+except Exception:
+    GENAI_AVAILABLE = False
+
+# Page config
+st.set_page_config(page_title="Generate the story of videos:", layout="wide")
+
+DATA_DIR = Path("./data")
+DATA_DIR.mkdir(exist_ok=True)
+
+# Session state defaults
+st.session_state.setdefault("videos", "")
+st.session_state.setdefault("downloaded", [])
+st.session_state.setdefault("loop_video", True)
+
+# Sidebar UI (all controls live here)
+st.sidebar.header("Video Input")
+st.sidebar.text_input("Video URL (or local .mp4 path)", key="url", placeholder="Enter Video URL or path")
+
+settings_exp = st.sidebar.expander("Settings", expanded=False)
+env_api_key = os.getenv("GOOGLE_API_KEY", "")
+API_KEY = settings_exp.text_input("Google API Key", value=env_api_key, placeholder="Set GOOGLE_API_KEY in .env or enter here")
+
+if API_KEY and GENAI_AVAILABLE:
+    genai.configure(api_key=API_KEY)
 
-# Config
-if "videos" not in st.session_state:
-    st.session_state["videos"] = ""
+if not API_KEY:
+    settings_exp.warning("No Google API key provided. Uploading/processing will not work.", icon="⚠️")
 
-if "downloaded" not in st.session_state:
-    st.session_state["downloaded"] = []
+model_id = settings_exp.text_input("Gemini Model", "gemini-2.0-flash-lite")
+analysis_prompt = settings_exp.text_area("Enter analysis", "watch entire video and describe")
+settings_exp.text_input("Video Password", key="video-password", placeholder="Enter Video Password (if needed)")
 
+# Safety settings (kept but optional)
+safety_settings = [
+    {"category": "HARM_CATEGORY_HARASSMENT", "threshold": "OFF"},
+    {"category": "HARM_CATEGORY_HATE_SPEECH", "threshold": "OFF"},
+    {"category": "HARM_CATEGORY_SEXUALLY_EXPLICIT", "threshold": "OFF"},
+    {"category": "HARM_CATEGORY_DANGEROUS_CONTENT", "threshold": "OFF"},
+]
 
-# Callbacks
-def _remove_video():
-    if st.session_state["videos"]:
-        try:
-            os.remove(st.session_state["videos"])
-            st.session_state["videos"] = ""
-        except FileNotFoundError:
-            print("Couldn't delete file")
-
-
-def _remove_all_videos():
-    videos = (
-        glob(os.path.join(".", "data", "*.mp4"))
-        + glob(os.path.join(".", "data", "*.webm"))
-        + glob(os.path.join(".", "data", "*.mov"))
-        + glob(os.path.join(".", "data", "*.m4a"))
-        + glob(os.path.join(".", "data", "*.mkv"))
-    )
-    for video in videos:
+# Initialize phi agent (cached)
+@st.cache_resource
+def initialize_agent(model_id: str):
+    if not PHI_AVAILABLE:
+        return None
+    return Agent(name="Video AI summarizer", model=Gemini(id=model_id), tools=[DuckDuckGo()], markdown=True)
+
+multimodal_Agent = None
+if PHI_AVAILABLE:
+    multimodal_Agent = initialize_agent(model_id)
+
+def sanitize_filename(path_str: str):
+    name = Path(path_str).name
+    name = name.lower().translate(str.maketrans("", "", string.punctuation)).replace(" ", "_")
+    return name
+
+def convert_video_to_mp4(video_path: str) -> str:
+    target_path = str(Path(video_path).with_suffix(".mp4"))
+    if os.path.exists(target_path):
+        return target_path
+    try:
+        ffmpeg.input(video_path).output(target_path).run(overwrite_output=True, quiet=True)
         try:
-            os.remove(video)
+            os.remove(video_path)
         except FileNotFoundError:
-            print("Couldn't delete file:", video)
-
-#Sidebar Content
-st.sidebar.text_input(
-    "**Video URL**",
-    key="url",
-    placeholder="Enter Video URL",
-    on_change=_remove_video,
-)
-
-expander = st.sidebar.expander("Settings", expanded=False)
-# ✅ Configure Gemini API
-API_KEY = expander.text_input("Google API Key", "AIzaSyB2uDj6IeuNszSuk80feW-eBgpIDH0WFSY")
-# ✅ Check if API key is available
-if API_KEY:
-    genai.configure(api_key=API_KEY)
-
-else:
-    expander.error("No GOOGLE_API_KEY", icon="⚠️")
-    # ✅ Load API key securely
-    API_KEY = os.getenv("GOOGLE_API_KEY")
-
-#Safety Settings
-safety_settings = [ { "category": "HARM_CATEGORY_HARASSMENT", "threshold": "OFF" }, { "category": "HARM_CATEGORY_HATE_SPEECH", "threshold": "OFF" }, { "category": "HARM_CATEGORY_SEXUALLY_EXPLICIT", "threshold": "OFF" }, { "category": "HARM_CATEGORY_DANGEROUS_CONTENT", "threshold": "OFF" }, ]
+            pass
+    except Exception as e:
+        st.error(f"FFmpeg conversion failed: {e}")
+        raise
+    return target_path
 
-#Gemini Model
-model_id = expander.text_input("Gemini Model", "gemini-2.0-flash-lite")
+def download_video_ytdlp(url: str, save_dir: str, video_password: str = None) -> str:
+    if not url:
+        raise ValueError("No URL provided")
 
-#Analysis prompt
-analysis_prompt = expander.text_area("Enter analysis", "watch entire video and describe")
+    # If it's a local file path and exists, just return it (convert if needed)
+    if os.path.exists(url) and os.path.isfile(url):
+        return convert_video_to_mp4(url)
 
-#Password
-expander.text_input("Video Password", key="video-password", placeholder="Enter Video Password")
+    outtmpl = os.path.join(save_dir, "%(id)s.%(ext)s")
+    ydl_opts = {"outtmpl": outtmpl, "format": "best"}
+    if video_password:
+        ydl_opts["videopassword"] = video_password
 
-@st.cache_resource
-def initialize_agent():
-    return Agent(
-        name="Video AI summarizer",
-        model=Gemini(id=model_id),
-        tools=[DuckDuckGo()],
-        markdown=True,
-    )
-
-# Initialize the agent
-multimodal_Agent = initialize_agent()
-
-def download_video(url: str, output_path: str):
-    """Downloads a video using yt-dlp."""
-    ydl_opts = {
-        'outtmpl': output_path,  # Path where the video will be saved
-        'format': 'best',        # Download the best quality available
-    }
-    
     try:
         with yt_dlp.YoutubeDL(ydl_opts) as ydl:
             ydl.download([url])
-        return True, f"Video downloaded successfully to {output_path}"
     except Exception as e:
-        return False, str(e)
-
-# --- Main Functions ---
-def download_video(url: str, save_path: str, **kwargs):
-    video_id = url.split("/")[-1]
-
-    if len(video_id) < 1:
-        video_id = url.split("/")[-2]
-
-    ydl_opts = {
-        "outtmpl": f"{save_path}/{video_id}.%(ext)s",
-    }
-    for opt in kwargs.keys():
-        ydl_opts[opt] = kwargs[opt]
-    print(ydl_opts)
+        raise RuntimeError(f"yt-dlp download failed: {e}")
 
-    with yt_dlp.YoutubeDL(ydl_opts) as ydl:
-        ydl.download(url)
+    video_id = url.rstrip("/").split("/")[-1] or url.rstrip("/").split("/")[-2]
+    matches = glob(os.path.join(save_dir, f"{video_id}.*"))
+    if not matches:
+        matches = sorted(glob(os.path.join(save_dir, "*")), key=os.path.getmtime, reverse=True)[:1]
 
-    videos = glob(os.path.join(".", "data", f"{video_id}.*"))
+    if not matches:
+        raise FileNotFoundError("Downloaded video not found")
 
-    print("videos found:", videos)
-    for video in videos:
-        if video_id in video:
-            video_path = video
-            converted_video_path = convert_video_to_mp4(video_path)
-            return converted_video_path
+    return convert_video_to_mp4(matches[0])
 
-
-def convert_video_to_mp4(video_path):
-    target_path = f"{os.path.splitext(video_path)[0]}.mp4"
-    if not os.path.exists(target_path):  # Convert only if target doesn't exist
-        (ffmpeg.input(video_path).output(target_path).run(overwrite_output=True))
-        try:
-            os.remove(video_path)
-        except FileNotFoundError:
-            print("Couldn't delete file:", video_path)
-    return target_path
-
-if st.session_state["url"]:
-    download = st.sidebar.button("**Load Video** 🚀", use_container_width=True)
-    if download:
-        download_options = {}
-        if st.session_state["video-password"]:
-            download_options["videopassword"] = st.session_state["video-password"]
-        video_path = download_video(
-            st.session_state["url"], save_path=os.path.join(".", "data"), **download_options
-        )
+# Sidebar actions
+if st.sidebar.button("Load Video", use_container_width=True):
+    try:
+        video_password = st.session_state.get("video-password", "")
+        video_path = download_video_ytdlp(st.session_state.get("url", ""), str(DATA_DIR), video_password)
         st.session_state["videos"] = video_path
+    except Exception as e:
+        st.sidebar.error(f"Failed to load video: {e}")
 
+# Preview & options in sidebar
 if st.session_state["videos"]:
-    
     try:
         st.sidebar.video(st.session_state["videos"], loop=st.session_state.get("loop_video", True))
-    except Exception as e:
-        st.write("Couldn't show video")
-    
+    except Exception:
+        st.sidebar.write("Couldn't preview video")
+
     with st.sidebar.expander("Options", expanded=False):
-        #Loop checkbox
-        loop_checkbox = st.checkbox("Enable Loop", value=st.session_state.get("loop_video", True), key="loop_checkbox")
-        st.session_state["loop_video"] = loop_checkbox  # Update session state based on checkbox
+        loop_checkbox = st.checkbox("Enable Loop", value=st.session_state.get("loop_video", True))
+        st.session_state["loop_video"] = loop_checkbox
+
+        if st.button("Clear Video(s)"):
+            for f in glob(str(DATA_DIR / "*")):
+                try:
+                    os.remove(f)
+                except Exception:
+                    pass
+            st.session_state["videos"] = ""
 
-        #Clear button
-        st.button("Clear Video 🔥", on_click=_remove_all_videos, type="secondary", help="Clear all downloaded videos", use_container_width=True,)
-   
         try:
-            video_file = open(st.session_state["videos"], "rb")
-            st.download_button(
-                "Download Video 📁",
-                data=video_file,
-                file_name=f"{st.session_state['videos'].lower().translate(str.maketrans('', '', string.punctuation)).replace(' ', '_')}.mp4",
-                mime="video/mp4",
-                type="secondary",
-                use_container_width=True,
-            )
+            with open(st.session_state["videos"], "rb") as video_file:
+                st.download_button(
+                    "Download Video",
+                    data=video_file,
+                    file_name=sanitize_filename(st.session_state["videos"]),
+                    mime="video/mp4",
+                    use_container_width=True,
+                )
+        except Exception:
+            st.sidebar.error("Failed to prepare download")
+
+    st.sidebar.write("Title:", Path(st.session_state["videos"]).name)
+
+# Main area action button
+if st.button("Generate the story", type="primary"):
+    if not st.session_state.get("videos"):
+        st.error("No video loaded. Use 'Load Video' in the sidebar.")
+    elif not GENAI_AVAILABLE or not API_KEY:
+        st.error("Google generative API not configured. Provide API key in Settings.")
+    elif not PHI_AVAILABLE or multimodal_Agent is None:
+        st.error("Phi/Gemini agent not available in this environment.")
+    else:
+        try:
+            with st.spinner("Uploading video to Google for processing..."):
+                processed_video = upload_file(st.session_state["videos"])
+                while getattr(processed_video, "state", None) and processed_video.state.name == "PROCESSING":
+                    time.sleep(2)
+                    processed_video = get_file(processed_video.name)
+
+            with st.spinner("Running Gemini analysis..."):
+                response = multimodal_Agent.run(analysis_prompt, videos=[processed_video], safety_settings=safety_settings)
+
+            st.subheader("Analysis Result")
+            st.markdown(response.content if hasattr(response, "content") else str(response))
         except Exception as e:
-            st.error("Failed downloading the video", icon="😔")
-            
-        st.write("**Title**:", st.session_state["videos"].split("/")[-1])        
-# --- Streamlit App ---
-
-# Main content
-if st.button('**Generate the story**', type="primary"):
-    
-            try:
-                with st.spinner("Generating the story of the video"):
-                    # Upload and process the video
-                    processed_video = upload_file(st.session_state["videos"])
-
-                    while processed_video.state.name == "PROCESSING":
-                        time.sleep(2)
-                        processed_video = get_file(processed_video.name)
-
-                    # AI agent processing
-                    response = multimodal_Agent.run(analysis_prompt, videos=[processed_video], safety_settings=safety_settings)
-
-                st.subheader('Analysis Result')
-                st.markdown(response.content)
-            
-            except Exception as error:
-                st.error(f"An error occurred: {error}")
-            
+            st.error(f"An error occurred: {e}")