Update streamlit_app.py

streamlit_app.py

@@ -8,6 +8,9 @@ from glob import glob
 from pathlib import Path
 from difflib import SequenceMatcher
 
+import requests
+from bs4 import BeautifulSoup
+
 import yt_dlp
 import ffmpeg
 import streamlit as st
@@ -15,30 +18,25 @@ from dotenv import load_dotenv
 
 load_dotenv()
 
-# phi agent disabled to avoid phi IndexError
-try:
-    from phi.agent import Agent  # noqa: F401
-    from phi.model.google import Gemini  # noqa: F401
-    from phi.tools.duckduckgo import DuckDuckGo  # noqa: F401
-    HAS_PHI = True
-except Exception:
-    HAS_PHI = False
-
+# phi agent removed to avoid fragile imports in varied environments
 HAS_PHI = False
 
+# google generative ai SDK (may be absent or partial in some runtimes)
 try:
-    import google.generativeai as genai
+    import google.generativeai as genai  # type: ignore
     from google.generativeai import upload_file, get_file  # type: ignore
     HAS_GENAI = True
 except Exception:
     genai = None
-    upload_file = get_file = None
+    upload_file = None
+    get_file = None
     HAS_GENAI = False
 
 st.set_page_config(page_title="Generate the story of videos", layout="wide")
 DATA_DIR = Path("./data")
 DATA_DIR.mkdir(exist_ok=True)
 
+# Session defaults
 st.session_state.setdefault("videos", "")
 st.session_state.setdefault("loop_video", False)
 st.session_state.setdefault("uploaded_file", None)
@@ -119,6 +117,7 @@ def maybe_configure_genai(key):
     if not key or not HAS_GENAI:
         return False
     try:
+        # defensive configuration (some envs require configure)
         genai.configure(api_key=key)
         return True
     except Exception:
@@ -138,43 +137,67 @@ def clear_all_video_state():
         except Exception:
             pass
 
-current_url = st.session_state.get("url", "")
-if current_url != st.session_state.get("last_url_value"):
-    clear_all_video_state()
-    st.session_state["last_url_value"] = current_url
-
-st.sidebar.header("Video Input")
-st.sidebar.text_input("Video URL", key="url", placeholder="https://")
-
-settings_exp = st.sidebar.expander("Settings", expanded=False)
-settings_exp.text_input("Gemini Model (short name)", "gemini-2.5-flash-lite", key="model_input")
-settings_exp.text_input("Google API Key", key="api_key", value=os.getenv("GOOGLE_API_KEY", ""), type="password")
-default_prompt = (
-    "Watch the video and provide a detailed behavioral report focusing on human actions, interactions, posture, movement, and apparent intent. Keep language professional. Include a list of observations for notable events."
-)
-settings_exp.text_area("Enter analysis", value=default_prompt, height=140, key="analysis_prompt")
-settings_exp.text_input("Video Password (if needed)", key="video-password", placeholder="password", type="password")
-settings_exp.checkbox("Fast mode (skip compression, smaller model, fewer tokens)", key="fast_mode")
-
-key_source = "session" if st.session_state.get("api_key") else ".env" if os.getenv("GOOGLE_API_KEY") else "none"
-settings_exp.caption(f"Using API key from: **{key_source}**")
-
-if not get_effective_api_key():
-    settings_exp.warning("No Google API key provided; upload/generation disabled.", icon="⚠️")
-
-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"},
-]
-
+# URL expand + extraction helpers (integrated into Load Video)
+def expand_url(short_url, timeout=10):
+    try:
+        r = requests.get(short_url, allow_redirects=True, timeout=timeout, headers={"User-Agent":"Mozilla/5.0"})
+        final_url = r.url
+        return final_url, r.text
+    except Exception as e:
+        return None, f"error: {e}"
+
+def extract_video_from_html(html, base_url=None):
+    soup = BeautifulSoup(html, "html.parser")
+    # 1) Open Graph video
+    og = soup.find("meta", property="og:video")
+    if og and og.get("content"):
+        return og.get("content")
+    # 2) Look for video tags
+    vtag = soup.find("video")
+    if vtag:
+        src = vtag.get("src")
+        if src:
+            return src
+        # source children
+        source = vtag.find("source")
+        if source and source.get("src"):
+            return source.get("src")
+    # 3) JSON-LD or structured data with video
+    for script in soup.find_all("script", type="application/ld+json"):
+        try:
+            import json
+            data = json.loads(script.string or "{}")
+            # common pattern
+            if isinstance(data, dict):
+                video = data.get("video") or data.get("videoObject") or data.get("mainEntity")
+                if isinstance(video, dict):
+                    for k in ("contentUrl", "url"):
+                        if video.get(k):
+                            return video.get(k)
+                # top-level contentUrl
+                if data.get("contentUrl"):
+                    return data.get("contentUrl")
+        except Exception:
+            continue
+    # 4) look for meta property site-specific fallbacks
+    for meta_name in ("twitter:player:stream", "twitter:player"):
+        m = soup.find("meta", attrs={"name": meta_name})
+        if m and m.get("content"):
+            return m.get("content")
+    # fallback: search for direct links to common video hosts (youtube, vimeo) in anchor tags
+    for a in soup.find_all("a", href=True):
+        href = a["href"]
+        if any(domain in href for domain in ("youtube.com", "youtu.be", "vimeo.com")):
+            return href
+    return None
+
+# When SDK has upload_file/get_file, use them; else raise when needed
 def upload_video_sdk(filepath: str):
     key = get_effective_api_key()
     if not key:
         raise RuntimeError("No API key provided")
     if not HAS_GENAI or upload_file is None:
