Create app.py

app.py

@@ -0,0 +1,183 @@
+import streamlit as st
+import google.generativeai as genai
+import os
+import tempfile
+import time
+import mimetypes
+from pathlib import Path
+
+# Hard‑coded API key (make sure this is okay with your security setup)
+GEMINI_API_KEY = "AIzaSyDCMPwXHagWqYTQB3HL7FceHEmKUv3v4wc"
+genai.configure(api_key=GEMINI_API_KEY)
+
+# Page config
+st.set_page_config(
+    page_title="Video RAG with Gemini",
+    page_icon="🎬",
+    layout="wide"
+)
+
+# ===========================
+#   Video Processing Class
+# ===========================
+class VideoProcessor:
+    def __init__(self):
+        # Using the flash model for video context
+        self.model = genai.GenerativeModel("gemini-2.0-flash")
+
+    def upload_video(self, video_path: str, display_name: str = None):
+        try:
+            return genai.upload_file(path=video_path, display_name=display_name or "uploaded_video")
+        except Exception as e:
+            st.error(f"Video upload error: {e}")
+            return None
+
+    def wait_for_file_processing(self, video_file):
+        try:
+            while video_file.state.name == "PROCESSING":
+                time.sleep(2)
+                video_file = genai.get_file(video_file.name)
+            if video_file.state.name == "FAILED":
+                raise ValueError("Video failed to process")
+            return video_file
+        except Exception as e:
+            st.error(f"Processing error: {e}")
+            return None
+
+    def chat_with_video(self, video_file, prompt: str):
+        try:
+            resp = self.model.generate_content([video_file, prompt])
+            return resp.text
+        except Exception as e:
+            st.error(f"Generation error: {e}")
+            return None
+
+# Helper functions
+def is_video_file(file) -> bool:
+    return mimetypes.guess_type(file.name)[0].startswith("video/") if file else False
+
+def get_size_mb(file) -> float:
+    return len(file.getvalue()) / (1024**2)
+
+def reset_chat():
+    st.session_state.messages = []
+    if st.session_state.get("video_file"):
+        try:
+            genai.delete_file(st.session_state.video_file.name)
+        except:
+            pass
+        del st.session_state["video_file"]
+    for key in ["proc", "video_name"]:
+        if key in st.session_state:
+            del st.session_state[key]
+
+def display_video(bytes_, name):
+    st.markdown(f"### 🎬 {name}")
+    st.video(bytes_)
+
+# Initialize session state
+if "messages" not in st.session_state:
+    st.session_state.messages = []
+if "video_file" not in st.session_state:
+    st.session_state.video_file = None
+if "proc" not in st.session_state:
+    st.session_state.proc = None
+if "video_name" not in st.session_state:
+    st.session_state.video_name = None
+
+# Sidebar for upload and controls
+with st.sidebar:
+    st.header("📹 Upload a Video")
+    if st.session_state.proc is None:
+        st.session_state.proc = VideoProcessor()
+
+    upload = st.file_uploader("Video file", type=['mp4','mov','avi','mkv','webm'])
+    if upload:
+        if not is_video_file(upload):
+            st.error("Invalid video type.")
+        else:
+            size = get_size_mb(upload)
+            st.info(f"File size: {size:.2f} MB")
+            if size > 100:
+                st.warning("Large files may fail or take longer to process.")
+
+            if (st.session_state.video_file is None
+                or st.session_state.video_name != upload.name):
+                with st.spinner("Uploading & processing…"):
+                    tmp = tempfile.NamedTemporaryFile(delete=False, suffix=Path(upload.name).suffix)
+                    tmp.write(upload.getvalue())
+                    tmp.flush()
+                    tmp.close()
+                    vf = st.session_state.proc.upload_video(tmp.name, upload.name)
+                    if vf:
+                        pf = st.session_state.proc.wait_for_file_processing(vf)
+                        if pf:
+                            st.session_state.video_file = pf
+                            st.session_state.video_name = upload.name
+                            st.session_state.messages = []
+                            st.success("✅ Video ready!")
+                    os.unlink(tmp.name)
+
+            if st.session_state.video_file:
+                display_video(upload.getvalue(), upload.name)
+
+    st.markdown("---")
+    col1, col2 = st.columns(2)
+    with col1:
+        if st.button("🗑️ Clear Chat"):
+            st.session_state.messages = []
+            st.rerun()
+    with col2:
+        if st.button("🔄 Reset All"):
+            reset_chat()
+            st.rerun()
+
+# Main interface
+st.title("🎬 Video RAG with Gemini")
+st.markdown("Upload video & chat with its contents!")
+
+if st.session_state.video_file is None:
+    st.info("👈 Upload a video to start chatting.")
+else:
+    st.success(f"✅ Chatting on: **{st.session_state.video_name}**")
+    for msg in st.session_state.messages:
+        with st.chat_message(msg["role"]):
+            st.markdown(msg["content"])
+
+    if not st.session_state.messages:
+        st.markdown("### 💡 Examples:")
+        for ex in [
+            "What is happening in this video?",
+            "Summarize the main events",
+            "Who is present?"
+        ]:
+            if st.button(ex, key=ex):
+                st.session_state.messages.append({"role": "user", "content": ex})
+                st.rerun()
+
+    if prompt := st.chat_input("Ask a question..."):
+        st.session_state.messages.append({"role": "user", "content": prompt})
+        with st.chat_message("user"):
+            st.markdown(prompt)
+        with st.chat_message("assistant"):
+            ph = st.empty()
+            with st.spinner("Thinking…"):
+                resp = st.session_state.proc.chat_with_video(
+                    st.session_state.video_file, prompt
+                )
+            if resp:
+                full = ""
+                for word in resp.split():
+                    full += word + " "
+                    ph.markdown(full + "▌")
+                    time.sleep(0.02)
+                ph.markdown(resp)
+                st.session_state.messages.append({"role":"assistant","content":resp})
+            else:
+                st.error("No response — please try again.")
+
+# Footer
+st.markdown("""---
+<p style='text-align:center; color:#666'>
+Built with ❤️ using Gemini Video API
+</p>""", unsafe_allow_html=True)