Update streamlit_app.py

streamlit_app.py

@@ -5,6 +5,8 @@ import string
 import hashlib
 import traceback
 import inspect
+import re
+import json
 from glob import glob
 from pathlib import Path
 from difflib import SequenceMatcher
@@ -57,6 +59,8 @@ st.session_state.setdefault("api_key", os.getenv("GOOGLE_API_KEY", ""))
 st.session_state.setdefault("last_model", "")
 st.session_state.setdefault("last_url_value", "")
 
+HEADERS = {"User-Agent": "Mozilla/5.0 (compatible)"}
+
 def sanitize_filename(path_str: str):
     return Path(path_str).name.lower().translate(str.maketrans("", "", string.punctuation)).replace(" ", "_")
 
@@ -142,16 +146,24 @@ def clear_all_video_state():
         except Exception:
             pass
 
-# URL expand + extraction helpers
+# --- Twitter (t.co / X) helpers integrated into expand/extract flow ---
 def expand_url(short_url, timeout=10):
+    """
+    General URL expander. For t.co/twitter shortlinks we try multiple variants
+    and return final URL and HTML if available.
+    """
     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
+        r = requests.get(short_url, allow_redirects=True, timeout=timeout, headers=HEADERS)
+        r.raise_for_status()
+        final = r.url
+        return final, r.text
     except Exception as e:
         return None, f"error: {e}"
 
 def extract_video_from_html(html, base_url=None):
+    """
+    Generic extractor tries og:video, <video>, LD+JSON, twitter tags, and links to common hosts.
+    """
     soup = BeautifulSoup(html, "html.parser")
     og = soup.find("meta", property="og:video")
     if og and og.get("content"):
@@ -166,7 +178,6 @@ def extract_video_from_html(html, base_url=None):
             return source.get("src")
     for script in soup.find_all("script", type="application/ld+json"):
         try:
-            import json
             data = json.loads(script.string or "{}")
             if isinstance(data, dict):
                 video = data.get("video") or data.get("videoObject") or data.get("mainEntity")
@@ -188,6 +199,117 @@ def extract_video_from_html(html, base_url=None):
             return href
     return None
 
+def extract_video_from_twitter_html(html):
+    """
+    Attempt to pull direct MP4 URL from Twitter/X HTML by searching JSON blobs and OG tags.
+    This is a best-effort extractor and may fail if Twitter/X obfuscates content.
+    """
+    soup = BeautifulSoup(html, "html.parser")
+
+    # 1) Open Graph video tag
+    og_video = soup.find("meta", property="og:video")
+    if og_video and og_video.get("content"):
+        return og_video["content"]
+
+    # 2) Look for JSON blobs in <script> tags and search for variants/urls
+    scripts = soup.find_all("script")
+    for s in scripts:
+        txt = s.string
+        if not txt:
+            continue
+        # crude detect for embedded JSON-ish blobs that include "video_info" or "variants"
+        if "video_info" in txt or "variants" in txt or "playbackUrl" in txt or "media" in txt:
+            # try to extract a JSON object within the script text
+            m = re.search(r"(?s)(\{.+\})", txt)
+            if not m:
+                continue
+            try:
+                blob = json.loads(m.group(1))
+            except Exception:
+                # sometimes it's not strict JSON; skip
+                continue
+
+            # deep search for urls and variants
+            def find_media_urls(obj):
+                if isinstance(obj, dict):
+                    for k, v in obj.items():
+                        if isinstance(v, str):
+                            if v.startswith("https://") and v.endswith(".mp4"):
+                                yield v
+                        else:
+                            yield from find_media_urls(v)
+                elif isinstance(obj, list):
+                    for it in obj:
+                        yield from find_media_urls(it)
+
+            for url in find_media_urls(blob):
+                return url
+
+            # also look for variant lists
+            def find_variants(obj):
+                if isinstance(obj, dict):
+                    for k, v in obj.items():
+                        if k == "variants" and isinstance(v, list):
+                            for vi in v:
+                                if isinstance(vi, dict):
+                                    url = vi.get("url") or vi.get("playbackUrl")
+                                    ct = vi.get("content_type", "") or vi.get("contentType", "")
+                                    if url and url.startswith("http") and ("mp4" in url or "video" in ct or "video" in url):
+                                        yield url
+                        else:
+                            yield from find_variants(v)
+                elif isinstance(obj, list):
+                    for it in obj:
+                        yield from find_variants(it)
+
+            for url in find_variants(blob):
+                return url
+
+    return None
+
+def extract_direct_twitter_video(url):
+    """
+    Expand t.co and try several page variants (mobile, amp, x.com) and oEmbed.
+    Returns (direct_video_url or None, info_string)
+    """
+    final, html_or_err = expand_url(url)
+    if final is None:
+        return None, html_or_err
+
+    # Try several variants (mobile, x.com, with query params)
+    variants = [
+        final,
+        final.replace("://twitter.com/", "://mobile.twitter.com/"),
+        final.replace("://twitter.com/", "://x.com/"),
+        final + "?s=20",
+        final + "?ref_src=twsrc%5Etfw",
+    ]
+    for u in variants:
+        try:
+            r = requests.get(u, allow_redirects=True, headers=HEADERS, timeout=10)
+            r.raise_for_status()
+            direct = extract_video_from_twitter_html(r.text)
+            if direct:
+                return direct, u
+        except Exception:
+            continue
+
+    # Try oEmbed as last resort
+    try:
+        oembed = requests.get("https://publish.twitter.com/oembed?url=" + final, headers=HEADERS, timeout=6)
+        if oembed.ok:
+            j = oembed.json()
+            html = j.get("html", "")
+            soup = BeautifulSoup(html, "html.parser")
+            video = soup.find("video")
+            if video and video.get("src"):
+                return video["src"], final
+    except Exception:
+        pass
+
+    return None, "not found"
+
+# --- Upload helpers for Generative AI SDK + HTTP fallback (fixed endpoint/use patterns) ---
 def upload_video_sdk(filepath: str):
     key = get_effective_api_key()
     if not key:
