Update streamlit_app.py

streamlit_app.py

@@ -10,6 +10,7 @@ import re
 from glob import glob
 from pathlib import Path
 from difflib import SequenceMatcher
+from typing import Optional, Tuple, Any
 
 import requests
 from bs4 import BeautifulSoup
@@ -21,8 +22,11 @@ from dotenv import load_dotenv
 
 load_dotenv()
 
-# Feature flags
-HAS_PHI = False
+# Try import google.generativeai (optional)
+HAS_GENAI = False
+genai = None
+upload_file = None
+get_file = None
 try:
     import google.generativeai as genai  # type: ignore
     try:
@@ -42,27 +46,26 @@ DATA_DIR = Path("./data")
 DATA_DIR.mkdir(exist_ok=True)
 
 # Session defaults
-for k, v in {
-    "videos": "",
-    "loop_video": False,
-    "uploaded_file": None,
-    "processed_file": None,
-    "busy": False,
-    "last_loaded_path": "",
-    "analysis_out": "",
-    "last_error": "",
-    "file_hash": None,
-    "fast_mode": False,
-    "api_key": os.getenv("GOOGLE_API_KEY", ""),
-    "last_model": "",
-    "last_url_value": "",
-}.items():
-    st.session_state.setdefault(k, v)
+st.session_state.setdefault("videos", "")
+st.session_state.setdefault("loop_video", False)
+st.session_state.setdefault("uploaded_file", None)
+st.session_state.setdefault("processed_file", None)
+st.session_state.setdefault("busy", False)
+st.session_state.setdefault("last_loaded_path", "")
+st.session_state.setdefault("analysis_out", "")
+st.session_state.setdefault("last_error", "")
+st.session_state.setdefault("file_hash", None)
+st.session_state.setdefault("fast_mode", False)
+st.session_state.setdefault("api_key", os.getenv("GOOGLE_API_KEY", ""))
+st.session_state.setdefault("last_model", "")
+st.session_state.setdefault("last_url_value", "")
+# allow disabling SSL verify for HTTP fallback (not recommended)
+st.session_state.setdefault("http_skip_ssl_verify", False)
 
 HEADERS = {"User-Agent": "Mozilla/5.0 (compatible)"}
 
-# Utilities --------------------------------------------------------------------
-def sanitize_filename(path_str: str):
+# ----------------- Utilities -----------------
+def sanitize_filename(path_str: str) -> str:
     return Path(path_str).name.lower().translate(str.maketrans("", "", string.punctuation)).replace(" ", "_")
 
 def file_sha256(path: str, block_size: int = 65536) -> str:
@@ -83,15 +86,15 @@ def convert_video_to_mp4(video_path: str) -> str:
         pass
     return target_path
 
-def compress_video(input_path: str, target_path: str, crf: int = 28, preset: str = "fast"):
+def compress_video(input_path: str, target_path: str, crf: int = 28, preset: str = "fast") -> str:
     try:
         ffmpeg.input(input_path).output(target_path, vcodec="libx264", crf=crf, preset=preset).run(overwrite_output=True, quiet=True)
         return target_path
     except Exception:
         return input_path
 
-# Downloader / extractor ------------------------------------------------------
-def expand_url(short_url, timeout=10):
+# ----------------- Download / Extract -----------------
+def expand_url(short_url: str, timeout: int = 10) -> Tuple[Optional[str], Optional[str]]:
     try:
         r = requests.get(short_url, allow_redirects=True, timeout=timeout, headers=HEADERS)
         r.raise_for_status()
@@ -99,7 +102,7 @@ def expand_url(short_url, timeout=10):
     except Exception as e:
         return None, f"error: {e}"
 
-def extract_video_from_html(html, base_url=None):
+def extract_video_from_html(html: str, base_url: Optional[str] = None) -> Optional[str]:
     soup = BeautifulSoup(html, "html.parser")
     og = soup.find("meta", property="og:video")
     if og and og.get("content"):
@@ -115,7 +118,6 @@ def extract_video_from_html(html, base_url=None):
     for script in soup.find_all("script", type="application/ld+json"):
         try:
             data = json.loads(script.string or "{}")