-        raise RuntimeError("google.generativeai SDK not available; cannot upload")
+        raise RuntimeError("google.generativeai SDK upload not available; cannot upload")
     genai.configure(api_key=key)
     return upload_file(filepath)
 
@@ -221,16 +244,66 @@ def remove_prompt_echo(prompt: str, text: str, check_len: int = 600, ratio_thres
             return b_full[len(ph):].lstrip(" \n:-")
     return text
 
+# UI layout
+current_url = st.session_state.get("url", "")
+if current_url != st.session_state.get("last_url_value"):
+    clear_all_video_state()
+    st.session_state["last_url_value"] = current_url
+
+st.sidebar.header("Video Input")
+st.sidebar.text_input("Video URL", key="url", placeholder="https://")
+
+settings_exp = st.sidebar.expander("Settings", expanded=False)
+settings_exp.text_input("Gemini Model (short name)", "gemini-2.5-flash-lite", key="model_input")
+settings_exp.text_input("Google API Key", key="api_key", value=os.getenv("GOOGLE_API_KEY", ""), type="password")
+default_prompt = (
+    "Watch the video and provide a detailed behavioral report focusing on human actions, interactions, posture, movement, and apparent intent. Keep language professional. Include a list of observations for notable events."
+)
+settings_exp.text_area("Enter analysis", value=default_prompt, height=140, key="analysis_prompt")
+settings_exp.text_input("Video Password (if needed)", key="video-password", placeholder="password", type="password")
+settings_exp.checkbox("Fast mode (skip compression, smaller model, fewer tokens)", key="fast_mode")
+
+key_source = "session" if st.session_state.get("api_key") else ".env" if os.getenv("GOOGLE_API_KEY") else "none"
+settings_exp.caption(f"Using API key from: **{key_source}**")
+
+if not get_effective_api_key():
+    settings_exp.warning("No Google API key provided; upload/generation disabled.", icon="⚠️")
+
+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"},
+]
+
 col1, col2 = st.columns([1, 3])
 with col1:
     generate_now = st.button("Generate the story", type="primary", disabled=not bool(get_effective_api_key()))
 with col2:
     pass
 
+# Load Video flow: expand short URLs and try to extract direct video links from HTML before yt-dlp
 if st.sidebar.button("Load Video", use_container_width=True):
     try:
         vpw = st.session_state.get("video-password", "")
-        path = download_video_ytdlp(st.session_state.get("url", ""), str(DATA_DIR), vpw)
+        url_val = st.session_state.get("url", "").strip()
+        # If URL present, try to expand and extract video from HTML first
+        final_url = url_val
+        html_text = None
+        if url_val:
+            expanded, html_or_err = expand_url(url_val)
+            if expanded:
+                final_url = expanded
+                html_text = html_or_err
+            else:
+                # expansion failed but html_or_err contains error message; ignore
+                html_text = None
+        # If we have HTML, try to find direct video link
+        extracted = None
+        if html_text:
+            extracted = extract_video_from_html(html_text, base_url=final_url)
+        target_url_for_ytdlp = extracted or final_url
+        path = download_video_ytdlp(target_url_for_ytdlp, str(DATA_DIR), vpw)
         st.session_state["videos"] = path
         st.session_state["last_loaded_path"] = path
         st.session_state.pop("uploaded_file", None)
@@ -270,6 +343,7 @@ if st.session_state["videos"]:
     except Exception:
         pass
 
+# Generation flow (robust handling of google.generativeai variants)
 if generate_now and not st.session_state.get("busy"):
     if not st.session_state.get("videos"):
         st.error("No video loaded. Use 'Load Video' in the sidebar.")
@@ -336,7 +410,12 @@ if generate_now and not st.session_state.get("busy"):
                     try:
                         if not HAS_GENAI or genai is None:
                             raise RuntimeError("Responses API not available; install google.generativeai SDK.")
-                        genai.configure(api_key=key_to_use)
+                        # ensure configured
+                        try:
+                            genai.configure(api_key=key_to_use)
+                        except Exception:
+                            pass
+
                         fname = file_name_or_id(processed)
                         if not fname:
                             raise RuntimeError("Uploaded file missing name/id")
@@ -345,10 +424,12 @@ if generate_now and not st.session_state.get("busy"):
                         user_msg = {"role": "user", "content": "Please summarize the attached video."}
 
                         response = None
+                        diagnostics = {"attempts": []}
 
-                        # try modern responses.generate if available
+                        # Attempt #1: genai.responses.generate (modern)
                         try:
                             if hasattr(genai, "responses") and hasattr(genai.responses, "generate"):
+                                diagnostics["attempts"].append("responses.generate")
                                 response = genai.responses.generate(
                                     model=model_used,
                                     messages=[system_msg, user_msg],
@@ -356,129 +437,135 @@ if generate_now and not st.session_state.get("busy"):
                                     safety_settings=safety_settings,
                                     max_output_tokens=max_tokens,
                                 )
-                        except Exception:
+                        except Exception as e:
+                            diagnostics["responses.generate_error"] = str(e)
                             response = None
 
-                        # try GenerativeModel (0.8.x)
+                        # Attempt #2: GenerativeModel variants (0.8.x+)
                         if response is None:
                             try:
                                 if hasattr(genai, "GenerativeModel"):
+                                    diagnostics["attempts"].append("GenerativeModel")
                                     gm = genai.GenerativeModel(model=model_used)
                                     if hasattr(gm, "generate_content"):
                                         response = gm.generate_content([system_msg, user_msg], files=[{"name": fname}], max_output_tokens=max_tokens)
                                     elif hasattr(gm, "generate"):
                                         response = gm.generate([system_msg, user_msg], files=[{"name": fname}], max_output_tokens=max_tokens)
-                            except Exception:
+                            except Exception as e:
+                                diagnostics["GenerativeModel_error"] = str(e)
                                 response = None
 
-                        # try top-level legacy helpers
+                        # Attempt #3: top-level legacy helpers
                         if response is None:
                             try:
                                 if hasattr(genai, "generate"):
+                                    diagnostics["attempts"].append("top.generate")
                                     response = genai.generate(model=model_used, input=[{"text": prompt_text, "files": [{"name": fname}]}], max_output_tokens=max_tokens)
                                 elif hasattr(genai, "create"):
+                                    diagnostics["attempts"].append("top.create")
                                     response = genai.create(model=model_used, input=[{"text": prompt_text, "files": [{"name": fname}]}], max_output_tokens=max_tokens)
-                            except Exception:
+                            except Exception as e:
+                                diagnostics["top_level_error"] = str(e)
                                 response = None
 
-                        # if still None, fall back to best-effort: some environments may have the module but missing callables
+                        # Final defensive tries for known objects
                         if response is None:
-                            # attempt to call GenerativeModel regardless if present (wrapped defensively)
                             try:
                                 if hasattr(genai, "GenerativeModel"):
+                                    diagnostics["attempts"].append("GenerativeModel_last")
                                     gm = genai.GenerativeModel(model=model_used)
-                                    # try generate_content/generate without raising
                                     try:
                                         response = gm.generate_content([system_msg, user_msg], files=[{"name": fname}], max_output_tokens=max_tokens)
                                     except Exception:
-                                        try:
-                                            response = gm.generate([system_msg, user_msg], files=[{"name": fname}], max_output_tokens=max_tokens)
-                                        except Exception:
-                                            response = None
-                            except Exception:
+                                        response = gm.generate([system_msg, user_msg], files=[{"name": fname}], max_output_tokens=max_tokens)
+                            except Exception as e:
+                                diagnostics["GenerativeModel_last_error"] = str(e)
                                 response = None
 
                         if response is None:
-                            # don't raise here; provide helpful diagnostics instead and return gracefully
-                            raise RuntimeError("No supported generate method found on google.generativeai in this runtime. See Environment diagnostics below.")
-
-                        # normalize outputs
-                        outputs = []
-                        try:
-                            if isinstance(response, dict):
-                                for key in ("output", "candidates", "items", "responses"):
-                                    val = response.get(key)
-                                    if isinstance(val, (list, tuple)) and val:
-                                        outputs = list(val)
-                                        break
-                                if not outputs:
-                                    for v in response.values():
-                                        if isinstance(v, (list, tuple)) and v:
-                                            outputs = list(v)
-                                            break
-                            else:
-                                for attr in ("output", "candidates", "items", "responses"):
-                                    val = getattr(response, attr, None)
-                                    if isinstance(val, (list, tuple)) and val:
-                                        try:
-                                            outputs = list(val)
-                                        except Exception:
-                                            outputs = val
-                                        break
-                        except Exception:
+                            # Instead of raising the runtime error seen previously, attach diagnostics to last_error and return gracefully
+                            diag_text = f"No supported generate method found on google.generativeai in this runtime. Diagnostics: {diagnostics}"
+                            st.session_state["last_error"] = diag_text
+                            st.error("Responses API not supported in this runtime. See Last Error for details.")
+                            out = ""
+                        else:
+                            # Normalize outputs into text pieces
                             outputs = []
+                            try:
+                                if isinstance(response, dict):
+                                    for key in ("output", "candidates", "items", "responses"):
+                                        val = response.get(key)
+                                        if isinstance(val, (list, tuple)) and val:
+                                            outputs = list(val)
+                                            break
+                                    if not outputs:
+                                        for v in response.values():
+                                            if isinstance(v, (list, tuple)) and v:
+                                                outputs = list(v)
+                                                break
+                                else:
+                                    for attr in ("output", "candidates", "items", "responses"):
+                                        val = getattr(response, attr, None)
+                                        if isinstance(val, (list, tuple)) and val:
+                                            try:
+                                                outputs = list(val)
+                                            except Exception:
+                                                outputs = val
+                                            break
+                            except Exception:
+                                outputs = []
+
+                            if not outputs:
+                                candidate_text = None
+                                if isinstance(response, dict):
+                                    candidate_text = response.get("text") or response.get("message")
+                                else:
+                                    candidate_text = getattr(response, "text", None) or getattr(response, "message", None)
+                                if candidate_text:
+                                    outputs = [{"text": candidate_text}]
+
+                            text_pieces = []
+                            for item in outputs:
+                                if not item:
+                                    continue
+                                if isinstance(item, dict):
+                                    for k in ("content", "text", "message", "output_text", "output"):
+                                        v = item.get(k)
+                                        if v:
+                                            if isinstance(v, str):
+                                                text_pieces.append(v.strip())
+                                            elif isinstance(v, (list, tuple)):
+                                                for e in v:
+                                                    if isinstance(e, str):
+                                                        text_pieces.append(e.strip())
+                                                    elif isinstance(e, dict):
+                                                        t = e.get("text") or e.get("content")
+                                                        if t:
+                                                            text_pieces.append(str(t).strip())
+                                            break
+                                else:
+                                    for k in ("content", "text", "message", "output", "output_text"):
+                                        v = getattr(item, k, None)
+                                        if v:
+                                            if isinstance(v, str):
+                                                text_pieces.append(v.strip())
+                                            elif isinstance(v, (list, tuple)):
+                                                for e in v:
+                                                    if isinstance(e, str):
+                                                        text_pieces.append(e.strip())
+                                                    else:
+                                                        t = getattr(e, "text", None) or getattr(e, "content", None)
+                                                        if t:
+                                                            text_pieces.append(str(t).strip())
+                                            break
 
-                        if not outputs:
-                            candidate_text = None
-                            if isinstance(response, dict):
-                                candidate_text = response.get("text") or response.get("message")
-                            else:
-                                candidate_text = getattr(response, "text", None) or getattr(response, "message", None)
-                            if candidate_text:
-                                outputs = [{"text": candidate_text}]
-
-                        text_pieces = []
-                        for item in outputs:
-                            if not item:
-                                continue
-                            if isinstance(item, dict):
-                                for k in ("content", "text", "message", "output_text", "output"):
-                                    v = item.get(k)
-                                    if v:
-                                        if isinstance(v, str):
-                                            text_pieces.append(v.strip())
-                                        elif isinstance(v, (list, tuple)):
-                                            for e in v:
-                                                if isinstance(e, str):
-                                                    text_pieces.append(e.strip())
-                                                elif isinstance(e, dict):
-                                                    t = e.get("text") or e.get("content")
-                                                    if t:
-                                                        text_pieces.append(str(t).strip())
-                                        break
-                            else:
-                                for k in ("content", "text", "message", "output", "output_text"):
-                                    v = getattr(item, k, None)
-                                    if v:
-                                        if isinstance(v, str):
-                                            text_pieces.append(v.strip())
-                                        elif isinstance(v, (list, tuple)):
-                                            for e in v:
-                                                if isinstance(e, str):
-                                                    text_pieces.append(e.strip())
-                                                else:
-                                                    t = getattr(e, "text", None) or getattr(e, "content", None)
-                                                    if t:
-                                                        text_pieces.append(str(t).strip())
-                                        break
-
-                        seen = set()
-                        filtered = []
-                        for t in text_pieces:
-                            if t and t not in seen:
-                                filtered.append(t)
-                                seen.add(t)
-                        out = "\n\n".join(filtered)
+                            seen = set()
+                            filtered = []
+                            for t in text_pieces:
+                                if t and t not in seen:
+                                    filtered.append(t)
+                                    seen.add(t)
+                            out = "\n\n".join(filtered)
 
                     except Exception as e:
                         tb = traceback.format_exc()