@@ -240,40 +362,34 @@ def remove_prompt_echo(prompt: str, text: str, check_len: int = 600, ratio_thres
             return b_full[len(ph):].lstrip(" \n:-")
     return text
 
-# Helper: try to call GenerativeModel with compatible signature
 def generative_model_call_flexible(model_name, messages, files=None, max_output_tokens=1024):
     """
     Try different call patterns for genai.GenerativeModel depending on its constructor/signature.
-    Returns the response object or raises if none work.
+    Do NOT pass unsupported keywords called 'files' into generate_content() if the SDK rejects them.
     """
     if not HAS_GENAI or genai is None:
         raise RuntimeError("genai not available")
 
-    # Inspect GenerativeModel if present
     GM = getattr(genai, "GenerativeModel", None)
     if GM is None:
         raise RuntimeError("GenerativeModel not available")
 
-    # Inspect constructor signature
+    # Construct instance robustly
     try:
         sig = inspect.signature(GM)
         params = sig.parameters
-        # prefer 'model' if available
         if "model" in params:
             gm = GM(model=model_name)
         elif "model_name" in params:
             gm = GM(model_name=model_name)
         else:
-            # fallback to no-arg constructor
             gm = GM()
-            # attempt to set attribute if accepted
             try:
                 if hasattr(gm, "model"):
                     setattr(gm, "model", model_name)
             except Exception:
                 pass
     except Exception:
-        # if signature inspection fails, try common constructors defensively
         try:
             gm = GM(model=model_name)
         except TypeError:
@@ -282,28 +398,40 @@ def generative_model_call_flexible(model_name, messages, files=None, max_output_
             except TypeError:
                 gm = GM()
 
-    # Now try available methods
+    # Now attempt supported generate methods but avoid unsupported kwargs
+    # 1) generate_content(messages...) may accept just messages and options (no files)
     if hasattr(gm, "generate_content"):
-        return gm.generate_content(messages, files=files, max_output_tokens=max_output_tokens)
-    if hasattr(gm, "generate"):
-        # some versions use generate(messages,...)
         try:
-            return gm.generate(messages, files=files, max_output_tokens=max_output_tokens)
+            return gm.generate_content(messages, max_output_tokens=max_output_tokens)
         except TypeError:
-            # try positional
+            # generate_content signature doesn't accept our args; try positional single string fallback
+            try:
+                # some versions expect a string prompt
+                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}")
+    # 2) generate(...) variants
+    if hasattr(gm, "generate"):
+        try:
             return gm.generate(messages, max_output_tokens=max_output_tokens)
+        except TypeError:
+            try:
+                return gm.generate(messages)
+            except Exception as e:
+                raise RuntimeError(f"GenerativeModel.generate unusable: {e}")
+
     raise RuntimeError("No usable generate method on GenerativeModel instance")
 
-# Fallback HTTP call using the REST Responses endpoint if the SDK is present but broken.
-# This requires an API key and uses the public Responses API endpoint.
 def responses_http_call(api_key, model, messages, file_name=None, max_output_tokens=1024, safety_settings=None):
     """
-    Minimal fallback: POST to the Responses API /v1/responses.
-    This constructs a small payload; note: some runtimes may block direct HTTP to Google or expect different endpoints.
+    Fallback to the public Responses API v1 endpoint (modern). Construct a minimal request body.
+    Note: endpoint and schema may change; this uses a simple v1-compatible payload.
     """
-    # Basic endpoint; adjust if your environment needs a different base URL
-    url = "https://generativeai.googleapis.com/v1beta2/models/{model}:generateMessage".format(model=model)
+    # Use the modern Responses v1 endpoint format
+    url = "https://api.generativeai.googleapis.com/v1/models/{model}:generateMessage".format(model=model)
     headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
+    # Build minimal 'messages' style payload expected by many GenAI endpoints
     payload = {
         "messages": [{"role": m.get("role", "user"), "content": [{"type": "text", "text": m.get("content", "")}]} for m in messages],
         "maxOutputTokens": max_output_tokens,
@@ -311,7 +439,7 @@ def responses_http_call(api_key, model, messages, file_name=None, max_output_tok
     if safety_settings:
         payload["safetySettings"] = safety_settings
     if file_name:
-        # Attach file ref format used by some SDKs
+        # Some endpoints accept files as references
         payload["files"] = [{"name": file_name}]
     try:
         r = requests.post(url, json=payload, headers=headers, timeout=60)
@@ -365,15 +493,27 @@ if st.sidebar.button("Load Video", use_container_width=True):
         url_val = st.session_state.get("url", "").strip()
         final_url = url_val
         html_text = None
+        extracted = None
         if url_val:
-            expanded, html_or_err = expand_url(url_val)
-            if expanded:
-                final_url = expanded
-                html_text = html_or_err
+            # Special handling for t.co / twitter shortlinks
+            if "t.co/" in url_val or ("twitter.com" in url_val or "x.com" in url_val):
+                extracted, src_info = extract_direct_twitter_video(url_val)
+                if extracted:
+                    final_url = extracted
+                    html_text = None
+                else:
+                    # fallback to expand_url to get final page HTML
+                    expanded, html_or_err = expand_url(url_val)
+                    if expanded:
+                        final_url = expanded
+                        html_text = html_or_err
             else:
-                html_text = None
-        extracted = None
-        if html_text:
+                expanded, html_or_err = expand_url(url_val)
+                if expanded:
+                    final_url = expanded
+                    html_text = html_or_err
+
+        if html_text and not extracted:
             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)
@@ -388,6 +528,7 @@ if st.sidebar.button("Load Video", use_container_width=True):
     except Exception as e:
         st.sidebar.error(f"Failed to load video: {e}")
 
+# Player / sidebar controls
 if st.session_state["videos"]:
     try:
         st.sidebar.video(st.session_state["videos"], loop=st.session_state.get("loop_video", False))
@@ -499,7 +640,7 @@ if generate_now and not st.session_state.get("busy"):
                         response = None
                         diagnostics = {"attempts": []}
 
-                        # Attempt #1: genai.responses.generate (modern)
+                        # Attempt #1: genai.responses.generate (modern public SDK)
                         try:
                             if hasattr(genai, "responses") and hasattr(genai.responses, "generate"):
                                 diagnostics["attempts"].append("responses.generate")
@@ -519,6 +660,7 @@ if generate_now and not st.session_state.get("busy"):
                             try:
                                 if hasattr(genai, "GenerativeModel"):
                                     diagnostics["attempts"].append("GenerativeModel")
+                                    # generative_model_call_flexible avoids passing unsupported 'files' kwarg
                                     response = generative_model_call_flexible(model_used, [system_msg, user_msg], files=[{"name": fname}], max_output_tokens=max_tokens)
                             except Exception as e:
                                 diagnostics["GenerativeModel_error"] = str(e)
@@ -537,7 +679,7 @@ if generate_now and not st.session_state.get("busy"):
                                 diagnostics["top_level_error"] = str(e)
                                 response = None
 
-                        # Attempt #4: fallback HTTP Responses call
+                        # Attempt #4: fallback HTTP Responses call (modern endpoint)
                         if response is None:
                             try:
                                 diagnostics["attempts"].append("http_fallback")
@@ -562,10 +704,22 @@ if generate_now and not st.session_state.get("busy"):
                                             outputs = list(val)
                                             break
                                     if not outputs:
-                                        for v in response.values():
-                                            if isinstance(v, (list, tuple)) and v:
-                                                outputs = list(v)
-                                                break
+                                        # some Responses v1 return {'message': {...}}
+                                        msg = response.get("message") or response.get("response") or response.get("output")
+                                        if isinstance(msg, dict):
+                                            # try to extract text from structured message
+                                            c = msg.get("content")
+                                            if isinstance(c, list):
+                                                for part in c:
+                                                    if isinstance(part, dict) and part.get("type") == "output_text":
+                                                        outputs.append({"text": part.get("text")})
+                                                    elif isinstance(part, dict) and part.get("type") == "text":
+                                                        outputs.append({"text": part.get("text")})
+                                        else:
+                                            # fallback: join string values
+                                            for v in response.values():
+                                                if isinstance(v, str) and v.strip():
+                                                    outputs.append({"text": v.strip()})
                                 else:
                                     for attr in ("output", "candidates", "items", "responses"):
                                         val = getattr(response, attr, None)
@@ -581,7 +735,7 @@ if generate_now and not st.session_state.get("busy"):
                             if not outputs:
                                 candidate_text = None
                                 if isinstance(response, dict):
-                                    candidate_text = response.get("text") or response.get("message")
+                                    candidate_text = response.get("text") or response.get("message") or response.get("output_text")
                                 else:
                                     candidate_text = getattr(response, "text", None) or getattr(response, "message", None)
                                 if candidate_text:
@@ -592,6 +746,7 @@ if generate_now and not st.session_state.get("busy"):
                                 if not item:
                                     continue
                                 if isinstance(item, dict):
+                                    # common dict shapes
                                     for k in ("content", "text", "message", "output_text", "output"):
                                         v = item.get(k)
                                         if v: