Update streamlit_app.py

streamlit_app.py

@@ -44,13 +44,11 @@ MODEL_OPTIONS = [
 # Helper utilities
 # ----------------------------------------------------------------------
 def _sanitize_filename(url: str) -> str:
-    """Make a safe filename from a URL."""
     name = Path(url).name.lower()
     return name.translate(str.maketrans("", "", string.punctuation)).replace(" ", "_")
 
 
 def _file_sha256(path: Path) -> Optional[str]:
-    """Return SHA‑256 hex digest of *path* or None on error."""
     try:
         h = hashlib.sha256()
         with path.open("rb") as f:
@@ -62,7 +60,6 @@ def _file_sha256(path: Path) -> Optional[str]:
 
 
 def _convert_to_mp4(src: Path) -> Path:
-    """Convert *src* to MP4 (ffmpeg) and delete the original."""
     dst = src.with_suffix(".mp4")
     if dst.exists():
         return dst
@@ -79,7 +76,6 @@ def _convert_to_mp4(src: Path) -> Path:
 
 
 def _compress_video(inp: Path, crf: int = 28, preset: str = "fast") -> Path:
-    """Compress *inp* with libx264; return the new file."""
     out = inp.with_name(f"{inp.stem}_compressed.mp4")
     try:
         ffmpeg.input(str(inp)).output(
@@ -91,7 +87,6 @@ def _compress_video(inp: Path, crf: int = 28, preset: str = "fast") -> Path:
 
 
 def _maybe_compress(path: Path, limit_mb: int) -> Tuple[Path, bool]:
-    """Compress *path* if it exceeds *limit_mb*."""
     size_mb = path.stat().st_size / (1024 * 1024)
     if size_mb <= limit_mb:
         return path, False
@@ -99,7 +94,6 @@ def _maybe_compress(path: Path, limit_mb: int) -> Tuple[Path, bool]:
 
 
 def _download_direct(url: str, dst: Path) -> Path:
-    """Simple HTTP GET download."""
     r = requests.get(url, stream=True, timeout=30)
     r.raise_for_status()
     out = dst / _sanitize_filename(url.split("/")[-1])
@@ -111,7 +105,6 @@ def _download_direct(url: str, dst: Path) -> Path:
 
 
 def _download_with_yt_dlp(url: str, dst: Path, password: str = "") -> Path:
-    """Download via yt‑dlp, returning the newest MP4."""
     tmpl = str(dst / "%(id)s.%(ext)s")
     fmt = "best[ext=mp4]/best"
 
@@ -156,7 +149,6 @@ def _download_with_yt_dlp(url: str, dst: Path, password: str = "") -> Path:
         raise RuntimeError("No MP4 file was created.")
     newest = max(mp4_files, key=lambda p: p.stat().st_mtime)
 
-    # Deduplicate via SHA‑256 cache
     sha = _file_sha256(newest)
     if sha:
         for existing in dst.iterdir():
@@ -167,7 +159,6 @@ def _download_with_yt_dlp(url: str, dst: Path, password: str = "") -> Path:
 
 
 def download_video(url: str, dst: Path, password: str = "") -> Path:
-    """Unified download entry point."""
     video_exts = (".mp4", ".mov", ".webm", ".mkv", ".avi", ".flv")
 
     if url.lower().endswith(video_exts):
@@ -188,12 +179,10 @@ def download_video(url: str, dst: Path, password: str = "") -> Path:
 
 
 def _encode_video_b64(path: Path) -> str:
-    """Base64‑encode a file."""
     return base64.b64encode(path.read_bytes()).decode()
 
 
 def generate_report(video_path: Path, prompt: str, model_id: str, timeout: int = 300) -> str:
-    """Send video + prompt to Gemini and return the response text."""
     b64 = _encode_video_b64(video_path)
     video_part = {"inline_data": {"mime_type": "video/mp4", "data": b64}}
     model = genai.GenerativeModel(model_name=model_id)
@@ -207,7 +196,6 @@ def generate_report(video_path: Path, prompt: str, model_id: str, timeout: int =
 
 
 def _strip_prompt_echo(prompt: str, text: str, threshold: float = 0.68) -> str:
-    """Trim the prompt if the model repeats it at the start."""
     if not prompt or not text:
         return text
     clean_prompt = " ".join(prompt.lower().split())
@@ -222,7 +210,6 @@ def _strip_prompt_echo(prompt: str, text: str, threshold: float = 0.68) -> str:
 # UI helpers
 # ----------------------------------------------------------------------
 def _expand_sidebar(width: int = 380) -> None:
-    """Make the Streamlit sidebar wider."""
     st.markdown(
         f"""
         <style>
@@ -235,11 +222,128 @@ def _expand_sidebar(width: int = 380) -> None:
         unsafe_allow_html=True,
     )
 
+
+# ----------------------------------------------------------------------
+# Streamlit UI
+# ----------------------------------------------------------------------
+def main() -> None:
+    st.set_page_config(page_title="Video Analysis", layout="wide")
+    _expand_sidebar()
+
+    # ---------- Sidebar ----------
+    st.sidebar.header("Video Input")
+    st.sidebar.text_input("Video URL", key="url", placeholder="https://")
+
+    if st.sidebar.button("Load Video"):
+        try:
+            with st.spinner("Downloading video…"):
+                raw_path = download_video(
+                    st.session_state["url"], DATA_DIR, st.session_state["video_password"]
+                )
+                mp4_path = _convert_to_mp4(Path(raw_path))
+                # Optional compression (if size exceeds user‑defined limit)
+                mp4_path, _ = _maybe_compress(mp4_path, st.session_state["compress_mb"])
+            st.session_state["video_path"] = str(mp4_path)
+            st.session_state["last_error"] = ""
+            st.toast("Video ready")
+            st.experimental_rerun()
+        except Exception as e:
+            st.session_state["last_error"] = f"Download failed: {e}"
+            st.sidebar.error(st.session_state["last_error"])
+
+    # ---------- Settings ----------
+    with st.sidebar.expander("Settings", expanded=False):
+        model = st.selectbox(
+            "Model", MODEL_OPTIONS, index=MODEL_OPTIONS.index(DEFAULT_MODEL)
+        )
+        if model == "custom":
+            model = st.text_input("Custom model ID", value=DEFAULT_MODEL, key="custom_model")
+        st.session_state["model_input"] = model
+
+        # API key handling
+        secret_key = os.getenv("GOOGLE_API_KEY", "")
+        if secret_key:
+            st.session_state["api_key"] = secret_key
+        st.text_input("Google API Key", key="api_key", type="password")
+
+        st.text_area(
+            "Analysis prompt",
+            value=DEFAULT_PROMPT,
+            key="prompt",
+            height=140,
+        )
+        st.text_input(
+            "Video password (if needed)",
+            key="video_password",
+            type="password",
+        )
+        st.number_input(
+            "Compress if > (MB)",
+            min_value=10,
+            max_value=2000,
+            value=st.session_state.get("compress_mb", 200),
+            step=10,
+            key="compress_mb",
+        )
+
+        if st.sidebar.button("Clear Video"):
+            for f in DATA_DIR.iterdir():
+                try:
+                    f.unlink()
+                except Exception:
+                    pass
+            st.session_state["video_path"] = ""
+            st.session_state["analysis_out"] = ""
+            st.session_state["raw_output"] = ""
+            st.toast("All cached videos cleared")
+            st.experimental_rerun()
+
+    # ---------- Main panel ----------
+    if st.session_state["last_error"]:
+        st.error(st.session_state["last_error"])
+
+    if st.session_state.get("video_path"):
+        st.video(st.session_state["video_path"])
+
+        if st.button("Run Analysis"):
+            st.session_state["busy"] = True
+            st.session_state["analysis_out"] = ""
+            st.session_state["raw_output"] = ""
+            try:
+                with st.spinner("Generating report…"):
+                    raw = generate_report(
+                        Path(st.session_state["video_path"]),
+                        st.session_state["prompt"],
+                        st.session_state["model_input"],
+                    )
+                    cleaned = _strip_prompt_echo(st.session_state["prompt"], raw)
+                    st.session_state["analysis_out"] = cleaned
+                    st.session_state["raw_output"] = raw
+            except Exception as e:
+                st.session_state["last_error"] = f"Analysis failed: {e}"
+                st.session_state["last_error_detail"] = traceback.format_exc()
+            finally:
+                st.session_state["busy"] = False
+                st.experimental_rerun()
+
+        # Show results
+        if st.session_state.get("analysis_out"):
+            st.subheader("📝 Analysis")
+            st.write(st.session_state["analysis_out"])
+
+            with st.expander("Show raw model output"):
+                st.code(st.session_state["raw_output"], language="text")
+
+        # Optional error details
+        if st.session_state.get("last_error_detail"):
+            with st.expander("Show error details"):
+                st.code(st.session_state["last_error_detail"], language="text")
+
+
 # ----------------------------------------------------------------------
 # Session‑state defaults
 # ----------------------------------------------------------------------
 def _init_state() -> None:
-    """Populate Streamlit's session_state with sensible defaults."""
     defaults = {
         "url": "",
         "video_path": "",
@@ -253,23 +357,16 @@ def _init_state() -> None:
         "analysis_out": "",
         "raw_output": "",
         "last_error_detail": "",
-        "show_raw_on_error": False,
-        "show_analysis": False,
     }
     for k, v in defaults.items():
         st.session_state.setdefault(k, v)
 
 
 # ----------------------------------------------------------------------
-# Main entry point
+# Entry point
 # ----------------------------------------------------------------------
 if __name__ == "__main__":
-    # Initialise session state before any UI code runs
     _init_state()
-
-    # Initialise the Gemini API – the key can be supplied via the sidebar or env var
     if st.session_state["api_key"]:
         genai.configure(api_key=st.session_state["api_key"])
-
-    # Run the Streamlit app
     main()