-            video = None
             if isinstance(data, dict):
                 video = data.get("video") or data.get("videoObject") or data.get("mainEntity")
                 if isinstance(video, dict):
@@ -127,16 +129,16 @@ def extract_video_from_html(html, base_url=None):
         except Exception:
             continue
     for mname in ("twitter:player:stream", "twitter:player"):
-        meta = soup.find("meta", attrs={"name": mname})
-        if meta and meta.get("content"):
-            return meta.get("content")
+        m = soup.find("meta", attrs={"name": mname})
+        if m and m.get("content"):
+            return m.get("content")
     for a in soup.find_all("a", href=True):
         href = a["href"]
-        if any(d in href for d in ("youtube.com", "youtu.be", "vimeo.com")):
+        if any(domain in href for domain in ("youtube.com", "youtu.be", "vimeo.com")):
             return href
     return None
 
-def extract_video_from_twitter_html(html):
+def extract_video_from_twitter_html(html: str) -> Optional[str]:
     soup = BeautifulSoup(html, "html.parser")
     og_video = soup.find("meta", property="og:video")
     if og_video and og_video.get("content"):
@@ -185,10 +187,10 @@ def extract_video_from_twitter_html(html):
                 return url
     return None
 
-def extract_direct_twitter_video(url):
+def extract_direct_twitter_video(url: str) -> Tuple[Optional[str], str]:
     final, html_or_err = expand_url(url)
     if final is None:
