Update streamlit_app.py

streamlit_app.py

@@ -1,24 +1,32 @@
-# streamlit_app.py
-import requests          # for direct‑video download
-import snscrape.modules.twitter as sntwitter  # Twitter scraper
+#!/usr/bin/env python3
+import os
 import base64
 import hashlib
-import os
 import string
 import traceback
-from glob import glob
 from pathlib import Path
+from typing import List, Tuple, Optional
 
+import requests
+import streamlit as st
 import ffmpeg
 import google.generativeai as genai
-import streamlit as st
 import yt_dlp
-from dotenv import load_dotenv
 
-load_dotenv()
+# ----------------------------------------------------------------------
+# Optional imports – give a clear message if missing
+# ----------------------------------------------------------------------
+try:
+    import snscrape.modules.twitter as sntwitter
+except ImportError:  # pragma: no cover
+    st.error(
+        "The package `snscrape` is required for Twitter video extraction. "
+        "Install it with `pip install snscrape`."
+    )
+    st.stop()
 
 # ----------------------------------------------------------------------
-# Configuration & constants
+# Configuration & defaults
 # ----------------------------------------------------------------------
 DATA_DIR = Path("./data")
 DATA_DIR.mkdir(exist_ok=True)
@@ -37,11 +45,8 @@ DEFAULT_PROMPT = (
     "Include a list of observations for notable events."
 )
 
-# ----------------------------------------------------------------------
-# Session defaults
-# ----------------------------------------------------------------------
-st.set_page_config(page_title="Video Analysis Tool", layout="wide")
-for key, val in {
+# Session‑state defaults (kept in one dict for readability)
+DEFAULT_SESSION_STATE = {
     "url": "",
     "videos": "",
     "loop_video": False,
@@ -55,18 +60,24 @@ for key, val in {
     "processing_timeout": 900,
     "generation_timeout": 300,
     "compress_threshold_mb": 200,
-}.items():
-    st.session_state.setdefault(key, val)
+}
+for k, v in DEFAULT_SESSION_STATE.items():
+    st.session_state.setdefault(k, v)
 
 # ----------------------------------------------------------------------
 # Helper utilities
 # ----------------------------------------------------------------------
+from difflib import SequenceMatcher  # imported once for clarity
+
+
 def sanitize_filename(p: str) -> str:
+    """Return a lower‑case, punctuation‑free filename."""
     name = Path(p).name.lower()
     return name.translate(str.maketrans("", "", string.punctuation)).replace(" ", "_")
 
 
-def file_sha256(p: str, block: int = 65536) -> str | None:
+def file_sha256(p: str, block: int = 65536) -> Optional[str]:
+    """SHA‑256 hash of a file; returns None on error."""
     try:
         h = hashlib.sha256()
         with open(p, "rb") as f:
@@ -78,73 +89,126 @@ def file_sha256(p: str, block: int = 65536) -> str | None:
 
 
 def convert_to_mp4(src: str) -> str:
+    """Convert *src* to MP4 with ffmpeg; returns the MP4 path."""
     dst = str(Path(src).with_suffix(".mp4"))
     if os.path.exists(dst):
         return dst
-    ffmpeg.input(src).output(dst).run(overwrite_output=True, quiet=True)
-    if os.path.exists(dst):
+    try:
+        ffmpeg.input(src).output(dst).overwrite_output().run(
+            capture_stdout=True, capture_stderr=True
+        )
+    except ffmpeg.Error as e:
+        raise RuntimeError(f"ffmpeg conversion failed: {e.stderr.decode()}") from e
+
+    # Delete source only if conversion succeeded and output is non‑empty
+    if os.path.exists(dst) and os.path.getsize(dst) > 0:
         os.remove(src)
     return dst
 
 
 def compress_video(inp: str, out: str, crf: int = 28, preset: str = "fast") -> str:
-    ffmpeg.input(inp).output(out, vcodec="libx264", crf=crf, preset=preset).run(
-        overwrite_output=True, quiet=True
-    )
+    """Compress *inp* to *out* using libx264."""
+    try:
+        ffmpeg.input(inp).output(
+            out, vcodec="libx264", crf=crf, preset=preset
+        ).overwrite_output().run(capture_stdout=True, capture_stderr=True)
+    except ffmpeg.Error as e:
+        raise RuntimeError(f"ffmpeg compression failed: {e.stderr.decode()}") from e
     return out if os.path.exists(out) else inp
 
 
+def maybe_compress(path: str, limit_mb: int) -> Tuple[str, bool]:
+    """Compress *path* if its size exceeds *limit_mb*.
+    Returns (final_path, was_compressed)."""
+    size_mb = os.path.getsize(path) / (1024 * 1024)
+    if size_mb <= limit_mb:
+        return path, False
+    out = str(Path(path).with_name(f"{Path(path).stem}_compressed.mp4"))
+    return compress_video(path, out), True
+
+
+def strip_prompt_echo(prompt: str, text: str, threshold: float = 0.68) -> str:
+    """Remove the prompt if the model repeats it at the start of *text*."""
+    if not prompt or not text:
+        return text
+    clean_prompt = " ".join(prompt.lower().split())
+    snippet = " ".join(text.lower().split()[:600])
+
+    if SequenceMatcher(None, clean_prompt, snippet).ratio() > threshold:
+        cut = max(len(clean_prompt), int(len(prompt) * 0.9))
+        return text[cut:].lstrip(" \n:-")
+    return text
+
+
+def generate_inline(
+    video_path: str, prompt: str, model_id: str, timeout: int
+) -> str:
+    """Encode *video_path* as base64 and call Gemini."""
+    with open(video_path, "rb") as f:
+        b64 = base64.b64encode(f.read()).decode()
+
+    video_part = {"inline_data": {"mime_type": "video/mp4", "data": b64}}
+    contents = [prompt, video_part]
+
+    model = genai.GenerativeModel(model_name=model_id)
+    resp = model.generate_content(
+        contents,
+        generation_config={"max_output_tokens": 1024},
+        request_options={"timeout": timeout},
+    )
+    return getattr(resp, "text", str(resp))
+
+
 def download_video(url: str, dst_dir: str, password: str = "") -> str:
     """
     Download a video from *url*.
-    - If the URL ends with a known video extension → plain HTTP download.
-    - If it is a YouTube / archive URL → try yt‑dlp (will raise a clear error if DNS fails).
-    - If it is a Twitter status URL → scrape the tweet for the first video URL.
+    1️⃣ Direct video file → HTTP GET.
+    2️⃣ Twitter status → scrape for video URL.
+    3️⃣ Fallback → yt‑dlp (YouTube, archive.org, etc.).
+    Returns the path to an MP4 file.
     """
     video_exts = (".mp4", ".mov", ".webm", ".mkv", ".avi", ".flv")
+    dst_dir = Path(dst_dir)
 
     # --------------------------------------------------------------
-    # 1️⃣ Direct video file (no yt‑dlp needed)
+    # 1️⃣ Direct video file
     # --------------------------------------------------------------
     if url.lower().endswith(video_exts):
         try:
             r = requests.get(url, stream=True, timeout=30)
             r.raise_for_status()
             filename = sanitize_filename(url.split("/")[-1])
-            out_path = Path(dst_dir) / filename
+            out_path = dst_dir / filename
             with open(out_path, "wb") as f:
                 for chunk in r.iter_content(chunk_size=8192):
                     if chunk:
                         f.write(chunk)
-            return str(out_path)          # already an MP4 (or other supported format)
+            return str(out_path)
         except Exception as e:
-            raise RuntimeError(f"Direct download failed: {e}")
+            raise RuntimeError(f"Direct download failed: {e}") from e
 
     # --------------------------------------------------------------
-    # 2️⃣ Twitter status – scrape for video URL
+    # 2️⃣ Twitter status
     # --------------------------------------------------------------
     if "twitter.com" in url and "/status/" in url:
         try:
             tweet_id = url.split("/")[-1].split("?")[0]
             for tweet in sntwitter.TwitterTweetScraper(tweet_id).get_items():
-                # Look for media URLs in the tweet JSON
-                media = tweet.media
-                if media:
-                    for m in media:
+                if getattr(tweet, "media", None):
+                    for m in tweet.media:
                         if getattr(m, "video_url", None):
-                            return download_video(m.video_url, dst_dir)  # recurse as a direct URL
-                # If no media field, fall back to expanded URLs
-                for u in tweet.urls:
+                            return download_video(m.video_url, str(dst_dir))
+                for u in getattr(tweet, "urls", []):
                     if u.expandedUrl.lower().endswith(video_exts):
-                        return download_video(u.expandedUrl, dst_dir)
+                        return download_video(u.expandedUrl, str(dst_dir))
             raise RuntimeError("No video found in the tweet.")
         except Exception as e:
-            raise RuntimeError(f"Twitter scrape failed: {e}")
+            raise RuntimeError(f"Twitter scrape failed: {e}") from e
 
     # --------------------------------------------------------------
-    # 3️⃣ Fallback to yt‑dlp (YouTube, archive.org, etc.)
+    # 3️⃣ yt‑dlp fallback
     # --------------------------------------------------------------
-    tmpl = str(Path(dst_dir) / "%(id)s.%(ext)s")
+    tmpl = str(dst_dir / "%(id)s.%(ext)s")
     opts = {"outtmpl": tmpl, "format": "best"}
     if password:
         opts["videopassword"] = password
@@ -153,282 +217,251 @@ def download_video(url: str, dst_dir: str, password: str = "") -> str:
         with yt_dlp.YoutubeDL(opts) as ydl:
             info = ydl.extract_info(url, download=True)
     except Exception as e:
-        # Re‑raise with a short, user‑friendly message
         raise RuntimeError(
             f"yt‑dlp could not download the URL. "
             f"Common reasons: DNS failure, unsupported site, or missing video. "
             f"Original error: {e}"
-        )
+        ) from e
 
-    # Locate the downloaded file (same logic as before)
-    if isinstance(info, dict) and info.get("id"):
+    # If yt‑dlp gave us a predictable filename, use it
+    if isinstance(info, dict) and "id" in info:
         vid_id = info["id"]
         ext = info.get("ext", "mp4")
-        candidate = str(Path(dst_dir) / f"{vid_id}.{ext}")
-        if os.path.exists(candidate):
-            return convert_to_mp4(candidate)
-
-    # Fallback: newest file in the folder
-    newest = max(glob(os.path.join(dst_dir, "*")), key=os.path.getmtime)
-    return convert_to_mp4(newest)
-
-
-def maybe_compress(path: str, limit_mb: int) -> str:
-    size_mb = os.path.getsize(path) / (1024 * 1024)
-    if size_mb <= limit_mb:
-        return path
-    out = str(Path(path).with_name(f"{Path(path).stem}_compressed.mp4"))
-    return compress_video(path, out)
+        candidate = dst_dir / f"{vid_id}.{ext}"
+        if candidate.exists():
+            return convert_to_mp4(str(candidate))
 
-
-def strip_prompt_echo(prompt: str, text: str) -> str:
-    if not prompt or not text:
-        return text
-    clean_prompt = " ".join(prompt.lower().split())
-    snippet = " ".join(text.lower().split()[:600])
-    from difflib import SequenceMatcher
-
-    if SequenceMatcher(None, clean_prompt, snippet).ratio() > 0.68:
-        cut = max(len(clean_prompt), int(len(prompt) * 0.9))
-        return text[cut:].lstrip(" \n:-")
-    return text
-
-
-def generate_inline(video_path: str, prompt: str, model_id: str, timeout: int) -> str:
-    # 1️⃣ read video and encode as base64
-    with open(video_path, "rb") as f:
-        b64 = base64.b64encode(f.read()).decode()
-
-    # 2️⃣ build the “inline” part that the Gemini API expects
-    video_part = {
-        "inline_data": {"mime_type": "video/mp4", "data": b64}
-    }
-    contents = [prompt, video_part]
-
-    # 3️⃣ create a GenerativeModel object (no Client class any more)
-    model = genai.GenerativeModel(model_name=model_id)
-
-    # 4️⃣ call generate_content – the timeout is passed via the request options
-    resp = model.generate_content(
-        contents,
-        generation_config={"max_output_tokens": 1024},
-        # `timeout` is a kw‑arg of the underlying HTTP request; the SDK forwards it
-        request_options={"timeout": timeout},
-    )
-
-    # 5️⃣ the response object has a `.text` attribute (or `.parts` for multi‑part)
-    return getattr(resp, "text", str(resp))
+    # Fallback: newest file in the folder (yt‑dlp sometimes uses different naming)
+    if not any(dst_dir.iterdir()):
+        raise RuntimeError("yt‑dlp did not download any files.")
+    newest = max(dst_dir.iterdir(), key=lambda p: p.stat().st_mtime)
+    return convert_to_mp4(str(newest))
 
 
 # ----------------------------------------------------------------------
-# Sidebar – inputs & settings
+# Streamlit UI
 # ----------------------------------------------------------------------
-st.sidebar.header("Video Input")
-st.sidebar.text_input("Video URL", key="url", placeholder="https://")
-
-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
-
-    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(
-        "Processing timeout (s)",
-        min_value=60,
-        max_value=3600,
-        value=st.session_state["processing_timeout"],
-        step=30,
-        key="processing_timeout",
-    )
-    st.number_input(
-        "Generation timeout (s)",
-        min_value=30,
-        max_value=1800,
-        value=st.session_state["generation_timeout"],
-        step=10,
-        key="generation_timeout",
-    )
-    st.number_input(
-        "Compress if > (MB)",
-        min_value=10,
-        max_value=2000,
-        value=st.session_state["compress_threshold_mb"],
-        step=10,
-        key="compress_threshold_mb",
-    )
+def main() -> None:
+    st.set_page_config(page_title="Video Analysis Tool", layout="wide")
 
-# ----------------------------------------------------------------------
-# Load / preview video
-# ----------------------------------------------------------------------
-if st.sidebar.button("Load Video"):
-    try:
-        path = download_video(
-            st.session_state["url"], str(DATA_DIR), st.session_state["video_password"]
-        )
-        st.session_state["videos"] = path
-        st.session_state["last_error"] = ""
-    except Exception as e:
-        st.session_state["last_error"] = f"Download failed: {e}"
-        st.sidebar.error(st.session_state["last_error"])
-
-# --------------------------------------------------------------
-# 📥 Twitter video extractor – hidden inside an expander
-# --------------------------------------------------------------
-with st.sidebar.expander("🔎 Extract video(s) from a Tweet", expanded=False):
-    tweet_url = st.text_input(
-        "Tweet URL (e.g. https://twitter.com/user/status/1234567890)",
-        key="tweet_url",
-    )
+    # ---------- Sidebar inputs ----------
+    st.sidebar.header("Video Input")
+    st.sidebar.text_input("Video URL", key="url", placeholder="https://")
 
-    # ------------------------------------------------------------------
-    # Step 1 – Scrape the tweet and collect every video URL
-    # ------------------------------------------------------------------
-    if st.button("Find videos in tweet"):
-        if not tweet_url:
-            st.error("Paste a tweet URL first.")
-        else:
-            try:
-                tweet_id = tweet_url.split("/")[-1].split("?")[0]
-                video_urls: list[str] = []
-
-                for tweet in sntwitter.TwitterTweetScraper(tweet_id).get_items():
-                    # media objects (video, gif, etc.)
-                    if getattr(tweet, "media", None):
-                        for m in tweet.media:
-                            if getattr(m, "video_url", None):
-                                video_urls.append(m.video_url)
-
-                    # expanded URLs that already end with a video extension
-                    for u in getattr(tweet, "urls", []):
-                        if u.expandedUrl.lower().endswith(
-                            (".mp4", ".mov", ".webm", ".mkv", ".avi", ".flv")
-                        ):
-                            video_urls.append(u.expandedUrl)
-
-                # deduplicate while preserving order
-                video_urls = list(dict.fromkeys(video_urls))
-
-                if not video_urls:
-                    raise RuntimeError("No video URLs detected in this tweet.")
-
-                # store a friendly label + raw URL for the selector
-                st.session_state["tweet_video_options"] = [
-                    (f"Video {i+1} – {url.split('/')[-1][:30]}...", url)
-                    for i, url in enumerate(video_urls)
-                ]
-                st.success(f"Found {len(video_urls)} video(s). Choose one below.")
-            except Exception as e:
-                st.session_state["tweet_video_options"] = []
-                st.error(f"Tweet scrape failed: {e}")
-
-    # ------------------------------------------------------------------
-    # Step 2 – Let the user pick one of the discovered videos
-    # ------------------------------------------------------------------
-    if st.session_state.get("tweet_video_options"):
-        display_names, raw_urls = zip(*st.session_state["tweet_video_options"])
-        sel_idx = st.selectbox(
-            "Select video to download",
-            options=range(len(display_names)),
-            format_func=lambda i: display_names[i],
-            key="tweet_video_select",
+    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
 
-        # ------------------------------------------------------------------
-        # Step 3 – Download the chosen video
-        # ------------------------------------------------------------------
-        if st.button("Download selected video"):
-            chosen_url = raw_urls[sel_idx]
-            try:
-                path = download_video(chosen_url, str(DATA_DIR))
-                st.session_state["videos"] = path
-                st.session_state["last_error"] = ""
-                st.success("Video downloaded and loaded.")
-            except Exception as e:
-                st.session_state["last_error"] = f"Download failed: {e}"
-                st.error(st.session_state["last_error"])
-    else:
-        st.info(
-            "Paste a tweet URL and click **Find videos in tweet** to discover available videos."
-        )
+        # API key – prefer Streamlit secrets, fall back to env / session state
+        secret_key = st.secrets.get("google_api_key")
+        if secret_key:
+            st.session_state["api_key"] = secret_key
+        st.text_input("Google API Key", key="api_key", type="password")
 
-if st.session_state["videos"]:
-    try:
-        mp4 = convert_to_mp4(st.session_state["videos"])
-        with open(mp4, "rb") as f:
-            video_bytes = f.read()
-        st.sidebar.video(video_bytes, format="video/mp4")
-    except Exception:
-        st.sidebar.write("Preview unavailable")
+        st.text_area(
+            "Analysis prompt", value=DEFAULT_PROMPT, key="prompt", height=140
+        )
+        st.text_input(
+            "Video password (if needed)", key="video_password", type="password"
+        )
 
-    if st.sidebar.button("Clear Video"):
-        for f in glob(str(DATA_DIR / "*")):
-            try:
-                os.remove(f)
-            except Exception:
-                pass
-        st.session_state.update(
-            {
-                "videos": "",
-                "analysis_out": "",
-                "last_error": "",
-                "busy": False,
-            }
+        st.number_input(
+            "Processing timeout (s)",
+            min_value=60,
+            max_value=3600,
+            value=st.session_state["processing_timeout"],
+            step=30,
+            key="processing_timeout",
+        )
+        st.number_input(
+            "Generation timeout (s)",
+            min_value=30,
+            max_value=1800,
+            value=st.session_state["generation_timeout"],
+            step=10,
+            key="generation_timeout",
+        )
+        st.number_input(
+            "Compress if > (MB)",
+            min_value=10,
+            max_value=2000,
+            value=st.session_state["compress_threshold_mb"],
+            step=10,
+            key="compress_threshold_mb",
         )
 
-# ----------------------------------------------------------------------
-# Generation
-# ----------------------------------------------------------------------
-col1, col2 = st.columns([1, 3])
-with col1:
-    generate_now = st.button("Generate analysis", type="primary")
-with col2:
-    if not st.session_state["videos"]:
-        st.info("Load a video first.", icon="ℹ️")
-
-if generate_now and not st.session_state["busy"]:
-    if not st.session_state["videos"]:
-        st.error("No video loaded.")
-    elif not (st.session_state["api_key"] or os.getenv("GOOGLE_API_KEY")):
-        st.error("Google API key missing.")
-    else:
-        st.session_state["busy"] = True
+    # ---------- Load video ----------
+    if st.sidebar.button("Load Video"):
         try:
-            genai.configure(api_key=st.session_state["api_key"] or os.getenv("GOOGLE_API_KEY"))
-            model_id = st.session_state["model_input"]
-            prompt = st.session_state["prompt"]
-
-            # Optional compression to stay under inline limits
-            video_path = maybe_compress(
-                st.session_state["videos"], st.session_state["compress_threshold_mb"]
-            )
-
-            with st.spinner("Generating…"):
-                raw_out = generate_inline(
-                    video_path, prompt, model_id, st.session_state["generation_timeout"]
+            with st.spinner("Downloading video…"):
+                path = download_video(
+                    st.session_state["url"], str(DATA_DIR), st.session_state["video_password"]
                 )
-            out = strip_prompt_echo(prompt, raw_out)
-            st.session_state["analysis_out"] = out
-            st.subheader("Analysis Result")
-            st.markdown(out or "No output.")
+            st.session_state["videos"] = path
+            st.session_state["last_error"] = ""
+            st.success("Video loaded successfully.")
         except Exception as e:
-            # Capture full traceback for debugging while showing a short message to the user
-            tb = traceback.format_exc()
-            st.session_state["last_error"] = f"Generation error: {e}"
-            st.error("An error occurred during generation. Check the log below.")
-            st.code(tb, language="text")
-        finally:
-            st.session_state["busy"] = False
+            st.session_state["last_error"] = f"Download failed: {e}"
+            st.sidebar.error(st.session_state["last_error"])
+
+    # ---------- Twitter extractor ----------
+    with st.sidebar.expander("🔎 Extract video(s) from a Tweet", expanded=False):
+        tweet_url = st.text_input(
+            "Tweet URL (e.g. https://twitter.com/user/status/1234567890)",
+            key="tweet_url",
+        )
+        if st.button("Find videos in tweet"):
+            if not tweet_url:
+                st.error("Paste a tweet URL first.")
+            else:
+                try:
+                    tweet_id = tweet_url.split("/")[-1].split("?")[0]
+                    video_urls: List[str] = []
+                    for tweet in sntwitter.TwitterTweetScraper(tweet_id).get_items():
+                        if getattr(tweet, "media", None):
+                            for m in tweet.media:
+                                if getattr(m, "video_url", None):
+                                    video_urls.append(m.video_url)
+                        for u in getattr(tweet, "urls", []):
+                            if u.expandedUrl.lower().endswith(
+                                (".mp4", ".mov", ".webm", ".mkv", ".avi", ".flv")
+                            ):
+                                video_urls.append(u.expandedUrl)
+
+                    video_urls = list(dict.fromkeys(video_urls))  # dedupe, preserve order
+                    if not video_urls:
+                        raise RuntimeError("No video URLs detected in this tweet.")
+
+                    st.session_state["tweet_video_options"] = [
+                        (f"Video {i+1} – {url.split('/')[-1][:30]}...", url)
+                        for i, url in enumerate(video_urls)
+                    ]
+                    st.success(f"Found {len(video_urls)} video(s).")
+                except Exception as e:
+                    st.session_state["tweet_video_options"] = []
+                    st.error(f"Tweet scrape failed: {e}")
+
+        # selector & download
+        if st.session_state.get("tweet_video_options"):
+            labels, urls = zip(*st.session_state["tweet_video_options"])
+            sel = st.selectbox(
+                "Select video to download",
+                options=range(len(labels)),
+                format_func=lambda i: labels[i],
+                key="tweet_video_select",
+            )
+            if st.button("Download selected video"):
+                try:
+                    with st.spinner("Downloading selected video…"):
+                        path = download_video(urls[sel], str(DATA_DIR))
+                    st.session_state["videos"] = path
+                    st.session_state["last_error"] = ""
+                    st.success("Video downloaded and loaded.")
+                except Exception as e:
+                    st.session_state["last_error"] = f"Download failed: {e}"
+                    st.error(st.session_state["last_error"])
+        else:
+            st.info(
+                "Paste a tweet URL and click **Find videos in tweet** to discover available videos."
+            )
 
-# ----------------------------------------------------------------------
-# Show results / errors
-# ----------------------------------------------------------------------
-if st.session_state["analysis_out"]:
-    st.subheader("📝 Analysis")
-    st.markdown(st.session_state["analysis_out"])
+    # ---------- Video preview ----------
+    if st.session_state["videos"]:
+        try:
+            mp4_path = convert_to_mp4(st.session_state["videos"])
+            st.sidebar.video(str(mp4_path))
+        except Exception:
+            st.sidebar.write("Preview unavailable")
+
+        if st.sidebar.button("Clear Video"):
+            for f in DATA_DIR.iterdir():
+                try:
+                    f.unlink()
+                except Exception:
+                    pass
+            st.session_state.update(
+                {
+                    "videos": "",
+                    "analysis_out": "",
+                    "last_error": "",
+                    "busy": False,
+                }
+            )
+            st.success("Session cleared.")
 
-if st.session_state["last_error"]:
-    st.subheader("❗️ Error log")
-    st.code(st.session_state["last_error"], language="text")
+    # ---------- Generation ----------
+    col1, col2 = st.columns([1, 3])
+    with col1:
+        generate_now = st.button(
+            "Generate analysis", type="primary", disabled=st.session_state["busy"]
+        )
+    with col2:
+        if not st.session_state["videos"]:
+            st.info("Load a video first.", icon="ℹ️")
+
+    if generate_now and not st.session_state["busy"]:
+        api_key = st.session_state["api_key"] or os.getenv("GOOGLE_API_KEY")
+        if not st.session_state["videos"]:
+            st.error("No video loaded.")
+        elif not api_key:
+            st.error("Google API key missing.")
+        else:
+            try:
+                st.session_state["busy"] = True
+                genai.configure(api_key=api_key)
+                model_id = st.session_state["model_input"]
+                prompt = st.session_state["prompt"]
+
+                # ---- optional compression ----
+                with st.spinner("Checking video size…"):
+                    video_path, was_compressed = maybe_compress(
+                        st.session_state["videos"],
+                        st.session_state["compress_threshold_mb"],
+                    )
+
+                # ---- generation ----
+                with st.spinner("Generating analysis…"):
+                    raw_out = generate_inline(
+                        video_path,
+                        prompt,
+                        model_id,
+                        st.session_state["generation_timeout"],
+                    )
+
+                # clean up temporary compressed file
+                if was_compressed:
+                    try:
+                        os.remove(video_path)
+                    except OSError:
+                        pass
+
+                out = strip_prompt_echo(prompt, raw_out)
+                st.session_state["analysis_out"] = out
+                st.success("Analysis generated successfully.")
+                st.markdown(out or "No output.")
+            except Exception as exc:
+                tb = traceback.format_exc()
+                st.session_state["last_error"] = f"Generation error: {exc}"
+                st.error("An error occurred during generation.")
+                st.code(tb, language="text")
+            finally:
+                st.session_state["busy"] = False
+
+    # ---------- Results / errors ----------
+    if st.session_state["analysis_out"]:
+        st.subheader("📝 Analysis")
+        st.markdown(st.session_state["analysis_out"])
+
+    if st.session_state["last_error"]:
+        with st.expander("❗️ Error details"):
+            st.code(st.session_state["last_error"], language="text")
+
+
+if __name__ == "__main__":
+    main()