-        return None, html_or_err
+        return None, html_or_err or "expand failed"
     variants = [
         final,
         final.replace("://twitter.com/", "://mobile.twitter.com/"),
@@ -218,7 +220,7 @@ def extract_direct_twitter_video(url):
         pass
     return None, "not found"
 
-def download_video_ytdlp(url: str, save_dir: str, video_password: str = None) -> str:
+def download_video_ytdlp(url: str, save_dir: str, video_password: Optional[str] = None) -> str:
     if not url:
         raise ValueError("No URL provided")
     outtmpl = str(Path(save_dir) / "%(id)s.%(ext)s")
@@ -237,11 +239,11 @@ def download_video_ytdlp(url: str, save_dir: str, video_password: str = None) ->
         raise FileNotFoundError("Downloaded video not found")
     return convert_video_to_mp4(matches[0])
 
-# Generative AI helpers -------------------------------------------------------
-def get_effective_api_key():
+# ----------------- Generative AI helpers -----------------
+def get_effective_api_key() -> Optional[str]:
     return st.session_state.get("api_key") or os.getenv("GOOGLE_API_KEY")
 
-def maybe_configure_genai(key):
+def maybe_configure_genai(key: str) -> bool:
     if not key or not HAS_GENAI:
         return False
     try:
@@ -250,7 +252,7 @@ def maybe_configure_genai(key):
     except Exception:
         return False
 
-def upload_video_sdk(filepath: str):
+def upload_video_sdk(filepath: str) -> Any:
     key = get_effective_api_key()
     if not key:
         raise RuntimeError("No API key provided")
@@ -259,18 +261,11 @@ def upload_video_sdk(filepath: str):
     genai.configure(api_key=key)
     return upload_file(filepath)
 
-def wait_for_processed(file_obj, timeout=180):
+def wait_for_processed(file_obj: Any, timeout: int = 180) -> Any:
     if not HAS_GENAI or get_file is None:
         return file_obj
     start = time.time()
-    name = None
-    if isinstance(file_obj, dict):
-        name = file_obj.get("name") or file_obj.get("id")
-    else:
-        for attr in ("name", "id", "fileId", "file_id"):
-            if hasattr(file_obj, attr):
-                name = getattr(file_obj, attr)
-                break
+    name = file_name_or_id(file_obj)
     if not name:
         return file_obj
     backoff = 1.0
@@ -287,7 +282,19 @@ def wait_for_processed(file_obj, timeout=180):
         time.sleep(backoff)
         backoff = min(backoff * 2, 8.0)
 
-def remove_prompt_echo(prompt: str, text: str, check_len: int = 600, ratio_threshold: float = 0.68):
+def file_name_or_id(file_obj: Any) -> Optional[str]:
+    if file_obj is None:
+        return None
+    if isinstance(file_obj, dict):
+        return file_obj.get("name") or file_obj.get("id")
+    for attr in ("name", "id", "fileId", "file_id"):
+        if hasattr(file_obj, attr):
+            val = getattr(file_obj, attr)
+            if val:
+                return val
+    return str(file_obj)
+
+def remove_prompt_echo(prompt: str, text: str, check_len: int = 600, ratio_threshold: float = 0.68) -> str:
     if not prompt or not text:
         return text
     a = " ".join(prompt.strip().lower().split())
@@ -309,13 +316,19 @@ def remove_prompt_echo(prompt: str, text: str, check_len: int = 600, ratio_thres
             return b_full[len(ph):].lstrip(" \n:-")
     return text
 
-def generative_model_call_flexible(model_name, messages, files=None, max_output_tokens=1024):
+def generative_model_call_flexible(model_name: str, messages: list, files: Optional[list] = None, max_output_tokens: int = 1024) -> Any:
+    """
+    Robustly call GenerativeModel with different signatures.
+    This function anticipates generate_content taking different arg shapes.
+    """
     if not HAS_GENAI or genai is None:
         raise RuntimeError("genai not available")
     GM = getattr(genai, "GenerativeModel", None)
     if GM is None:
         raise RuntimeError("GenerativeModel not available")
-    # robust constructor
+
+    # Construct instance robustly
+    gm = None
     try:
         sig = inspect.signature(GM)
         params = sig.parameters
@@ -325,34 +338,58 @@ def generative_model_call_flexible(model_name, messages, files=None, max_output_
             gm = GM(model_name=model_name)
         else:
             gm = GM()
-            if hasattr(gm, "model"):
-                try:
+            try:
+                if hasattr(gm, "model"):
                     setattr(gm, "model", model_name)
-                except Exception:
-                    pass
+            except Exception:
+                pass
     except Exception:
         try:
             gm = GM(model=model_name)
         except Exception:
-            gm = GM()
-    # try generate methods but avoid unsupported kwargs
+            try:
+                gm = GM(model_name=model_name)
+            except Exception:
+                gm = GM()
+
+    # Try generate_content with multiple call shapes:
+    #  - generate_content(messages=..., files=..., max_output_tokens=...)
+    #  - generate_content(messages, files, max_output_tokens)
+    #  - generate_content(messages)
     if hasattr(gm, "generate_content"):
+        # try keyword style first
         try:
-            # many versions accept 'messages' and 'files'
+            return gm.generate_content(messages=messages, files=files, max_output_tokens=max_output_tokens)
+        except TypeError:
+            # try positional: some versions expect (messages, max_output_tokens)
             try:
-                return gm.generate_content(messages=messages, files=files, max_output_tokens=max_output_tokens)
-            except TypeError:
                 return gm.generate_content(messages, max_output_tokens)
+            except TypeError:
+                # maybe (prompt_str,) shape
+                try:
+                    prompt = messages[-1].get("content") if isinstance(messages, (list, tuple)) and messages else str(messages)
+                    return gm.generate_content(prompt)
+                except Exception as e:
+                    raise RuntimeError(f"GenerativeModel.generate_content unusable: {e}")
         except Exception as e:
             raise RuntimeError(f"generate_content failed: {e}")
+
+    # Try generate with files kw / positional
     if hasattr(gm, "generate"):
         try:
             return gm.generate(messages=messages, files=files, max_output_tokens=max_output_tokens)
         except TypeError:
-            return gm.generate(messages, max_output_tokens=max_output_tokens)
+            try:
+                return gm.generate(messages, max_output_tokens)
+            except Exception as e:
+                raise RuntimeError(f"GenerativeModel.generate unusable: {e}")
+
     raise RuntimeError("No usable generate method on GenerativeModel instance")
 
-def responses_http_call(api_key, model, messages, file_name=None, max_output_tokens=1024, safety_settings=None):
+def responses_http_call(api_key: str, model: str, messages: list, file_name: Optional[str] = None, max_output_tokens: int = 1024, safety_settings: Optional[list] = None) -> dict:
+    """
+    HTTP fallback to Responses v1 endpoint. Attempts retries and allows optional SSL skip.
+    """
     url = f"https://api.generativeai.googleapis.com/v1/models/{model}:generateMessage"
     headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
     payload = {
@@ -363,25 +400,36 @@ def responses_http_call(api_key, model, messages, file_name=None, max_output_tok
         payload["safetySettings"] = safety_settings
     if file_name:
         payload["files"] = [{"name": file_name}]
-    r = requests.post(url, json=payload, headers=headers, timeout=60)
-    r.raise_for_status()
-    return r.json()
+    # allow skipping SSL verify in special environments (not recommended)
+    verify = not bool(st.session_state.get("http_skip_ssl_verify", False))
+    last_err = None
+    for attempt in range(1, 4):
+        try:
+            r = requests.post(url, json=payload, headers=headers, timeout=30, verify=verify)
+            r.raise_for_status()
+            return r.json()
+        except requests.exceptions.SSLError as e:
+            last_err = e
+            # If SSL hostname mismatch or cert issues, surface helpful message once
+            raise RuntimeError(f"SSL error when calling Responses HTTP endpoint: {e}. If you are behind a proxy intercepting TLS, set 'Skip HTTP SSL verify' in Settings (not recommended) or fix your CA bundle.")
+        except Exception as e:
+            last_err = e
+            time.sleep(0.8 * attempt)
+            continue
+    raise RuntimeError(f"HTTP responses fallback failed after retries: {last_err}")
 
-def normalize_response_to_text(response) -> str:
+def normalize_response_to_text(response: Any) -> str:
     """Extract text from SDK or HTTP responses into a single string."""
     if not response:
         return ""
-    # dict-like responses (HTTP fallback or genai.responses)
+    # dict-like
     if isinstance(response, dict):
-        # modern Responses v1 may include 'output' or 'message'
-        # search keys for lists of candidates/items/responses
-        for list_key in ("output", "candidates", "items", "responses"):
-            val = response.get(list_key)
+        for key in ("output", "candidates", "items", "responses"):
+            val = response.get(key)
             if isinstance(val, (list, tuple)) and val:
                 pieces = []
                 for el in val:
                     if isinstance(el, dict):
-                        # content field may be list of {type, text}
                         c = el.get("content") or el.get("message") or el.get("text")
                         if isinstance(c, list):
                             for part in c:
@@ -397,7 +445,7 @@ def normalize_response_to_text(response) -> str:
                         pieces.append(el)
                 if pieces:
                     return "\n\n".join(pieces)
-        # message/content path
+        # message path
         msg = response.get("message") or response.get("response") or response.get("output")
         if isinstance(msg, dict):
             c = msg.get("content")
@@ -430,14 +478,12 @@ def normalize_response_to_text(response) -> str:
                 else:
                     pieces.append(str(el))
             return "\n\n".join([p for p in pieces if p])
-    # fallback
     text = getattr(response, "text", None) or getattr(response, "message", None)
     return text or ""
 
-# UI --------------------------------------------------------------------------
+# ----------------- UI -----------------
 current_url = st.session_state.get("url", "")
 if current_url != st.session_state.get("last_url_value"):
-    # clear when user changes URL
     st.session_state.update({"videos": "", "uploaded_file": None, "processed_file": None, "last_loaded_path": "", "analysis_out": "", "last_error": "", "file_hash": None})
     st.session_state["last_url_value"] = current_url
 
@@ -453,9 +499,11 @@ default_prompt = (
 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")
+settings_exp.checkbox("Skip HTTP SSL verify (only if you trust the network)", key="http_skip_ssl_verify")
 
 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="⚠️")
 
@@ -472,7 +520,7 @@ with col1:
 with col2:
     pass
 
-# Load Video button
+# Load Video
 if st.sidebar.button("Load Video", use_container_width=True):
     try:
         vpw = st.session_state.get("video-password", "")
@@ -507,7 +555,7 @@ if st.sidebar.button("Load Video", use_container_width=True):
     except Exception as e:
         st.sidebar.error(f"Failed to load video: {e}")
 
-# Sidebar preview + controls
+# Sidebar preview
 if st.session_state["videos"]:
     try:
         st.sidebar.video(st.session_state["videos"], loop=st.session_state.get("loop_video", False))
@@ -537,7 +585,7 @@ if st.session_state["videos"]:
     except Exception:
         pass
 
-# Generation flow --------------------------------------------------------------
+# Generation flow
 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.")
@@ -560,7 +608,6 @@ if generate_now and not st.session_state.get("busy"):
 
                 upload_path = current_path
                 if reupload_needed:
-                    # compress if large and not fast
                     fast_mode = st.session_state.get("fast_mode", False)
                     try:
                         file_size_mb = os.path.getsize(current_path) / (1024 * 1024)
@@ -580,20 +627,13 @@ if generate_now and not st.session_state.get("busy"):
                 prompt_text = (st.session_state.get("analysis_prompt", "") or default_prompt).strip()
                 system_msg = {"role": "system", "content": prompt_text}
                 user_msg = {"role": "user", "content": "Please summarize the attached video."}
-                fname = None
-                if processed:
-                    if isinstance(processed, dict):
-                        fname = processed.get("name") or processed.get("id")
-                    else:
-                        for attr in ("name", "id", "fileId", "file_id"):
-                            if hasattr(processed, attr):
-                                fname = getattr(processed, attr)
-                                break
-                # prefer SDK methods that support 'files' / file references
+
+                fname = file_name_or_id(processed)
+
                 response = None
                 diagnostics = {"attempts": []}
 
-                # 1) genai.responses.generate (if available) - supports files param
+                # 1) genai.responses.generate (supports files)
                 if response is None and HAS_GENAI and genai is not None and hasattr(genai, "responses") and hasattr(genai.responses, "generate"):
                     try:
                         diagnostics["attempts"].append("responses.generate")
@@ -622,15 +662,22 @@ if generate_now and not st.session_state.get("busy"):
                     try:
                         if hasattr(genai, "generate"):
                             diagnostics["attempts"].append("top.generate")
-                            response = genai.generate(model=model_id, input=[{"text": prompt_text, "files": [{"name": fname}]}], max_output_tokens=(256 if st.session_state.get("fast_mode") else 1024))
+                            # don't assume exact param shapes; try best-effort
+                            try:
+                                response = genai.generate(model=model_id, input=[{"text": prompt_text, "files": [{"name": fname}]}], max_output_tokens=(256 if st.session_state.get("fast_mode") else 1024))
+                            except TypeError:
+                                response = genai.generate(model=model_id, input=prompt_text)
                         elif hasattr(genai, "create"):
                             diagnostics["attempts"].append("top.create")
-                            response = genai.create(model=model_id, input=[{"text": prompt_text, "files": [{"name": fname}]}], max_output_tokens=(256 if st.session_state.get("fast_mode") else 1024))
+                            try:
+                                response = genai.create(model=model_id, input=[{"text": prompt_text, "files": [{"name": fname}]}], max_output_tokens=(256 if st.session_state.get("fast_mode") else 1024))
+                            except TypeError:
+                                response = genai.create(model=model_id, input=prompt_text)
                     except Exception as e:
                         diagnostics["top_level_error"] = str(e)
                         response = None
 
-                # 4) HTTP fallback to Responses endpoint (supports file references)
+                # 4) HTTP fallback
                 if response is None:
                     try:
                         diagnostics["attempts"].append("http_fallback")
@@ -646,7 +693,6 @@ if generate_now and not st.session_state.get("busy"):
                 else:
                     out = normalize_response_to_text(response)
                     out = remove_prompt_echo(prompt_text, out).strip()
-                    # additional cleanup of obvious echoes/placeholders
                     placeholders = ["enter analysis", "enter your analysis", "enter analysis here", "please enter analysis"]
                     low = out.strip().lower()
                     for ph in placeholders: